Example of a dynamic alerting script in Flux

dynamic-alert.flux

import "experimental/http/requests"
import "influxdata/influxdb/secrets"
import "json"
import ejson "experimental/json"
import "slack"

option task = {name: "example-task", every: 10m}

slackToken = secrets.get(key: "SLACK_TOKEN")

// Create a custom function to pull data from a custom API.
getHistoricalData = (depth, lat, lon) => {
    _response = requests.post(
        url: "http://example-pyapp-url.com",
        body: json.encode(v: {depth: depth, lat: lat, lon: lon}),
    )

    // Parse the returned JSON body into a Flux record and output the record.
    return ejson.parse(data: _response.body)
}

from(bucket: "example-bucket")
    |> range(start: -task.every)
    |> filter(fn: (r) => r._measurement == "example-measurement")
    |> filter(fn: (r) => r._field == "example-field")
    // Use data in each input row to request historical data
    |> map(fn: (r) => {
        hData = getHistoricalData(depth: r.depth, lat: r.lat, lon: r.lon)

        // Assign a status to each input row based on how the value compares
        // to the historical mean and standard devation
        return {r with
            status: if ((r._value - hData.mean) / hData.stddev) > 1.0 then "error" else "ok" 
        }
    })
    // Only return rows with the "error" status
    |> filter(fn: (r) => r.status == "error")
    // Use map() to interate over rows and send a message to Slack for each row.
    |> map(fn: (r) => ({r with 
        sent: slack.message(
            url: "https://slack.com/api/chat.postMessage",
            token: slackToken,
            channel: "#example-channel",
            text: "Sensor ${r.id} is reporting anomalous readings.",
            color: "warning",
        )
    }))