Контакты
IBSurgeon
Эта страница переведена машинным переводом. Читайте английский оригинал. English

Служба (демон) fb_streaming

← Стриминг / CDC Firebird

Служба (демон) fb_streaming предназначена для асинхронной обработки событий репликации. Она отслеживает новые файлы журнала репликации в указанном каталоге, анализирует их, и генерирует события, которые обрабатываются одним из плагинов.

Принцип работы

Служба (демон) fb_streaming проверяет содержимое директории, указанной для задачи в конфигурационном файле fb_streaming.conf, и если в этой директории присутствует ещё не обработанные файлы журнала репликации, то она анализирует их, и генерирует соответствующие события, которые обрабатываются указанным плагином. Последний номер обработанного сегмента репликации записывается в контрольный файл с именем <database guid>. Расположение этого файла можно указать в параметре конфигурации lockDir. Если этот параметр не указан, то по умолчанию контрольный файл <database guid> будет создан в директории, указанной в качестве директории для журналов архивов для задачи.

[!IMPORTANT] Важно: Файлы журналов репликации должны быть в архивном состоянии.

Контрольный файл необходим для восстановления работы после внезапного завершения работы службы (например выключили свет или перезагрузили сервер). Он содержит следующую информацию:

  • контрольную информацию (GUID базы данных, размер и т.д.);
  • номер последнего обработанного сегмента и позицию в нём;
  • номер сегмента с самой старой активной транзакцией (транзакция может начаться в одном сегменте, а завершиться в другом);
  • список всех активных транзакций в виде пар {tnxNumber, segmentNumber}, где segmentNumber номер сегмента в котором транзакция началась.

Если произошла внештатная ситуация и служба fb_streaming была остановлена, то при следующем старте она читает контрольный файл, и перечитывает все сегменты репликации, начиная с номера сегмента с самой старой активной транзакцией, и заканчивая номером последнего обработанного сегмента. В процессе чтения этих сегментов fb_streaming повторяет все события из списка активных транзакций, после чего переходит в штатный режим работы.

Файл сегмента репликации удаляется, если его номер меньше чем номер сегмента с самой старой активной транзакцией.

В процессе анализа журналов репликации могут возникать следующие события:

  • старт транзакции (START);
  • выполнение первой фазы подтверждения двухфазной транзакции (PREPARE);
  • подтверждение транзакции (COMMIT);
  • откат транзакции (ROLLBACK);
  • установка точки сохранения (SAVE);
  • откат точки сохранения (UNDO);
  • освобождение точки сохранения (RELEASE);
  • установка значения генератора (SET SEQUENCE);
  • выполнение SQL оператора (EXECUTE SQL). Такие события происходят только для DDL операторов;
  • сохранение BLOB (BLOB);
  • вставка новой записи в таблицу (INSERT);
  • обновление записи в таблице (UPDATE);
  • удаление записи из таблицы (DELETE).

Какие из них будут обработаны, а какие просто проигнорированы зависит от выбранного плагина.

Установка и запуск службы в Windows

Перейдите в каталог установки fb_streaming.

Выполните настойку в файле конфигурации fb_streaming.conf.

После настройки файла конфигурации, вы можете установить и запустить службу.

Для получения справки по использованию и установке службы наберите

fb_streaming help

Вам будет выведена информация по установке, удаления, старту и остановки службы:

================================================================================
 Firebird streaming service

 Copyright (c) 2023 IBSurgeon ltd.
================================================================================

usage: fb_streaming <command> [service_name]
Possible commands are:
    install            install service
    remove             remove service
    start              start service
    stop               stop service

После настройки файла конфигурации, вы можете установить и запустить службу, выполнив следующие команды:

fb_streaming install

fb_streaming start

После установки вы можете управлять службой через графическую утилиту - службы (services.msc).

Для удаления службы выполните следующие команды:

fb_streaming stop

fb_streaming remove

Замечание. Если у вас будут функционировать несколько служб fb_streaming, то при установке, запуске, остановке и удалении им надо дать уникальные имена.

Файлы и папки fb_streaming на Windows

  • fb_streaming.exe - служба fb_streaming;
  • fbclient.dll - клиентская библиотека Firebird (версия 4.0 и выше);
  • icudt63.dll, icuin63.dll, icuuc63.dll, icudt63l.dat - файлы библиотеки ICU. В некоторых версиях Windows уже установлена системная библиотека ICU, в таком случае эти файлы необязательны;
  • fb_streaming.conf - файл конфигурации службы fb_streaming;
  • fb_streaming.log - журнал службы fb_streaming (создаётся самостоятельно);
  • doc - папка с файлами документации;
  • stream_plugins - папка для размещения файлов библиотек плагинов.

Установка и запуск демона в Linux

Перейдите в каталог установки fb_streaming. В Linux это обычно директория /opt/fb_streaming.

