如何使用 InfluxDB 和 Flux 查询语言创建 Grafana 告警

作者：Grant Pinkos

最后更新于 2023 年 7 月 18 日

高级

引言

Grafana 告警代表了一种强大的新方法，用于系统可观测性和事件响应管理。虽然告警平台可能以其与 Prometheus 的强大集成而闻名，但该系统支持包括 InfluxDB 在内的众多流行数据源。在本教程中，我们将学习如何使用 InfluxDB 和更新的 Flux 查询语言创建 Grafana 告警。我们将涵盖从最基本到最复杂的五种常见场景。这五种场景将为使用 Grafana 和 Flux 创建几乎任何类型的告警查询提供极好的指导。

在深入探讨我们的告警场景之前，有必要回顾一下 InfluxDB 的两种流行查询语言：InfluxQL 和 Flux 的发展历程。最初，InfluxDB 使用 InfluxQL 作为其查询语言，它使用类似于 SQL 的语法。但从 InfluxDB v1.8 开始，该公司引入了 Flux，“一种开源函数式数据脚本语言，旨在用于查询、分析和处理数据。”其官方文档进一步指出，“Flux 将查询、处理、写入和处理数据的代码统一到单个语法中。该语言设计得易于使用、可读、灵活、可组合、可测试、可贡献和可共享。”

在接下来的五个示例中，我们将看到新的 Flux 查询语言是多么强大和灵活。我们还将看到 Flux 与 Grafana 告警配合得多么出色。

示例 1：当值高于或低于设定阈值时创建告警

我们的第一个示例使用了 InfluxDB 和 Grafana 告警的一个常见实际场景。InfluxDB 在 IoT 和边缘应用中很受欢迎，擅长现场实时可观测性。在本例以及后续的许多示例中，我们将考虑一种假设场景：我们正在监控制造工厂中的一些流体罐。此场景基于 InfluxDB 和告警的实际应用，将允许我们逐步完成 Grafana 的各种告警设置，从最简单到最复杂。

对于示例 1，让我们考虑以下场景：我们正在监控一个罐体 A5，并存储其实时温度数据。我们需要确保该罐体的温度始终大于 30 °C 且小于 60 °C。

我们想编写一个 Grafana 告警，该告警将在罐体 A5 中的温度低于 30 °C 或高于 60 °C 时触发。

为此，我们将：创建 Grafana 告警规则，添加 Flux 查询，然后向告警规则添加表达式。

创建 Grafana 告警规则

1. 打开 Grafana 告警菜单并选择告警规则。
2. 点击新建告警规则。
3. 为您的告警规则命名，然后选择 Grafana 管理告警。对于 InfluxDB，您将始终创建一个Grafana 管理规则。

向告警规则添加初始 Flux 查询

仍在告警规则页面步骤 2 部分，您将看到三个框：一个查询编辑器 (A)，以及两个标记为 B 和 C 的部分。您将使用这三个部分来构建您的规则。让我们逐一讲解。

首先，我们想要查询我们的虚拟 InfluxDB 实例中的数据，以获取罐体 A5 温度的时间序列图。为此，您将从下拉列表中选择您的 InfluxDB 数据源，然后编写一个如下所示的查询

```
 from(bucket: "RetroEncabulator")
|> range(start: v.timeRangeStart, stop: v.timeRangeStop)
|> filter(fn: (r) => r["_measurement"] == "TemperatureData")
|> filter(fn: (r) => r["Tank"] == "A5")
|> filter(fn: (r) => r["_field"] == "Temperature")
|> aggregateWindow(every: v.windowPeriod, fn: mean, createEmpty: false)
|> yield(name: "mean")
```

这是一个相当典型的 Flux 查询。让我们逐个函数进行解释。我们首先使用from() 函数选择存储我们罐体数据的正确 bucket。然后，我们使用range() 函数根据时间限制过滤行。然后，我们通过三个filter() 函数来缩小结果范围。我们选择一个特定的 measurement（InfluxDB 中的一个特殊关键字），然后是我们查询的罐体 (A5)，再然后是一个特定的 field（InfluxDB 中的另一个特殊关键字）。之后，我们将数据传递到aggregateWindow() 函数中，该函数将我们的数据下采样到特定的时间段，最后是yield() 函数，该函数指定我们想要的最终结果：mean（均值）。

这个 Flux 查询将生成一个如下所示的时间序列图

向 Grafana 告警规则添加表达式

数据现在出现在我们的规则设置中，下一步是创建一个表达式。移动到 B 部分。对于此场景，我们希望创建一个 Reduce 表达式，将上述结果减少到单个值。在此图中，您可以看到我们选择了将输入 A 中的时间序列数据缩减为Last 值。在本例中，它返回罐体 A5 的温度值 53 摄氏度

