跨任务执行跟踪状态更改

问题

通常使用 InfluxDB 任务 来评估和分配时间序列数据的状态，然后检测这些状态的变化。任务以批处理方式处理数据，但是如果跨批处理边界发生状态更改会怎样？如果不知道上一次任务执行的最终状态，任务将无法识别它。本指南将引导您创建一个任务，该任务为行分配状态，然后使用上一次任务执行的结果来检测跨批处理边界的任何状态更改，这样您就不会错过任何状态更改。

解决方案

根据阈值显式分配数据的级别。

解决方案优势

如果您从未使用 monitor 包编写过任务，那么这是最容易理解的解决方案。

解决方案缺点

您必须显式定义阈值，这可能需要更多代码。

解决方案概述

创建一个任务，您可以在其中

1. 样板代码。导入包并定义任务选项。
2. 查询您的数据。
3. 根据阈值将状态分配给您的数据。将此数据存储在变量中，例如“states”。
4. 将“states”写入存储桶。
5. 查找上一次任务运行的最新值，并将其存储在变量“last_state_previous_task”中。
6. 联合“states”和“last_state_previous_task”。将此数据存储在变量“unioned_states”中。
7. 在“unioned_states”中发现状态更改。将此数据存储在变量“state_changes”中。
8. 通知跨越最后两个任务的状态更改，以捕获在任务执行之间发生的任何状态更改。

解决方案详解

1. 导入包并定义任务选项和密钥。导入以下包

• Flux Telegram 包：此包

• Flux InfluxDB 密钥包：此包包含 secrets.get() 函数，该函数允许您从 InfluxDB 密钥存储中检索密钥。了解如何在 InfluxDB 中 管理密钥 以使用此包。

• Flux InfluxDB 监控包：此包包含用于监控数据的功能和工具。

  import "contrib/sranka/telegram"
  import "influxdata/influxdb/secrets"
  import "influxdata/influxdb/monitor"
  
  option task = {name: "State changes across tasks", every: 30m, offset: 5m}
  
  telegram_token = secrets.get(key: "telegram_token")
  telegram_channel_ID = secrets.get(key: "telegram_channel_ID")
  

2. 查询您要监控的数据。

   data = from(bucket: "example-bucket")
       // Query for data from the last successful task run or from the 1 every duration ago.
       // This ensures that you won’t miss any data.
       |> range(start: tasks.lastSuccess(orTime: -task.every))
       |> filter(fn: (r) => r._measurement == "example-measurement")
       |> filter(fn: (r) => r.tagKey1 == "example-tag-value")
       |> filter(fn: (r) => r._field == "example-field")
   

   其中 data 可能看起来像

   _measurement	tagKey1	_field	_value	_time
   example-measurement	example-tag-value	example-field	30.0	2022-01-01T00:00:00Z
   example-measurement	example-tag-value	example-field	50.0	2022-01-01T00:00:00Z

3. 根据阈值将状态分配给您的数据。将此数据存储在变量中，例如“states”。为了简化此示例，只有两个状态：“ok”和“crit”。将状态存储在 _level 列中（monitor 包要求）。

   states =
       data
           |> map(fn: (r) => ({r with _level: if r._value > 40.0 then "crit" else "ok"}))
   

   其中 states 可能看起来像

   _measurement	tagKey1	_field	_value	_level	_time
   example-measurement	example-tag-value	example-field	30.0	ok	2022-01-01T00:00:00Z
   example-measurement	example-tag-value	example-field	50.0	crit	2022-01-01T00:01:00Z

4. 将“states”写回 InfluxDB。您可以将数据写入新测量或新存储桶。要将数据写入新测量，请使用 set() 来更新“states”数据中 _measurement 列的值。

   states
       // (Optional) Change the measurement name to write the data to a new measurement
       |> set(key: "_measurement", value: "new-measurement")
       |> to(bucket : "example-bucket") 
   

5. 查找上一次任务运行的最新值，并将其存储在变量“last_state_previous_task”中，

   last_state_previous_task =
       from(bucket: "example-bucket")
           |> range(start: date.sub(d: task.every, from: tasks.lastSuccess(orTime: -task.every))
           |> filter(fn: (r) => r._measurement == "example-measurement")
           |> filter(fn: (r) => r.tagKey == "example-tag-value")
           |> filter(fn: (r) => r._field == "example-field")
           |> last() 
   

   其中 last_state_previous_task 可能看起来像

   _measurement	tagKey1	_field	_value	_level	_time
   example-measurement	example-tag-value	example-field	55.0	crit	2021-12-31T23:59:00Z

6. 联合“states”和“last_state_previous_task”。将此数据存储在变量“unioned_states”中。使用 sort() 确保行按时间排序。

   unioned_states =
       union(tables: [states, last_state_previous_task])
           |> sort(columns: ["_time"], desc: true)
   

   其中 unioned_states 可能看起来像

   _measurement	tagKey1	_field	_value	_level	_time
   example-measurement	example-tag-value	example-field	55.0	crit	2021-12-31T23:59:00Z
   example-measurement	example-tag-value	example-field	30.0	ok	2022-01-01T00:00:00Z
   example-measurement	example-tag-value	example-field	50.0	crit	2022-01-01T00:01:00Z

7. 使用 monitor.stateChangesOnly() 仅返回“unioned_states”中状态已更改的行。将此数据存储在变量“state_changes”中。

   state_changes =
       unioned_states 
           |> monitor.stateChangesOnly()
   

   其中 state_changes 可能看起来像

   _measurement	tagKey1	_field	_value	_level	_time
   example-measurement	example-tag-value	example-field	30.0	ok	2022-01-01T00:00:00Z
   example-measurement	example-tag-value	example-field	50.0	crit	2022-01-01T00:01:00Z

8. 通知跨越最后两个任务的状态更改，以捕获在任务执行之间发生的任何状态更改。

   state_changes =
       data
           |> map(
               fn: (r) =>
                   ({
                       _value:
                           telegram.message(
                               token: telegram_token,
                               channel: telegram_channel_ID,
                               text: "state change at ${r._value} at ${r._time}",
                           ),
                   }),
           )
   

   使用联合数据，以下警报将发送到 Telegram

   • 状态在 2022-01-01T00:00:00Z 时更改为 30.0
   • 状态在 2022-01-01T00:01:00Z 时更改为 50.0