Мониторинг бизнес-процессов 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-client.

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

У Cockpit есть подробная документация о том, как писать для него плагины.

Уточню лишь, что для вывода логов на фронтенде нас интересует 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 проекта, и приглашаю туда всех заинтересовавшихся.)