Трикове за обработка на метрики в Kapacitor.

Вероятно, днес вече никой не поставя под въпрос необходимостта от събиране на метрики на услугите. Следващата логична стъпка е да настроите оповестяване за събраните метрики, което ще уведомява за всяко отклонение в данните по удобни за вас канали (имейл, Slack, Telegram). В услугата за онлайн резервация на хотели Ostrovok.ru всички метрики на нашите услуги се събират в InfluxDB и се показват в Grafana, там също е настроено основно оповестяване. За задачи като „треба да се изчисли нещо и да се сравни с това“ използваме Kapacitor.

Kapacitor е част от TICK стека, който може да обработва метрики от InfluxDB. Той може да свърже няколко измерения помежду си (join), да изчисли нещо полезно от получените данни, да запише резултата обратно в InfluxDB, да изпрати оповестяване в Slack/Twitter/имейл.

Целият стек има впечатляваща и подробна документацията, но винаги има полезни неща, които не са ясно указани в ръководствата. В тази статия реших да събера серия от полезни и неочевидни съвети (основният синтаксис на TICKscipt е описан тук.) и да покажа как могат да се прилагат, на примера на решаване на една от нашите задачи.

Хайде да започваме!

float & int, грешки в изчисленията

Абсолютно стандартен проблем, решава се чрез cast:

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 в join (inner vs outer)

По подразбиране join отхвърля точките, където няма данни (inner).
При fill(‘null’) ще бъде изпълнен outer join, след което е необходимо да се направи default() и да се запълнят празните стойности:

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

Тук все пак има нюанс. Ако в примера по-горе една от сериите (res1 или res2) е празна, крайната серия (data) също ще бъде празна. Има няколко тикаета на GitHub по тази тема (1633, 1871, 6967) – чакаме поправки и малко страдаме.

Използване на условия в изчисленията (if в lambda)

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

Последните пет минути от пайплайна за период

Например, необходимо е да сравните стойностите на последните пет минути с предходната седмица. Можете да вземете две партиди данни в две отделни партиди или да извлечете част от данните от по-голям период:

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

Алтернатива за последните пет минути може да бъде използването на BarrierNode, който отрязва данни по-рано от зададеното време:

|barrier()
        .period(5m)

Примери за използване на Go шаблони в съобщение

Шаблоните отговарят на формата от пакета text.template, по-долу няколко често срещани задачи.

if-else

Слагаме ред, без да предизвикваме хората с ненужен текст:

|alert()
    ...
    .message(
        '{{ if eq .Level "OK" }}Сега е ок{{ else }}Шефе, всичко е счупено{{end}}'
    )

Две цифри след запетаята в съобщението

Подобряваме четивността на съобщението:

|alert()
    ...
    .message(
        'сега стойността е {{ index .Fields "value" | printf "%0.2f" }}'
    )

Разгръщане на променливи в съобщението

Извеждаме повече информация в съобщението за отговора на въпроса "Защо вика?"

var warnAlert = 10
  |alert()
    ...
    .message(
       'Днешната стойност е по-малка от ' + string(warnAlert) + '%'
    )

Уникален идентификатор на алерта

Необходимо е, когато в данните има повече от една група, иначе ще се генерира само един алерт:

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

Персонализирани обработчици

В голям списък от обработчици има exec, който позволява изпълнението на собствен скрипт с предадените параметри (stdin) – само творчество!

Един от нашите персонализирани е малък python скрипт за изпращане на известия в Slack.
Първоначално искахме да изпратим в съобщението картинка от Grafana с авторизация. След това – да пишем OK в треда на предишния алерт от същата група, а не като отделно съобщение. Малко по-късно – да добавяме в съобщението най-честата грешка за последните Х минути.

Отделна тема е свързаността с други услуги и каквито и да е действия, инициирани от алерта (само ако вашият мониторинг работи достатъчно добре).
Пример за описание на обработчика, където 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("нещо")

Гледайте (cli): kapacitor -url host-or-ip:9092 logs lvl=error

Вариант с httpOut

Показва данни в текущия пайплайн:

