Aggregate

Analytics State Management

Synopsis

Folds a high-volume stream into per-window summary events: group by some fields, apply count/sum/min/max/avg over a tumbling window, and emit the summaries to a named target.

Schema

- aggregate:
    group_by: <string[]>
    window_sec: <numeric>
    aggregations:
      - func: <enum>
        field: <ident>
        as: <ident>
    destination: <string>
    pass_through: <boolean>
    max_groups: <numeric>
    filter: <script>
    description: <text>
    if: <script>
    ignore_failure: <boolean>
    on_failure: <processor[]>
    on_success: <processor[]>
    tag: <string>

Configuration

Field	Required	Default	Description
aggregations	Y	-	The summaries to compute. At least one is required
destination	Y	-	Named target of the route that receives the summary events, exactly as Reroute works
group_by	N	-	Fields whose distinct value combinations define a group. Omitted, every record falls in one group
window_sec	N	60	Tumbling window length in seconds
pass_through	N	false	Also let the source records continue down the pipeline. By default they are consumed
max_groups	N	10000	Cap on live groups per window
filter	N	-	Cribl-style JavaScript truthiness expression evaluated after if. A falsy result skips the processor
description	N	-	Explanatory note
if	N	-	Condition to run
ignore_failure	N	false	See Handling Failures. This does not mean "keep filtering and ignore errors" — see the warning below.
on_failure	N	-	See Handling Failures
on_success	N	-	See Handling Success
tag	N	-	Identifier
disabled	N	false	When true, the processor is skipped and the event continues to the next one. Lets you take a processor out of the path without removing its configuration

warning

ignore_failure: true silently turns this processor off. The drop is signalled to the pipeline as an error value, and the ignore_failure check runs before that value is inspected — so a matched event is kept instead of dropped, no error is logged, and the pipeline reports success. The processor appears to run normally while filtering nothing.

Use it only if you genuinely want a pass-through. To tolerate real errors without losing the filtering, leave ignore_failure unset and handle the failure with on_failure instead.

Aggregation

Field	Required	Default	Description
func	Y	-	count, sum, min, max or avg
field	Y*	-	Field to aggregate. Required by every function except count, which must not have one
as	Y	-	Output field name on the summary event. Must be unique within the processor

* = count takes no field and supplying one is a configuration error; every other function requires one.

Details

The source records are consumed by default. aggregate replaces a stream with its summaries — that is the point, and it is why the volume drops. Set pass_through: true when the raw records are still needed downstream.

The summaries do not continue down the pipeline. They are emitted to destination, a named target of the route, the same mechanism Reroute uses. destination is required for that reason: without a target there is nowhere for a summary to go.

window_sec is a tumbling window: each window closes and emits, then the next begins. Windows do not overlap, so every record is counted exactly once.

window_sec: 0 and max_groups: 0 mean "use the default", not "unlimited". They resolve to 60 seconds and 10000 groups. A negative value is a configuration error, rejected at load. This is the same convention Schema Drift follows.

Every requirement above is checked at load: an unknown func, a missing as, a duplicate as, a count with a field, or any other function without one all fail the configuration rather than misbehaving at runtime.

Examples

Counting by Group

Turning per-event firewall denials into a per-minute count...	- aggregate: group_by: ["source.ip", "destination.port"] window_sec: 60 aggregations: - func: count as: denies destination: metrics_target
one summary per group per window, and the raw records are consumed:	{ "source": {"ip": "10.4.2.17"}, "destination": {"port": 22}, "denies": 1043 }

Several Functions at Once

Each aggregation writes its own output field...	- aggregate: group_by: ["service.name"] window_sec: 300 aggregations: - func: count as: requests - func: avg field: duration_ms as: avg_duration - func: max field: duration_ms as: slowest destination: metrics_target
with as names that must be distinct:	{ "service": {"name": "checkout"}, "requests": 8811, "avg_duration": 42.7, "slowest": 1904 }

Keeping the Source Records

Summarizing without giving up the raw stream...	- aggregate: group_by: ["host.name"] aggregations: - func: count as: events destination: metrics_target pass_through: true
the summaries go to the target and the records carry on:	# Without pass_through the records stop here, which is # the volume reduction aggregate exists to provide.