最后，我们需要创建一个数学表达式，Grafana 将根据它触发告警。在我们的示例中，我们将编写一个包含两个条件，并使用 OR || 运算符分隔的表达式。我们希望当 B 部分的结果小于 30 或大于 60 时触发告警。表达式如下：$B < 30 || $B > 60

将告警条件设置为 C - expression。我们现在可以预览我们的告警了。这是当状态为 Normal 时的告警预览

这是当状态为 Alerting 时的告警预览

请注意，需要上述 Reduce 表达式。否则，在预览结果时，Grafana 将显示 invalid format of evaluation results for the alert definition B: looks like time series data, only reduced data can be alerted on。

💡提示：如果您所在区域仍然固执地使用华氏度，我们可以修改上述 Flux 查询，在 aggregateWindow 语句之前添加一个 map() 函数，将值从 °C 转换为 °F。请注意，我们并未创建新字段。我们只是重新映射了现有值。

flux 复制

|> map(fn: (r) => ({r with _value: r._value * 1.8 + 32.0}))

结论

通过这三个步骤，您可以创建一个基于 Flux 的 Grafana 告警，该告警将根据单个数据源的两个阈值中的任何一个触发。但是，如果需要根据多个条件和多个时间序列触发告警怎么办？在示例二中，我们将涵盖这个具体的场景。

示例 2：如何通过两次查询和两个条件创建 Grafana 告警

让我们在示例二中稍微改变一下场景，离开我们假想的制造工厂。想象一下，您是《回到未来》中伟大的 Emmett Brown 博士的助手，布朗博士给您布置了以下挑战：“我希望每当满足时间旅行的两个条件时，都给我发送一个告警：当车辆速度达到每小时 88 英里，并且一个物体产生 1.21 千兆瓦电力时。”

假设我们在 InfluxDB 和 Grafana 中跟踪这些数据。再假设上述每个数据源都来自不同的桶。我们如何对此进行告警？我们如何使用 Grafana 和 Flux 对来自两个不同数据源的两个不同条件进行告警？

向 Grafana 告警规则添加两次 Flux 查询

像示例 1 中那样，我们首先模拟我们的查询。车辆数据的查询与上一个查询非常相似。我们使用 from()、range() 和一系列 filter() 函数。然后我们使用 AggregateWindow() 和 yield() 函数进一步缩小数据范围。在本例中，结果是一个跟踪我们的 1983 年 DeLorean 车辆速度的时间序列

flux 复制

from(bucket: "vehicles")
|> range(start: v.timeRangeStart, stop: v.timeRangeStop)
|> filter(fn: (r) => r["_measurement"] == "VehicleData")
|> filter(fn: (r) => r["VehicleType"] == "DeLorean")
|> filter(fn: (r) => r["VehicleYear"] == "1983")
|> filter(fn: (r) => r["_field"] == "velocity")
|> aggregateWindow(every: v.windowPeriod, fn: mean, createEmpty: false)
|> yield(name: "mean")

当我们的电力资源（希尔谷钟楼的闪电）达到所需的 1.21 千兆瓦时，我们的第二个查询将触发告警。类似这样的查询将与我们的车辆速度查询非常相似

flux 复制

from(bucket: "HillValley")
|> range(start: v.timeRangeStart, stop: v.timeRangeStop)
|> filter(fn: (r) => r["_measurement"] == "ElectricityData")
|> filter(fn: (r) => r["Location"] == "clocktower")
|> filter(fn: (r) => r["Source"] == "lightning")
|> filter(fn: (r) => r["_field"] == "power")
|> aggregateWindow(every: v.windowPeriod, fn: mean, createEmpty: false)
|> yield(name: "mean")

我们现在可以使用表达式来修改这些数据了。

向 Grafana 告警规则添加表达式

1. 现在，我们使用相同的步骤将每个查询缩减到最后一个（最新的）值。将查询 A 缩减为单个值可能如下所示

2. 以下是我们将查询 B 缩减后的结果

3. 现在，在 C 部分，我们需要创建一个数学表达式进行告警。在这种情况下，我们将使用 AND && 运算符来指定必须满足两个条件：C 的值（查询 A 的缩减值）必须大于 88.0，而 D 的值（查询 B 的缩减值）必须大于 1.21。我们将其写成 $C > 88.0 && $D > 1.21

这是我们的告警预览

💡提示：如果您的 InfluxDB 数据小数位过多（例如上面显示的 1.2104705741732575），并且您希望 Grafana 告警更易读，可以尝试使用 {{ printf “%.2f” $values.D.Value }}。例如，在注释摘要中，我们可以写下以下内容

