DdDataStreams

Datadog Data Streams Monitoring for Elixir pipelines running on 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 says exactly what ships.

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:

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), 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 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 is the short list, and Environment variables has every variable with its default.

3. Attach

Attach once at start up, before Broadway starts:

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), 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:

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), 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, a client that takes a header list under Producing from a Broadway pipeline, the SQS _datadog attribute under SQS wrapping, and a stage publishing to a different bus than it consumed from under 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 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), and a pipeline that does not run on Broadway drives the same two checkpoints itself (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 - it reports how long the message waited in the bus. When nothing shows up, 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.

Ground truth is the tree itself: lib/dd_data_streams/extractor/ lists every shipped extractor, lib/dd_data_streams/carrier/ every shipped carrier, and 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 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 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.

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.

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:

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

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:

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:

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.

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:

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, and the default extractor for Pub/Sub is described under Google Pub/Sub.

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.

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 makes:

# 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:

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, 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:

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 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:

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:

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:

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.

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

# 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:

{: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:

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:

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:

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

# 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:

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:

# 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:

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:

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.

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) 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

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:

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:

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).
• 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

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.

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.

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:

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:

elixir e2e/stub_agent.exs

In a second shell, also from the repository root:

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:

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.