Kapacitor에서 메트릭을 처리하는 요령

오늘날에는 서비스 지표를 수집하는 것이 왜 필요한지 의문을 품은 사람은 아무도 없을 것입니다. 다음 논리적 단계는 수집된 지표에 대한 알림을 설정하는 것입니다. 이를 통해 편리한 채널(메일, Slack, Telegram)을 통해 데이터의 편차를 알려줍니다. 온라인 호텔 예약 서비스에서 Ostrovok.ru 모든 서비스 지표는 InfluxDB로 스트리밍되고 Grafana에 표시되며, 기본 알림도 구성됩니다. "무언가를 계산해서 이것과 비교해야 합니다"와 같은 작업에는 Kapacitor를 사용합니다.

Kapacitor는 InfluxDB의 메트릭을 처리할 수 있는 TICK 스택의 일부입니다. 여러 차원을 연결(조인)하고, 수신된 데이터에서 유용한 것을 계산하고, 결과를 InfluxDB에 쓰고, Slack/Telegram/메일에 알림을 보낼 수 있습니다.

전체 스택에는 멋지고 자세한 내용이 있습니다. 문서하지만 매뉴얼에 명확하게 표시되어 있지 않은 유용한 정보도 항상 있습니다. 이 기사에서는 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)

조인 채우기(내부 대 외부)

기본적으로 join은 데이터가 없는 지점(내부)을 삭제합니다.
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)

파이프라인의 마지막 5분 동안의 기간

예를 들어, 지난 5분 동안의 값을 지난주 값과 비교해야 합니다. 두 개의 별도 배치로 두 개의 데이터 배치를 가져오거나 더 긴 기간의 데이터 일부를 추출할 수 있습니다.

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

마지막 5분에 대한 대안은 지정된 시간 전에 데이터를 차단하는 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" }}')

사용자 정의 핸들러

핸들러의 큰 목록에는 전달된 매개변수(stdin)로 스크립트를 실행할 수 있게 해주는 exec가 포함되어 있습니다. 창의성만 있으면 됩니다!

우리의 관례 중 하나는 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 로그 레벨=오류

httpOut 옵션

현재 파이프라인의 데이터를 표시합니다.

|httpOut('something')

시청(받기): 호스트 또는 IP:9092/kapacitor/v1/작업/작업_이름/뭔가

구현 계획

• 각 작업은 다음 형식으로 유용한 숫자가 포함된 실행 트리를 반환합니다. 그래프비즈.
• 우리는 블록을 가져갑니다 점.
• 뷰어에 붙여넣기 즐기자.

또 어디에서 긁어모을 수 있나요?

influxdb의 writeback 타임스탬프

예를 들어, 우리는 시간당 요청 합계에 대한 알림을 설정하고(groupBy(1h)) 발생한 알림을 influxdb에 기록하려고 합니다(grafana의 그래프에서 문제의 사실을 잘 보여주기 위해).

influxDBOut()은 경고의 시간 값을 타임스탬프에 기록하므로 차트의 지점은 경고가 도착하기 전/후로 기록됩니다.

정밀도가 요구되는 경우 사용자 정의 핸들러를 호출하여 현재 타임스탬프와 함께 influxdb에 데이터를 기록함으로써 이 문제를 해결합니다.

도커, 빌드 및 배포

시작할 때 kapacitor는 config의 [load] 블록에서 지정된 디렉토리에서 작업, 템플릿 및 핸들러를 로드할 수 있습니다.

작업을 올바르게 생성하려면 다음 사항이 필요합니다.

1. 파일 이름 - ID/스크립트 이름으로 확장됩니다.
2. 유형 – 스트림/배치
3. dbrp – 스크립트가 작동하는 데이터베이스 + 정책을 지정하기 위한 키워드(dbrp "supplier"."autogen")

일부 배치 작업에 dbrp가 포함된 줄이 없으면 전체 서비스가 시작을 거부하고 로그에 솔직하게 기록하지 않습니다.

반면에 크로노그래피에서는 이런 선이 존재해서는 안 됩니다. 인터페이스를 통해 허용되지 않으며 오류를 반환합니다.

