Kapacitor でメトリクスを処理するためのコツ

おそらく、今日では、なぜサービス メトリクスを収集する必要があるのか​​を尋ねる人はいないでしょう。 次の論理的なステップは、収集したメトリクスに対するアラートを設定することです。これにより、都合の良いチャネル (メール、Slack、電報) でデータの逸脱が通知されます。 オンラインホテル予約サービスでは Ostrovok.ru 当社のサービスのすべてのメトリクスは InfluxDB に注がれて Grafana に表示され、基本的なアラートもそこで設定されます。 「何かを計算して比較する必要がある」といったタスクには、Kapacitor を使用します。

Kapacitor は、InfluxDB からのメトリクスを処理できる TICK スタックの一部です。 複数の測定値を結合 (結合) し、受信したデータから有用な情報を計算し、結果を InfluxDB に書き戻し、Slack/Telegram/mail にアラートを送信できます。

スタック全体がクールで詳細です ドキュメンテーション、しかし、マニュアルに明示的に示されていない便利なものは常にあります。 この記事では、そのような便利で自明ではないヒントをいくつか収集することにしました (TICKscipt の基本的な構文について説明します) ここで)、問題の XNUMX つを解決する例を使用して、それらをどのように適用できるかを示します。

行こう！

float と int、計算エラー

カーストを通じて解決される、まったく標準的な問題です。

var alert_float = 5.0
var alert_int = 10
data|eval(lambda: float("value") > alert_float OR float("value") < float("alert_int"))

デフォルト()の使用

タグ/フィールドが入力されていない場合、計算エラーが発生します。

|default()
        .tag('status', 'empty')
        .field('value', 0)

フィルイン結合 (内部と外部)

デフォルトでは、結合はデータのないポイント (内部) を破棄します。
fill('null') を使用すると、外部結合が実行され、その後、default() を実行して空の値を埋める必要があります。

