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

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

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

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

Да вървим!

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

Ако етикет/поле не е попълнено, ще възникнат грешки в изчислението:

|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) – чакаме корекции и малко страдаме.

Използване на условия в изчисленията (ако е в ламбда)

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

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

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

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

докер, изграждане и внедряване

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

За да създадете правилно задача, имате нужда от следните неща:

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

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

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

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

присъединете се към един към много

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

Не можете да направите присъединяване "един към много", последната/средната стойност/медианата над група от точки превръща възела в поток, ще бъде върната грешката „не може да добави дъщерни несъответстващи ръбове: партида -> поток“.

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

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

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

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

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

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

Grafana не знае как да изчислява между измервания + сигнализиране, но ни трябва процент (заявки-грешки)/заявки.

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

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

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

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

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

Как го направи?

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

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

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

Бариерата за влизане не е много висока - опитайте, ако графана или други инструменти не задоволяват напълно вашите желания.

Източник: www.habr.com