Мониторинг бизнес-процессов Camunda

Привет, Хабр.

Меня зовут Антон и я техлид в компании ДомКлик. Создаю и поддерживаю микросервисы позволяющие обмениваться данными инфраструктуре ДомКлик с внутренними сервисами Сбербанка.

Это продолжение цикла статей о нашем опыте использования движка для работы с диаграммами бизнес-процессов Camunda. Предыдущая статья была посвящена разработке плагина для Bitbucket позволяющего просматривать изменения BPMN-схем. Сегодня я расскажу о мониторинге проектов, в которых используется Camunda, как с помощью сторонних инструментов (в нашем случае это стек Elasticsearch из Kibana и Grafana), так и «родного» для Camunda — Cockpit. Опишу сложности, возникшие при использовании Cockpit, и наши решения.

Когда у тебя много микросервисов, то хочется знать об их работе и текущем статусе всё: чем больше мониторинга, тем увереннее ты себя чувствуешь как в штатных, так и внештатных ситуациях, во время релиза и так далее. В качестве средств мониторинга мы используем стек Elasticsearch: Kibana и Grafana. В Kibana смотрим логи, а в Grafana — метрики. Также в БД имеются исторические данные по процессам Camunda. Казалось бы, этого должно хватать для понимания, работает ли сервис штатно, и если нет, то почему. Загвоздка в том, что данные приходится смотреть в трёх разных местах, и они далеко не всегда имеют четкую связь друг с другом. На разбор и анализ инцидента может уходить много времени. В частности, на анализ данных из БД: Camunda имеет далеко не очевидную схему данных, некоторые переменные хранит в сериализованном виде. По идее, облегчить задачу может Cockpit — инструмент Camunda для мониторинга бизнес-процессов.


Интерфейс Cockpit.

Главная проблема в том, что Cockpit не может работать по кастомному URL. Об этом на их форуме есть множество реквестов, но пока такой функциональности из коробки нет. Единственный выход: сделать это самим. У Cockpit есть Sring Boot-автоконфигурация CamundaBpmWebappAutoConfiguration, вот её-то и надо заменить на свою. Нас интересует CamundaBpmWebappInitializer— основной бин, который инициализирует веб-фильтры и сервлеты Cockpit.

Нам необходимо передать в основной фильтр (LazyProcessEnginesFilter) информацию об URL, по которому он будет работать, а в ResourceLoadingProcessEnginesFilter — информацию о том, по каким URL он будет отдавать JS- и CSS-ресурсы.

Для этого в нашей реализации CamundaBpmWebappInitializer меняем строчку:

registerFilter("Engines Filter", LazyProcessEnginesFilter::class.java, "/api/*", "/app/*")

на:

registerFilter("Engines Filter", CustomLazyProcessEnginesFilter::class.java, singletonMap("servicePath", servicePath), *urlPatterns)

servicePath — это наш кастомный URL. В самом же CustomLazyProcessEnginesFilter указываем нашу реализацию ResourceLoadingProcessEnginesFilter:

class CustomLazyProcessEnginesFilter:
       LazyDelegateFilter<ResourceLoaderDependingFilter>
       (CustomResourceLoadingProcessEnginesFilter::class.java)

В CustomResourceLoadingProcessEnginesFilter добавляем servicePath ко всем ссылкам на ресурсы, которые мы планируем отдавать клиентской стороне:

override fun replacePlaceholder(
       data: String,
       appName: String,
       engineName: String,
       contextPath: String,
       request: HttpServletRequest,
       response: HttpServletResponse
) = data.replace(APP_ROOT_PLACEHOLDER, "$contextPath$servicePath")
           .replace(BASE_PLACEHOLDER,
                   String.format("%s$servicePath/app/%s/%s/", 
contextPath, appName, engineName))
           .replace(PLUGIN_PACKAGES_PLACEHOLDER,
                   createPluginPackagesString(appName, contextPath))
           .replace(PLUGIN_DEPENDENCIES_PLACEHOLDER,
                   createPluginDependenciesString(appName))

Теперь мы можем указывать нашему Cockpit, по какому URL он должен слушать запросы и отдавать ресурсы.