var data = res1
    |join(res2)
        .as('res1', 'res2)
        .fill('null')
    |default()
        .field('res1.value', 0.0)
        .field('res2.value', 100.0)

ここにはまだニュアンスがあります。 上の例では、系列 (res1 または res2) の XNUMX つが空の場合、結果の系列 (データ) も空になります。 Github にはこのトピックに関するチケットがいくつかあります (1633, 1871, 6967) – 私たちは修正を待っていますが、少し苦しんでいます。

計算での条件の使用 (ラムダの場合)

|eval(lambda: if("value" > 0, true, false)

期間中のパイプラインからの最後の XNUMX 分間

たとえば、過去 XNUMX 分間の値を前の週と比較する必要があります。 データの XNUMX つのバッチを XNUMX つの別々のバッチで取得することも、より大きな期間からデータの一部を抽出することもできます。

 |where(lambda: duration((unixNano(now()) - unixNano("time"))/1000, 1u) < 5m)

最後の XNUMX 分間の代替方法は、指定された時間の前にデータを遮断する BarrierNode を使用することです。

|barrier()
        .period(5m)

メッセージでの Go テンプレートの使用例

テンプレートはパッケージの形式に対応します テキスト.テンプレート以下によくあるパズルをいくつか紹介します。

そうでなければ

私たちは物事を整理し、再びテキストで人々を刺激することはありません。

|alert()
    ...
    .message(
        '{{ if eq .Level "OK" }}It is ok now{{ else }}Chief, everything is broken{{end}}'
    )

メッセージの小数点以下 XNUMX 桁

メッセージの読みやすさの向上:

|alert()
    ...
    .message(
        'now value is {{ index .Fields "value" | printf "%0.2f" }}'
    )

メッセージ内の変数を展開する

「なぜ叫んでいるのか?」という質問に答えるために、メッセージに詳細情報が表示されます。

var warnAlert = 10
  |alert()
    ...
    .message(
       'Today value less then '+string(warnAlert)+'%'
    )

一意のアラート識別子

これは、データ内に複数のグループが存在する場合に必要です。そうでない場合、アラートは XNUMX つだけ生成されます。

|alert()
      ...
      .id('{{ index .Tags "myname" }}/{{ index .Tags "myfield" }}')

カスタムハンドラー

ハンドラーの大きなリストには exec が含まれており、これを使用すると、渡されたパラメーター (標準入力) を使用してスクリプトを実行できます。創造性だけで十分です。

私たちのカスタムの XNUMX つは、Slack に通知を送信するための小さな Python スクリプトです。
最初は、認証で保護された grafana 画像をメッセージで送信したいと考えていました。 その後、別のメッセージとしてではなく、同じグループからの前のアラートに対するスレッドに OK を書き込みます。 少し後、過去 X 分間に最も多かった間違いをメッセージに追加します。

別のトピックとして、他のサービスとの通信、およびアラートによって開始されるアクション (監視が十分に機能している場合のみ) について説明します。
ハンドラーの説明の例。slack_handler.py は自分で書いたスクリプトです。

topic: slack_graph
id: slack_graph.alert
match: level() != INFO AND changed() == TRUE
kind: exec
options:
  prog: /sbin/slack_handler.py
  args: ["-c", "CHANNELID", "--graph", "--search"]

デバッグ方法は?

ログ出力付きオプション

|log()
      .level("error")
      .prefix("something")

ウォッチ (cli): kapacitor -url ホストまたはIP:9092 ログ lvl=エラー

httpOut のオプション

現在のパイプライン内のデータを表示します。

|httpOut('something')

見る(入手する): ホストまたはIP:9092/kapacitor/v1/tasks/task_name/something

実行スキーム

• 各タスクは、有用な数値を含む実行ツリーを形式で返します。 graphviz.
• ブロックを取る ドット.
• それをビューアに貼り付けて、 楽しむ.

他に熊手はどこで入手できますか？

ライトバック時の influxdb のタイムスタンプ

たとえば、1 時間あたりのリクエストの合計 (groupBy(XNUMXh)) のアラートを設定し、発生したアラートを influxdb に記録したいとします (問題の事実を grafana のグラフに美しく表示するため)。

influxDBOut() はアラートからの時間値をタイムスタンプに書き込みます。これに応じて、チャート上のポイントはアラートの到着よりも早く/遅く書き込まれます。

正確さが必要な場合: 現在のタイムスタンプでデータを influxdb に書き込むカスタム ハンドラーを呼び出すことで、この問題を回避します。

Docker、ビルドとデプロイメント

起動時に、kapacitor は [load] ブロックの設定で指定されたディレクトリからタスク、テンプレート、およびハンドラーをロードできます。

タスクを正しく作成するには、次のものが必要です。

1. ファイル名 – スクリプト ID/名前に展開されます
2. タイプ - ストリーム/バッチ
3. dbrp – スクリプトが実行されるデータベースとポリシーを示すキーワード (dbrp “supplier.” “autogen”)

一部のバッチ タスクに dbrp を含む行が含まれていない場合、サービス全体が開始を拒否し、それについて正直にログに書き込みます。

逆に、クロノグラフでは、この行は存在すべきではなく、インターフェースを通じて受け入れられず、エラーが生成されます。

コンテナーのビルド時のハック: //.+dbrp を含む行がある場合、Dockerfile は -1 で終了します。これにより、ビルドのアセンブル時に失敗の理由をすぐに理解できるようになります。

XNUMX対多に参加する

タスク例: 95 週間のサービス稼働時間の 10 パーセンタイルを取得し、最後の XNUMX 分の各分をこの値と比較する必要があります。

XNUMX 対多の結合は実行できません。ポイントのグループにわたる最終/平均/中央値がノードをストリームに変換すると、「子の不一致エッジを追加できません: バッチ -> ストリーム」というエラーが返されます。

ラムダ式の変数としてのバッチの結果も置換されません。

最初のバッチから必要な数を udf 経由でファイルに保存し、このファイルをサイドロード経由でロードするオプションがあります。

これで何が解決したのでしょうか？

当社には約 100 社のホテル サプライヤーがあり、それぞれが複数の接続 (チャネルと呼ぶことにします) を持つことができます。 これらのチャネルは約 300 あり、各チャネルが脱落する可能性があります。 記録されたすべてのメトリクスのうち、エラー率 (リクエストとエラー) を監視します。

なぜグラファナではないのでしょうか？

Grafana で設定されたエラー アラートにはいくつかの欠点があります。 状況に応じて、重大なものもあれば、目をつぶっても問題ないものもあります。

Grafana は測定とアラートの間の計算方法を知りませんが、レート (リクエスト - エラー)/リクエストが必要です。

エラーは厄介なようです:

成功したリクエストで見ると、悪影響は少なくなります。

さて、grafana の前にサービスでレートを事前に計算することができ、場合によってはこれが機能します。 しかし、私たちのものではそうではありません、なぜなら... 各チャネルでは独自の比率が「正常」とみなされ、アラートは静的な値に従って機能します（アラートを目で探し、頻繁にアラートがある場合は変更します）。

以下は、さまざまなチャネルの「通常」の例です。

私たちは前の点を無視し、「通常の」状況はすべてのサプライヤーで同様であると仮定します。 これですべてがうまくいき、grafana でアラートを表示できるようになりますか?
可能ですが、次のオプションの XNUMX つを選択する必要があるため、実際にはそうしたくありません。
a) チャネルごとに個別に多数のグラフを作成します (そして、それらを伴うのが苦痛です)
b) すべてのチャネルを含む XNUMX つのチャートを残す (カラフルな線とカスタマイズされたアラートに夢中になる)

