2.14. Конфигурация системы (config.yml)¶

2.14.1.1. connector¶

input - настройки для получения и первоначальной обработки (parsing) входящих запросов. Поддерживаемые варианты (параметр type):

• http — для запросов в формате JSON по HTTP
• soap — для запросов формате XML (SOAP) по HTTP
• kafka — интеграция с шиной данных Apache Kafka.

output - настройки end-point’ов для отправки объектов. Поддерживаемые варианты (параметр type):

• input_processor — отправка объекта на роль input_processor
• http — отправка объекта в формате JSON во внешнюю систему по HTTP
• soap — отправка объекта в формате XML (SOAP) во внешнюю систему по HTTP
• kafka — интеграция с шиной данных Apache Kafka
• dummy — для использования тестовой заглушки.

routing — маршрутизация объекта для отправки через определенный output. Output определяется в зависимости от ключа маршрутизации, заданного параметром key.

2.14.1.2. input_processor¶

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

routing — маршрутизация объекта в определенный пайплайн обработки по ключу маршрутизации (параметр key).

storage — настройки сохранения объекта в роли storage по ключу маршрутизации (параметр key). Параметр type определяет тип бизнес-объекта (агрегата).

Пример маршрута (порядка обработки) объекта

1. После попадания в систему на роль connector, объект будет обработан в соответствии с настройками конфигурации input в секции connector.

В нашем примере мы видим, что для всех возможных вариантов input (JSON via HHTP, XML via SOAP, Kafka) конвейер обработки указан один и тот же: pipeline: t_connect_input_handle.

2. В секции pipelines находим этот конвейер — он состоит из выполнения одной функции

pipelines:
  t_connect_input_handle:
    - focus_routing

3. В секции functions мы видим, что исполняемый код этой функции находится в файле focus_routing.lua

functions:
  ...
  focus_routing: {__file: focus_routing.lua}

Листинг кода:

local first = ...

local ret = {obj = first, priority = 1, routing_key = 'input_processor'}

return ret

Необработанный объект должен быть помещен в поле obj. Объекту будет присвоен ключ маршрутизации «input_processor».

4. Дальнейший маршрут объекта определяется настройками routing в секции connector. Для объектов с ключом маршрутизации «input_processor» это будет output: to_input_processor:

connector:

  routing:
    - key: input_processor
      output: to_input_processor

5. В секции connector находим данный output:

connector:

  output:
    - name: to_input_processor
      type: input_processor

Значение параметра type означает, что объект будет направлен на инстанс с ролью input_processor.

6. При попадании на роль input_processor объект прежде всего будет обрабатываться в соответствие с настройками из секции ниже:

input_processor:
  classifiers:
    - name: focus
      pipeline: focus_classifiers

В указанном здесь пайплайне будет выполняться следующий код из focus_classifier.lua:

local param = ...

local type_table =
    {
        ["Initiation"] =  "focus_initiation",
        ["Coupon Payment"] =  "focus_couponpayment",
        ["Exercise"] =  "focus_exercise",
        ["Rate Change"] =  "focus_ratechange"
    }

local node = lom.get_by_path(param.obj, 'message.header.routedata.event_type')

if node == nil then
    return param.obj
end

local event_type = string.strip(node[1])

param.routing_key = type_table[event_type]

return param

где lom.get_by_path — функция, которая берет значение из объекта по указанному пути. При этом путь записывается с разделением при помощи точки.

Допустим, что после обработки на данном этапе наш объект получит ключ маршрутизации «focus_couponpayment».

7. Дальнейшая обработка объекта с таким ключом маршрутизации будет определяться следующими настройками в секции input_processor:

input_processor:

  routing:
    - key: focus_couponpayment
      pipeline: coupon_payment_handle

  storage:
    - key: focus_couponpayment
      type: CouponPayment

Это означает, что сначала объект будет обработан в еще одном пайплайне — coupon_payment_handle, а потом направлен на роль storage как объект типа «CouponPayment». Данный тип должен быть описан в модели данных, которая определена в секции

types: {__file: model.avsc}

После успешной валидации по модели данных, объект будет сохранен на роли storage.

8. Если для данного ключа маршрутизации указаны еще какие-либо настройки в других секциях config.yml, объекты с этим ключом будут обработаны далее (см. пример про секцию output_processor).

2.14.1.3. output_processor¶

Задается логика обработки объектов, которые будут реплицироваться во внешние системы. Для объектов с определенным ключом маршрутизации определяется:

• каким пайплайном будет обработан объект;
• посредством какого output он будет отправляться во внешнюю систему.

В рассматриваемом примере:

output_processor:
  focus_couponpayment:
    pipeline: notification_filter
    output: to_external_http_service

Данные настройки означают, что объекты, которые на более ранних этапах обработки получили ключ маршрутизации «focus_couponpayment», будут вначале обработаны функциями из пайплайна notification_filter, а затем отосланы в соответствие с настройками указанного здесь output (смотрим эти настройки в секции connector):

connector:

  output:
    - name: to_external_http_service
      type: http
      url: http://localhost:8021/test_json_endpoint
      format: json

Аналогично определяется поведение роли output_processor в отношение объектов с другими ключами маршрутизации.

2.14.1.4. repair_queue¶

repair_queue:
  on_object_added:
    __unclassified__:
        postprocess_with_routing_key: unclassified

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

При попадании такого объекта в ремонтную очередь, согласно настройкам postprocess_with_routing_key: unclassified, объекту присваивается другой ключ маршрутизации «unclassified», и с этим ключом он направляется на роль output_processor. Дальнейшая обработка объекта будет выполняться в соответствие с настройками в секции output_processor:

output_processor:
  unclassified:
    pipeline: notification_filter
    output: to_external_http_service

2.14.1.5. logger¶

В этой секции определяются настройки для роли logger. В нашем примере:

logger:
  rotate_log: true
  max_msg_in_log: 500000
  max_log_size: 10485760
  delete_by_n_msg: 1000

где

• rotate_log — указатель, осуществлять ли ротацию лога (значения true/false);
• max_msg_in_log — максимальной количество сообщений, сохраняемых в логе;
• max_log_size — максимальный размер файла лога, в байтах;
• delete_by_n_msg — количество одновременно удаляемых сообщений при ротации лога. При превышении значений параметров max_msg_in_log или max_msg_log_size наиболее старые n сообщений в логе удаляются за раз, что повышает производительность по сравнению с режимом, когда старые сообщения удаляются по одному.

2.14.1.6. tasks¶

В этой секции настраивается конфигурация задач, выполняемых при помощи ролей scheduler и task_runner. В нашем примере:

tasks:
  task_1:
    kind: single_shot
    pipeline: single_task
  task_2:
    kind: continuous
    pipeline: long_task
    pause_sec: 10
  task_3:
    kind: periodical
    pipeline: long_task
    schedule: "0 */5 * * * *"

где

• task_N — имя задачи;
• kind — вид задачи;
• pipeline — пайплайн, определяющий, что именно делается в рамках задачи;
• pause_sec — пауза (в секундах) в выполнении задачи вида «continuous»;
• schedule — расписание выполнения для задач вида «periodical». Задается в формате cron. Используется расширенный синтаксис, в котором минимальной величиной является секунда:

* * * * * *
| | | | | |
| | | | | ----- День недели (0-6) (Воскресенье = 0)
| | | | ------- Месяц (1-12)
| | | --------- День (1-31)
| | ----------- Час (0-23)
| ------------- Минута (0-59)
--------------- Секунда (0-59)

В примере выше значение параметра schedule: "0 */5 * * * *" означает, что задача будет запускаться периодически каждые 5 минут.