TL;DR InfluxDB 技术提示 — 使用和理解 InfluxDB Cloud 使用量模板

作者：Anais Dotis-Georgiou / 用例, 开发者, 产品
2021 年 6 月 29 日

假设您正在使用 InfluxDB Cloud，并且正在向您的帐户写入数百万个指标。无论您是在 InfluxDB 之上构建 IoT 应用程序，还是使用 InfluxDB 监控您的生产环境，您的时间序列操作最终都运行顺畅。您希望保持这种状态。您可能是 免费计划 Cloud 用户 或 按使用量付费计划用户，但无论哪种方式，您都需要了解实例大小，以便管理资源和成本。使用 InfluxDB Cloud 使用量模板 来了解您的使用情况，并在达到限制之前采取积极措施。

什么是 InfluxDB Cloud 使用量模板？

如果您是 InfluxData 的新手，InfluxDB 模板 是仪表板、任务、警报、Telegraf 配置等的预配置且可共享的集合。您可以通过以下两种方式应用它们：1) 使用 CLI 和 influx apply 命令，或 2) 从 社区模板 复制感兴趣的模板的 URL，并将其粘贴到 UI 中。今天，我们将应用 InfluxDB Cloud 使用量模板到我们的 InfluxDB Cloud 帐户，以监控我们的使用情况和速率限制事件。当用户超出其 Cloud 帐户的使用限制时，会发生速率限制事件。使用以下命令通过 CLI 应用 InfluxDB Cloud 使用量模板

influx apply --file https://raw.githubusercontent.com/influxdata/community-templates/master/usage_dashboard/usage_dashboard.yml

或者，您可以使用 UI 来应用模板。导航到“设置”页面上的 模板 选项卡，并将要应用的模板的 YAML 或 JSON 链接粘贴到此处。

InfluxDB Cloud 使用量模板包含以下资源

• 1 个 仪表板：使用量仪表板。此仪表板使您能够跟踪 InfluxDB Cloud 数据使用情况和限制事件。
• 1 个 任务：基数限制警报（使用量仪表板），每小时运行一次。此任务负责确定您的基数是否已超过与您的 Cloud 帐户关联的基数限制。
• 1 个 标签：usage_dashboard。此标签可帮助您在 UI 和通过 API 更轻松地找到属于此模板的资源。

使用量仪表板说明

InfluxDB Cloud 使用量模板包含使用量仪表板，该仪表板使您可以了解数据传入、查询计数、存储、数据传出和速率限制事件。

仪表板中的单元格显示以下信息

• 数据传入（图表）可视化通过 /write 端点 写入 InfluxDB Org 的总字节数。
• 查询计数（图表）可视化执行的查询数量。
• 存储（按存储桶）可视化每个存储桶以及所有存储桶的总字节数。
• 数据传出（图表）可视化通过 /query 端点 从您的 InfluxDB 组织查询的总字节数。
• 速率限制事件可视化达到限制的次数
• 速率限制事件在右下角的单元格中按事件类型拆分，以显示写入限制、查询限制和基数限制事件的数量以及组织的基数限制总数。

所有这些可视化都自动将数据聚合到 1 小时分辨率。从本质上讲，此仪表板补充了 InfluxDB UI 中“使用量”页面中存在的数据，以便您更深入地了解您的使用情况。

<figcaption> InfluxDB UI 中的“使用量”页面包含有关您的使用情况的常规信息。</figcaption>

理解使用量仪表板背后的 Flux

为了充分理解 InfluxDB Cloud 使用量模板，您必须理解使用量仪表板背后的 Flux。让我们看一下生成第一个单元格中“数据传入（图表）”可视化的查询。

import "math"
import "experimental/usage"
 
