Documentation

EC2AutoscaleNode

ec2Autoscale 节点为 AWS Autoscaling 组上的组触发自动扩展事件。该节点还输出触发事件的点。

示例

// Target 80% cpu per ec2 instance
var target = 80.0
var min = 1
var max = 10
var period = 5m
var every = period
stream
  |from()
    .measurement('cpu')
    .groupBy('host_name','group_name')
    .where(lambda: "cpu" == 'cpu-total')
  |eval(lambda: 100.0 - "usage_idle")
    .as('usage_percent')
  |window()
    .period(period)
    .every(every)
  |mean('usage_percent')
    .as('mean_cpu')
  |groupBy('group_name')
  |sum('mean_cpu')
    .as('total_cpu')
  |ec2Autoscale()
    // Get the group name of the VM(EC2 instance) from "group_name" tag.
    .groupNameTag('group_name')
    .min(min)
    .max(max)
    // Set the desired number of replicas based on target.
    .replicas(lambda: int(ceil("total_cpu" / target)))
  |influxDBOut()
    .database('deployments')
    .measurement('scale_events')
    .precision('s')

上面的示例计算了按 host_namenamegroup_name 分组的 CPU usage_percent 的平均值。平均 CPU 使用率的总和计算为 total_cpu。使用过去一段时间内的 total_cpu,根据 CPU 的目标百分比使用率计算所需的副本数。

如果所需的副本数已更改,Kapacitor 会调用适当的 API 调用 AWS 自动扩展组以更新副本的规范。

每当 Ec2Autoscale 节点更改副本计数时,它都会发出一个点。该点使用 groupName 标记组名。此外,分组依据标记将保留在发出的点上。该点包含两个字段,表示副本的更改:oldnew

可用统计信息

  • increase_events:副本计数增加的次数。
  • decrease_events:副本计数减少的次数。
  • cooldown_drops:由于冷却计时器而丢弃事件的次数。
  • errors:遇到的错误数,通常与与 AWS 自动扩展 API 通信有关。

构造函数

链式方法描述
ec2Autoscale ( )创建一个可以为 EC2 Autoscale 组触发自动扩展事件的节点。

属性方法

设置器描述
cluster ( value string)Cluster 是要使用的 EC2 Autoscale 组的 ID。集群的 ID 在 kapacitor 配置中指定。
currentField ( value string)CurrentField 是将当前副本计数设置为 int 的字段名称。如果为空,则不会设置任何字段。用于计算当前状态的增量很有用。
decreaseCooldown ( value time.Duration)每个资源在每个 DecreaseCooldown 间隔内只能触发一个减少事件。
groupName ( value string)GroupName 是要自动扩展的自动扩展组的名称。
groupNameTag ( value string)GroupName 是一个标记的名称,其中包含要自动扩展的自动扩展组的名称。
increaseCooldown ( value time.Duration)每个资源在每个 IncreaseCooldown 间隔内只能触发一个增加事件。
max ( value int64)要设置的最大比例因子。如果为 0,则没有上限。默认值:0,即无限制。
min ( value int64)要设置的最小比例因子。默认值:1
outputGroupNameTag ( value string)OutputGroupName 是一个标记的名称,组名称将写入该标记以用于输出自动扩展事件。如果 GroupNameTag 不为空,则默认为 GroupNameTag 的值。
quiet ( )禁止来自此节点的所有错误日志记录事件。
replicas ( value ast.LambdaNode)Replicas 是一个 lambda 表达式,应评估为资源的所需副本数。

链式方法

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, Mean, Median, Mode, MovingAverage, Percentile, Sample, Shift, Sideload, Spread, StateCount, StateDuration, Stats, Stddev, Sum, SwarmAutoscale, Top, Trickle, Union, Where, Window


属性

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

Cluster

Cluster 是要使用的 ec2 自动扩展组的 ID。集群的 ID 在 kapacitor 配置中指定。

ec2Autoscale.cluster(value string)

CurrentField

CurrentField 是将当前副本计数设置为 int 的字段名称。如果为空,则不会设置任何字段。用于计算当前状态的增量很有用。

示例

    |ec2Autoscale()
        .currentField('replicas')
        // Increase the replicas by 1 if the qps is over the threshold
        .replicas(lambda: if("qps" > threshold, "replicas" + 1, "replicas"))
ec2Autoscale.currentField(value string)

DecreaseCooldown

每个资源在每个 DecreaseCooldown 间隔内只能触发一个减少事件。

ec2Autoscale.decreaseCooldown(value time.Duration)

GroupName

GroupName 是要自动扩展的自动扩展组的名称。

ec2Autoscale.groupName(value string)

GroupNameTag

GroupName 是一个标记的名称,其中包含要自动扩展的自动扩展组的名称。

ec2Autoscale.groupNameTag(value string)

IncreaseCooldown

每个资源在每个 IncreaseCooldown 间隔内只能触发一个增加事件。

ec2Autoscale.increaseCooldown(value time.Duration)

Max

要设置的最大比例因子。如果为 0,则没有上限。默认值:0,即无限制。

ec2Autoscale.max(value int64)

Min

要设置的最小比例因子。默认值:1

ec2Autoscale.min(value int64)

OutputGroupNameTag

OutputGroupName 是一个标记的名称,组名称将写入该标记以用于输出自动扩展事件。如果 GroupNameTag 不为空,则默认为 GroupNameTag 的值。

ec2Autoscale.outputGroupNameTag(value string)

Quiet

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

ec2Autoscale.quiet()

Replicas

