CombineNode

Warning! This page documents an old version of Kapacitor, which is no longer actively developed. Kapacitor v1.4 is the most recent stable version of Kapacitor.

Combine the data from a single node with itself.
Points with the same time are grouped and then combinations are created.
The size of the combinations is defined by how many expressions are given.
Combinations are order independent and will not ever include the same point multiple times.

Example:

stream
|from()
.measurement('request_latency')
|combine(lambda: "service" == 'login', lambda: TRUE)
.as('login', 'other')
// points that are within 1 second are considered the same time.
.tolerance(1s)
// delimiter for new field and tag names
.delimiter('.')
// Change group by to be new other.service tag
|groupBy('other.service')
// Both the "value" fields from each data point have been prefixed
// with the respective names 'login' and 'other'.
|eval(lambda: "login.value" / "other.value")
.as('ratio')
...

In the above example the data points for the login service are combined with the data points from all other services.

Properties

Property methods modify state on the calling node.
They do not add another node to the pipeline, and always return a reference to the calling node.
Property methods are marked using the . operator.

As

Prefix names for all fields from the respective nodes.
Each field from the parent nodes will be prefixed with the provided name and a '.'.
See the example above.

The names cannot have a dot '.' character.

node.as(names ...string)

Delimiter

The delimiter between the As names and existing field an tag keys.
Can be the empty string, but you are responsible for ensuring conflicts are not possible if you use the empty string.

node.delimiter(value string)

Max

Maximum number of possible combinations.
Since the number of possible combinations can grow very rapidly
you can set a maximum number of combinations allowed.
If the max is crossed, an error is logged and the combinations are not calculated.
Default: 10,000

node.max(value int64)

Tolerance

The maximum duration of time that two incoming points
can be apart and still be considered to be equal in time.
The joined data point's time will be rounded to the nearest
multiple of the tolerance duration.

node.tolerance(value time.Duration)

Chaining Methods

Chaining methods create a new node in the pipeline as a child of the calling node.
They do not modify the calling node.
Chaining methods are marked using the | operator.

Elapsed

Eval

Create an eval node that will evaluate the given transformation function to each data point.
A list of expressions may be provided and will be evaluated in the order they are given.
The results are available to later expressions.

HttpOut

Create an HTTP output node that caches the most recent data it has received.
The cached data are available at the given endpoint.
The endpoint is the relative path from the API endpoint of the running task.
For example, if the task endpoint is at /kapacitor/v1/tasks/&lt;task_id&gt; and endpoint is
top10, then the data can be requested from /kapacitor/v1/tasks/&lt;task_id&gt;/top10.

Spread

Stats

Create a new stream of data that contains the internal statistics of the node.
The interval represents how often to emit the statistics based on real time.
This means the interval time is independent of the times of the data points the source node is receiving.