usage.from(
start: v.timeRangeStart,
stop: v.timeRangeStop,
)
|> filter(fn: (r) =>
r._measurement == "http_request"
and (r.endpoint == "/api/v2/write" or r.endpoint == "/write")
and r._field == "req_bytes"
)
|> group()
|> keep(columns: ["_value", "_field", "_time"])
|> fill(column: "_value", value: 0)
|> map(fn: (r) =>
({r with
write_mb: math.round(x: float(v: r._value) / 10000.0) / 100.0
}))

首先，我们导入所有相关包以执行 Flux 查询，该查询生成我们所需的可视化效果。Flux math 包提供基本的数学函数和常量。Flux experimental/usage 包提供用于收集与您的 InfluxDB Cloud 帐户相关的使用情况和使用量限制数据的功能。InfluxDB Cloud 使用量模板中的所有单元格、仪表板和任务都使用 Flux experimental/usage 包。usage.from() 函数返回使用情况数据。我们从任何写入中筛选出“req_bytes”字段的使用情况数据。然后，我们对写入数据进行分组，用 0 填充任何空行，并计算兆字节数，并将该值存储在新列“write_mb”中。

探索 experimental/usage Flux 包

experimental/usage 包包含两个函数：usage.from() 和 usage.limits()。

usage.from() 函数映射到 InfluxDB v2 API 中的 org/{orgID}/usage 端点。usage.from() 函数返回您的 InfluxDB 组织的使用情况数据。usage.from() 函数要求您指定 start 和 stop 参数。默认情况下，该函数返回聚合的使用情况数据作为每小时总和。此默认的降采样使您可以有效地查询长时间段的使用情况数据。如果您想查看原始高分辨率数据，请在短时间范围（几个小时）内查询您的数据，并将 raw:true 设置为 true。默认情况下， usage.from() 从您当前使用的 InfluxDB Org 返回使用情况数据。要查询外部 InfluxDB Cloud Org，请提供 host、orgID 和 token 参数。usage.from() 返回具有以下架构的使用情况数据

• _time：使用情况事件的时间（raw:true）或降采样使用情况数据（raw:false）。
• _bucket_id：每个存储桶特定用量的存储桶 ID，例如 storage_usage_bucket_bytes 测量 — 标签。
• _org_id：从中查询使用情况数据的组织 ID — 标签。
• _endpoint：从中收集使用情况数据的 InfluxDB v2 API 端点 — 标签
• _status：HTTP 状态代码 — 标签。
• _measurement：测量值。
  • http_request
  • storage_usage_buckets_bytes
  • query _count
  • events
• _field：字段键
  • resp_bytes：一个响应中的字节数或 raw:true。在降采样数据中，raw:false 是每个端点每小时的总和。
  • gauge：_time 指示的时间点存储桶 ID 标签指示的存储桶中的字节数。如果您的存储桶具有无限保留策略，那么从您开始将数据写入存储桶时起，您的 req_bytes 总和将等于您的 gauge 值。
  • req_bytes：raw:true 的一个请求中的字节数。在降采样数据中，raw:false 是每个端点每小时的总和。
  • event_type_limited_cardinality：基数限制事件的数量。
  • event_type_limited_query：查询限制事件的数量。
  • event_type_limited_write：写入限制事件的数量。

usage.limits() 函数映射到 InfluxDB v2 API 中的 org/{orgID}/limits 端点。usage.limits() 返回记录，特别是与您的 InfluxDB Cloud Org 关联的使用量限制的 json 对象。要返回 VS Code 或 InfluxDB UI 中的记录，我建议使用 array.from 函数，如下所示

import "experimental/usage"
import "array"
limits = usage.limits()
array.from(rows: [{
    orgID: limits.orgID,
    wrte_rate: limits.rate.writeKBs,
    query_rate: limits.rate.readKBs,
    bucket: limits.bucket.maxBuckets,
    task: limits.maxTasks,
    dashboard: limits.dashboard.maxDashboards,
    check: limits.check.maxChecks,
    notificationRule: limits.notificationRule.maxNotifications,
}])