复制

{{  $values.D.Labels.Source }} at the {{  $values.D.Labels.Location }} has generated {{ printf "%.2f"  $values.D.Value }} jigowatts.`

这将显示如下： )

您可以参考我们的告警消息模板文档，以了解更多关于此强大功能的信息。

结论

在本示例中，我们展示了如何创建基于 Flux 的告警，该告警使用来自两个不同数据源的两个不同查询中的两个不同条件。对于示例三，我们将改变思路，解决另一个常见的告警场景：如何基于聚合（每日）值创建告警。

示例 3：如何基于聚合（每日）值创建 Grafana 告警

Grafana 社区论坛中最常见的请求之一涉及绘制每日用电量和发电量的图表。此类数据通常存储在 InfluxDB 中。在本示例中，我们将展示如何将时间序列数据聚合为每日值，然后根据该值触发告警。

假设我们的电表每小时向 InfluxDB 发送一个读数，其中包含该小时的总用电量 (kWh)。我们希望编写一个查询，将这些每小时值聚合为每日值，然后创建一个告警，当用电量 (kWh) 超过每天 5,000 kWh 时触发。

向 Grafana 告警规则添加初始 Flux 查询

1. 让我们首先检查一个典型查询及其在 7 天期间内每小时数据的生成时间图。如下所示的查询：

   flux 复制

   from(bucket: "RetroEncabulator")
   |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
   |> filter(fn: (r) => r["_measurement"] == "ElectricityData")
   |> filter(fn: (r) => r["Location"] == "PlantD5")
   |> filter(fn: (r) => r["_field"] == "power_consumed")
   |> aggregateWindow(every: v.windowPeriod, fn: mean, createEmpty: false)
   |> yield(name: "power")

   我们可以看到与示例 1 和 2 中相同的 Flux 函数模式。这样的查询会生成类似于下面的图表

2. 现在让我们调整查询来计算每日用量。对于许多数据源，这可能是一个相当复杂的操作。但对于 Flux，只需更改 aggregateWindow 函数参数，我们就可以计算相同 7 天期间的每日用量

   flux 复制

   from(bucket: "RetroEncabulator")
     |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
     |> filter(fn: (r) => r["_measurement"] == "ElectricityData")
     |> filter(fn: (r) => r["Location"] == "PlantD5")
     |> filter(fn: (r) => r["_field"] == "power_consumed")
     |> aggregateWindow(every: 1d, fn: sum)
     |> yield(name: "power")

   请注意我们是如何将 aggregateWindow() 函数调整为 aggregateWindow(every: 1d, fn: sum) 的。这将生成一个如下所示的图表

3. 将表达式添加到 Grafana 告警规则中。

   既然我们的每日查询正确了，我们可以像以前一样继续使用相同的模式，添加表达式来缩减和对我们的结果执行数学运算。

   像之前一样，我们将查询缩减为一个单值

   现在创建要触发告警的数学表达式并设置评估行为。在本例中，我们希望编写 $B > 5000

   现在，只要我们的每日用电量超过 5000 kWh，我们就会触发告警。这是我们的告警预览

结论

绘制和聚合电量消耗是 InfluxDB 和 Grafana 结合使用的常见用例。使用 Flux，我们看到了按天分组数据并根据该每日值触发告警是多么容易。在接下来的两个示例中，我们将探讨更复杂的 Grafana 告警形式：多维告警。

示例 4：如何使用 Flux 创建动态（多维）Grafana 告警

让我们回到示例 1 中的流体罐，但这次假设我们有 5 个罐体（A5、B4、C3、D2 和 E1）。我们现在正在监控五个罐体：A5、B4、C3、D2 和 E1 的温度。

我们希望创建一个多维告警，当任何罐体的温度低于 30 °C 或高于 60 °C 时通知我们。

向 Grafana 告警规则添加初始 Flux 查询

我们一如既往地从编写初始查询开始。这与示例 1 中的查询非常相似，但请注意我们的第三个 filter() 函数如何捕获所有五个罐体的数据，而不仅仅是 A5

flux 复制

from(bucket: "HyperEncabulator")
|> range(start: v.timeRangeStart, stop: v.timeRangeStop)
|> filter(fn: (r) => r["_measurement"] == "TemperatureData")
|> filter(fn: (r) => r["MeasType"] == "actual")
|> filter(fn: (r) => r["Tank"] == "A5" or r["Tank"] == "B4" or r["Tank"] == "C3" or r["Tank"] == "D2" or r["Tank"] == "E1")
|> aggregateWindow(every: v.windowPeriod, fn: mean, createEmpty: false)
|> yield(name: "mean")

💡提示：如果每天晚上 23:00 到 07:00 罐体关闭，它们的温度可能会降至 30 °C 阈值以下。如果不想在这些时间段收到告警，可以使用 Flux 函数 hourSelection()，该函数按指定小时范围内的时值过滤行。

flux 复制

|> hourSelection(start: 7, stop: 23)`