Но ведь не может быть всё так просто? В нашем случае Cockpit способен работать из коробки на нескольких экземплярах приложения (например, в подах Kubernetes), так как вместо OAuth2 и JWT используется старый добрый jsessionid, который хранится в локальном кэше. Это значит, что если попытаться залогиниться в Cockpit, подключенный к Camunda, запущенной сразу в нескольких экземплярах, имея на руках ей же выданный jsessionid, то при каждом запросе ресурсов от клиента можно получить ошибку 401 с вероятностью х, где х = (1 — 1/количество_под). Что с этим можно сделать? У Cockpit во всё том же CamundaBpmWebappInitializer объявлен свой Authentication Filter, в котором и происходит вся работа с токенами; надо заменить его на свой. В нём из кеша сессии берём jsessionid, сохраняем его в базу данных, если это запрос на авторизацию, либо проверяем его валидность по базе данных в остальных случаях. Готово, теперь мы можем смотреть инциденты по бизнес-процессам через удобный графический интерфейс Cockpit, где сразу видно stacktrace-ошибки и переменные, которые были у процесса на момент инцидента.

И в тех случаях, когда причина инцидента ясна по stacktrace исключения, Cockpit позволяет сократить время разбора инцидента до 3-5 минут: зашел, посмотрел, какие есть инциденты по процессу, глянул stacktrace, переменные, и вуаля — инцидент разобран, заводим баг в JIRA и погнали дальше. Но что если ситуация немного сложнее, stacktrace является лишь следствием более ранней ошибки или процесс вообще завершился без создания инцидента (то есть технически всё прошло хорошо, но, с точки зрения бизнес-логики, передались не те данные, либо процесс пошел не по той ветке схемы). В этом случае надо снова идти в Kibana, смотреть логи и пытаться связать их с процессами Camunda, на что опять-таки уходит много времени. Конечно, можно добавлять к каждому логу UUID текущего процесса и ID текущего элемента BPMN-схемы (activityId), но это требует много ручной работы, захламляет кодовую базу, усложняет рецензирование кода. Весь этот процесс можно автоматизировать.

Проект Sleuth позволяет трейсить логи уникальным идентификатором (в нашем случае — UUID процесса). Настройка Sleuth-контекста подробно описана в документации, здесь я покажу лишь, как запустить его в Camunda.

Во-первых, необходимо зарегистрировать customPreBPMNParseListeners в текущем processEngine Camunda. В слушателе переопределить методы parseStartEvent (добавление слушателя на событие запуска верхнеуровневого процесса) и parseServiceTask (добавление слушателя на событие запуска ServiceTask).

В первом случае мы создаем Sleuth-контекст:

customContext[X_B_3_TRACE_ID] = businessKey
customContext[X_B_3_SPAN_ID] = businessKeyHalf
customContext[X_B_3_PARENT_SPAN_ID] = businessKeyHalf
customContext[X_B_3_SAMPLED] = "0" 
val contextFlags: TraceContextOrSamplingFlags = tracing.propagation()
       .extractor(OrcGetter())
       .extract(customContext)
val newSpan: Span = tracing.tracer().nextSpan(contextFlags)
tracing.currentTraceContext().newScope(newSpan.context())

… и сохраняем его в переменную бизнес-процесса:

execution.setVariable(TRACING_CONTEXT, sleuthService.tracingContextHeaders)

Во втором случае мы его из этой переменной восстанавливаем:

val storedContext = execution
       .getVariableTyped<ObjectValue>(TRACING_CONTEXT)
       .getValue(HashMap::class.java) as HashMap<String?, String?>
val contextFlags: TraceContextOrSamplingFlags = tracing.propagation()
       .extractor(OrcGetter())
       .extract(storedContext)
val newSpan: Span = tracing.tracer().nextSpan(contextFlags)
tracing.currentTraceContext().newScope(newSpan.context())

Нам нужно трейсить логи вместе с дополнительными параметрами, такими как activityId (ID текущего BPMN-элемента), activityName (его бизнес-название) и scenarioId (ID схемы бизнес-процесса). Такая возможность появилась только с выходом Sleuth 3.

Для каждого параметра нужно объявить BaggageField:

companion object {
   val HEADER_BUSINESS_KEY = BaggageField.create("HEADER_BUSINESS_KEY")
   val HEADER_SCENARIO_ID = BaggageField.create("HEADER_SCENARIO_ID")
   val HEADER_ACTIVITY_NAME = BaggageField.create("HEADER_ACTIVITY_NAME")
   val HEADER_ACTIVITY_ID = BaggageField.create("HEADER_ACTIVITY_ID")
}

