文档文档

SideloadNode

sideload 节点根据来自各种来源的分层数据,向数据点添加字段和标签。

SideloadNode 示例

以下示例展示了如何基于从分层源加载的值,向每个数据点添加字段 cpu_threshold 和标签 foo.order() 属性中的模板列表使用数据点的标签进行评估。文件路径会按顺序检查指定的键,并使用找到的第一个值。

|sideload()
  .source('file:///path/to/dir')
  .order('host/{{.host}}.yml', 'hostgroup/{{.hostgroup}}.yml')
  .field('cpu_threshold', 0.0)
  .tag('foo', 'unknown')

构造函数

链式方法描述
sideload ( )创建一个可以从外部源加载数据的节点

属性方法

Setter描述
field ( f stringv interface{})Field 是要从源加载的字段的名称及其默认值。加载的类型必须与默认值的类型匹配。否则,将记录错误并使用默认值。
order ( order ...string)Order 是指示分层顺序的路径列表。路径相对于源,并且可以具有模板标记,例如 {{.tagname}},这些标记将被数据点的标签值替换。然后按顺序搜索路径以查找键,并使用找到的第一个值。这允许基于标签的层次结构覆盖值。
quiet ( )禁止来自此节点的所有错误日志记录事件。
source ( value string)数据源,例如,file://http://
tag ( t stringv string)Tag 是要从源加载的标签的名称及其默认值。加载的值必须是字符串,否则将记录错误并使用默认值。

链式方法

Alert, Barrier, Bottom, ChangeDetect, Combine, Count, CumulativeSum, Deadman, Default, Delete, Derivative, Difference, Distinct, Ec2Autoscale, Elapsed, Eval, First, Flatten, GroupBy, HoltWinters, HoltWintersWithFit, HttpOut, HttpPost, InfluxDBOut, Join, K8sAutoscale, KapacitorLoopback, Last, Log, Max, Mean, Median, Min, Mode, MovingAverage, Percentile, Sample, Shift, Sideload, Spread, StateCount, StateDuration, Stats, Stddev, Sum, SwarmAutoscale, Top, Trickle, Union, Where, Window


属性

属性方法修改调用节点上的状态。它们不会向管道添加另一个节点,并且始终返回对调用节点的引用。属性方法使用 . 运算符标记。

字段 (Field)

Field 是要从源加载的字段的名称及其默认值。加载的类型必须与默认值的类型匹配。否则,将记录错误并使用默认值。

sideload.field(f string, v interface{})

顺序 (Order)

Order 是指示分层顺序的路径列表。路径相对于源,并且可以具有模板标记,例如 {{.tagname}},这些标记将被数据点的标签值替换。然后按顺序搜索路径以查找键,并使用找到的第一个值。这允许基于标签的层次结构覆盖值。

sideload.order(order ...string)

静默 (Quiet)

禁止来自此节点的所有错误日志记录事件。

sideload.quiet()

源 (Source)