上面的 Flux 本质上是将 json 输出转换为 Annotated CSV，使用了 array.from 函数。以下是 usage.limits() 返回的 json 对象的示例，可帮助您了解嵌套 json 键的来源。

{
  orgID: "123",
  rate: {
    readKBs: 1000,
    concurrentReadRequests: 0,
    writeKBs: 17,
    concurrentWriteRequests: 0,
    cardinality: 10000
  },
  bucket: {
    maxBuckets: 2,
    maxRetentionDuration: 2592000000000000
  },
  task: {
    maxTasks: 5
  },
  dashboard: {
    maxDashboards: 5
  },
  check: {
    maxChecks: 2
  },
  notificationRule: {
    maxNotifications: 2,
    blockedNotificationRules: "comma, delimited, list"
  },
  notificationEndpoint: {
    blockedNotificationEndpoints: "comma, delimited, list"
  }
}

使用基数限制警报任务

基数限制警报任务负责确定您何时达到或超过基数限制，并将警报发送到 Slack 端点。为了使此任务正常运行并将警报发送到您的 Slack，您必须编辑该任务并提供 webhook。您可以直接通过 UI 编辑任务，也可以使用 CLI 和 Visual Studio Code 的 Flux 扩展。对于这一行编辑，我建议使用 UI。

<figcaption> 导航到“任务”页面，单击基数限制警报任务进行编辑，编辑第 7 行并提供您的 Slack webhook，然后单击“保存”以使基数限制警报任务正常运行。</figcaption>

使用 CLI、Visual Studio Code 和 Flux 扩展进行 Flux 开发

InfluxDB UI 是许多初级和中级 InfluxDB 用户的一站式商店。它有助于简单的仪表板、笔记本、警报和任务创建。我通常建议用户从 InfluxDB UI 开始他们的 InfluxDB 之旅，因为它提供了许多用户可以利用的无代码或低代码解决方案来完成有意义的警报和可视化工作。但是，我建议使用 VS Code 和 Flux 扩展进行复杂的 Flux 开发，原因如下：

1. 许多开发人员更熟悉使用命令行和代码编辑器。
2. 它为用户提供了保存脚本并轻松查看带注释的 CSV 输出以及轻松查询不同组织的能力。

因此，我们将借此机会重点介绍在 VS Code 中使用 CLI 编辑和更新任务的工作流程。

首先，使用 influx task list 命令列出您的任务以获取您的任务 ID，如下所示

./influx task list

接下来，使用 jq（一个轻量级命令行 json 解析器）漂亮地打印任务的 Flux，如下所示

./influx task list -i 0786fb9aa2bab000 --json | jq -r '.[].flux

现在将 Flux 复制并粘贴到 VS Code 中，并使用 Flux 扩展使用您的 Slack webhook 更新 slackWebhook 变量。从左侧面板中选择您要使用的 InfluxDB 配置和组织，右键单击以运行查询并在左侧查看输出。您会注意到输出是一个单列表，值为 0。我们将在下一节“理解基数限制警报任务背后的 Flux”中讨论此输出。

<figcaption> 在 VS Code 中使用 Flux 扩展编辑任务的示例。轻松切换并在左侧面板的现有配置中选择您要使用的 InfluxDB 组织。相应的输出显示在左侧的选项卡中。我为我的个人 InfluxDB Cloud 帐户和组织进行了配置。我也是我的同事 John 的组织的用户。</figcaption>

在针对我的同事 John 的 InfluxDB 组织（我是其成员）运行查询后，我收到了 Slack 警报。我还更改了任务的查询持续时间，以查询过去 30 天的所有基数限制事件（第 17 行），因为我们在过去一个月中调用了数百个基数事件。选择此长查询时间范围只是为了检查任务是否成功向 Slack 发送警报。

最后，既然我已经确信我的任务按预期工作，我将 start 参数的值返回到 -1h，然后再使用 influx update task 命令更新任务，如下所示