上面的查询将生成如下所示的时间序列图

向 Grafana 告警规则添加表达式

1. 我们创建了一个 Reduce 表达式，将每个罐体的时间序列数据缩减为单个值。这将得到五个不同的温度值

   )

2. 创建一个数学表达式用于触发告警。这与示例 1 中的表达式完全相同，即 $B < 30 || $B > 60

正如我们所见，三个罐体在可接受的阈值内，而两个罐体已超过上限。这将触发罐体 D2 和 E1 的告警。

结论

使用多维告警，我们可以避免重复。但是，如果场景更复杂呢？在下一个也是最后一个示例中，我们将探讨如何使用多维告警创建最动态的告警。

示例 5：如何使用 Flux 通过多次查询和多个阈值创建动态（多维）Grafana 告警

对于最后一个示例，让我们继续使用我们的五个流体罐及其五个数据集。再次假设每个罐体都有一个温度控制器，其设定点值存储在 InfluxDB 中。让我们稍微改变一下，假设每个罐体都有一个不同的设定点，并且我们始终需要将温度控制在设定点正负 3 度以内。

我们希望创建一个多维告警，该告警将涵盖每个罐体的独特场景，当任何罐体的温度超出其独特的允许范围时触发告警。

为了更好地可视化这个挑战，这里有一个表格，代表了我们的五个罐体、它们的温度设定点以及它们的允许范围

使用 Grafana Alerting，我们可以创建一个单一的多维规则来覆盖所有 5 个罐体，并且可以使用 Flux 来比较每个罐体的设定点和实际值。换句话说，一个多维告警可以监控 5 个独立的罐体，每个罐体都有不同的设定点和实际值，但都遵循一个共同的“允许阈值”（即 ±3 度的温差）。

向 Grafana 告警规则添加初始 Flux 查询

让我们从数据查询开始。它与我们过去的查询相似，但现在更复杂。我们必须添加额外的函数以将数据转换为正确的格式，包括 pivot()、map()、rename()、keep() 和 drop() 函数

flux 复制

from(bucket: "HyperEncabulator")
 |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
 |> filter(fn: (r) => r["_measurement"] == "TemperatureData")
 |> filter(fn: (r) => r["MeasType"] == "actual" or r["MeasType"] == "setpoint")
 |> filter(fn: (r) => r["Tank"] == "A5" or r["Tank"] == "B4" or r["Tank"] == "C3" or r["Tank"] == "D2" or r["Tank"] == "E1")
 |> filter(fn: (r) => r["_field"] == "Temperature")
 |> aggregateWindow(every: v.windowPeriod, fn: mean, createEmpty: false)
 |> pivot(rowKey:["_time"], columnKey: ["MeasType"], valueColumn: "_value")
 |> map(fn: (r) => ({ r with _value: (r.setpoint - r.actual)}))
 |> rename(columns: {_value: "difference"})
 |> keep(columns: ["_time", "difference", "Tank"])
 |> drop(columns: ["actual", "setpoint"])
 |> yield(name: "mean")

请注意，在上面的查询中，我们正在计算实际值与设定点之间的差值。Grafana 解析 InfluxDB 结果的方式是，如果找到 _value 列，则假定它是时间序列数据。一个快速的解决方案是添加以下 rename() 函数

flux 复制

 |> rename(columns: {_value: "something"})

上述查询生成如下时间序列

向 Grafana 告警规则添加表达式

1. 同样，我们为上述查询创建了一个 Reduce 表达式，将上述每个系列缩减为一个单值。这个值代表了每个罐体的设定点与其实际实时温度之间的温差

2. 现在我们创建一个数学表达式进行告警。这次我们将创建一个条件，检查我们的缩减计算的绝对值是否大于 3，即 abs($(B))>3.0

我们现在可以看到，两个罐体，D2 和 E1，正在评估为真。当我们预览告警时，我们可以看到这两个罐体将触发通知并将其状态从 Normal 更改为 Alerting

结论

Flux 查询和 Grafana 统一告警是强大的组合，可用于识别数据集中或整个系统中的几乎任何可触发告警的条件。有关 Grafana Alerting 的更多信息，请访问此处文档。有关 Flux 查询语言的更多信息，您也可以访问该文档。