定义数据源。支持以下来源

  • 文件 URI (file://)
  • URL (http://)
  • 端点名称。使用纯字符串(不带 file://http://)以确保源在 Kapacitor 配置的 [[httpost]] 部分中被解释为端点。

源示例

文件源
|sideload()
  .source('file:///path/to/dir')
  .order('host/{{.host}}.yml', 'hostgroup/{{.hostgroup}}.yml')
  .field('cpu_threshold', 0.0)
  .tag('foo', 'unknown')
URL 源
|sideload()
  .source('https://127.0.0.1:5000/threshold/')
  .order('host/{{.host}}.yml', 'hostgroup/{{.hostgroup}}.yml')
  .field('cpu_threshold', 0.0)
  .tag('foo', 'unknown')
端点源
|sideload()
  .source('host1')
  .order('host/{{.host}}.yml', 'hostgroup/{{.hostgroup}}.yml')
  .field('cpu_threshold', 0.0)
  .tag('foo', 'unknown')

当任务启用时,Kapacitor 使用 HTTP GET 请求加载 URL端点 源一次,然后在后续调用 /sideload/reload 端点时加载。

HTTP 源端点应返回一个 JSON 对象,其中每个属性都是 order 语句中指定的键名,其值是一个包含一组键值对的对象。

{
    "host1": {
        "cpu_threshold": 99.9,
        "disable": "false"
    },
    "host2": {
        "cpu_threshold": 85,
        "disable": "true"
    },
    "host3": {
        "cpu_threshold": 56,
        "disable": "false"
    }
}

标签 (Tag)

Tag 是要从源加载的标签的名称及其默认值。加载的值必须是字符串,否则将记录错误并使用默认值。

sideload.tag(t string, v string)

链式方法

链式方法在管道中创建新节点作为调用节点的子节点。它们不修改调用节点。链式方法使用 | 运算符标记。

Alert (警报)

创建一个警报节点,可以触发警报。

sideload|alert()

返回: AlertNode

Barrier (屏障)

创建一个新的 Barrier 节点,该节点定期发出 BarrierMessage。

每个周期持续时间将发出一个 BarrierMessage。

sideload|barrier()

返回: BarrierNode

Bottom (底部)

选择 field 的底部 num 个数据点,并按任何额外的标签或字段排序。

sideload|bottom(num int64, field string, fieldsAndTags ...string)

返回: InfluxQLNode

ChangeDetect (变更检测)

创建一个新节点,仅当与上一个数据点不同时才发出新的数据点。

sideload|changeDetect(field string)

返回: ChangeDetectNode

Combine (合并)

将此节点与其自身合并。数据在时间戳上合并。

sideload|combine(expressions ...ast.LambdaNode)

返回: CombineNode

Count (计数)

计算数据点的数量。

sideload|count(field string)

返回: InfluxQLNode

CumulativeSum (累积和)

计算接收到的每个数据点的累积和。为收集的每个数据点发出一个数据点。

sideload|cumulativeSum(field string)

返回: InfluxQLNode

Deadman (死信)

用于在低吞吐量(又名死信开关)上创建警报的辅助函数。

  • Threshold (阈值): 如果吞吐量降至阈值以下(点/间隔),则触发警报。
  • Interval (间隔): 检查吞吐量的频率。
  • Expressions (表达式): 也要评估的可选表达式列表。对于一天中的时间警报很有用。

示例

    var data = stream
        |from()...
    // Trigger critical alert if the throughput drops below 100 points per 10s and checked every 10s.
    data
        |deadman(100.0, 10s)
    //Do normal processing of data
    data...

以上内容等效于此示例

    var data = stream
        |from()...
    // Trigger critical alert if the throughput drops below 100 points per 10s and checked every 10s.
    data
        |stats(10s)
            .align()
        |derivative('emitted')
            .unit(10s)
            .nonNegative()
        |alert()
            .id('node \'stream0\' in task \'{{ .TaskName }}\'')
            .message('{{ .ID }} is {{ if eq .Level "OK" }}alive{{ else }}dead{{ end }}: {{ index .Fields "emitted" | printf "%0.3f" }} points/10s.')
            .crit(lambda: "emitted" <= 100.0)
    //Do normal processing of data
    data...

idmessage 警报属性可以通过 'deadman' 配置部分全局配置。

由于 AlertNode 是最后一部分,因此可以像往常一样进一步修改它。示例

    var data = stream
        |from()...
    // Trigger critical alert if the throughput drops below 100 points per 10s and checked every 10s.
    data
        |deadman(100.0, 10s)
            .slack()
            .channel('#dead_tasks')
    //Do normal processing of data
    data...

您可以指定其他 lambda 表达式,以进一步约束何时触发死信开关。示例

    var data = stream
        |from()...
    // Trigger critical alert if the throughput drops below 100 points per 10s and checked every 10s.
    // Only trigger the alert if the time of day is between 8am-5pm.
    data
        |deadman(100.0, 10s, lambda: hour("time") >= 8 AND hour("time") <= 17)
    //Do normal processing of data
    data...
sideload|deadman(threshold float64, interval time.Duration, expr ...ast.LambdaNode)

返回: AlertNode

Default (默认值)

创建一个可以为缺失的标签或字段设置默认值的节点。

sideload|default()

返回: DefaultNode

Delete (删除)

创建一个可以删除标签或字段的节点。

sideload|delete()

返回: DeleteNode

Derivative (导数)

创建一个新的节点,该节点计算相邻数据点的导数。

sideload|derivative(field string)

返回: DerivativeNode

Difference (差值)

计算数据点之间的差值,与经过的时间无关。

sideload|difference(field string)

返回: InfluxQLNode

Distinct (去重)

生成仅包含去重数据点的批处理。

sideload|distinct(field string)

返回: InfluxQLNode

Ec2Autoscale

创建一个可以为 ec2 autoscalegroup 触发自动扩展事件的节点。

sideload|ec2Autoscale()

返回: Ec2AutoscaleNode

Elapsed (经过时间)

计算数据点之间经过的时间。

sideload|elapsed(field string, unit time.Duration)

返回: InfluxQLNode

Eval (评估)

创建一个 eval 节点,该节点将评估给定转换函数到每个数据点。可以提供表达式列表,并将按给定的顺序进行评估。结果可用于以后的表达式。

sideload|eval(expressions ...ast.LambdaNode)

返回: EvalNode

First (第一个)

选择第一个数据点。

sideload|first(field string)

返回: InfluxQLNode

Flatten (展平)

将具有相似时间的数据点展平为单个数据点。

sideload|flatten()

返回: FlattenNode

GroupBy (分组)

按一组标签对数据进行分组。

可以传递字面量 * 以按所有维度分组。示例

    |groupBy(*)
sideload|groupBy(tag ...interface{})

返回: GroupByNode

HoltWinters

计算数据集的 Holt-Winters (/influxdb/v1/query_language/functions/#holt-winters) 预测。

sideload|holtWinters(field string, h int64, m int64, interval time.Duration)

返回: InfluxQLNode

HoltWintersWithFit

计算数据集的 Holt-Winters (/influxdb/v1/query_language/functions/#holt-winters) 预测。此方法还输出用于拟合数据的所有数据点以及预测数据。

sideload|holtWintersWithFit(field string, h int64, m int64, interval time.Duration)

返回: InfluxQLNode

HttpOut

创建一个 HTTP 输出节点,该节点缓存其接收到的最新数据。缓存的数据在给定的端点可用。端点是正在运行的任务的 API 端点的相对路径。例如,如果任务端点位于 /kapacitor/v1/tasks/<task_id>,并且端点为 top10,则可以从 /kapacitor/v1/tasks/<task_id>/top10 请求数据。

sideload|httpOut(endpoint string)

返回: HTTPOutNode

HttpPost

创建一个 HTTP Post 节点,该节点将接收到的数据 POST 到提供的 HTTP 端点。HttpPost 期望 0 个或 1 个参数。如果提供 0 个参数,则必须指定端点属性方法。

sideload|httpPost(url ...string)

返回: HTTPPostNode

InfluxDBOut

创建一个 influxdb 输出节点,该节点将传入数据存储到 InfluxDB 中。

sideload|influxDBOut()

返回: InfluxDBOutNode

Join (连接)

将此节点与其他节点连接。数据在时间戳上连接。

sideload|join(others ...Node)

返回: JoinNode

K8sAutoscale

创建一个可以为 kubernetes 集群触发自动扩展事件的节点。

sideload|k8sAutoscale()

返回: K8sAutoscaleNode

KapacitorLoopback

创建一个 kapacitor loopback 节点,该节点会将数据作为流发送回 Kapacitor。

sideload|kapacitorLoopback()

返回: KapacitorLoopbackNode

Last (最后一个)

选择最后一个数据点。

sideload|last(field string)

返回: InfluxQLNode

Log

创建一个记录其接收的所有数据的节点。

sideload|log()

返回: LogNode

Max (最大值)

选择最大数据点。

sideload|max(field string)

返回: InfluxQLNode

Mean (平均值)

计算数据的平均值。

sideload|mean(field string)

返回: InfluxQLNode

Median (中位数)

计算数据的中位数。

注意: 此方法不是选择器。如果您想要中位数点,请使用 .percentile(field, 50.0)

sideload|median(field string)

返回: InfluxQLNode

Min (最小值)

选择最小数据点。

sideload|min(field string)

返回: InfluxQLNode

Mode (众数)

计算数据的众数。

sideload|mode(field string)

返回: InfluxQLNode

MovingAverage (移动平均值)

计算最后 window 个数据点的移动平均值。在窗口填满之前,不会发出任何数据点。

sideload|movingAverage(field string, window int64)

返回: InfluxQLNode

Percentile (百分位数)

选择给定百分位数的数据点。这是一个选择器函数,不执行数据点之间的插值。

sideload|percentile(field string, percentile float64)

返回: InfluxQLNode

Sample (采样)

创建一个新节点,该节点对传入的数据点或批处理进行采样。

将每计数或指定的持续时间发出一个数据点。

sideload|sample(rate interface{})

返回: SampleNode

Shift (偏移)

创建一个新节点,该节点在时间上偏移传入的数据点或批处理。

sideload|shift(shift time.Duration)

返回: ShiftNode

Sideload (侧加载)

创建一个可以从外部源加载数据的节点。

sideload|sideload()

返回: SideloadNode

Spread (范围)

计算 minmax 数据点之间的差值。

sideload|spread(field string)

返回: InfluxQLNode

StateCount (状态计数)

创建一个节点,该节点跟踪给定状态下连续数据点的数量。

sideload|stateCount(expression ast.LambdaNode)

返回: StateCountNode

StateDuration (状态持续时间)

创建一个节点,该节点跟踪给定状态下的持续时间。

sideload|stateDuration(expression ast.LambdaNode)

返回: StateDurationNode

Stats (统计信息)

创建一个新的数据流,其中包含节点的内部统计信息。间隔表示基于实时发出统计信息的频率。这意味着间隔时间独立于源节点正在接收的数据点的时间。

sideload|stats(interval time.Duration)

返回: StatsNode

Stddev (标准差)

计算标准差。

sideload|stddev(field string)

返回: InfluxQLNode

Sum (总和)

计算所有值的总和。

sideload|sum(field string)

返回: InfluxQLNode

SwarmAutoscale

创建一个可以为 Docker swarm 集群触发自动扩展事件的节点。

sideload|swarmAutoscale()

返回: SwarmAutoscaleNode

Top (顶部)

选择 field 的顶部 num 个数据点,并按任何额外的标签或字段排序。

sideload|top(num int64, field string, fieldsAndTags ...string)

返回: InfluxQLNode

Trickle (涓流)

创建一个新节点,该节点将批处理数据转换为流数据。

sideload|trickle()

返回: TrickleNode

Union (联合)

执行此节点和所有其他给定节点的联合。

sideload|union(node ...Node)

返回: UnionNode

Where (条件)

创建一个新节点,该节点通过给定的表达式过滤数据流。

sideload|where(expression ast.LambdaNode)

返回: WhereNode

Window (窗口)

创建一个新节点,该节点按时间窗口化数据流。

注意:Window 只能应用于流边缘。

sideload|window()

返回: WindowNode


此页是否对您有帮助?

感谢您的反馈!


Flux 的未来

Flux 即将进入维护模式。您可以继续像当前一样使用它,而无需对您的代码进行任何更改。

阅读更多

InfluxDB 3 开源版现已公开发布 Alpha 版

InfluxDB 3 开源版现已可用于 Alpha 测试,根据 MIT 或 Apache 2 许可获得许可。

我们正在发布两个产品作为 Alpha 版本的一部分。

InfluxDB 3 Core 是我们新的开源产品。它是一个用于时间序列和事件数据的最新数据引擎。InfluxDB 3 Enterprise 是一个商业版本,它建立在 Core 的基础上,增加了历史查询功能、读取副本、高可用性、可扩展性和细粒度的安全性。

有关如何开始使用的更多信息,请查看