|httpOut('нещо')

Гледайте (get): host-or-ip:9092\/kapacitor\/v1\/tasks\/task_name\/something

Схема на изпълнение

• Всяка задача връща дърво на изпълнение с полезни цифри във формат graphviz.
• Взимаме блока dot.
• Вмъкваме в viewer, наслаждаваме се.

Къде още можем да получим грабли

timestamp в influxdb при обратно записване

Например, ние настройваме аларма за броя на заявките на час (groupBy(1h)) и искаме да запишем настъпилата аларма в influxdb (за да покажем красиво факта за наличие на проблем на графика в grafana).

influxDBOut() ще запише в timestamp стойността time от алармата, съответно, точката на графика ще бъде записана по-рано/по-късно от получената аларма.

Когато е необходима прецизност: заобикаляме този проблем чрез извикване на персонализиран handler, който ще запише данните в influxdb с текущия timestamp.

docker, сборка и деплой

При стартиране на kapacitor, той може да зарежда задачи, шаблони и хендлери от директорията, зададена в конфигурацията, в блока [load].

За коректно създаване на задача са необходими следните неща:

1. Име на файла – разширява се в id/име на скрипта
2. Тип – stream/batch
3. dbrp – ключова дума за указване в коя база + политика работи скрипта (dbrp «supplier».«autogen»)

Ако в някаква batch-задача няма ред с dbrp, цялата услуга ще откаже да стартира и честно ще напише за това в логовете.

В chronograf обаче, противно на това, този ред не трябва да присъства, интерфейсът не го приема и издава грешка.

Хак при сборка на контейнера: Dockerfile излиза с -1, ако има редове с /.+dbrp, което ще позволи веднага да се разбере причината за неуспеха при сборка на билда.

join един към много

Задача-пример: трябва да вземем 95-ти перцентил на времето за работа на услугата за седмица, да сравним всяка минута от последните 10 с тази стойност.

Не може да се направи join един към много, last/mean/median по група точки превръщат нода в stream, ще се върне грешка «cannot add child mismatched edges: batch -> stream».

Резултатът от batch-а, като променлива в lambda-израз, също не се подставя.

Има вариант да се съхранят нужните цифри от първия batch в файл чрез udf и да се зареди този файл чрез sideload.

Какво решавахме с това?

Имаме около 100 доставчици на хотели, към всеки от тях може да има няколко свързвания, наречени канали. Техните канали са около 300, всеки от каналите може да отпадне. От всички записвани метрики ще наблюдаваме процента на грешки (requests и errors).

Защо не графана?

Алармите за грешки, настроени в графана, имат няколко недостатъка. Някои от тях са критични, на някои може да се затворят очи, в зависимост от ситуацията.

Графана не умее да прави изчисления между измервания + алармиране, а ние се нуждаем от процента (requests-errors)/requests.

Грешките изглеждат злобно:

И по-малко злобно, ако се гледа с успешните заявки:

Добре, можем предварително да изчислим рейт в услугата преди графаната, и в някои случаи това би било подходящо. Но не в нашия случай, тъй като за всеки канал се изчислява „нормално“ соотношение, а алармите работят по статични стойности (търсим на око, променяме, ако алармира често).

Това са примери за „нормално“ за различни канали:

Пренебрегваме предишната точка и приемаме, че при всички доставчици картината е „нормална“. Сега всичко е наред, и можем ли да се справим с алармите в графаната?
Можем, но не ми се иска, защото трябва да изберем един от вариантите:
а) да направим множество графики за всеки канал поотделно (и мъчително да ги поддържаме)
б) да оставим една графика с всички канали (и да се загубим в цветни линии и настроени аларми)

Какво направихме?

Отново, в документацията има добър стартов пример (Изчисляване на рейтове за свързани серии), можем да надникнем или да използваме за основа в аналогични задачи.

Какво направихме в крайна сметка:

• обединение на две серии за няколко часа, групиране по канали;
• попълваме сериите по групи, ако няма данни;
• сравняваме медианата на последните 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)  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=UTC0300'
 )
 .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