InfluxDBOutNode

stream
  |from()
    .measurement('requests')
  |eval(lambda: "errors" / "total")
    .as('error_percent')
  // Write the transformed data to InfluxDB
  |influxDBOut()
    .database('mydb')
    .retentionPolicy('myrp')
    .measurement('errors')
    .tag('kapacitor', 'true')
    .tag('version', '0.2')

influxDBOut.buffer(value int64)

influxDBOut.cluster(value string)

influxDBOut.create()

influxDBOut.database(value string)

influxDBOut.flushInterval(value time.Duration)

influxDBOut.measurement(value string)

influxDBOut.precision(value string)

influxDBOut.quiet()

influxDBOut.retentionPolicy(value string)

influxDBOut.tag(key string, value string)

influxDBOut.writeConsistency(value string)

    var data = stream
        |from()...
    // Trigger critical alert if the throughput drops below 100 points per 10s and checked every 10s.
    data
        |deadman(100.0, 10s)
    //Do normal processing of data
    data...

    var data = stream
        |from()...
    // Trigger critical alert if the throughput drops below 100 points per 10s and checked every 10s.
    data
        |stats(10s)
            .align()
        |derivative('emitted')
            .unit(10s)
            .nonNegative()
        |alert()
            .id('node \'stream0\' in task \'{{ .TaskName }}\'')
            .message('{{ .ID }} is {{ if eq .Level "OK" }}alive{{ else }}dead{{ end }}: {{ index .Fields "emitted" | printf "%0.3f" }} points/10s.')
            .crit(lambda: "emitted" <= 100.0)
    //Do normal processing of data
    data...

    var data = stream
        |from()...
    // Trigger critical alert if the throughput drops below 100 points per 10s and checked every 10s.
    data
        |deadman(100.0, 10s)
            .slack()
            .channel('#dead_tasks')
    //Do normal processing of data
    data...

    var data = stream
        |from()...
    // Trigger critical alert if the throughput drops below 100 points per 10s and checked every 10s.
    // Only trigger the alert if the time of day is between 8am-5pm.
    data
        |deadman(100.0, 10s, lambda: hour("time") >= 8 AND hour("time") <= 17)
    //Do normal processing of data
    data...

influxDBOut|deadman(threshold float64, interval time.Duration, expr ...ast.LambdaNode)

influxDBOut|stats(interval time.Duration)