将 Flux 任务与 Kapacitor 结合使用

1. 在 InfluxDB 中设置 Flux 任务数据库
2. 配置 Kapacitor Flux 任务
3. 创建 Flux 任务

在 InfluxDB 中设置 Flux 任务数据库

(可选，但建议配置)

当 Kapacitor 执行 Flux 任务时，它可以将有关任务执行（run）的信息存储在 InfluxDB 数据库中。要存储此数据，请执行以下操作：

1. 在 InfluxDB 中创建一个新数据库，用于存储 Flux 任务的运行记录和日志数据。

   CREATE DATABASE kapacitorfluxtasks

2. 为防止磁盘上积累大量的 Kapacitor Flux 任务日志数据，请更新默认的 autogen 保留策略（RP）并设置有限的保留期，或者创建一个具有有限保留期的新保留策略。

-- Syntax
ALTER RETENTION POLICY <rp-name> ON <db-name> DURATION <new-retention-duration>

-- Example
ALTER RETENTION POLICY autogen ON kapacitorfluxtasks DURATION 3d

-- Syntax
CREATE RETENTION POLICY <rp-name> on <db-name> DURATION <retention-duration>

-- Example 
CREATE RETENTION POLICY threedays on kapacitorfluxtasks DURATION 3d

配置 Kapacitor Flux 任务

在 kapacitor.conf 的 [fluxtask] 下更新或添加以下设置：

• enabled: true
• task-run-influxdb: 用于存储 Flux 任务数据的 InfluxDB 配置（位于 kapacitor.conf 中）的名称。若要禁用 Flux 任务日志记录，请设置为 "none"。
• task-run-bucket: 用于存储 Flux 任务数据和日志的 InfluxDB 数据库。建议将其留空。默认情况下，数据会写入 kapacitor_fluxtask_logs 数据库。要指定另一个数据库来写入任务日志数据，请使用 "db-name" 命名规范（不支持包含保留策略的 "db-name/rp" 格式）。如果指定的数据库在 InfluxDB 中尚不存在，Kapacitor 将尝试创建该数据库。如果开启了身份验证，则需要 CREATE DATABASE 的权限。更多信息，请参阅 InfluxDB 中的身份验证和授权。
• 提供以下任一选项：
  • task-run-org: 留为空字符串 ("")
  • task-run-orgid: 留为空字符串 ("")
• task-run-measurement: 用于存储任务运行和日志数据的 InfluxDB measurement。默认值为 "runs"。

Kapacitor Flux 任务配置示例

# ...

[fluxtask]
  enabled = true
  task-run-influxdb = "default"
  task-run-bucket = "kapacitor_fluxtask_logs"
  task-run-org = ""
  task-run-orgid = ""
  task-run-measurement = "runs"

# ...

有关 Kapacitor [fluxtask] 配置选项的更多信息，请参阅 配置 Kapacitor。

创建 Flux 任务

1. 创建一个 Flux 任务脚本。在脚本中包含 任务选项 以配置 Kapacitor Flux 任务。有关编写 Flux 任务的更多信息，请参阅：

   • Flux 任务入门
   • 常见数据处理任务

   提供 InfluxDB 连接凭据

   from() 和 to() 函数需要你的 InfluxDB 主机地址 (host) 和 令牌 (token)。

   • host: InfluxDB URL。
   • token: 如果 启用了 InfluxDB 身份验证，请使用 用户名:密码 语法。否则，令牌请使用空字符串 ("")。

   Bucket 名称语法

   使用 Flux 查询或写入 InfluxDB 1.x 时，请使用 数据库名称/保留策略名称 的模式来指定你的 bucket。

   example-task.flux

   option task = {
     name: "example-task-name",
     every: 1h,
     offset: 10m
   }
   
   host = "https://:8086"
   token = ""
   
   from(bucket: "example-db/example-rp", host: host, token: token)
     |> range(start: -task.every)
     |> filter(fn: (r) => r._measurement == "example-measurement")
     |> aggregateWindow(every: 10m, fn: mean)
     |> to(bucket: "example-db/example-rp-downsampled", host: host, token: token)