Replicas 是一个 lambda 表达式,应评估为资源的所需副本数。

ec2Autoscale.replicas(value ast.LambdaNode)

链式方法

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

Alert

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

ec2Autoscale|alert()

返回:AlertNode

Barrier

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

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

ec2Autoscale|barrier()

返回:BarrierNode

Bottom

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

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

返回:InfluxQLNode

ChangeDetect

创建一个新节点,该节点仅在与前一个点不同时才发出新点。

ec2Autoscale|changeDetect(field string)

返回:ChangeDetectNode

Combine

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

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

返回:CombineNode

Count

计算点数。

ec2Autoscale|count(field string)

返回:InfluxQLNode

CumulativeSum

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

ec2Autoscale|cumulativeSum(field string)

返回:InfluxQLNode

Deadman

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

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

示例

    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...
ec2Autoscale|deadman(threshold float64, interval time.Duration, expr ...ast.LambdaNode)

返回:AlertNode

Default

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

ec2Autoscale|default()

返回:DefaultNode

Delete

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

ec2Autoscale|delete()

返回:DeleteNode

Derivative

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

ec2Autoscale|derivative(field string)

返回:DerivativeNode

Difference

计算点之间的差异,与经过的时间无关。

ec2Autoscale|difference(field string)

返回:InfluxQLNode

Distinct

生成仅包含不同点的批次。

ec2Autoscale|distinct(field string)

返回:InfluxQLNode

Ec2Autoscale

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

ec2Autoscale|ec2Autoscale()

返回:Ec2AutoscaleNode

Elapsed

计算点之间经过的时间。

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

返回:InfluxQLNode

Eval

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

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

返回:EvalNode

First

选择第一个点。

ec2Autoscale|first(field string)

返回:InfluxQLNode

Flatten

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

ec2Autoscale|flatten()

返回:FlattenNode

GroupBy

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

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

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

返回:GroupByNode

HoltWinters

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

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

返回:InfluxQLNode

HoltWintersWithFit

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

ec2Autoscale|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 请求数据。

ec2Autoscale|httpOut(endpoint string)

返回:HTTPOutNode

HttpPost

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

ec2Autoscale|httpPost(url ...string)

返回:HTTPPostNode

InfluxDBOut

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

ec2Autoscale|influxDBOut()

返回:InfluxDBOutNode

Join

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

ec2Autoscale|join(others ...Node)

返回:JoinNode

K8sAutoscale

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

ec2Autoscale|k8sAutoscale()

返回:K8sAutoscaleNode

KapacitorLoopback

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

ec2Autoscale|kapacitorLoopback()

返回:KapacitorLoopbackNode

Last

选择最后一个点。

ec2Autoscale|last(field string)

返回:InfluxQLNode

Log

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

ec2Autoscale|log()

返回:LogNode

Mean

计算数据的平均值。

ec2Autoscale|mean(field string)

返回:InfluxQLNode

Median

计算数据的中位数。

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

ec2Autoscale|median(field string)

返回:InfluxQLNode

Mode

计算数据的众数。

ec2Autoscale|mode(field string)

返回:InfluxQLNode

MovingAverage

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

ec2Autoscale|movingAverage(field string, window int64)

返回:InfluxQLNode

Percentile

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

ec2Autoscale|percentile(field string, percentile float64)

返回:InfluxQLNode

Sample

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

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

ec2Autoscale|sample(rate interface{})

返回:SampleNode

Shift

创建一个新节点,该节点及时移动传入的点或批次。

ec2Autoscale|shift(shift time.Duration)

返回:ShiftNode

Sideload

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

ec2Autoscale|sideload()

返回:SideloadNode

Spread

计算 minmax 点之间的差异。

ec2Autoscale|spread(field string)

返回:InfluxQLNode

StateCount

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

ec2Autoscale|stateCount(expression ast.LambdaNode)

返回:StateCountNode

StateDuration

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

ec2Autoscale|stateDuration(expression ast.LambdaNode)

返回:StateDurationNode

Stats

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

ec2Autoscale|stats(interval time.Duration)

返回:StatsNode

Stddev

计算标准偏差。

ec2Autoscale|stddev(field string)

返回:InfluxQLNode

Sum

计算所有值的总和。

ec2Autoscale|sum(field string)

返回:InfluxQLNode

SwarmAutoscale

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

ec2Autoscale|swarmAutoscale()

返回:SwarmAutoscaleNode

Top

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

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

返回:InfluxQLNode

Trickle

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

ec2Autoscale|trickle()

返回:TrickleNode

Union

执行此节点和所有其他给定节点的并集。

ec2Autoscale|union(node ...Node)

返回:UnionNode

Where

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

ec2Autoscale|where(expression ast.LambdaNode)

返回:WhereNode

Window

创建一个新节点,该节点按时间对流进行窗口化。

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

ec2Autoscale|window()

返回:WindowNode


此页面是否对您有帮助?

感谢您的反馈!


Flux 的未来

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

阅读更多

现已全面上市

InfluxDB 3 Core 和 Enterprise

快速启动。更快扩展。

获取更新

InfluxDB 3 Core 是一个开源、高速、最近数据引擎,可实时收集和处理数据,并将其持久保存到本地磁盘或对象存储。InfluxDB 3 Enterprise 构建在 Core 的基础上,增加了高可用性、读取副本、增强的安全性以及数据压缩,以实现更快的查询和优化的存储。InfluxDB 3 Enterprise 的免费层可供非商业家庭或业余爱好者使用。

有关更多信息,请查看