ProHoster > 部落格 > 管理 > Kapacitor 中處理指標的技巧

Kapacitor 中處理指標的技巧

最有可能的是，今天沒有人質疑為什麼有必要收集服務指標。下一個合乎邏輯的步驟是為收集的指標設定警報，它將透過方便您的管道（郵件、Slack、Telegram）通知您資料中的任何偏差。線上飯店預訂服務 奧斯特羅沃克 我們服務的所有指標都流入 InfluxDB 並顯示在 Grafana 中，其中還配置了基本警報。對於「需要計算某些東西並將其與此進行比較」之類的任務，我們使用 Kapacitor。

Kapacitor 中處理指標的技巧
Kapacitor 是 TICK 堆疊的一部分，可以處理來自 InfluxDB 的指標。它可以將多個維度連接在一起（連接），從接收到的資料中計算出一些有用的信息，將結果寫回 InfluxDB，並向 Slack/Telegram/mail 發送警報。

整個堆疊有一個很酷且詳細的 文件但是總有一些有用的東西沒有在手冊中明確指出。在本文中，我決定收集一些這樣有用的、不太明顯的技巧（TICKscipt 的基本語法描述如下 這裡) 並透過解決我們的一個問題的例子來展示如何應用它們。

我們走吧！

浮點數和整數，計算錯誤

一個絕對標準的問題，透過轉換解決：

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

使用 default()

如果標籤/欄位沒有填寫，將會出現計算錯誤：

|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）為空，則產生的系列（資料）也將為空。 GitHub 上關於此主題有幾張票（1633, 1871, 6967) – 我們正在等待修復，並且遭受了一點痛苦。

在計算中使用條件（如果在 lambda 中）

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

管道最後五分鐘的時間段

例如，您需要將最近五分鐘的值與前一周的值進行比較。您可以分兩個批次獲取兩批數據，或從更長的時間段中提取部分數據：

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

最後五分鐘的替代方法是使用 BarrierNode 節點，該節點會在指定時間之前切斷資料：

|barrier()
        .period(5m)

在訊息中使用 Go 模板的範例

範本與套件中的格式相對應 文字.模板，以下是一些經常遇到的問題。

如果別的

讓我們把事情理順，不要用不必要的文字來刺激人們：

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

訊息中的小數點後兩位數字

提高訊息的可讀性：

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

展開訊息中的變數

我們在訊息中顯示更多資訊來回答「它為什麼尖叫？」這個問題。

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

唯一警報標識符

當資料中有多個群組時這是必要的，否則只會產生一個警報：

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

自訂處理程序

大量的處理程序清單包括 exec，它允許您使用傳遞的參數（stdin）執行腳本 - 創造力，僅此而已！

我們的一個習慣是編寫一個小型 Python 腳本，用於向 Slack 發送通知。
一開始我們想透過訊息發送 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=error

帶有 httpOut 的選項

顯示目前管道中的數據：

|httpOut('something')

觀看（獲取）： 主機或 IP：9092/kapacitor/v1/任務/task_name/某事

實施方案

  • 每個任務都會傳回一個執行樹，其中包含有用的數字，格式如下 Graphviz.
  • 我們採取了街區 點.
  • 貼到檢視器中， 讓我們享受吧.

您還能在哪裡獲得佣金？

influxdb 寫回時的時間戳

例如，我們設定了每小時要求總數的警報（groupBy（1h）），並希望記錄在 influxdb 中發生的警報（以便在 grafana 中的圖表上很好地顯示問題的事實）。

influxDBOut() 會將警報的時間值寫入時間戳，因此圖表上的點將比警報到達的時間更早/更晚寫入。

當需要精確度時：我們透過呼叫自訂處理程序來解決這個問題，該處理程序會使用當前時間戳將資料寫入 influxdb。

docker，建置與部署

啟動時，kapacitor 可以從設定中指定的目錄（在 [load] 區塊中）載入任務、範本和處理程序。

要正確建立任務，您需要以下內容：

  1. 檔案名稱 - 擴充為 id/腳本名稱
  2. 類型 – 流/批次
  3. dbrp – 用於指定腳本在哪個資料庫 + 策略中執行的關鍵字（dbrp“supplier”.“autogen”）

如果某個批次任務沒有dbrp的一行，整個服務就會拒絕啟動，並且如實的在日誌中記錄。

相反，在 chronograf 中，這條線不應該存在；不透過介面接受，回傳錯誤。

容器建置技巧：如果 Dockerfile 中有 //.+dbrp 行，則它會以 -1 退出，以便您可以立即了解建置失敗的原因。

一對多連接

範例任務：您需要取得一週內服務正常運作時間的第 95 個百分位數，並將過去 10 分鐘的每一分鐘與該值進行比較。

您不能進行一對多連接，一組點上的最後一個/平均值/中位數會將節點變成一個流，錯誤“無法添加子不匹配的邊：批量->流”將返回。

批次的結果作為 lambda 表達式中的變量，也沒有被取代。

有一個選項可以透過 udf 將第一批所需的數字儲存到檔案中，然後透過 sideload 載入該檔案。

我們用這個來解決什麼問題？

我們有大約100家酒店供應商，每個供應商可以有多個連接，我們稱之為通路。這樣的通道大約有300個，每個通道都可以脫落。在所有記錄的指標中，我們將監控錯誤率（請求和錯誤）。

為什麼不使用 Grafana？

Grafana 中配置的錯誤警報有幾個缺點。有些是至關重要的，有些可以忽略，這取決於具體情況。

Grafana 無法進行跨維度計算 + 警報，但我們需要一個速率（請求-錯誤）/請求。

這些錯誤看起來很糟：

Kapacitor 中處理指標的技巧

如果你看一下成功的查詢，那麼危害就小了：

Kapacitor 中處理指標的技巧

好的，我們可以在 Grafana 之前預先計算服務中的速率，在某些情況下這會起作用。但在我們的系統中不是這樣，因為對於每個頻道來說，其自己的比率被認為是“正常的”，並且警報根據靜態值工作（我們用眼睛看，如果它經常發出警報，我們會改變它）。

以下是不同頻道「正常」的範例：

Kapacitor 中處理指標的技巧

Kapacitor 中處理指標的技巧

讓我們忽略前一點，假設所有供應商的「正常」情況都是類似的。現在一切都很好，我們可以透過 grafana 中的警報來解決嗎？
可以，但我們真的不想，因為我們必須選擇以下選項之一：
a）為每個通道分別製作大量圖表（並痛苦地維護它們）
b) 保留一張包含所有頻道的圖表（並迷失在彩色線條和自訂警報中）

Kapacitor 中處理指標的技巧

你是怎麼做到的？

同樣，文件中有一個很好的入門範例（計算連接序列的比率)，大家可以看一下或是在遇到類似問題時可以作為參考。

我們最終採取的措施是：

  • 幾個小時內觀看兩集，按頻道分組；
  • 如果沒有數據，我們按組別填寫系列；
  • 將過去 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 或其他工具不能完全滿足您的願望，請嘗試一下。

來源： www.habr.com

為具有 DDoS 保護、VPS VDS 服務器的站點購買可靠的主機 🔥 購買具備 DDoS 防護的可靠網站寄存服務，包括 VPS 和 VDS 伺服器 | ProHoster