Затем объявить три бина для обработки этих полей:

@Bean
open fun propagateBusinessProcessLocally(): BaggagePropagationCustomizer =
       BaggagePropagationCustomizer { fb ->
           fb.add(SingleBaggageField.local(HEADER_BUSINESS_KEY))
           fb.add(SingleBaggageField.local(HEADER_SCENARIO_ID))
           fb.add(SingleBaggageField.local(HEADER_ACTIVITY_NAME))
           fb.add(SingleBaggageField.local(HEADER_ACTIVITY_ID))
       }

/** [BaggageField.updateValue] now flushes to MDC  */
@Bean
open fun flushBusinessProcessToMDCOnUpdate(): CorrelationScopeCustomizer =
       CorrelationScopeCustomizer { builder ->
           builder.add(SingleCorrelationField.newBuilder(HEADER_BUSINESS_KEY).flushOnUpdate().build())
           builder.add(SingleCorrelationField.newBuilder(HEADER_SCENARIO_ID).flushOnUpdate().build())
           builder.add(SingleCorrelationField.newBuilder(HEADER_ACTIVITY_NAME).flushOnUpdate().build())
           builder.add(SingleCorrelationField.newBuilder(HEADER_ACTIVITY_ID).flushOnUpdate().build())
       }

/** [.BUSINESS_PROCESS] is added as a tag only in the first span.  */
@Bean
open fun tagBusinessProcessOncePerProcess(): SpanHandler =
       object : SpanHandler() {
           override fun end(context: TraceContext, span: MutableSpan, cause: Cause): Boolean {
               if (context.isLocalRoot && cause == Cause.FINISHED) {
                   Tags.BAGGAGE_FIELD.tag(HEADER_BUSINESS_KEY, context, span)
                   Tags.BAGGAGE_FIELD.tag(HEADER_SCENARIO_ID, context, span)
                   Tags.BAGGAGE_FIELD.tag(HEADER_ACTIVITY_NAME, context, span)
                   Tags.BAGGAGE_FIELD.tag(HEADER_ACTIVITY_ID, context, span)
               }
               return true
           }
       }

После чего мы можем сохранять дополнительные поля в контекст Sleuth:

HEADER_BUSINESS_KEY.updateValue(businessKey)
HEADER_SCENARIO_ID.updateValue(scenarioId)
HEADER_ACTIVITY_NAME.updateValue(activityName)
HEADER_ACTIVITY_ID.updateValue(activityId)

Когда мы можем видеть логи отдельно по каждому бизнес-процессу по его ключу, разбор инцидентов проходит гораздо быстрее. Правда, всё равно приходится переключаться между Kibana и Cockpit, вот бы их объединить в рамках одного UI.

И такая возможность имеется. Cockpit поддерживает пользовательские расширения — плагины, в Kibana есть Rest API и две клиентские библиотеки для работы с ним: elasticsearch-rest-low-level-client и elasticsearch-rest-high-level-clientelasticsearch-rest-high-level-client.

Плагин представляет из себя проект на Maven, наследуемый от артефакта camunda-release-parent, с бэкендом на Jax-RS и фронтендом на AngularJS. Да-да, AngularJS, не Angular.

У Cockpit есть подробная документация о том, как писать для него плагины: https://docs.camunda.org/manual/latest/webapps/cockpit/extend/plugins/.

Уточню лишь, что для вывода логов на фронтенде нас интересует tab-панель на странице просмотра информации о Process Definition (cockpit.processDefinition.runtime.tab) и странице просмотра Process Instance (cockpit.processInstance.runtime.tab). Для них регистрируем наши компоненты:

ViewsProvider.registerDefaultView('cockpit.processDefinition.runtime.tab', {
   id: 'process-definition-runtime-tab-log',
   priority: 20,
   label: 'Logs',
   url: 'plugin://log-plugin/static/app/components/process-definition/processDefinitionTabView.html'
});

ViewsProvider.registerDefaultView('cockpit.processInstance.runtime.tab', {
   id: 'process-instance-runtime-tab-log',
   priority: 20,
   label: 'Logs',
   url: 'plugin://log-plugin/static/app/components/process-instance/processInstanceTabView.html'
});

У Cockpit есть UI-компонент для вывода информации в табличном виде, однако ни в одной документации про него не сказано, информацию о нем и о его использовании можно найти, только читая исходники Cockpit. Если вкратце, то использование компонента выглядит следующим образом:

