InfluxDB告警配置需求：两类response_code统计值不等时触发

我已经配置了一个正常运行的InfluxDB告警，当值大于5时触发，对应的Flux查询代码如下：

import "influxdata/influxdb/monitor"
import "influxdata/influxdb/v1"

data = from(bucket: "my_bucket")
    |> range(start: -15s)
    |> filter(fn: (r) => r["_measurement"] == "my_mes")
    |> filter(fn: (r) => r["service"] == "test_service")
    |> filter(fn: (r) => r["action"] == "differ")
    |> filter(fn: (r) => r["env"] == "production")
    |> filter(fn: (r) => r["response_code"] == "400")
    |> filter(fn: (r) => r["_field"] == "value")
    |> aggregateWindow(every: 15s, fn: sum, createEmpty: false)

option task = {name: "differ_task", every: 15s, offset: 0s}

check = {_check_id: "12345", _check_name: "check_differ", _type: "threshold", tags: {differ_tag: "1"}}
crit = (r) => r["value"] > 5.0
messageFn = (r) => "Check: ${ r._check_name } is: ${ r._level }"

data |> v1["fieldsAsCols"]() |> monitor["check"](data: check, messageFn: messageFn, crit: crit)

现在需要修改告警规则，要求15秒内，response_code为"0"和"200"的value字段求和结果不相等时触发告警。我尝试编写了如下自定义查询但未达到预期效果，请问该需求是否可行？能否通过变量比较实现？

尝试的代码：

import "influxdata/influxdb/monitor"
import "influxdata/influxdb/v1"

data = from(bucket: "my_bucket")
    |> range(start: -15s)
    |> filter(fn: (r) => r["_measurement"] == "my_mes")
    |> filter(fn: (r) => r["service"] == "test_service")
    |> filter(fn: (r) => r["action"] == "differ")
    |> filter(fn: (r) => r["env"] == "production")
    |> filter(fn: (r) => r["response_code"] == "0")
    |> filter(fn: (r) => r["_field"] == "value")
    |> aggregateWindow(every: 15s, fn: sum, createEmpty: false)

data2 = from(bucket: "my_bucket")
    |> range(start: -15s)
    |> filter(fn: (r2) => r2["_measurement"] == "my_mes")
    |> filter(fn: (r2) => r2["service"] == "test_service")
    |> filter(fn: (r2) => r2["action"] == "differ")
    |> filter(fn: (r2) => r2["env"] == "production")
    |> filter(fn: (r2) => r2["response_code"] == "200")
    |> filter(fn: (r2) => r2["_field"] == "value")
    |> aggregateWindow(every: 15s, fn: sum, createEmpty: false)

option task = {name: "differ_task", every: 15s, offset: 0s}

check = {_check_id: "12345", _check_name: "check_differ", _type: "threshold", tags: {differ_tag: "1"}}
crit = (r) => r["value"] != r2["value"]
messageFn = (r) => "Check: ${r._check_name} is: ${r._level}"

data |> v1["fieldsAsCols"]() |> monitor["check"](data: check, messageFn: messageFn, crit: crit)

这个需求完全可行，你之前的代码问题在于两个独立的数据集data和data2没有关联，crit函数里无法直接访问r2变量。需要将两个数据集合并，把不同response_code的求和结果放到同一行中，再进行比较。

以下是修正后的Flux查询代码：

import "influxdata/influxdb/monitor"
import "influxdata/influxdb/v1"

// 一次性获取response_code为0和200的数据，避免重复查询
raw_data = from(bucket: "my_bucket")
    |> range(start: -15s)
    |> filter(fn: (r) => r["_measurement"] == "my_mes")
    |> filter(fn: (r) => r["service"] == "test_service")
    |> filter(fn: (r) => r["action"] == "differ")
    |> filter(fn: (r) => r["env"] == "production")
    |> filter(fn: (r) => r["response_code"] == "0" or r["response_code"] == "200")
    |> filter(fn: (r) => r["_field"] == "value")
    |> aggregateWindow(every: 15s, fn: sum, createEmpty: false)
    |> pivot(rowKey: ["_time"], columnKey: ["response_code"], valueColumn: "_value")

// 重命名列，方便后续比较
data = raw_data
    |> rename(columns: {"0": "value_0", "200": "value_200"})

option task = {name: "differ_task", every: 15s, offset: 0s}

check = {_check_id: "12345", _check_name: "check_differ", _type: "threshold", tags: {differ_tag: "1"}}
// 当两个值不相等时触发告警，同时处理其中一方无数据的情况（可选）
crit = (r) => r["value_0"] != r["value_200"] or exists r["value_0"] != exists r["value_200"]
messageFn = (r) => "Check: ${r._check_name} is CRITICAL. value_0: ${r.value_0}, value_200: ${r.value_200}"

data |> v1["fieldsAsCols"]() |> monitor["check"](data: check, messageFn: messageFn, crit: crit)

关键修改说明： #

• 合并查询：不再分两次查询，而是一次性过滤出response_code为0和200的数据，提升效率
• 使用pivot函数：将不同response_code的求和结果转为同一行的不同列，这样就能在crit函数中直接比较两个值
• 处理边界情况：crit函数中额外判断了其中一方无数据的场景（如果不需要可以去掉），避免漏告警
• 清晰的列名：重命名列后更直观，便于编写比较逻辑和告警消息

如果坚持要分开查询两个数据集，也可以用join.inner函数关联它们，代码示例如下：

import "influxdata/influxdb/monitor"
import "influxdata/influxdb/v1"
import "join"

data0 = from(bucket: "my_bucket")
    |> range(start: -15s)
    |> filter(fn: (r) => r["_measurement"] == "my_mes")
    |> filter(fn: (r) => r["service"] == "test_service")
    |> filter(fn: (r) => r["action"] == "differ")
    |> filter(fn: (r) => r["env"] == "production")
    |> filter(fn: (r) => r["response_code"] == "0")
    |> filter(fn: (r) => r["_field"] == "value")
    |> aggregateWindow(every: 15s, fn: sum, createEmpty: false)
    |> rename(columns: {"_value": "value_0"})

data200 = from(bucket: "my_bucket")
    |> range(start: -15s)
    |> filter(fn: (r) => r["_measurement"] == "my_mes")
    |> filter(fn: (r) => r["service"] == "test_service")
    |> filter(fn: (r) => r["action"] == "differ")
    |> filter(fn: (r) => r["env"] == "production")
    |> filter(fn: (r) => r["response_code"] == "200")
    |> filter(fn: (r) => r["_field"] == "value")
    |> aggregateWindow(every: 15s, fn: sum, createEmpty: false)
    |> rename(columns: {"_value": "value_200"})

// 按时间字段关联两个数据集
data = join.inner(
    tables: {t1: data0, t2: data200},
    on: ["_time"]
)
// 合并重复的标签列
data = data |> drop(columns: ["_measurement_t2", "service_t2", "action_t2", "env_t2", "response_code_t1", "response_code_t2", "_field_t1", "_field_t2"])

option task = {name: "differ_task", every: 15s, offset: 0s}

check = {_check_id: "12345", _check_name: "check_differ", _type: "threshold", tags: {differ_tag: "1"}}
crit = (r) => r["value_0"] != r["value_200"]
messageFn = (r) => "Check: ${r._check_name} is CRITICAL. value_0: ${r.value_0}, value_200: ${r.value_200}"

data |> v1["fieldsAsCols"]() |> monitor["check"](data: check, messageFn: messageFn, crit: crit)

这种方式适合两个数据集过滤条件差异较大的场景，但相比合并查询效率稍低。

内容的提问来源于stack exchange，提问作者Sergey