influx task update --file /path/to/example-task.flux

理解基数限制警报任务背后的 Flux 并绕过 InfluxDB 检查和通知系统

现在我们了解了与复杂 Flux 开发相关的最佳实践，因此让我们花一点时间回顾一下基数限制警报任务背后的 Flux。与任何 Flux 任务或脚本一样，首先导入所有相关包。接下来，指定任务选项，包括名称、every 和 offset 参数。every 间隔指定查询开始时间，而 offset 间隔指定任务应何时运行。在您的任务选项中包含 offset 可确保您捕获任何延迟传入的数据，并帮助您避免读/写冲突。

import "array"
import "experimental/usage"
import "slack"
option task = { name: "Cardinality Limit Alert (usage dashboard)", every: 1h0m0s, offset: 5m0s }

接下来，指定您的 slackWebhook URL 和 slackChannel。如果您要发送直接消息，请提供空引号作为您的 slackChannel。

slackWebhook = "https://hooks.slack.com/services/####/####/####"
slackChannel = ""

现在定义一个自定义函数 alert()，以便在基数事件数 eventValue 超过您的阈值时触发 Slack 消息。我们使用条件逻辑和 slack.message() 函数来实现此目的。

alert = (eventValue, threshold) =>
(if eventValue >= threshold then slack.message(
url: slackWebhook,
channel: slackChannel,
text: "A Cardinality Event has occured! The number of cardinality events = \"${string(v: eventValue)}\".",
color: "warning",
) else 0)

现在我们希望能够使用 usage.from() 查询我们的基数事件，并简单地将此函数应用于该查询的输出，以便在我们的事件超过阈值时触发 slack 警报。但是，我们必须解决两个问题，这稍微复杂化了我们的 Flux 脚本。

1. 首先，我们必须处理查询基数事件未返回任何结果的情况。为了初始化我们的查询以返回一个值为 0 的表，应用了 array.from() 函数。
2. 其次，flux 查询的输出是带注释的 CSV 格式的表流，而不是标量，标量是我们 alert() 函数的预期输入。为了从带注释的 CSV 输出返回标量，应用了 findRecord() 函数。

我们使用 array.from() 函数将一个包含一列 (_value) 和一个值为 0 的行的表存储到变量 data_0 中。我们通过应用适当的过滤器，对事件求和并存储输出在变量 data 中，使用 usage.from() 查询我们的基数事件数据。接下来，我们在两个表流 data_0 和 data 之间执行联合，以确保表输出和随后成功应用我们的 alert() 函数。group() 和 sum() 函数应用于联合，以将联合的输出减少到单列、单行表。请注意，sum() 函数只是将 0 添加到基数事件值。现在始终返回表，应用 findRecord() 函数来检索标量值，我们将输入到我们的 alert() 函数中。

data_0 = array.from(rows: [{_value: 0}])
data = usage.from(start: -1h, stop: now())
|> filter(fn: (r) =>
(r._measurement == "events" and r._field == "event_type_limited_cardinality" and exists r._value))
|> sum()
events = union(tables: [data_0, data])
|> group()
|> sum()
|> findRecord(fn: (key) =>
(true), idx: 0)
eventTotal = events._value
alert(eventValue: eventTotal, threshold: 1)

最后，任务需要返回表流。如果任务没有输出表流，即使它成功查询了我们的使用数据并执行了警报逻辑，也会失败。为了满足任务的表输出要求，我们输出了 data_0，即我们的初始化表。

data_0
|> yield(name: "ignore")f

这种特殊性引入了此任务最重要的方面，即它完全绕过了 InfluxDB 检查和通知系统。任务需要表输出的原因是，通常检查任务和通知任务使用 monitor.check() 和 monitor.notify() 函数，这些函数输出并将关于您的检查和通知的元数据写入系统存储桶“_monitoring”。这些函数及其元数据输出提供了对您的检查和通知任务的可见性，以及用于所有检查和通知数据的统一组织系统。然而，正如我们的基数限制警报（使用仪表板）任务所演示的那样，用户并非必须订阅 InfluxDB 检查和通知系统，并且如果他们认为合适，可以选择完全绕过它。要了解更多信息，请阅读 InfluxDB 的检查和通知系统 博客或查看下面的“延伸阅读”部分。

