Хотите использовать несколько окон в одном потоке с kapacitor

Question

Цель: я хотел бы получить уведомление через 5 минут после состояния предупреждения, а затем каждые 30 минут после этого.

Я поиграл с .count() и функцией времени, но это ни к чему не привело, мне не хотелось вычислять этот беспорядок, и я не мог придумать, как сделать его удобным и надежным.

Решение, с которым я сейчас работаю, состоит в том, чтобы иметь два потока с отдельными окнами.

var initialData = stream
        |from()
            .database(db)
            .measurement(metricType)
            .retentionPolicy(rPolicy)
            .groupBy(group)
            .where(lambda: "cpu" == 'cpu-total')
            .where(lambda: "host" =~ hostFilter)
        |mean(metric)
            .as('initialStat')
        |window()
            .period(10m)
            .every(5m)
            .align()

var continuousData = stream
    |from()
        .database(db)
        .measurement(metricType)
        .retentionPolicy(rPolicy)
        .groupBy(group)
        .where(lambda: metricType == 'cpu-total')
        .where(lambda: "host" =~ hostFilter)
    |mean(metric)
        .as('continuousStat')
    |window()
        .period(10m)
        .every(30)
        .align()

Помимо того, что это кажется странным, мне нужно было бы рассчитать значения для каждого, и мне также нужно отдельно |alert() узлы. Первый узел будет уведомлять только об изменении состояния, но второй узел не может иметь этого, поэтому я получаю напоминание о предупреждении каждые N минут. У меня также проблема в том, что первый |alert() узел отправит OK уведомление, а второй также отправит дупе OK N минут спустя.

Я чувствую, что должен быть лучший способ сделать это. Я думаю, что я могу использовать if заявление во втором |alert() узел, чтобы не отправлять уведомление о OK как первый |window справлюсь с этим. На данный момент я еще не понял, как это сделать, но я уверен, что это возможно. Я также не хочу бороться с тикстингом, я знаю, что он не предназначен для использования в качестве полноценного языка в выпуске 741

Полный текст ниже

// CONFIGURATION PARAMETERS

// Alerting

var emailAddress = '$EMAIL'
var pagerdutyKey = '$PD'
var slackChannel = '$SLACK'

// Static Thresholds in percent cpu steal used
var warn = 85
var crit = 95

// Dynamic thresholds in number of std deviations
var warnSig = 2.5
var critSig = 3.5

// Print INFO level (every result will be an alert)
// AlertNode.StateChangesOnly will also need to be disabled
// NOTE:
// INFO level alerts will be disregarded by the pagerduty handler, this is not configurable.
var debug = FALSE

// Datastream
// Define the data that will be acted upon
var db           = 'telegraf'
var group        = 'host'
var metricType   = 'cpu'
var metric       = 'time_steal'
var rPolicy      = 'default'

// Regex used to filter on a subset of hosts
var hostFilter = /.+/

// Window
var dataPeriod            = 10m
var initialFrequency      = 5m
var continuousFrequency   = 30m

// DATAFRAME
var initialData = stream
    |from()
        .database(db)
        .measurement(metricType)
        .retentionPolicy(rPolicy)
        .groupBy(group)
        .where(lambda: metricType == 'cpu-total')
        .where(lambda: "host" =~ hostFilter)
    |mean(metric)
        .as('initialStat')
    |window()
        .period(dataPeriod)
        .every(initialFrequency)
        .align()

var continuousData = stream
    |from()
        .database(db)
        .measurement(metricType)
        .retentionPolicy(rPolicy)
        .groupBy(group)
        .where(lambda: metricType == 'cpu-total')
        .where(lambda: "host" =~ hostFilter)
    |mean(metric)
        .as('continuousStat')
    |window()
        .period(dataPeriod)
        .every(continuousFrequency)
        .align()

// Calculations
var initialCalculation = initialData
    |eval(lambda: sigma("initialStat"))
        .as('intialSigma')
        .keep()

var continuousCalculation = continuousData
    |eval(lambda: sigma("continuousStat"))
        .as('continuousSigma')
        .keep()

// ALERT CONDITIONS
var initialCondition = initialCalculation
    |alert()
        .id('{{ index .Tags "host"  }}')
        .message('{{ .ID  }} is {{ .Level  }}: CPU STEAL USAGE {{ index .Fields "initialStat" }}% SHORT')
        .details('this is an alert')
        .stateChangesOnly()
        .info(lambda: debug)
        .warn(lambda: "stat" < warn OR
            "sigma" > warnSig)
        .crit(lambda: "stat" < crit OR
            "sigma" > critSig)

var continuousCondition = continuousCalculation
    |alert()
        .id('{{ index .Tags "host"  }}')
        .message('{{ .ID  }} is {{ .Level  }}: CPU STEAL USAGE {{ index .Fields "continuousStat" }}% LONG')
        .details('this is an alert')
        .info(lambda: debug)
        .warn(lambda: "stat" < warn OR
            "sigma" > warnSig)
        .crit(lambda: "stat" < crit OR
            "sigma" > critSig)

// ACTIONS
continuousCondition
        // .log('/tmp/alerts/cpu_steal_usage_alerts')
        // .slack()
        // .channel(slackChannel)
        .email(emailAddress)
        .pagerDuty()
                .serviceKey(pagerdutyKey)

initialCondition
        // .log('/tmp/alerts/cpu_steal_usage_alerts')
        // .slack()
        // .channel(slackChannel)
        .email(emailAddress)
        .pagerDuty()
                .serviceKey(pagerdutyKey)

Answer 1

Так что, очевидно, я могу сделать несколько окон в одном узле потока.

stream
    |from()
        .database(db)
        .measurement(metricType)
        .retentionPolicy(rPolicy)
        .groupBy(group)
        .where(lambda: metricType == metricFilter)
        .where(lambda: "host" =~ hostFilter)
    |window()
        .period(dataPeriod)
        .every(initialFrequency)
        .align()
    |mean(metric)
        .as('initialStat')
    |window()
        .period(dataPeriod)
        .every(continuousFrequency)
        .align()
    |mean(metric)
        .as('continuousStat')

Все еще работает через OK вопрос хотя.