# DdDataStreams

[![CI](https://github.com/rockneurotiko/dd_data_streams/actions/workflows/ci.yml/badge.svg)](https://github.com/rockneurotiko/dd_data_streams/actions/workflows/ci.yml)

[Datadog Data Streams Monitoring][dsm] for Elixir pipelines running on
[Broadway][broadway].

Data Streams Monitoring follows a payload from the service that produced it to
the service that consumed it, so it can report how long the payload waited in
the queue between them and how the services are connected. This library adds the
Broadway side of that: it records a checkpoint on each side of every edge,
writes the pathway onto the outgoing message, and ships the aggregated
statistics to the Datadog Agent.

It is a library, not an integration package: it has no dependency on Broadway
and none on any broker client. It hangs itself on the telemetry events Broadway
already emits, so the pathway and propagation layer works with any producer,
and the two things that are broker-specific - what a message's metadata says
about the edge it arrived on, and the `type:` technology to report - are one
module behind `DdDataStreams.Extractor`. That layer ships one extractor today,
for `BroadwayCloudPubSub` v2: tagging a message from any other broker needs an
extractor module of your own (the carriers `Flat` and `Sqs` are provided), and
an application can list several extractors to run pipelines on more than one bus
at once. [Broker support](#broker-support) says exactly what ships.

[dsm]: https://docs.datadoghq.com/data_streams/
[broadway]: https://hexdocs.pm/broadway

## What's in the box

* `DdDataStreams`: the entry point. `attach/0` installs the Broadway telemetry
  handlers, `produce/3` records a produce checkpoint and returns the `{key,
  value}` pair to add to the outgoing message, `produce_batch/3` does the same
  for a batch, and `consume/1` and `checkpoint/1` cover a pipeline that does not
  run on Broadway (`produce/3` takes `extractor:` when the stage publishes to a
  different bus than it consumed from).
* `DdDataStreams.Extractor`: the behaviour that names the edge of a message from
  its metadata. Configured with `config :dd_data_streams, extractors:`.
* `DdDataStreams.Extractor.GooglePubSub`: the extractor for
  `BroadwayCloudPubSub` v2, and the default.
* `DdDataStreams.Checkpoints`: the consume and produce checkpoints of a
  pathway, given edge tags. It knows no broker.
* `DdDataStreams.Pathway`: the pathway struct and its hashing, the identity of a
  node and of an edge.
* `DdDataStreams.Propagator`: the pathway header codec (default key
  `dd-pathway-ctx-base64`), byte compatible with the official Datadog tracers
  (dd-trace-go v2.11.2, dd-trace-java 1.67.1 and dd-trace-py 3.17.0 verified)
  for every timestamp a real clock produces; the widest varint the format
  allows diverges, and that divergence is recorded in the interop fixture. It
  works over an attribute map or a header list.
* `DdDataStreams.Carrier`: broker-specific carriers for the pathway header.
  `Flat` handles attribute maps and header lists; `Sqs` unwraps the `_datadog`
  JSON message attribute BroadwaySQS exposes.
* `DdDataStreams.Aggregator`: a GenServer that buckets checkpoints and flushes
  them to the Agent every ten seconds.
* `DdDataStreams.Payload` and `DdDataStreams.Sketch`: the StatsPayload MsgPack
  encoder and the DDSketch distributions the intake expects.
* `DdDataStreams.Tags`: the canonical tag vocabulary.
* `DdDataStreams.Context`: the pathway the current process produces from.
* `DdDataStreams.Config`: the environment and application settings that
  configure the library.
* `DdDataStreams.Backend`: the behaviour that decides whether the calling
  process records and where one checkpoint goes. The default,
  `DdDataStreams.Backend.Aggregator`, is production; tests can configure
  `DdDataStreams.Testing.Backend` for per-test ownership.
* `DdDataStreams.Transport`: the behaviour for posting to the Agent, with a Req
  implementation as the default and a `DdDataStreams.Transport.Capture`
  implementation for tests.

## Installation

Add `:dd_data_streams` to your dependencies:

```elixir
def deps do
  [
    {:dd_data_streams, "~> 0.3.0"}
  ]
end
```

`:req` is an optional dependency of this library. It provides the default
transport to the Agent; if you pass your own transport (see
[Application configuration](#application-configuration)), you do not need it, and
if you do not have it the startup logs an error telling you to configure one.
Any Req `~> 0.5` is supported: the transport only passes options Req has
accepted since 0.5, because Req raises `ArgumentError` on an option its own
version does not know.

The library requires Elixir 1.17 or later.

## Checklist

- The Agent is reachable at `DD_TRACE_AGENT_URL` (or `DD_AGENT_HOST` /
  `DD_TRACE_AGENT_PORT`) and has an API key.
- `DD_SERVICE` is set (defaults to `unnamed-elixir-service`); `DD_ENV` is
  optional and an empty value is sent as-is.
- `:req` is a dependency, or a custom `DdDataStreams.Transport` is configured.
- `DdDataStreams.attach/0` runs once before Broadway starts.
- The producer and consumer use the same `DdDataStreams.Config.header/0` value.
- The broker preserves the attribute or header.

## Quickstart

One normative path, from nothing to points on the Data Streams map: install,
start the Agent, attach, produce, consume, verify. Every later section
elaborates one of these steps, and none of them adds a second way to do it.

### 1. Install

Add `:dd_data_streams` as [Installation](#installation) describes; `:req` gives
the default transport to the Agent.

### 2. Start the Agent

The Datadog Agent has to be reachable over HTTP, with an API key configured in
its own config or environment, and `DD_SERVICE` names the service the nodes are
reported under. [Checklist](#checklist) is the short list, and
[Environment variables](#environment-variables) has every variable with its
default.

### 3. Attach

Attach once at start up, before Broadway starts:

```elixir
def start(_type, _args) do
  DdDataStreams.attach()
  Supervisor.start_link([MyApp.Pipeline], strategy: :one_for_one)
end
```

`attach/0` takes no options: the configured extractors decide what each message
means. It installs two telemetry handlers:

* `[:broadway, :processor, :message, :start]` records the consume checkpoint of
  each message, once per message, including for a leaf consumer that produces
  nothing.
* `[:broadway, :batch_processor, :start]` samples one of the batch's pathways so
  a `handle_batch/4` that publishes continues a pathway that connects back to
  the consumed messages. Fan-in keeps one pathway, not a join of all of them.

Both events are emitted by Broadway itself, for every producer, so a pipeline
that runs on Kafka, SQS, RabbitMQ or anything else is attached the same way.
What changes is how messages are tagged: Kafka, SQS and RabbitMQ have shipped
extractors and Pub/Sub is the default (see
[Broker support](#broker-support)), so another broker needs an extractor module
of your own before checkpoints are recorded.

Calling `attach/0` twice is harmless: telemetry keeps one handler per id, so a
hot code upgrade does not double the checkpoints.

`attach/0` validates the configured extractors before it attaches and raises
`ArgumentError` when `config :dd_data_streams, extractors:` is not a non-empty
list of extractor modules. That validation runs even while the library is
disabled, so a configuration mistake fails when the application starts instead
of waiting for the flag to be flipped.

To stop recording without restarting the application, `DdDataStreams.detach/0`
removes the handlers `attach/0` installed. It returns `:ok` whether or not
anything was attached, so a test can end with it unconditionally; checkpoints
already handed to the aggregator are still flushed.
`DdDataStreams.status/0` answers "is it recording?" from the local VM - no
Datadog Agent needed - with the enabled flag, whether the handlers are attached,
whether the aggregator is running, and the effective transport, agent URL,
header, service, environment, service namer and extractors.

### 4. Produce

Measure the payload, ask the library for the metadata pairs, and add them to the
message your client sends. For Pub/Sub:

```elixir
defmodule MyApp.Orders do
  def publish(order) do
    data = Jason.encode!(order)

    attributes = Enum.into(DdDataStreams.produce("orders", data), existing_attributes)

    MyApp.PubSub.publish("orders", data, attributes: attributes)
  end
end
```

`produce/3` records the produce checkpoint and returns a list of `{key, value}`
pairs; `[]` means the library is disabled, so there is nothing to add. The first
pair is the pathway under `DdDataStreams.Config.header/0`,
`dd-pathway-ctx-base64` unless configured (see
[Configuration](#configuration)), and its value is the encoded pathway: the node
hash and the pathway and edge start times, base64 encoded. The second pair is
the topic under `dd-topic`, written whenever the outgoing extractor names a
topic, so a consumer whose metadata carries only its subscription can still
report the topic. Merge the pairs into the metadata the message already carries
- never replace the other attributes or headers - and the broker has to
preserve them on the way through. The consumer reads the pathway automatically;
a missing or malformed value is not an error: the message starts a new pathway,
with zero latency.

Call `produce/3` immediately before the send. The checkpoint is recorded when
the function runs, so a send that never happens still reports a produce.

That block is the only full one; every variant of it is written out exactly
once elsewhere: a batch published to one topic under [Batching](#batching), a
client that takes a header list under
[Producing from a Broadway pipeline](#producing-from-a-broadway-pipeline), the
SQS `_datadog` attribute under [SQS wrapping](#sqs-wrapping), and a stage
publishing to a different bus than it consumed from under
[Choosing extractors](#choosing-extractors).

### 5. Consume

The handler from step 3 records the consume checkpoint for every message, using
the producer's pathway value when it is present. In `handle_message/3`, answer
with `produce/3` exactly as the producer did in step 4: measure the outgoing
payload, take the pairs it returns, and merge them into the outgoing message.

The pathway is taken from the process context the handler filled in for this
message, so the produce edge continues the consumed one. In `handle_batch/4`
the context holds one pathway sampled from the batch - [Batching](#batching)
shows one send answering the whole batch. A stage that publishes to a different
bus than it consumed from names the outgoing one by passing `extractor:` to
`produce/3` ([Choosing extractors](#choosing-extractors)), and a pipeline that
does not run on Broadway drives the same two checkpoints itself
([Without Broadway](#without-broadway)).

### 6. Verify

Once the Agent receives the flushed stats (about 10 seconds, then a few minutes
for the first UI data), Data Streams Monitoring shows:

- the service and topic/subscription nodes on the Data Streams map;
- throughput per edge;
- `data_streams.latency` distributions (p50/p75/p95/p99) for the edge and the
  full pathway;
- `data_streams.payload_size` distributions.

There are no APM traces or spans, and no message-level detail. A consumer-only
deployment starts a new pathway: its edge latency is zero unless the broker
stamps a publish time the extractor believes, in which case - as with
[Pub/Sub](#google-pub-sub) - it reports how long the message waited in the bus.
When nothing shows up,
[Requirements and troubleshooting](#requirements-and-troubleshooting) lists the
usual causes.

## Guides

Per-broker and per-shape guidance: the quickstart's one path, spelled out.

### Broker support

What actually ships, as opposed to what the library's design allows. Each cell
names the module or the test that proves it, so the table stays true as those
change.

| Broker | Shipped extractor | Carrier | Tested package / version | Status & known limits |
| --- | --- | --- | --- | --- |
| Google Pub/Sub | `DdDataStreams.Extractor.GooglePubSub`, the default (`lib/dd_data_streams/extractor/google_pub_sub.ex`) | `DdDataStreams.Carrier.Flat` - attribute maps, header lists, AMQP 3-tuples | `broadway_cloud_pub_sub ~> 2.0.0-rc.3` (`e2e/mix.exs`); `test/dd_data_streams/google_pub_sub_test.exs`, `test/integration/pubsub_emulator_test.exs` | Shipped and tested. Consume checkpoints need 2.0.0-rc or newer: on stable 1.0.0 the metadata carries no `:subscription`, so nothing is recorded. |
| Amazon SQS | `DdDataStreams.Extractor.SQS` (`lib/dd_data_streams/extractor/sqs.ex`) | `DdDataStreams.Carrier.Sqs` - the `_datadog` message attribute | `broadway_sqs ~> 1.0.1` (`mix.exs`); `test/dd_data_streams/sqs_test.exs`, `test/integration/sqs_broker_test.exs` (goaws) | Shipped, unit-tested and run against a real SQS API. The queue comes from the producer's `queue_url`, and the pipeline must request `message_attribute_names: ["_datadog"]` for the pathway to arrive and `attribute_names: [:sent_timestamp]` for publish latency. The emulator behind the tier stamps its own receive time as `SentTimestamp`, so send latency is only proven against real SQS. |
| Kafka | `DdDataStreams.Extractor.Kafka` (`lib/dd_data_streams/extractor/kafka.ex`) | `DdDataStreams.Carrier.Flat` - header lists | `broadway_kafka ~> 0.7.1` (`mix.exs`); `test/dd_data_streams/kafka_test.exs`, `test/integration/kafka_broker_test.exs` (Apache Kafka 4.4) | Shipped, unit-tested and run against a real Kafka. `group:` is reported when the producer's `group_id` is known; `kafka_cluster_id` is not, because BroadwayKafka exposes no cluster id. |
| RabbitMQ | `DdDataStreams.Extractor.RabbitMQ` (`lib/dd_data_streams/extractor/rabbitmq.ex`) | `DdDataStreams.Carrier.Flat` - AMQP 3-tuples | `broadway_rabbitmq ~> 0.8.2` (`mix.exs`); `test/dd_data_streams/rabbitmq_test.exs`, `test/integration/rabbitmq_broker_test.exs` (RabbitMQ 4.4) | Shipped, unit-tested and run against a real RabbitMQ. The queue comes from the producer's `queue:`, and the pipeline must set `metadata: [:headers, :timestamp]` or no header, and no publish time, reaches the pipeline. On RabbitMQ 4 the producer also needs `declare: [durable: true]`: the default `declare: []` asks for a transient queue, which 4.x refuses. |

Ground truth is the tree itself: [`lib/dd_data_streams/extractor/`](lib/dd_data_streams/extractor/)
lists every shipped extractor, [`lib/dd_data_streams/carrier/`](lib/dd_data_streams/carrier/)
every shipped carrier, and [`mix.exs`](mix.exs) has no broker client as a
dependency - so a row names a tested package only where a test actually runs
against it.

Each row's tier is a real pipeline on the real client against a real broker:
`test/integration/{sqs,kafka,rabbitmq}_broker_test.exs` publish a pathway,
consume it through Broadway, and read the produced pair back off a second
queue or topic, so the metadata shape, the carrier round trip and the tag set
are all observed rather than derived. `just sqs`, `just kafka` and `just
rabbitmq` run them against the containers in `compose.yaml`; they are excluded
from `mix test`, so the default suite still needs no broker.

What the tiers do not cover: no AWS account, so SQS is a compatible emulator
and not the service, and its `SentTimestamp` is the receive time; nothing here
runs against a managed Kafka or a clustered broker; and no tier posts to a real
Datadog Agent (the [`agent`](#agent) tier covers that separately).

### Google Pub/Sub

The default extractor, `DdDataStreams.Extractor.GooglePubSub`, applies when
`extractors:` is absent, so a Pub/Sub pipeline configures nothing.
A message names its consume edge by its subscription, and
`BroadwayCloudPubSub` v2 stamps the subscription, the attributes, the ordering
key and the receive time, which is everything a checkpoint needs. Consume
checkpoints therefore need `broadway_cloud_pub_sub` 2.0.0-rc or newer, tested
with 2.0.0-rc.2: stable 1.0.0 (v1) metadata carries no `:subscription`, so the
default extractor returns `:error` for every message and a v1 pipeline records
nothing. A message that carries no subscription records no consume checkpoint,
because there is no name to report it under.

The pathway rides in an attribute, and `DdDataStreams.Carrier.Flat` reads and
writes attribute maps (and header lists), so the producer merges the pairs
`produce/3` returns into the attributes exactly as the
[quickstart](#quickstart) does. What the extractor reports for each edge -
including how Pub/Sub's full resource names are reduced to short ids - is under
[Edge tags](#edge-tags).

A consume edge normally knows only its subscription, because that is all the
message metadata carries. `produce/3` also writes the outgoing topic under the
`dd-topic` attribute, and the extractor reads it back, so a consume edge that
follows a producer using this library reports both tags. Datadog derives the
queues in the middle of a pathway - `visited_queues` on
`data_streams.latency` and the queue entries of `detailed_pathway` - from the
`topic` tag, so without it the queue between two services is missing. A message
from outside, or from a producer that did not write the attribute, keeps the
subscription-only edge; the topic is never guessed from the subscription name.

Pub/Sub stamps the time it accepted the message, and the extractor reports it as
the first-hop start, so a pipeline that only consumes - no upstream service
propagating to it - reports how long the message waited in the bus instead of
zero. A message that carries a pathway header keeps the start its producer
wrote; a publish time that is missing or that a clock could not have stamped is
dropped, as described under [Latency](#latency).

### Kafka

`DdDataStreams.Extractor.Kafka` names the consume edge by the record's topic and
the produce edge by the topic the host produces to, both as `type:kafka`, and
adds `group:<consumer group>` when the producer's options carry one. The
pathway rides in the record headers, so nothing has to be configured for it to
propagate; the pipeline does have to list this extractor, since Pub/Sub stays
the default.

The record timestamp is a publish time, so a first-hop message measures from it
and the consume latency does not include Broadway's buffer. `partition` and
`offset` are deliberately not tags: a per-partition tag would name whichever
partition arrived first.

One tag the official tracers report is missing: `kafka_cluster_id`. Nothing in
the record carries it, and it cannot be handed in through the producer either,
because BroadwayKafka rejects options it does not know. A node of this library
therefore differs from a Java or Go tracer node that reports a cluster id, and
the two do not join up; the pathway still connects through the header, so
latency and throughput are measured across them. The module's moduledoc states
the rest of what this cannot report.

### Amazon SQS

`DdDataStreams.Extractor.SQS` is the one extractor that cannot name its edge
from the message: SQS metadata carries no queue. It takes the queue from the
producer's `queue_url` and reports `topic:<queue>`, `type:sqs` on both edges, so
two queues in one application stay two nodes.

Two options belong on the producer, and without the first the pathway never
arrives:

```elixir
module:
  {BroadwaySQS.Producer,
   queue_url: "https://sqs.eu-west-1.amazonaws.com/123456789012/orders",
   message_attribute_names: ["_datadog"],
   attribute_names: [:sent_timestamp]}
```

`attribute_names: [:sent_timestamp]` is optional and only feeds publish latency.
The produce side writes the `_datadog` attribute itself, as described under
[SQS wrapping](#sqs-wrapping).

### SQS wrapping

For SQS, BroadwaySQS exposes the pathway under the `_datadog` message
attribute as a JSON object. `DdDataStreams.Carrier.Sqs` unwraps that JSON to
read the pathway and merges a produced header into it, so a producer wraps the
pairs returned by `DdDataStreams.produce/3` into the `_datadog` attribute's
JSON rather than adding them as separate attributes:

```elixir
attributes = existing_attributes

attributes =
  Map.put(
    attributes,
    "_datadog",
    DdDataStreams.produce("orders", data) |> Map.new() |> Jason.encode!()
  )
```

### RabbitMQ

`DdDataStreams.Extractor.RabbitMQ` names the consume edge by the queue from the
producer's `queue:` option - never by the exchange, which would split a queue
bound to several exchanges into several nodes - and reports `type:rabbitmq`.
The pipeline must ask for the properties it reads, because BroadwayRabbitMQ
copies none by default:

```elixir
module: {BroadwayRabbitMQ.Producer, queue: "orders-q", metadata: [:headers, :timestamp]}
```

An outgoing edge takes two values, the exchange and the routing key, so this is
the extractor that implements `produce/3` and reads them from the call's
options. A publish to the default exchange is named by its routing key, as the
key is the queue; a named exchange is named by `exchange:` with
`manual_checkpoint:false` and `has_routing_key:`, following dd-trace-java.

```elixir
DdDataStreams.produce("orders", payload,
  extractor: DdDataStreams.Extractor.RabbitMQ,
  exchange: "orders-x",
  routing_key: "created"
)
```

### Choosing extractors

An extractor turns a Broadway message's `metadata` into the edge it arrived on.
Configure a list of them, one per technology the application talks to:

```elixir
config :dd_data_streams,
  extractors: [MyApp.KafkaExtractor, DdDataStreams.Extractor.GooglePubSub]
```

Every message is offered to the extractors in order, and the first one that
recognises its metadata names its edge, so put the more specific one first.
What ships for which broker - shipped extractor, carrier, tested version - is
the matrix under [Broker support](#broker-support), and the default extractor
for Pub/Sub is described under [Google Pub/Sub](#google-pubsub).

Another broker is one module implementing `DdDataStreams.Extractor`. Its
moduledoc has a `BroadwayKafka` example: the metadata mapping, and the `type:`
Datadog files those streams under. How much mapping that is depends on the
broker: when the metadata alone names the edge it is those few lines, but a
broker whose edge lives in the producer's configuration instead (the SQS
queue, the Kafka consumer group) implements `consume/2` and takes the edge name
from `producer_info`, and SQS additionally needs the `_datadog` wrapping
under [SQS wrapping](#sqs-wrapping).

On the produce side there is no metadata to look at, only the topic the call is
made with and the options passed to it, so a produce is reported by the
extractor that consumed the message being answered. A stage that changes
technology names the outgoing one by passing it to the same call the
[quickstart](#quickstart) makes:

```elixir
# Consumed from Kafka, publishing to Pub/Sub.
DdDataStreams.produce("orders-out", data,
  extractor: DdDataStreams.Extractor.GooglePubSub
)
```

A broker whose outgoing edge takes more than a topic - a RabbitMQ publish is
named by its exchange and its routing key - is expressed by an extractor that
implements `produce/3`, which receives the options of the same call. The
library reads `:extractor` and `:payload_size` itself and hands the list over
unchanged, so the extra options are the host's vocabulary:

```elixir
DdDataStreams.produce(routing_key, data,
  extractor: MyApp.RabbitMQExtractor,
  exchange: "orders",
  routing_key: routing_key
)
```

The pair it returns is merged into the outgoing message exactly as in the
quickstart.

### Producing from a Broadway pipeline

In `handle_message/3`, call `DdDataStreams.produce/3` with the outgoing payload
and merge the pairs it returns into the outgoing message - the block is the one
in step 4 of the [quickstart](#quickstart), unchanged. It does not publish, and
the pathway is taken from the process context that `attach/0` filled in for this
message. A client that takes a header list uses the same pairs:

```elixir
headers = existing_headers ++ DdDataStreams.produce("orders-out", data)

MyApp.Kafka.produce("orders-out", data, headers: headers)
```

The `[...]` pattern assumes the library is enabled; the Pub/Sub example shows
the `[]` branch.

Fan-in re-derives each message's consume checkpoint from the batch
messages' metadata, so a `handle_message/3` that returns the message unchanged
keeps the batch linked. Replacing `message.metadata` entirely (for example by
building a new `%Broadway.Message{}`) drops the inbound carrier: the batch then
fans in without a parent pathway and the produce edge starts a new one. Preserve
the metadata, or at least carry its inbound pathway, when transforming a message.

Producing one message per consumed message means calling `produce/3` once per
message, not chaining one call's result into the next: all of them then share
one produce node with the consume node as their parent.

Call `produce/3` from the stage process itself, not from a `Task` it spawns: the
context is kept per process, so a task sees no pathway and starts a new one.

### Batching

One send can answer a whole batch. `produce_batch/3` records one checkpoint per
payload and returns the pairs for the whole batch - the batch shares one produce
node - and those pairs are merged into the outgoing message exactly as the
[quickstart](#quickstart) merges the single-message pairs, on the producer as
well as in the pipeline.

In `handle_batch/4` the context holds one pathway sampled from the batch, so
one send answers the batch and stays connected to the consumed messages:

```elixir
def handle_batch(_batcher, messages, _batch_info, _context) do
  payloads = Enum.map(messages, &encode(&1.data))

  attributes = Enum.into(DdDataStreams.produce_batch("orders-out", payloads), existing_attributes)

  MyApp.PubSub.publish_batch("orders-out", payloads, attributes: attributes)
  messages
end
```

The minimal batchers configuration is
`batchers: [default: [batch_size: 20, batch_timeout: 100]]`. Broadway builds
no batch stages without it, so `handle_batch/4` runs only when `batchers:` is
configured. Return every message you were given: Broadway logs an error when
the returned list does not match, and a message left out is never
acknowledged. When the batch's merged fan-in leaves no pathway, the next
produce starts a fresh one.

`produce_batch/3` returns `[]` for an empty batch; both functions return `[]`
when the library is disabled, so the call site needs no guard beyond that.

### Without Broadway

`consume/1` and `produce/3` are the same two checkpoints as functions, for a
pipeline that drives the library itself. `consume/1` takes a Broadway message,
or a map shaped like one - the body in `:data`, the producer's metadata in
`:metadata` - so the extractor names the edge exactly as it does for Broadway:

```elixir
case DdDataStreams.consume(%{
       data: message.data,
       metadata: %{
         subscription: message.metadata.subscription,
         attributes: message.metadata.attributes,
         orderingKey: nil,
         receivedAt: message.metadata.receivedAt
       }
     }) do
  nil ->
    # The extractor found no edge: nothing was recorded and the context
    # was cleared, so there is no pathway to continue.
    nil

  pathway ->
    # LATER, when the message is answered, in the same process: produce exactly
    # as the quickstart does - `DdDataStreams.produce/3` returns the pairs to add
    # to the outgoing message - and this pathway continues the edge.
    pathway
end
```

`consume/1` returns `nil` when the extractor finds no edge in the metadata: it
records nothing and clears the context, so the `nil` branch above has no
pathway to continue. On a match it leaves the pathway in the context of the
calling process; if the answer is produced in another process, hand the pathway
it returned over with `DdDataStreams.Context.put(pathway)` there, and pass
`extractor:` to `produce/3` when several extractors are configured.

The metadata is the extractor's own contract: for the Pub/Sub extractor those
are the values `BroadwayCloudPubSub` v2 exposes as
`message.metadata.subscription`, `.attributes`, `.orderingKey` and `.receivedAt` (a `DateTime`).

Do not call `consume/1` yourself while `attach/0` is installed: the telemetry
handler already records that checkpoint, and the message would be counted twice.

### Checkpoints for other stages

`checkpoint/1` records a checkpoint for a stage that is not a bus edge, such as
an internal queue, and takes its edge tags verbatim:

```elixir
DdDataStreams.Context.put(
  DdDataStreams.checkpoint(
    ["direction:in", "type:internal-queue", "manual_checkpoint:true"]
  )
)
```

The call leaves the process context alone; the example puts the pathway it
returned back, so a produce in that stage continues the pathway the checkpoint
returned and the new node gets its downstream edge instead of a dead end.

`manual_checkpoint:true` is an aggregation tag, so it splits the statistics of
a node: two checkpoints of the same node, meaning the same hash-tier tags and
therefore the same hash, that differ in their aggregation tags are counted
apart and reported as separate points that share that hash.

All checkpoints are handed to the aggregator, which is not running while the
library is disabled, so these calls need no guard at the call site.

## Testing your pipeline

What a test suite should do depends on what it is testing: the pipeline's own
code, the checkpoints the library derived from a message, or the wire-format
payload a real aggregator would send. The first two can run with `async: true`;
only the last needs the library's global switches and `async: false`.

### Owner-scoped tests: the default, `async: true`

Configure the test backend, start its ownership server once, and attach the
Broadway handlers once at boot. The server is the test backend's own process, so
production never starts it.

```elixir
# config/test.exs
config :dd_data_streams, backend: DdDataStreams.Testing.Backend
```

```elixir
# test/test_helper.exs
{:ok, _} = DdDataStreams.Testing.Backend.start_link()
DdDataStreams.attach()
ExUnit.start()
```

`nimble_ownership` is an optional dependency of this library. A suite that uses
the owner-scoped API adds it to its own dependencies:

```elixir
{:nimble_ownership, "~> 1.0"}
```

`attach/0` still decides whether to install handlers from the `enabled` flag, so
leave that flag at its default `true` for the owner-scoped recipe; do not combine
it with the `enabled: false` recipe below. Once installed, a handler asks the
backend whether the process that emitted the event is recording, so a test that
never opts in makes the same handler a no-op and never sees another test's
checkpoints.

A test opts in with `start/1` and reads the result with `checkpoints/1`:

```elixir
defmodule MyApp.OrdersTest do
  use ExUnit.Case, async: true
  use DdDataStreams.Testing

  test "the consume checkpoint links to the produce checkpoint" do
    DdDataStreams.Testing.start()

    pairs = DdDataStreams.produce("orders-out", Jason.encode!(order))

    DdDataStreams.consume(%{
      data: Jason.encode!(order),
      metadata: %{
        subscription: "projects/demo/subscriptions/orders-sub",
        attributes: Map.new(pairs)
      }
    })

    produce = DdDataStreams.Testing.assert_produced("orders-out")
    consume = DdDataStreams.Testing.assert_consumed("orders-sub")
    DdDataStreams.Testing.assert_chained(produce, consume)
  end
end
```

`use DdDataStreams.Testing` only picks the ownership mode: private for an
`async: true` case and shared for a sync one. A test that never calls `start/1`
owns nothing, so `produce/3` returns `[]` and nothing is recorded, even though
other tests in the same suite are recording at the same time. `start/1` is
idempotent.

Reading is synchronous. `checkpoints/1` returns the points recorded so far for
the owner, oldest first, so there is no flush, no ten-second bucket window and
nothing to wait for. Each point is a map with `:edge_tags`, `:hash`,
`:parent_hash`, `:pathway_latency`, `:edge_latency`, `:payload_size` and
`:timestamp_type`; the latencies and size are the integers the checkpoint
measured, and `:timestamp_type` is `"current"` because this is the checkpoint
itself rather than an aggregator bucket. The three assertion helpers are thin
wrappers over it: `assert_produced/1` and `assert_consumed/1` find a point by
its direction and topic or subscription tag, and `assert_chained/2` checks that
a consume point's `parent_hash` is a produce point's `hash`.

`subscribe/0` is the push alternative for a test that prefers `assert_receive`:
the owner then receives `{DdDataStreams.Testing, :checkpoint, point}` for every
point recorded after the call:

```elixir
DdDataStreams.Testing.start()
DdDataStreams.Testing.subscribe()

[_pair] = DdDataStreams.produce("orders-out", payload)

assert_receive {DdDataStreams.Testing, :checkpoint, %{edge_tags: tags}}
assert "topic:orders-out" in tags
```

#### Processes the test did not start

A `Task` inherits the test's `$callers`, so it records into the same owner
without any extra call. A process Broadway started does not, so allow it
explicitly. Both arguments accept a pid or a registered name, and Broadway
registers its stages under the pipeline's name:

```elixir
DdDataStreams.Testing.start()

Broadway.start_link(MyApp.Pipeline, name: MyApp.Pipeline)
DdDataStreams.Testing.allow(self(), :"MyApp.Pipeline.Broadway.Processor_default_0")
DdDataStreams.Testing.allow(self(), :"MyApp.Pipeline.Broadway.BatchProcessor_default")
```

The processors are `...Processor_default_<index>` and the batchers
`...BatchProcessor_default`; a pipeline with several processors allows each one.
There is no hidden framework hook: `allow/2` is the supported way, and shared
mode, which records every process into the test process, is the alternative for
an `async: false` case.

Nothing has to be cleaned up. `NimbleOwnership` monitors the owner and drops its
checkpoints when the test process exits, so there is no `on_exit` and no
detach. Tests using this API do not call `attach/0` or `detach/0` either: the
handlers are installed once at boot, and an unowned process makes them no-ops.

### Tests that do not record: the library off

```elixir
# config/test.exs
config :dd_data_streams, enabled: false
```

With the flag off, `attach/0` installs no handlers, `produce/3` and
`produce_batch/3` return `[]`, and no aggregator is started when the application
boots. This is the recipe for a suite that only tests the pipeline's own code;
every test can be `async: true`. A suite that uses the owner-scoped backend does
not need this flag: an unowned process already behaves as if the library were
disabled.

### Wire-format tests: the global path, `async: false`

Some tests need more than a checkpoint: they need the encoded payload, the
aggregator's buckets, or the exact bytes the Agent would receive. Those use the
global testing API and run with `async: false`, because it flips the application
environment, starts the one shared aggregator and attaches the VM-wide handler
id.

`DdDataStreams.Testing.enable!/1` flips the library on for one test and returns
the function that flips it back:

```elixir
defmodule MyApp.OrdersWireTest do
  use ExUnit.Case, async: false

  setup do
    on_exit(DdDataStreams.Testing.enable!(transport: DdDataStreams.Transport.Capture))
    :ok
  end

  test "the produce checkpoint reaches the aggregator" do
    [{_key, value} | _topic] = DdDataStreams.produce("orders-out", Jason.encode!(order))
    assert is_binary(value)

    assert DdDataStreams.status().aggregator_running
    assert :ok = DdDataStreams.Aggregator.flush(:all)
  end
end
```

`enable!/1` sets `enabled: true`, accepts `transport:`, `service:` and `env:`,
attaches the handlers `attach/0` installs, and starts the aggregator when it is
not already running. The cleanup it returns stops the aggregator the call
started, detaches the handlers it attached and restores the application
environment to what it was before the call; anything already running or already
attached when it was called is left alone.

`DdDataStreams.Transport.Capture` posts nothing: each flush is reported on the
`[:dd_data_streams, :transport, :captured]` telemetry event, whose metadata is
`%{url: url, headers: headers, body: body}` with the exact gzip-compressed
payload the Agent would have received (`:zlib.gunzip/1` reads it back).

`DdDataStreams.Testing.capture/1` then runs a function, calls
`DdDataStreams.Aggregator.flush(:all)` to send every bucket at once - including
the one still being filled, which `flush/0` keeps until its ten-second window
completes - and returns the points of everything that was flushed, decoded by
`DdDataStreams.Testing.decode_payload/1`:

```elixir
# In an async: false test: it observes the shared aggregator's flushes.
defmodule MyApp.OrdersCaptureTest do
  use ExUnit.Case, async: false

  test "the consume checkpoint links to the produce checkpoint" do
    points =
      DdDataStreams.Testing.capture(fn ->
        pairs = DdDataStreams.produce("orders-out", Jason.encode!(order))

        DdDataStreams.consume(%{
          data: Jason.encode!(order),
          metadata: %{
            subscription: "projects/demo/subscriptions/orders-sub",
            attributes: Map.new(pairs)
          }
        })
      end)

    produce = Enum.find(points, &("topic:orders-out" in &1.edge_tags))
    consume = Enum.find(points, &("subscription:orders-sub" in &1.edge_tags))

    assert consume.parent_hash == produce.hash
  end
end
```

Each point is a map with `:edge_tags`, `:hash`, `:parent_hash`,
`:pathway_latency`, `:edge_latency`, `:payload_size` and `:timestamp_type`;
the three distributions stay binaries, exactly as they went on the wire.
`decode_payload/1` is the decoder under it: it takes the gzip-compressed
MsgPack body from any handler on the captured event and returns the whole
payload as a plain map, so a test that wants the service or the bucket start as
well can decode a body it captured itself.

A test that attaches or detaches handlers itself also runs with `async: false`,
because the handler id is shared by the whole VM. Tests that leave the
boot-time handlers alone do not: with the owner-scoped backend, a test that did
not opt in makes every handler a no-op.

## Reference

What the library reports, emits and reads.

### Edge tags

Edge tags come from the extractor. The Pub/Sub one reports a consume edge as
`direction:in`, `subscription:<subscription>`, `topic:<topic>`,
`type:google-pubsub` when the producer wrote the `dd-topic` attribute, and as
`direction:in`, `subscription:<subscription>`, `type:google-pubsub` otherwise. A
produce edge is `direction:out`, `topic:<topic>`, `type:google-pubsub`. Pub/Sub
names subscriptions and topics as full resource names
(`projects/<project>/subscriptions/<id>`); the extractor reduces each to its
short id, and an already-short name is reported as it is.

`type:` is the queueing technology Datadog files the stream under, and it is
hashed into every node, so it has to be the same for every service in a pipeline,
and every edge of one bus is reported by one extractor.

Every tag from the extractor is reported. What the vocabulary of
`DdDataStreams.Tags` decides is what else each key does:

* hashed, identifying a node: `bus`, `direction`, `exchange`, `topic`, `type`,
  `subscription`, `kafka_cluster_id`
* aggregation, splitting a node's statistics without changing its identity:
  `ds.name`, `ds.namespace`, `manual_checkpoint`, `group`, `consumer_group`
* per-message values, reported but never hashed: `partition`, `has_routing_key`

Any other key is reported as it is: it hashes into no node and splits no
statistics.

A tag is always a `key:value` string. A tag that is not a string carries no
key and no value, so `checkpoint/1,2` rejects it at the boundary, the
extractors filter it out of a message's tags, and the aggregator drops any
that reach it before the flush.

A produce measured for one topic is one node even when the messages are spread
over partitions, which is what lets downstream produce edges join up.

### Service name

`service` (or `DD_SERVICE`) names one service for the whole application. A
pipeline that handles messages for several services - one per tenant, one per
Broadway pipeline, one per topic - names it per checkpoint instead, by
configuring a module that implements `DdDataStreams.ServiceName`:

```elixir
config :dd_data_streams, service_name: MyApp.ServiceNames

defmodule MyApp.ServiceNames do
  @behaviour DdDataStreams.ServiceName

  @impl true
  def call(%{direction: :in, metadata: metadata}) do
    case metadata do
      %{subscription: subscription} -> "my_service." <> short_id(subscription)
      _ -> nil
    end
  end

  def call(%{direction: :out, topic: topic}), do: "my_service." <> short_id(topic)

  defp short_id(resource), do: resource |> String.split("/") |> List.last()
end
```

It is asked once per checkpoint: the `:in` direction gets the message's broker
metadata and the `:out` direction the publish topic and the options passed to
`produce/3`, and both get the edge tags and the producer info. A `:service`
option on `produce/3` or `produce_batch/3` wins over the module; a module that
returns `nil`, `""` or anything else that is not a string names no service and
`service` applies. `checkpoint/1,2` reports `service`, since it has neither a
message nor a topic to name one from.

A produce can also name itself after what it answers. The context carries
`pathway`, and on the `:out` direction its `tags` are the tags of the consumed
edge, so the parent topic is one `DdDataStreams.Tags.get/2` away:

```elixir
def call(%{direction: :out, pathway: %DdDataStreams.Pathway{} = pathway}) do
  case DdDataStreams.Tags.get(pathway.tags, "topic") do
    nil -> nil
    topic -> "my_service." <> topic
  end
end
```

`pathway` is `nil` for `:in`, and for a produce that continues nothing: the
pathway header on the message carries the hashes and the two starts, not the
tags, so a consumer cannot read its producer's topic from what it receives.

Three things to know before using it:

* The service is hashed into the node's identity and it keys the aggregator's
  buckets: **one MsgPack payload and one agent request go out per service per
  ten-second flush**, so name a small, known set of logical services. The
  requests of one flush are posted in parallel, up to `flush_concurrency` at a
  time, which defaults to 8. A name derived from an order or a tenant id
  multiplies services and nodes instead of describing them.
* The module runs on the recording path, once per checkpoint per direction.
  Keep it a decision, not a lookup: no database, no HTTP call.
* The outgoing payload is not given to it, because `produce/3` never inspects
  the payload and a name derived from message content is the cardinality trap
  above. Use the topic, the tags, the metadata or the pipeline name.

### Latency

Checkpoint times are in native time units: the consume checkpoint's
`received_at` (the metadata's receive time) and the times stamped on produce.

A consume checkpoint is stamped with the time the message reached the pipeline,
not the time the processor picked it up. The consume edge therefore reports the
time the message spent in the queue, not Broadway's own backlog. A metadata
that carries no receive time has it filled in with the current time before the
checkpoint is taken, so the checkpoint reports no consume latency. A produce
checkpoint is stamped at the time of the produce call.

An extractor may additionally report the broker's publish time for a first-hop
message, as `:pathway_start` or `:edge_start` in the parts it returns (both in
native time units); the consume edge then reports publish-to-consume latency
instead of zero. These starts apply only to a message that starts a pathway: a
message carrying a pathway header keeps the start its producer wrote. A start is
only believed inside the window from 2000-01-01 to a minute ahead of this node's
clock - the same window the propagator applies to starts read from a header -
and an out-of-window value falls back to the checkpoint time.

A message that carried no pathway and no such publish time starts a new one at
its consume checkpoint, so both latencies are zero for it. A redelivery is a
new attempt: every one that reaches `handle_message/3` records its own
checkpoint, with the wait still measured from the receive time.

### Payload size

The consume size is the data, the ordering key, and every attribute's or
header's key and value, in bytes. It includes the inbound pathway header (about
49 bytes).

The produce size is the size of `data` by default, or the `:payload_size` you
pass to `produce/3`; `produce_batch/3` measures each payload separately. The
library adds the size of the `{key, value}` pair it returns - the configured
header key plus the encoded value. That is all produce can see: structurally
it cannot see the attributes the publisher put on the message, while consume
counts every attribute and the ordering key on the arriving one. The two sides
therefore agree only for a bare message; pass `payload_size:` to count the
rest. Measure the message without that pair.

### Flushing

`DdDataStreams.Aggregator` buckets checkpoints in ten-second windows and posts
one MsgPack, gzip-compressed payload per service to
`<agent-url>/v0.1/pipeline_stats` every ten seconds. Each checkpoint is counted
twice, in the bucket it belongs to and in the bucket that contains the start of
its pathway, as dd-trace-go v2.11.2 does; the two entries differ only in bucket
start and timestamp type. dd-trace-java 1.67.1 and dd-trace-py 4.16.0 emit one
bucket per checkpoint and no timestamp type at all. A bucket is dropped before
it is sent, so a failed send loses that bucket rather than growing memory; the
failure is reported via telemetry, and nothing is raised. A bucket whose start
sits more than a minute ahead of the clock (a replayed message, a skewed broker
clock) never becomes
due and is dropped at the next flush instead of being held forever; that drop
is reported on telemetry too. Stopping the aggregator sends
the partial bucket too, unless the agent has not answered within ten seconds,
in which case the node keeps shutting down without it.

`DdDataStreams.Aggregator.flush(:all)` sends every bucket immediately, the
partial one included; `flush/0` keeps its meaning of complete buckets only.
That is the seam a test uses to assert on a checkpoint without waiting for the
window - see [Testing your pipeline](#testing-your-pipeline).

The aggregator is started and supervised by `:dd_data_streams` itself, so the
host does not add it to its own supervision tree.

### Telemetry

The library emits five events:

* `[:dd_data_streams, :transport, :captured]`, once per flush through
  `DdDataStreams.Transport.Capture`, with `%{count: 1}` measurements and
  `%{url: url, headers: headers, body: body}` metadata. Only a suite that
  configured the capture transport (see
  [Testing your pipeline](#testing-your-pipeline)) ever sees it.

* `[:dd_data_streams, :flush, :stop]`, once per flush that ends without an
  error, with `%{buckets: buckets, points: points, dropped: dropped}`
  measurements and `%{service: service, reason: reason}` metadata. On a
  successful send, `buckets` and `points` are what was sent, `dropped` is `0`
  and `reason` is `nil`. When a bucket is dropped because its start sits beyond
  the clock-skew horizon and it would never become due, `buckets` and `points`
  are `0`, `dropped` counts the points that were discarded and `reason` is
  `:future_bucket`.

* `[:dd_data_streams, :flush, :error]`, once per failed Agent flush, with
  `%{count: 1}` measurements and `%{service: service, reason: reason}` metadata.

* `[:dd_data_streams, :checkpoint, :skip]`, once per checkpoint the library
  decides not to record, with `%{count: 1}` measurements and
  `%{reason: reason, extractors: extractors}` metadata. The reasons are:

  * `:no_extractor_match` - no configured extractor recognised the message's
    metadata, so nothing was recorded for it; `extractors` is the configured
    list that was asked.
  * `:malformed_header` - the message carried the pathway header but it could
    not be decoded, so the consumer started a new pathway; a message with no
    header at all is the ordinary start of a new pathway and is not reported.
    `extractors` is `nil`.
  * `:aggregator_not_running` - a checkpoint was handed to
    `DdDataStreams.Aggregator.add/1` while no aggregator was running (the
    library disabled at boot, or between a crash and its restart);
    `extractors` is `nil`.

* `[:dd_data_streams, :handler, :error]`, once per failure: one per failing
  extractor, one per edge the library cannot use, and one per message or batch
  the handler could not record. The measurements are `%{count: 1}` and the
  metadata is `%{event: event, extractor: extractor, reason: reason,
  stacktrace: stacktrace}`, with `nil` for the keys that do not apply. The
  reason is the exception that was raised, `:unusable_edge` for an edge the
  library cannot use, or `{kind, reason}` for a call that threw or exited.

The HTTP request is best effort and is not retried.

### Interoperability

The pathway travels in the `dd-pathway-ctx-base64` header by default, which is
the same format dd-trace-go, dd-trace-java and dd-trace-py use: the 64-bit FNV-1
hash of the node in little-endian byte order followed by two zigzag varints of
Unix milliseconds, base64 encoded. Headers written by those tracers decode
here, and headers written here decode there, for every timestamp a real clock
produces: the widest varint the format allows (timestamps of 2^56 milliseconds)
diverges. dd-trace-py reads it back exactly, but dd-trace-go's decoder
overflows its int64 millisecond-to-nanosecond conversion and reports
4611686018427, and dd-trace-java clamps the start to the largest signed long,
so neither re-encodes the same bytes. That divergence is recorded in
`test/fixtures/interop/reference_decoders.txt` rather than corrected. The
claims here were verified against dd-trace-go v2.11.2, dd-trace-java 1.67.1 and
dd-trace-py 3.17.0. A malformed header is ignored and the message starts a new
pathway.

For backwards compatibility, when the configured key is absent the library also
reads a legacy binary header under `dd-pathway-ctx`, decoding its raw
(non-base64) bytes. The configured key always wins when present.

The header key is configurable with `header:` / `DD_DATA_STREAMS_HEADER`, but the
value format is the official one; changing the key stops the official tracers
from finding the pathway.

The Agent payload matches the `StatsPayload` shape of dd-trace-go v2.11.2, and the
DDSketch bins use the same mapping as dd-trace-go v2.11.2 and
  dd-trace-java 1.67.1 (gamma 1.015625, offset 1.8761281912861705), which is
  self-describing in the payload; dd-trace-py uses offset 1338.5, so its bins do
  not line up with Go/Java.

For the Pub/Sub extractor there is one known naming difference: dd-trace-py
tags a consume edge with `topic:<subscription>` instead of
`subscription:<subscription>`, and reports the full resource path rather than
the short id. Both change the node's identity, so a Py consumer appears as a
different node rather than joining the one reported here; the pathway itself
still connects, because the link between two nodes is the hash carried on the
message. dd-trace-java reduces the resource name to its short id on both the
subscription and the topic, which is what this library reports.

#### Node hash

The node hash is FNV-1 over the service, environment, primary tag and the
hashable edge tags. The hashable set here is the same as dd-trace-java 1.67.1's
(`bus`, `direction`, `exchange`, `topic`, `type`, `subscription`,
`kafka_cluster_id`), but the tags are sorted rather than written in Java's
fixed order. dd-trace-go v2.11.2 hashes a different set (it includes `group`,
`event_type` and `segment_name` and omits `bus` and `subscription`) and adds
process tags and an optional container hash by default. dd-trace-py 3.17.0
hashes every tag, including per-message ones such as `partition`. None of the
four agree for an arbitrary set of tags, and Go and Java add runtime-specific
process tags, so numeric equality between tracers is not attainable; the
library does not try. The tags that are not hashed are still reported on the
edge, and those that split a node's statistics do so without changing its
identity.

### Configuration

Every value is read when it is needed, never at compile time: from the
application configuration below when that key is set, and from the process
environment otherwise. Application configuration wins over the environment, and
an unset, empty or blank value counts as unset, so one setting can be given in
either place.

#### Environment variables

| Variable | Default | Meaning |
| --- | --- | --- |
| `DD_DATA_STREAMS_ENABLED` | unset, on | Turns the library off when it reads as false: `false`, `0`, `no`, `off`. |
| `DD_SERVICE` | `unnamed-elixir-service` | The service name reported for this pipeline. |
| `DD_ENV` | empty | The deployment environment, sent as-is (the library does not normalize it). |
| `DD_VERSION` | empty | The application version, reported as `Version` in the payload. |
| `DD_PRIMARY_TAG` | empty | The primary tag, hashed into every node identity. |
| `DD_DATA_STREAMS_HEADER` | `dd-pathway-ctx-base64` | Attribute/header key the pathway travels under; the producer and consumer must use the same value. |
| `DD_TRACE_AGENT_URL` | unset | Agent base URL; wins over host and port when set and non-empty. |
| `DD_AGENT_HOST` | `localhost` | Agent host, when `DD_TRACE_AGENT_URL` is unset. |
| `DD_TRACE_AGENT_PORT` | `8126` | Agent port, when `DD_TRACE_AGENT_URL` is unset. |

Unix domain sockets are not supported. The Agent adds the API key itself, so no
key is configured here and no authentication header is sent.

#### Application configuration

Each environment variable in the table above has a key of the same meaning
here, and the key wins when both are set:

```elixir
config :dd_data_streams,
  # Defaults to true. A boolean here wins over DD_DATA_STREAMS_ENABLED.
  enabled: true,
  service: "checkout",
  env: "prod",
  version: "1.2.3",
  primary_tag: "region:eu",
  # The attribute/header key `produce/3` returns and `consume/1` reads.
  header: "dd-pathway-ctx-base64",
  # A port is accepted as a string or an integer.
  agent_url: "http://localhost:8126",
  agent_host: "localhost",
  agent_port: 8126,
  # Defaults to [DdDataStreams.Extractor.GooglePubSub].
  extractors: [MyApp.KafkaExtractor, DdDataStreams.Extractor.GooglePubSub],
  # Optional: names the service per checkpoint instead of `service`. See
  # "Service name" below.
  service_name: MyApp.ServiceNames,
  # How many services' payloads one flush posts at once. Defaults to 8.
  flush_concurrency: 8,
  # Defaults to DdDataStreams.Transport.Req when Req is available.
  transport: MyApp.AgentTransport,
  # Defaults to DdDataStreams.Backend.Aggregator. Tests can use
  # DdDataStreams.Testing.Backend instead.
  backend: DdDataStreams.Backend.Aggregator
```

When each setting is read:

| Setting | Read when |
|---|---|
| `enabled` | at attach, produce and consume time, and once when the application starts (for the aggregator) |
| `service`, `env`, `primary_tag` | per checkpoint; `service` and `env` again when building the payload |
| `service_name` | per checkpoint, when the caller passed no `:service` option of its own |
| `version` | when building the payload (on flush) |
| `header` | on every produce and consume call |
| `extractors` | on attach, on every consume, and on produce only as a fallback |
| `agent_url`, `agent_host`, `agent_port` | on every flush |
| `flush_concurrency` | on every flush |
| `transport` | on every flush, and once when the application starts (for the default) |
| `backend` | on every produce, consume and checkpoint call, and in the Broadway handlers |

So changing any of them at runtime with `Application.put_env/3` or
`System.put_env/2` affects the next use, except that a runtime change cannot
start or stop the aggregator; see below. When the transport key is absent,
`DdDataStreams.Transport.Req` is used if Req is available.

#### Turning it off

The library is on as soon as you depend on it: adding the dependency and calling
`attach/0` is the opt-in. Datadog's other clients default the other way and
require `DD_DATA_STREAMS_ENABLED=true`, but their tracer is already running and
data streams monitoring is one more product on top of it, while this library does
nothing else.

The flag stays supported, as a kill switch an operator can flip without a
release. Two ways to set it, in this order:

* `config :dd_data_streams, enabled:` when it is a boolean. This wins, which is
  what makes `config/test.exs` a sensible place to keep the aggregator out of a
  test suite.
* otherwise `DD_DATA_STREAMS_ENABLED`, read as false from `false`, `0`, `no` or
  `off` (case-insensitive). Unset, empty or anything else leaves the library on.

The flag is read in two different places, and the difference matters:

* `attach/0`, and the default backend on every `produce/3`, `produce_batch/3`,
  `consume/1` and `checkpoint/1` call, read it on every call, so they follow a
  runtime change: attaching after boot installs handlers, and disabling stops
  producing the pairs. A backend other than the default decides for itself; the
  owner-scoped test backend, for example, ignores this flag.
* `DdDataStreams.Application` reads it once, when the application starts. The
  aggregator is only started if the library was enabled at boot; turning the
  flag on later makes the handlers record into an aggregator that is not
  running, and turning it off later leaves the aggregator running but stops the
  recording until the flag is on again.

Restart the application to change whether the aggregator runs.

## Limitations

* A message is named by the first extractor that recognises its metadata, and a
  produce by the extractor that consumed the message being answered. A stage
  that produces without consuming, or that publishes to a different bus than it
  consumed from, says which extractor by passing `extractor:` to
  `DdDataStreams.produce/3`.
* Google Pub/Sub is the only shipped, tested extractor. Every other broker
  needs an extractor module of your own before a consume checkpoint is
  recorded (see [Broker support](#broker-support)).
* This is Data Streams Monitoring only: no APM traces, backlogs or transactions
  are collected.
* The Agent must be reachable over HTTP; Unix domain sockets are not supported.
* A message whose metadata names no edge records no consume checkpoint and
  leaves no pathway behind for a later produce to continue.
* Fan-in is sampled, not joined: a batch that consumed several pathways
  continues exactly one of them.

## Requirements and troubleshooting

* The Datadog Agent must be version 7.34 or later.
* The Agent must have an API key configured in its own config or environment;
  the library sends none.
* A `500` response means the Agent's forwarder is off (no API key to forward
  with); a `4xx` means the intake rejected the payload.
* The HTTP request is best effort and is not retried.
* Stats are flushed about every ten seconds, so telemetry
  (`[:dd_data_streams, :flush, :error]`) reports a failure at that cadence.

## Development

```sh
mix deps.get
mix test          # default tier: the offline protocol core
mix ci            # format, compile, credo, cover over the default tier
mix docs          # module documentation

just up           # start every broker the tiers use, and wait for health
just down         # stop them and drop their state
```

The suite has tiers for the parts that need something running. They are excluded
by default, so a plain `mix test` cannot report them as passing while nothing is
listening.

### `sqs`, `kafka`, `rabbitmq`

One tier per broker, each a real Broadway pipeline on the real client:
`test/integration/sqs_broker_test.exs` on goaws,
`test/integration/kafka_broker_test.exs` on Apache Kafka and
`test/integration/rabbitmq_broker_test.exs` on RabbitMQ. Each publishes a
pathway into the broker, consumes it through the pipeline, and reads the pairs
`produce/3` returns back off a second queue or topic, so the metadata the client
builds, the carrier and the tags are all checked against a running broker.

```sh
just sqs          # start the container, run `mix test --only sqs`, stop it
just kafka
just rabbitmq
just brokers      # all three, one after another
```

A recipe stops its container only when the tier passes, so a failure leaves the
broker running with its logs (`docker compose logs sqs`); `just down` cleans up.
The images are pinned in `compose.yaml`, because a tier asserts the shape one
client version produces against one broker version.

Each tier is `mix test --only <tag>` underneath, so they also run against a
broker you already have: `DSM_KAFKA_HOST`, `DSM_SQS_ENDPOINT` and
`DSM_RABBITMQ_HOST` / `DSM_RABBITMQ_PORT` point a tier elsewhere.

### `emulator`

A receive/publish round trip against the Google Pub/Sub emulator over its REST
API, with the message handed to the library in the shape
`BroadwayCloudPubSub` v2 produces. It exercises the broker side: resource-name
subscriptions, attributes surviving a publish, the receive time.

```sh
docker run --rm -d --name dsm-e2e-emulator \
  -p 127.0.0.1:8085:8085 \
  gcr.io/google.com/cloudsdktool/cloud-sdk:582.0.0-emulators \
  gcloud beta emulators pubsub start --project=dsm-test --host-port=0.0.0.0:8085

DSM_PAYLOAD_DECODER=e2e/decoder/decode \
  PUBSUB_EMULATOR_HOST=127.0.0.1:8085 \
  mix test --only emulator
```

`DSM_PAYLOAD_DECODER` is optional: it hands the flushed body to the reference
decoder under `e2e/decoder/decode` (set up as `e2e/README.md` describes) and the
tier skips that check when it is unset. Stop the emulator when you are done:

```sh
docker stop dsm-e2e-emulator
```

### `agent`

One flushed payload posted to `<DD_TRACE_AGENT_URL>/v0.1/pipeline_stats`. The
Agent proxies that route to the pipeline stats intake, so the status code is the
intake's own answer: `202` means the backend accepted for processing, `500` means
the Agent has no API key to forward with, and a `4xx` means the intake rejected
the payload. This is the tier that validates the wire format.

From the repository root, start the stub:

```sh
elixir e2e/stub_agent.exs
```

In a second shell, also from the repository root:

```sh
DD_TRACE_AGENT_URL=http://127.0.0.1:8126 mix test --only agent
```

`e2e/stub_agent.exs` answers `202` to any request, so the tier also runs on a
machine with no Datadog Agent. That proves the transport and nothing about the
payload: a stub accepts any body by design, so the schema is only checked by a
real Agent with an API key.

### The end-to-end harness

`e2e/README.md` describes it: a separate Mix project running a real Broadway
pipeline on `BroadwayCloudPubSub.Pull.Producer` against the same emulator, with a
capture transport instead of the Agent and a reference decoder that reads the
body back.

### Checking the transport against another Req

The library supports any Req `~> 0.5`, but this suite can only resolve Req
`0.7.x`, because `broadway_sqs` requires `~> 0.7.2`. To check a change to
`DdDataStreams.Transport.Req` against another version of Req, pin it in a
throwaway project that depends on the library by path:

```sh
mix new /tmp/req_check && cd /tmp/req_check
# in mix.exs: deps do [{:dd_data_streams, path: "/path/to/dd_data_streams"}, {:req, "== 0.5.0"}] end
mix deps.get
mix run -e 'IO.inspect(DdDataStreams.Transport.Req.post("http://127.0.0.1:1/", [], "x"))'
```

A Req that accepts every option the transport passes answers `{:error,
%Req.TransportError{}}` - the connection to port 1 is refused, which happens
after option validation; one that does not know an option fails with
`%ArgumentError{}` instead.

## License

Beerware. See [LICENSE](LICENSE).