使用您自己的任务警报来补充使用模板

虽然 InfluxDB Cloud 使用模板附带了基数限制警报任务，但您不仅限于此任务。我建议将该任务视为创建您自己的关于 InfluxDB Cloud 使用情况的任务和警报的跳板。例如，您可能决定希望将警报发送到 Slack 以外的端点。您可以编辑 Flux 代码以使用任何 <package>.message() 函数将警报发送到您选择的端点。请查看此问题 #2405 以获取 Flux 通知端点软件包的完整列表。

或者，您可能决定创建一个任务，以便在您达到写入限制事件时向您发出警报。该任务看起来与我们上面的基数限制事件任务相同，但只需对字段类型进行一次修改

data = usage.from(start: -1h, stop: now())
|> filter(fn: (r) =>
(r._measurement == "events" and r._field == "event_type_limited_write" and exists r._value))

延伸阅读

虽然这篇文章旨在全面概述 InfluxDB Cloud 使用模板及其驱动的 Flux，但以下资源也可能让您感兴趣

1. TL;DR InfluxDB 技术技巧 – 监控任务并查找失控基数的来源：这篇文章描述了如何使用操作监控模板来帮助您在遇到基数事件时识别基数的来源。我鼓励您尝试 操作监控模板 以及 InfluxDB Cloud 使用模板，以全面了解 InfluxDB Cloud 中可用的检测选项。
2. InfluxDB 的检查和通知系统：这篇文章描述了 InfluxDB 的检查和通知系统如何工作，以及如何创建自定义检查和通知任务。
3. TL;DR InfluxDB 技术技巧 – 如何解释带注释的 CSV：这篇文章描述了如何解释 带注释的 CSV，即 InfluxDB 的 Flux 查询结果格式。如果您是 InfluxDB 和 Flux 的新手，我建议您花时间了解带注释的 CSV 输出，然后再尝试编写复杂的自定义检查任务。
4. TL;DR InfluxDB 技术技巧 – 如何使用 Flux 和 InfluxDB 提取值、可视化标量和执行自定义聚合：这篇文章描述了如何使用 findRecord() 提取标量，以及一种使您能够在 UI 中可视化它们的技巧。
5. Flux Telegram 软件包：此视频描述了如何配置 Telegram 机器人，以及如何使用 Flux Telegram 软件包 向 Telegram 发送通知。
6. TL;DR InfluxDB 技术技巧：使用 InfluxDB 配置 Slack 通知：这篇文章描述了如何配置 Slack webhook 以使用 Flux 向 Slack 发送通知。

关于 InfluxDB Cloud 使用模板的最终想法

我希望这篇 InfluxDB 技术技巧文章能够启发您利用 InfluxDB Cloud 使用模板。最后，以下问题可能与您和您的 Flux 开发相关。我鼓励您发表评论并分享您的想法，以便 InfluxData 能够更好地为您提供支持

1. InfluxDB UI #1664：支持可视化除带注释的 CSV 以外的数据格式。
2. Flux Extension for VS Code #234：支持可视化除带注释的 CSV 以外的数据格式。
3. Flux #3726：绕过为任务调用流函数时的错误。

如果您正在使用社区模板或 Flux 并且需要帮助，请在我们的 社区站点 或 Slack 频道中寻求帮助。如果您正在 InfluxDB 之上开发一个很酷的 IoT 应用程序，我们很乐意了解它，因此请务必 分享您的故事！此外，请在评论区分享您的想法、疑虑或问题。我们很乐意获得您的反馈并帮助您解决遇到的任何问题！