2. 使用 kapacitor flux task create 命令将你的 Flux 脚本添加为 Kapacitor Flux 任务。

   kapacitor flux task create --file /path/to/example-task.flux

有关创建 Kapacitor Flux 任务的更多详细信息，请参阅 创建 Kapacitor Flux 任务。

1. 为 InfluxDB Cloud 或 2.x 设置 Kapacitor
2. 为 InfluxDB Cloud 或 2.x 配置 Kapacitor Flux 任务
3. 创建 Flux 任务

为 InfluxDB Cloud 或 2.x 设置 Kapacitor

配置 Kapacitor 以连接到 InfluxDB Cloud 或 InfluxDB OSS 2.x。详细说明请参阅以下内容：

• 在 InfluxDB Cloud 中使用 Kapacitor
• 在 InfluxDB 2.x OSS 中使用 Kapacitor

为 InfluxDB Cloud 或 2.x 配置 Kapacitor Flux 任务

在 kapacitor.conf 的 [fluxtask] 下更新或添加以下设置：

• enabled: true
• task-run-influxdb: 用于存储 Flux 任务数据的 InfluxDB 配置（位于 kapacitor.conf 中）的名称。若要禁用 Flux 任务日志记录，请设置为 "none"。
• task-run-bucket: 用于存储 Flux 任务数据和日志的 InfluxDB bucket。建议将其留空。默认情况下，数据写入 kapacitor_fluxtask_logs bucket。要指定另一个用于写入任务日志数据的 bucket，请使用 _tasks 系统 bucket 或 创建一个新 bucket。如果指定的 bucket 在 InfluxDB 中尚不存在，Kapacitor 将尝试通过 POST /api/v2/buckets 创建它，在这种情况下，你的 API 令牌必须拥有在 InfluxDB 中创建 bucket 的权限。更多信息，请参阅 管理 API 令牌。
• 提供以下任一选项：
  • task-run-org: InfluxDB 组织名称。
  • task-run-orgid: InfluxDB 组织 ID。
• task-run-measurement: 用于存储任务运行和日志数据的 InfluxDB measurement。默认值为 "runs"。

# ...

[fluxtask]
  enabled = true
  task-run-influxdb = "InfluxDB"
  task-run-bucket = "kapacitor_fluxtask_logs"
  task-run-org = "example-org"
  task-run-measurement = "runs"

# ...

创建 Flux 任务

1. 创建一个 Flux 任务脚本。在脚本中包含 任务选项 以配置 Kapacitor Flux 任务。有关编写 Flux 任务的更多信息，请参阅：

   • Flux 任务入门
   • 常见数据处理任务

   提供 InfluxDB 连接凭据

   from() 和 to() 函数需要你的 InfluxDB 主机地址 和 令牌。

   • host: InfluxDB URL。
   • token: 如果 启用了 InfluxDB 身份验证，请使用 用户名:密码 语法。否则，令牌请使用空字符串 ("")。

   example-task.flux

   option task = {
     name: "example-task-name",
     every: 1h,
     offset: 10m
   }
   
   host = "https://:8086"
   token = ""
   
   from(bucket: "example-bucket", host: host, token: token)
     |> range(start: -task.every)
     |> filter(fn: (r) => r._measurement == "example-measurement")
     |> aggregateWindow(every: 10m, fn: mean)
     |> to(bucket: "example-bucket-downsampled", host: host, token: token)

2. 使用 kapacitor flux task create 命令将你的 Flux 脚本添加为 Kapacitor Flux 任务。

   kapacitor flux task create --file /path/to/example-task.flux

有关创建 Kapacitor Flux 任务的更多详细信息，请参阅 创建 Kapacitor Flux 任务。