컨테이너 빌드 해킹: Dockerfile은 //.+dbrp가 포함된 줄이 있는 경우 -1을 반환하여 빌드 실패 이유를 즉시 파악할 수 있습니다.

일대다로 연결하다

예제 작업: 일주일 동안의 서비스 가동 시간의 95번째 백분위수를 구하고, 지난 10분의 각 분을 이 값과 비교해야 합니다.

일대다 조인을 수행할 수 없으며, 점 그룹에 대한 마지막/평균/중앙값을 사용하면 노드가 스트림으로 바뀌고 "자식 불일치 에지를 추가할 수 없습니다: 배치 -> 스트림" 오류가 반환됩니다.

람다 표현식의 변수인 배치 결과도 대체되지 않습니다.

첫 번째 배치에서 필요한 숫자를 udf를 통해 파일로 저장하고, 사이드로드를 통해 이 파일을 로드하는 옵션이 있습니다.

우리는 이것으로 무엇을 해결하고 있었는가?

호텔 공급업체는 약 100개 정도이고, 각 공급업체는 여러 개의 연결을 가질 수 있습니다. 이를 채널이라고 부르죠. 이런 채널은 약 300개가 있으며, 각 채널은 사라질 수 있습니다. 기록된 모든 지표 중에서 오류율(요청 및 오류)을 모니터링합니다.

왜 Grafana는 아닌가?

Grafana에서 구성된 오류 알림에는 여러 가지 단점이 있습니다. 상황에 따라 일부는 중요하고 일부는 무시해도 됩니다.

Grafana는 차원 간 계산 및 알림을 수행할 수 없지만 비율(요청-오류)/요청이 필요합니다.

오류가 심각해 보입니다.

성공적인 쿼리를 살펴보면 덜 사악합니다.

네, Grafana 이전에 서비스에서 요금을 미리 계산할 수 있고, 어떤 경우에는 이 방법이 효과적일 수도 있습니다. 하지만 우리 채널에서는 그렇지 않습니다. 각 채널의 비율이 "정상"으로 간주되고 알림은 정적 값에 따라 작동하기 때문입니다(우리는 눈으로 확인하고, 알림이 자주 울리면 변경합니다).

다음은 다양한 채널의 "정상"에 대한 예입니다.

이전 사항은 무시하고 모든 공급업체의 "정상적인" 상황이 비슷하다고 가정해 보겠습니다. 이제 모든 것이 괜찮고, 그라파나에서 알림을 받을 수 있을까요?
할 수는 있지만, 우리는 정말로 하고 싶지 않습니다. 왜냐하면 우리는 다음 옵션 중 하나를 선택해야 하기 때문입니다.
a) 각 채널에 대해 개별적으로 많은 그래프를 만듭니다(그리고 이를 유지하는 데 많은 시간이 걸립니다)
b) 모든 채널이 포함된 하나의 차트를 남겨두고(다채로운 선과 사용자 지정 알림에 묻혀버리세요)

어떻게 했나요?

다시 한번, 문서에 좋은 시작 예가 있습니다.결합된 시리즈에 대한 비율 계산), 살펴보거나 비슷한 문제에 대한 기초로 사용할 수 있습니다.

결국 우리가 한 일은 다음과 같습니다.

• 몇 시간 안에 두 개의 에피소드를 합치고 채널별로 그룹화합니다.
• 데이터가 없으면 그룹별로 시리즈를 채웁니다.
• 지난 10분 동안의 중간값을 이전 데이터와 비교합니다.
• 뭔가 발견하면 소리치자.
• 계산된 요율과 발생한 알림을 influxdb에 기록합니다.
• Slack에 유용한 메시지를 보내세요.

제 생각에, 우리는 가장 아름다운 방법으로(그리고 맞춤형 핸들러를 사용하면 그보다 더 많은 것을) 우리가 원했던 모든 것을 최종적으로 달성할 수 있었습니다.

github.com에서 살펴보실 수 있습니다. 예제 코드 и 최소 계획(그래프비즈) 받은 스크립트.

결과 코드의 예:

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