Мониторинг бизнес-процессов 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 под лицензией MIT. Пожелания, критика и пулл-реквесты приветствуются.
    ДомКлик
    Место силы

    Комментарии 2

      0

      Идея затащить логи в кокпит — интересная, хоть и немного странная. Спасибо за статью, хотелось бы все-таки посмотреть плагин подробнее на github'e.

        0
        Были примерно такие-же мысли при прочтении.
        Кроме того, Cockpit — административный интерфейс, можно иметь его в единственном экземпляре.

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

      Самое читаемое