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

아마도 오늘날 서비스 지표를 수집해야 하는 이유를 묻는 사람은 아무도 없을 것입니다. 다음 논리적 단계는 수집된 지표에 대해 편리한 채널(메일, Slack, Telegram)의 데이터 편차에 대해 알리는 경고를 설정하는 것입니다. 온라인 호텔 예약 서비스에서 Ostrovok.ru 우리 서비스의 모든 지표는 InfluxDB에 입력되어 Grafana에 표시되며 기본 경고도 거기에서 구성됩니다. "뭔가를 계산하고 비교해야 합니다"와 같은 작업에는 Kapacitor를 사용합니다.

Kapacitor는 InfluxDB의 메트릭을 처리할 수 있는 TICK 스택의 일부입니다. 여러 측정값을 함께 연결(조인)하고, 수신된 데이터에서 유용한 것을 계산하고, 결과를 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()
        .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)

해당 기간 동안 파이프라인에서 마지막 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}}'
    )

메시지 소수점 이하 두 자리

메시지 가독성 향상:

|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에 알림을 보내는 작은 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/작업/작업 이름/뭔가

실행 계획

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

갈퀴는 또 어디서 구할 수 있나요?

쓰기 저장 시 influxdb의 타임스탬프

예를 들어, 시간당 요청 합계(groupBy(1h))에 대한 알림을 설정하고 influxdb에 발생한 알림을 기록하려고 합니다(문제의 사실을 grafana의 그래프에 아름답게 표시하기 위해).

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

정확성이 필요한 경우: 현재 타임스탬프를 사용하여 influxdb에 데이터를 쓰는 사용자 정의 핸들러를 호출하여 이 문제를 해결합니다.

도커, 빌드 및 배포

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

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

1. 파일 이름 – 스크립트 ID/이름으로 확장됨
2. 유형 – 스트림/배치
3. dbrp – 스크립트가 실행되는 데이터베이스 + 정책을 나타내는 키워드(dbrp “supplier.”“autogen”)

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

반대로 chronograf에서는 이 줄이 존재해서는 안 되며, 인터페이스를 통해 허용되지 않고 오류가 발생합니다.

컨테이너 빌드 시 해킹: //.+dbrp가 포함된 행이 있는 경우 Dockerfile은 -1로 종료되므로 빌드를 어셈블할 때 실패 이유를 즉시 이해할 수 있습니다.

일대다에 합류하다

예제 작업: 일주일 동안 서비스 운영 시간의 95번째 백분위수를 가져와서 마지막 10분의 각 분위수를 이 값과 비교해야 합니다.

일대다 조인을 수행할 수 없습니다. 포인트 그룹에 대한 마지막/평균/중앙값은 노드를 스트림으로 전환합니다. "일치하지 않는 하위 가장자리를 추가할 수 없습니다: 배치 -> 스트림" 오류가 반환됩니다.

람다 식의 변수인 일괄 처리 결과도 대체되지 않습니다.

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

이것으로 무엇을 해결했습니까?

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

그라파나는 왜 안되나요?

Grafana에 구성된 오류 경고에는 몇 가지 단점이 있습니다. 상황에 따라 일부는 중요하고 일부는 눈을 감을 수 있습니다.

Grafana는 측정 + 알림 간 계산 방법을 모르지만 비율(요청-오류)/요청이 필요합니다.

오류가 심해 보입니다.

성공적인 요청으로 볼 때 덜 악합니다.

좋습니다. Grafana 이전에 서비스 요금을 미리 계산할 수 있으며 경우에 따라 이것이 작동할 수도 있습니다. 하지만 우리의 경우는 아니기 때문에... 각 채널마다 자체 비율이 "정상"으로 간주되며 경고는 정적 값에 따라 작동합니다(눈으로 찾고 경고가 자주 발생하면 변경합니다).

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

우리는 이전 요점을 무시하고 "정상적인" 그림이 모든 공급업체에 유사하다고 가정합니다. 이제 모든 것이 괜찮습니다. Grafana에서 경고를 통해 문제를 해결할 수 있습니까?
할 수 있지만 실제로는 원하지 않습니다. 왜냐하면 다음 옵션 중 하나를 선택해야 하기 때문입니다.
a) 각 채널에 대해 개별적으로 많은 그래프를 만듭니다(그리고 고통스럽게 동반합니다).
b) 모든 채널이 포함된 하나의 차트를 남깁니다. (그리고 다채로운 선과 맞춤형 알림에 빠져들게 됩니다.)

넌 어떻게 그걸 했니?

다시 한 번 문서에 좋은 시작 예제가 있습니다(결합된 계열 전체의 요율 계산), 유사한 문제에서 엿보거나 기초로 삼을 수 있습니다.

우리가 결국 한 일은:

• 채널별로 그룹화하여 몇 시간 안에 두 개의 시리즈에 참여하세요.
• 데이터가 없으면 그룹별로 시리즈를 채우십시오.
• 지난 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