# `DdDataStreams.Aggregator`
[🔗](https://github.com/rockneurotiko/dd_data_streams/blob/main/lib/dd_data_streams/aggregator.ex#L1)

Aggregates checkpoints into ten-second buckets and flushes them to the agent.

A checkpoint is counted twice, as dd-trace-go v2.11.2 does it: once in the bucket
that contains the checkpoint (`TimestampType: "current"`) and once in the
bucket that contains the start of the pathway (`TimestampType: "origin"`). The
two entries carry the same distributions; only the bucket start and the
timestamp type differ. A bucket is identified by its service, its start and
which of the two families it belongs to, and its points by their hash
together with their aggregation tags, so that one node whose statistics are
split by a `group:` or a `manual_checkpoint:` tag is reported as one point
per split.

The three values a point carries are the pathway latency and the edge latency
in seconds, and the payload size in bytes. Checkpoints measure both latencies
in milliseconds, so they are scaled here; the payload size defaults to zero.

`flush/0` sends the buckets that are complete, meaning their start is not
newer than ten seconds ago, and keeps the bucket still being filled.
`flush(:all)` sends every bucket, the still-being-filled one included, as
stopping the process does.

A bucket is removed before it is sent, so a failed send drops it: it is
counted in
`[:dd_data_streams, :flush, :error]` telemetry with a `count` measurement,
and nothing is raised. A successful send is counted in
`[:dd_data_streams, :flush, :stop]` telemetry with the buckets and points it
carried, and a bucket dropped because its start sits beyond the clock-skew
horizon is counted in the same event by its `dropped` points. Stopping the
process sends everything,
including the partial bucket, unless the agent has not answered within ten
seconds, in which case the send is abandoned and the node keeps shutting
down.

One payload and one request are built per service, because the intake accepts
one service per payload. The requests go out in parallel, up to
`DdDataStreams.Config.flush_concurrency/0` at a time, and the flush returns
once every one of them has been attempted. The transport and the agent URL
are read from `DdDataStreams.Config` on every flush, so a change to either
applies to the next one.

A checkpoint that cannot be aggregated is dropped instead of counted: it is
reported in `[:dd_data_streams, :checkpoint, :invalid]` telemetry with a
`count` measurement, and nothing is raised.

# `add`

```elixir
@spec add(DdDataStreams.Pathway.checkpoint()) :: :ok
```

Adds a checkpoint to its current and origin buckets.

Takes the map `DdDataStreams.Pathway.set_checkpoint/3` returns, with an
optional `:payload_size` in bytes. Does nothing when the aggregator is not
running, which is the case while the library is disabled; that skip is
reported in `[:dd_data_streams, :checkpoint, :skip]` telemetry. While the
library is enabled such a drop is a misconfiguration - the flag was flipped
after the application booted - so the first one in each process is also
logged at warning level instead of vanishing silently.

# `child_spec`

Returns a specification to start this module under a supervisor.

See `Supervisor`.

# `flush`

```elixir
@spec flush() :: :ok
```

Sends every bucket that is complete, dropping each one before it is sent.

Returns once the sends have finished, whether or not they succeeded. The wait
is bounded by the transport's own timeouts; the default transport aborts a
request after ten seconds.

# `flush`

```elixir
@spec flush(:all) :: :ok
```

Sends every bucket, including the one still being filled.

This is the test seam: a checkpoint recorded a moment ago lands in the
current bucket, which `flush/0` keeps until the window completes, while
`flush(:all)` sends it at once - so with a capture transport in place a test
asserts on it immediately. `flush/0` keeps its meaning: only complete
buckets go, and the partial one waits for the next flush. Like `flush/0`,
this does nothing when the aggregator is not running.

# `start_link`

```elixir
@spec start_link(keyword()) :: GenServer.on_start()
```

Starts the aggregator under its registered name.

The transport is not an option: it is read from `DdDataStreams.Config` on
every flush, so it can be changed while the aggregator runs.

---

*Consult [api-reference.md](api-reference.md) for complete listing*