カマンダを監視するビジネスプロセス

こんにちは、Habr。



私の名前はアントンです。DomClickのテクニカルリード です。DomClickインフラストラクチャがSberbankの内部サービスとデータを交換できるようにするマイクロサービスを作成および保守 しています。



これは、Camundaビジネスプロセスダイアグラムエンジンの使用経験に関する一連の記事の続きです 。 前回の記事では、BPMNスキーマの変更を表示できるBitbucketのプラグインの開発について説明しました。今日は、Camundaを使用するプロジェクトの監視、サードパーティツールの使用方法について説明します（この場合、これはからのElasticsearchスタック です） Kibanaと Grafana）ならびにCamundaための「ネイティブ」 - コックピット。 Cockpitを使用する際に発生した問題とその解決策について説明します。



マイクロサービスがたくさんある場合は、その作業と現在のステータスについてすべてを知りたいと考えています。監視を強化するほど、通常の状況と緊急の状況の両方で、リリース中などに自信を持てるようになります。 Elasticsearchスタック：KibanaとGrafanaを監視ツールとして使用します。 Kibanaではログを確認し、Grafanaではメトリックを確認します。データベースには、Camundaプロセスの履歴データも含まれています。これは、サービスが正常に機能しているかどうか、機能していない場合はその理由を理解するのに十分なはずです。欠点は、3つの異なる場所でデータを確認する必要があり、それらが常に明確に相互に関連しているとは限らないことです。インシデントの解析と分析には時間がかかる場合があります。特に、データベースからのデータの分析の場合：Camundaは、明白なデータスキームとはほど遠いため、いくつかの変数をシリアル化された形式で格納します。理論的には、ビジネスプロセスを監視するためのCamundaツールであるCockpitは、タスクを簡単にすることができます。





コックピットインターフェース。



主な問題は、CockpitがカスタムURLを処理できないことです。彼らのフォーラムではこれについて多くの要求がありますが、これまでのところ、箱から出してそのような機能はありません。唯一の方法は自分でやることです。CockpitにはSringBoot自動構成 CamundaBpmWebappAutoConfiguration



があるため、独自のものと交換する必要があります。CamundaBpmWebappInitializer



CockpitWebフィルターとサーブレットを初期化するメインBeanに関心が あります。



メインフィルター（LazyProcessEnginesFilter



）に、それが機能するURLに関する情報 を渡す必要がありResourceLoadingProcessEnginesFilter



ます。また、-JSおよびCSSリソースを提供するURLに関する情報を渡す必要があり ます。



これを行うには、実装で次 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))

      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

これで、コックピットがリクエストをリッスンしてリソースを提供するURLを指定できます。



しかし、それはそれほど単純なことではありませんね。この場合、OAuth2とJWTの代わりに、ローカルキャッシュに格納されている古き良きjsessionidが使用されるため、Cockpitはアプリケーションの複数のインスタンス（たとえば、Kubernetesポッド）でそのままでは機能しません。つまり、Camundaに接続されたCockpitにログインしようとすると、一度に複数のインスタンスで起動され、同じjsessionidが発行されます。その後、クライアントからのリソースの要求ごとに、確率xで401エラーが発生する可能性があります。ここで、x =（1-1 / number_pods）。あなたはそれについて何ができますか？コックピットも同じです CamundaBpmWebappInitializer



認証フィルターが宣言され、トークンの処理がすべて行われます。あなたはそれをあなた自身のものと取り替える必要があります。その中で、セッションキャッシュからjsessionidを取得し、承認要求の場合はデータベースに保存し、それ以外の場合はデータベースに対してその有効性を確認します。これで、便利なCockpitグラフィカルインターフェイスを介してビジネスプロセスごとにインシデントを監視できるようになりました。このインターフェイスでは、インシデント発生時にプロセスにあったスタックトレースエラーと変数をすぐに確認できます。



インシデントの原因が例外のスタックトレースから明らかな場合、Cockpitを使用すると、インシデントの分析にかかる時間を3〜5分に短縮できます。私が入って、プロセスのインシデントを調べ、スタックトレース、変数、および出来上がりを調べました。インシデントは整理され、JIRAにバグを入れました。と運転しました。しかし、状況がもう少し複雑な場合、スタックトレースが以前のエラーの結果であるか、プロセスがインシデントをまったく作成せずに終了した場合（つまり、技術的にはすべてうまくいきましたが、ビジネスロジックの観点からは、間違ったデータが転送されたか、プロセスが間違ったブランチに沿って進んだ場合はどうなりますか？スキーム）。この場合、もう一度Kibanaに移動し、ログを確認して、それらをCamundaプロセスに接続する必要があります。これも、非常に時間がかかります。もちろん、現在のプロセスのUUIDと現在のBPMNダイアグラム要素のID（activityId）を各ログに追加できますが、これには多くの手動作業が必要です。コードベースが乱雑になり、コードレビューが複雑になります。このプロセス全体を自動化できます。



Sleuthプロジェクトで は、一意の識別子（この場合はプロセスUUID）を使用してログをトレースできます。Sleuthコンテキストの設定については、ドキュメントで詳しく説明されています。ここでは、CamundaでSleuthコンテキストを開始する方法のみを示します。



まず、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)

      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

2番目のケースでは、次の変数から復元します。

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



（現在のBPMN要素のID）、 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")
}

      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

次に、これらのフィールドを処理する3つのBeanを宣言します。

@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)

      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

キーごとにビジネスプロセスごとに個別にログを表示できる場合、インシデントの分析ははるかに高速です。確かに、キバナとコックピットを切り替える必要があります。つまり、1つのUI内でそれらを組み合わせる必要があります。



そして、そのような機会があります。 Cockpitはカスタム拡張機能（プラグイン）をサポートします。KibanaにはRest APIと、それを操作するための2つのクライアントライブラリ（ elasticsearch-rest-low-level-clientと elasticsearch-rest-high-level-client）があります。



プラグインは、camunda-release-parentアーティファクトから継承されたMavenプロジェクトであり、Jax-RSバックエンドとAngularJSフロントエンドを備えています。はい、AngularではなくAngularJSです。



コックピットの詳細 そのためのプラグインを作成する方法に関するドキュメント。



フロントエンドにログを表示するには、プロセス定義情報ページ（cockpit.processDefinition.runtime.tab）のタブパネルとプロセスインスタンスビューページ（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'
});

      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

コックピットには、情報を表形式で表示するためのUIコンポーネントがありますが、ドキュメントにはそれについては記載されておらず、コックピットのソースコードを読むだけで情報とその使用法を見つけることができます。つまり、コンポーネントの使用は次のようになります。

<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. ソート機能。

バックエンドでは、KibanaAPIと連携するようにクライアントを構成する必要があります。これを行うには、elasticsearch-rest-high-level-clientライブラリのRestHighLevelClientを使用するだけです。そこで、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から直接、プロセスごとおよびアクティビティごとに個別にログを表示できます。次のようになります。





コックピットインターフェースでログを表示するためのタブ。



しかし、プロジェクトの開発のためのアイデアの計画では、そこで止まることはできません。まず、検索機能を拡張します。多くの場合、インシデントの解析の開始時に、手元にビジネスキープロセスはありませんが、他のキーパラメータに関する情報があり、それらの検索をカスタマイズする機能を追加すると便利です。また、ログに関する情報が表示されるテーブルはインタラクティブではありません。テーブルの対応する行をクリックして、必要なプロセスインスタンスに移動する方法はありません。要するに、開発の余地があります。（週末が終わり次第、プロジェクトのGithubへのリンクを投稿し、そこに興味のあるすべての人を招待します。）