<div cam-searchable-area (1)
    config="searchConfig" (2)
    on-search-change="onSearchChange(query, pages)" (3)
    loading-state="’Loading...’" (4)
    text-empty="Not found"(5)
    storage-group="'ANU'"
    blocked="blocked">
   <div class="col-lg-12 col-md-12 col-sm-12">
       <table class="table table-hover cam-table">
           <thead cam-sortable-table-header (6)
                  default-sort-by="time"
                  default-sort-order="asc" (7)
                  sorting-id="admin-sorting-logs"
                  on-sort-change="onSortChanged(sorting)"
                  on-sort-initialized="onSortInitialized(sorting)" (8)>
           <tr>
               <!-- headers -->
           </tr>
           </thead>
           <tbody>
           <!-- table content -->
           </tbody>
       </table>
   </div>
</div>

1. Атрибут для объявления компонента поиска.
2. Конфигурация компонента. Здесь имеем такую структуру:
   
   

   tooltips = { //здесь мы объявляем плейсхолдеры и сообщения, 
                      //которые будут выводиться в поле поиска в зависимости от результата
      'inputPlaceholder': 'Add criteria',
      'invalid': 'This search query is not valid',
      'deleteSearch': 'Remove search',
      'type': 'Type',
      'name': 'Property',
      'operator': 'Operator',
      'value': 'Value'
   },
   operators =  { //операторы, используемые для поиска, нас интересует сравнение строк
        'string': [
          {'key': 'eq',  'value': '='},
          {'key': 'like','value': 'like'}
      ]
   },
   types = [// поля, по которым будет производится поиск, нас интересует поле businessKey
      {
          'id': {
              'key': 'businessKey',
              'value': 'Business Key'
          },
          'operators': [
              {'key': 'eq', 'value': '='}
          ],
          enforceString: true
      }
   ]
   

   
   
3. Функция поиска данных используется как при изменении параметров поиска, так и при первоначальной загрузке.
4. Какое сообщение отображать во время загрузки данных.
5. Какое сообщение отображать, если ничего не найдено.
6. Атрибут для объявления таблицы отображения данных поиска.
7. Поле и тип сортировки по умолчанию.
8. Функции сортировок.

На бэкенде нужно настроить клиент для работы с Kibana API. Для этого достаточно воспользоваться RestHighLevelClient из библиотеки elasticsearch-rest-high-level-client. Там указать путь до Kibana, данные для аутентификации: логин и пароль, а если используется протокол шифрования, то надо указать подходящую реализацию X509TrustManager.

Для формирования запроса поиска используем QueryBuilders.boolQuery(), он позволяет составлять сложные запросы вида:

val boolQueryBuilder = QueryBuilders.boolQuery();

KibanaConfiguration.ADDITIONAL_QUERY_PARAMS.forEach((key, value) ->
       boolQueryBuilder.filter()
               .add(QueryBuilders.matchPhraseQuery(key, value))
);
if (!StringUtils.isEmpty(businessKey)) {
   boolQueryBuilder.filter()
           .add(QueryBuilders.matchPhraseQuery(KibanaConfiguration.BUSINESS_KEY, businessKey));
}
if (!StringUtils.isEmpty(procDefKey)) {
   boolQueryBuilder.filter()
           .add(QueryBuilders.matchPhraseQuery(KibanaConfiguration.SCENARIO_ID, procDefKey));
}
if (!StringUtils.isEmpty(activityId)) {
   boolQueryBuilder.filter()
           .add(QueryBuilders.matchPhraseQuery(KibanaConfiguration.ACTIVITY_ID, activityId));
}

Теперь мы прямо из Cockpit можем просматривать логи отдельно по каждому процессу и по каждой activity. Выглядит это так:


Таб для просмотра логов в интерфейсе Cockpit.

Но нельзя останавливаться на достигнутом, в планах идеи о развитии проекта. Во-первых, расширить возможности поиска. Зачастую в начале разбора инцидента business key процесса на руках отсутствует, но имеется информация о других ключевых параметрах, и было бы неплохо добавить возможность настройки поиска по ним. Также таблица, в которую выводится информация о логах, не интерактивна: нет возможности перехода в нужный Process Instance по клику в соответствующей ему строке таблицы. Словом, развиваться есть куда. (Как только закончатся выходные, я опубликую ссылку на Github проекта, и приглашаю туда всех заинтересовавшихся.)