Выполните настойку в файле конфигурации fb_streaming.conf.

После настройки файла конфигурации, вы можете установить и запустить службу, выполнив следующие команды:

sudo systemctl enable fb_streaming

sudo systemctl start fb_streaming

Для удаления службы выполните следующие команды:

sudo systemctl stop fb_streaming

sudo systemctl disable fb_streaming

Настройка конфигурации служба (демон) fb_streaming

Для конфигурации службы используется файл fb_streaming.conf, который должен быть расположен в корневой директории fb_streaming.

Синтаксис файла конфигурации точно такой же как для файлов конфигурации Firebird.

Символ # обозначает комментарий, то есть всё что следует за этим символом до конца строки игнорируется.

Структура файла конфигурации fb_streaming.conf выглядит так:

# logLevel = info

# pluginDir = $(root)/stream_plugins

task = <sourceDir_1>
{
# другие параметры задачи 1
}

task = <sourceDir_2>
{
# другие параметры задачи 2
}

Общие параметры службы (демона) fb_streaming:

  • logLevel - уровень журналирования (по умолчанию info). Допускаются уровни журналирования: trace, debug, info, warning, error, critical, off.
  • pluginDir - директория, где размещаются плагины fb_streaming. По умолчанию $(root)/stream_plugins. Макроподстановка $(root) обозначает корневую директорию службы (демона).

Далее следуют конфигурации задач. Один экземпляр службы может обслуживать сразу несколько задач. Каждая задача работает в своём потоке. На каждый каталог с файлами журнала репликации создаётся своя задача.

Здесь sourceDir_N - директория с файлами журнала репликации. Эти директории должны быть уникальны. Одну и ту же директорию нельзя обрабатывать более чем одной задачей.

Основные параметры задачи:

  • controlFileDir - директория в которой будет создан контрольный файл (по умолчанию та же директория, что и sourceDir);
  • errorTimeout - тайм-аут после ошибки, в секундах. По истечению этого таймаута будет произведено повторное сканирование сегментов и перезапуск задачи. По умолчанию равен 60 секунд;
  • database - строка подключения к базе данных (обязательный);
  • username - имя пользователя для подключения к базе данных;
  • password - пароль для подключения к базе данных;
  • plugin - плагин, который обрабатывает события, возникающие в процессе анализа журнала репликации (обязательный);
  • deleteProcessedFile - удалять ли файл журнала после обработки (по умолчанию true).

Задача может содержать также другие параметры, специфичные для используемого плагина.

Плагины для обработки событий из журнала репликации расположены в динамических библиотеках. Имя файла динамической библиотеки зависит от операционной системы и строится следующим образом:

  • для Windows <имя плагина>.dll
  • для Linux lib<имя плагина>.so

Файл динамической библиотеки должен быть расположен в директории stream_plugins, или в той, что указана в параметре pluginDir.

Трюки при конфигурации Firebird

Обратите внимание: папка с архивами журналов репликации должна обрабатываться только одной задачей службы fb_streaming. Если вы хотите чтобы, одновременно работала логическая репликация или с архивом журналов работало несколько служб, то необходимо дублирование архивных журналов в разные директории.

Это можно сделать в файле replication.conf следующим образом

...
journal_archive_directory = <archiveDir>
journal_archive_command = "copy $(pathname) $(archivepathname) && copy $(pathname) <archiveDirTask>
...

Здесь archiveDir - директория архивов для работы асинхронной репликации, archiveDirTask - директория архивов для работы задачи службы fb_streaming.

Плагины

Служба (демон) fb_streaming поставляет в комплекте с демонстрационным плагином simple_json_plugin. Остальные плагины мы можем написать под заказ. У нас существуют готовые плагины:

  • kafka_cdc_plugin - Запись событий изменения данных в Kafka (Change Data Capture);
  • mongodb_events_plugin - запись событий репликации в MongoDB;
  • rabbitmq_plugin - помещение событий репликации в сервер очередей RabbitMQ;
  • fts_lucene_plugin - обновление полнотекстовых индексов созданных с помощью LuceneFTS UDR.

Плагин simple_json_plugin

Плагин simple_json_plugin предназначен для автоматической трансляции бинарных файлов журнала репликации в эквивалентный JSON формат и сохранять журналы в заданную директорию с тем же именем, но с расширением .json.

Каждый json файл содержит в себе корневой объект, состоящий из двух полей: header и events.

Поле header представляет собой объект описывающий заголовок сегмента репликации. Он содержит следующие поля:

  • guid - GUID базы данных;
  • sequence - номер сегмента репликации;
  • state - состояние в котором находится сегмент. Возможные значения: free, used, full, archive;
  • version - версия протокола журнала репликации.