どうやってやったのですか？

繰り返しますが、ドキュメントに良い開始例があります (結合されたシリーズ全体のレートの計算) を覗いたり、同様の問題の基礎として使用したりできます。

最終的に私たちがやったこと:

• チャンネルごとにグループ化して、数時間で XNUMX つのシリーズに参加します。
• データがない場合はグループごとに系列を入力します。
• 過去 10 分間の中央値を以前のデータと比較します。
• 私たちは何かを見つけると叫びます。
• 計算されたレートと発生したアラートを influxdb に書き込みます。
• 役立つメッセージを Slack に送信します。

私の意見では、最終的に得たいと思っていたものはすべて (カスタム ハンドラーを使用してさらに少し) できるだけ美しく達成することができました。

github.com を見ることができます コード例 и 最小回路 (graphviz) 結果として得られるスクリプト。

結果のコードの例:

dbrp "supplier"."autogen"
var name = 'requests.rate'
var grafana_dash = 'pczpmYZWU/mydashboard'
var grafana_panel = '26'
var period = 8h
var todayPeriod = 10m
var every = 1m
var warnAlert = 15
var warnReset = 5
var reqQuery = 'SELECT sum("count") AS value FROM "supplier"."autogen"."requests"'
var errQuery = 'SELECT sum("count") AS value FROM "supplier"."autogen"."errors"'

var prevErr = batch
    |query(errQuery)
        .period(period)
        .every(every)
        .groupBy(1m, 'channel', 'supplier')

var prevReq = batch
    |query(reqQuery)
        .period(period)
        .every(every)
        .groupBy(1m, 'channel', 'supplier')

var rates = prevReq
    |join(prevErr)
        .as('req', 'err')
        .tolerance(1m)
        .fill('null')
    // заполняем значения нулями, если их не было
    |default()
        .field('err.value', 0.0)
        .field('req.value', 0.0)
    // if в lambda: считаем рейт, только если ошибки были
    |eval(lambda: if("err.value" > 0, 100.0 * (float("req.value") - float("err.value")) / float("req.value"), 100.0))
        .as('rate')

// записываем посчитанные значения в инфлюкс
rates
    |influxDBOut()
        .quiet()
        .create()
        .database('kapacitor')
        .retentionPolicy('autogen')
        .measurement('rates')

// выбираем данные за последние 10 минут, считаем медиану
var todayRate = rates
    |where(lambda: duration((unixNano(now()) - unixNano("time")) / 1000, 1u) < todayPeriod)
    |median('rate')
        .as('median')

var prevRate = rates
    |median('rate')
        .as('median')

var joined = todayRate
    |join(prevRate)
        .as('today', 'prev')
    |httpOut('join')

var trigger = joined
    |alert()
        .warn(lambda: ("prev.median" - "today.median") > warnAlert)
        .warnReset(lambda: ("prev.median" - "today.median") < warnReset)
        .flapping(0.25, 0.5)
        .stateChangesOnly()
        // собираем в message ссылку на график дашборда графаны
        .message(
            '{{ .Level }}: {{ index .Tags "channel" }} err/req ratio ({{ index .Tags "supplier" }})
{{ if eq .Level "OK" }}It is ok now{{ else }}
'+string(todayPeriod)+' median is {{ index .Fields "today.median" | printf "%0.2f" }}%, by previous '+string(period)+' is {{ index .Fields "prev.median" | printf "%0.2f" }}%{{ end }}
http://grafana.ostrovok.in/d/'+string(grafana_dash)+
'?var-supplier={{ index .Tags "supplier" }}&var-channel={{ index .Tags "channel" }}&panelId='+string(grafana_panel)+'&fullscreen&tz=UTC%2B03%3A00'
        )
        .id('{{ index .Tags "name" }}/{{ index .Tags "channel" }}')
        .levelTag('level')
        .messageField('message')
        .durationField('duration')
        .topic('slack_graph')

// "today.median" дублируем как "value", также пишем в инфлюкс остальные филды алерта (keep)
trigger
    |eval(lambda: "today.median")
        .as('value')
        .keep()
    |influxDBOut()
        .quiet()
        .create()
        .database('kapacitor')
        .retentionPolicy('autogen')
        .measurement('alerts')
        .tag('alertName', name)

そして、結論は何ですか？

Kapacitor は、多数のグループ化による監視アラートの実行、すでに記録されたメトリクスに基づく追加の計算の実行、カスタム アクションの実行、およびスクリプト (udf) の実行に優れています。

参入障壁はそれほど高くありません。grafana やその他のツールでは満足できない場合は、試してみてください。

出所： habr.com