Поле events представляет собой массив объектов, каждый из которых представляет одно из событий репликации. Объект события может содержать следующие поля:

  • event - тип события. Доступны следующие варианты:
    • SET SEQUENCE - установка значения последовательности (генератора);
    • START TRANSACTION - старт транзакции;
    • PREPARE TRANSACTION - выполнение первой фазы подтверждения двухфазной транзакции;
    • SAVEPOINT - установка точки сохранения;
    • RELEASE SAVEPOINT - освобождение точки сохранения;
    • ROLLBACK SAVEPOINT - откат точки сохранения;
    • COMMIT - подтверждение транзакции;
    • ROLLBACK - откат транзакции;
    • INSERT - вставка новой записи в таблицу;
    • UPDATE - обновление записи в таблице;
    • DELETE - удаление записи из таблицы;
    • EXECUTE SQL - выполнение SQL оператора. Такие события происходят только для DDL операторов;
    • STORE BLOB - сохранение BLOB.
  • tnx - номер транзакции. Доступно для всех событий кроме SET SEQUENCE, поскольку генераторы работают вне контекста транзакций;
  • sequence - имя последовательности. Доступно только в событии SET SEQUENCE;
  • value - значение последовательности. Доступно только в событии SET SEQUENCE;
  • sql - текст SQL запроса. Доступно только в событии EXECUTE SQL;
  • blobId - идентификатор BLOB. Доступно только в событии STORE BLOB;
  • data - данные сегмента BLOB в шестнадцатеричном представлении. Доступно только в событии STORE BLOB;
  • table - имя таблицы. Доступно в событиях INSERT, UPDATE и DELETE;
  • record - значения полей записи. Для событий INSERT и UPDATE это новая запись, а для DELETE - старая;
  • oldRecord - значения полей старой записи. Доступно только в событии UPDATE;
  • changedFields - массив имён изменённых полей. Доступно только в событии UPDATE.

Обратите внимание, что для BLOB полей в качестве значения указан идентификатор BLOB.

Пример содержимого файла .json:

{
    "header": {
        "version": 1,
        "guid": "{F396449D-F6E4-4812-875E-248AB7C2BEE7}",
        "sequence": 1,
        "state": "archive"
    },
    "events": [
        {
            "event": "START TRANSACTION",
            "tnx": 1937
        },
        {
            "event": "SAVEPOINT",
            "tnx": 1937
        },
        {
            "event": "SAVEPOINT",
            "tnx": 1937
        },
        {
            "event": "UPDATE",
            "table": "CUSTOMER",
            "tnx": 1937,
            "changedFields": [
                "ADDRESS"
            ],
            "oldRecord": {
                "CUSTOMER_ID": 8,
                "NAME": "Abigail Thomas",
                "ADDRESS": null,
                "ZIPCODE": null,
                "PHONE": "1-290-853-7531"
            },
            "record": {
                "CUSTOMER_ID": 8,
                "NAME": "Abigail Thomas",
                "ADDRESS": "555",
                "ZIPCODE": null,
                "PHONE": "1-290-853-7531"
            }
        },
        {
            "event": "RELEASE SAVEPOINT",
            "tnx": 1937
        },
        {
            "event": "RELEASE SAVEPOINT",
            "tnx": 1937
        },
        {
            "event": "COMMIT",
            "tnx": 1937
        }
    ]
}

Настройка плагина

Пример настройки плагина:

task = d:\fbdata\4.0\replication\examples\json_source
{
    deleteProcessedFile = true
    database = inet://localhost:3054/examples
    username = SYSDBA
    password = masterkey
    plugin = simple_json_plugin
    dumpBlobs = true
    register_ddl_events = true
    register_sequence_events = true
    outputDir = d:\fbdata\4.0\replication\examples\json_archive
    # include_tables =
    # exclude_tables =
}

Описание параметров:

  • controlFileDir - директория в которой будет создан контрольный файл (по умолчанию та же директория, что и sourceDir);
  • database - строка подключения к базе данных (обязательный);
  • username - имя пользователя для подключения к базе данных;
  • password - пароль для подключения к базе данных;
  • plugin - плагин, который обрабатывает события, возникающие в процессе анализа журнала репликации (обязательный);
  • deleteProcessedFile - удалять ли файл журнала после обработки (по умолчанию true);
  • outputDir - директория в которой будут находится готовые JSON файлы (обязательный);
  • dumpBlobs - публиковать ли данные BLOB полей (по умолчанию false);
  • register_ddl_events - регистрировать ли DDL события (по умолчанию true);
  • register_sequence_events - регистрировать ли события установки значения последовательности (по умолчанию true);
  • include_tables - регулярное выражение, определяющие имена таблиц для которых необходимо отслеживать события;
  • exclude_tables - регулярное выражение, определяющие имена таблиц для которых не надо отслеживать события.