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

Data Streams Monitoring for Broadway pipelines.

Datadog's 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.
Doing that needs a checkpoint on each side of every edge and a pathway carried
on the message itself. This module supplies the Broadway glue that decides
where a checkpoint belongs; the pathway hashing, propagation, aggregation and
transport live in the modules it calls.

Nothing here is specific to one broker, or to one broker per application. The
library hangs itself on the telemetry events Broadway already emits, so the
pathway and propagation layer works with any producer, and the parts that are
broker-specific - what a message's metadata says about the edge it arrived
on, and which technology to report - are modules behind
`DdDataStreams.Extractor`:

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

A message is named by the first configured extractor that recognises its
metadata, so an application can run a Kafka pipeline and a Pub/Sub pipeline
side by side. One extractor ships today -
`DdDataStreams.Extractor.GooglePubSub` - so tagging a message from any other
broker needs an extractor module of your own; the carriers `Flat` and `Sqs`
ship as well.

An extractor names the edge, not the service the checkpoint is reported under:
that is `DdDataStreams.Config.service/0`, unless a
`DdDataStreams.ServiceName` module is configured, which names it per message.

## Using it in a Broadway pipeline

A Broadway pipeline reaches the library through telemetry instead of through a
dependency, so attaching is the whole integration:

    # at start up, before Broadway does:
    DdDataStreams.attach()

    # in handle_message/3 or handle_batch/4:
    def handle_message(_processor, message, _context) do
      data = encode(message.data)

      attributes = existing_attributes

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

      MyApp.Publisher.publish("orders-out", data, attributes: attributes)
      message
    end

`produce/3` records the produce checkpoint and returns the attributes to add
to the outgoing message - the pathway header and the topic - without
publishing. `produce_batch/3` does the same for several payloads published to
one topic and returns the same attributes for the whole batch.

`attach/0` installs two telemetry handlers:

  * `[:broadway, :processor, :message, :start]`, which Broadway emits once for
    each message it is about to hand to `handle_message/3`. The handler
    records the consume checkpoint - one per message that reaches
    `handle_message/3`, including a leaf consumer that produces nothing - and
    leaves the resulting pathway in `DdDataStreams.Context` for a produce
    call in the same stage process.
  * `[:broadway, :batch_processor, :start]`, once per batch. The handler
    samples one of the pathways the batch's messages were consumed on and
    leaves that in `DdDataStreams.Context`, so a produce call in
    `handle_batch/4` continues a pathway that connects to the messages being
    answered. Fan-in is a sample rather than a join, which is what
    dd-trace-go v2.11.2 does in `Merge`/`MergeContexts`: it keeps one pathway.

Both handlers record only what each event is: the batch handler adds no
consume checkpoint, because every message of the batch already got one at its
own message event.

A message that never reaches `handle_message/3` therefore gets no consume
checkpoint. That happens when `prepare_messages/2` raises: Broadway marks
every message of the batch failed before the per-message start event
(`deps/broadway/lib/broadway/topology/processor_stage.ex`, the
`maybe_prepare_messages/2` wrapper), so the start event never fires and a
raise silences the checkpoint for the whole batch, not just the message that
raised. Marking a message failed with `Broadway.Message.failed/2` is a
different case: those messages still reach `handle_message/3`, so they still
checkpoint, and failed messages take Broadway's failed-ack path either way.
A redelivery is a new attempt and counts a new checkpoint; the wait is still
measured from the receive time, so per-attempt counting does not distort it.

## Messages the library cannot record

The handlers never raise. Telemetry detaches a handler that raises, for every
pipeline in the VM, and never reattaches it, so a single malformed message
would stop monitoring for the whole application. A message the library cannot
record is skipped instead: the pathway and the extractor of the previous
message are cleared, so a later produce does not continue them, and the reason
is logged and reported as a `[:dd_data_streams, :handler, :error]` telemetry
event.

## Pathways that are not a Broadway stage

`consume/1` and `produce/3` are the same two checkpoints as functions, for a
pipeline that drives the library itself, and `checkpoint/2` records one for a
stage that is neither a consume nor a produce.

Checkpoints are handed to the configured `DdDataStreams.Backend`, which drops
them when the calling process is not recording (the default backend delegates
to `DdDataStreams.Aggregator.add/1`, which does nothing when the aggregator
is not running), so none of these needs to be guarded at the call site.

# `propagation`

```elixir
@type propagation() :: {String.t(), String.t()}
```

The metadata key and base64 pathway value to add to an outgoing message, as
returned by `produce/3` and `produce_batch/3`.

The key is `DdDataStreams.Config.header/0`, `"dd-pathway-ctx-base64"` unless
configured; the value is the encoded pathway.

# `attach`

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

Installs the Broadway telemetry handlers that record consume checkpoints and
fan-in pathways.

Call it once, when the application starts, before Broadway starts. It does
nothing when the library is disabled, and calling it again is harmless:
telemetry keeps one handler per id and the second call returns `:ok` instead
of failing, so a hot code upgrade does not double the checkpoints.

Once installed, a handler records only when the configured
`DdDataStreams.Backend` says the emitting process is recording, so a test
suite can attach once at boot and leave the handlers attached: a test that
did not opt in emits events the handlers turn into no-ops. Only tests that
attach or detach themselves, or that use the global aggregator testing API,
still need `async: false`.

Which broker a message comes from, and which edge tags it is reported under,
is the configured `DdDataStreams.Extractor`'s business. Attaching fails with an
`ArgumentError` when `config :dd_data_streams, extractors:` is not a non-empty
list of extractors, rather than leaving a handler attached that cannot name an
edge. The extractors are validated even while the library is disabled, so a
configuration mistake raises when the application starts instead of waiting
for the flag to be flipped.

# `checkpoint`

```elixir
@spec checkpoint([String.t()]) :: DdDataStreams.Pathway.t()
```

Records a checkpoint for the stage reached by `tags` and returns the
checkpointed pathway. The checkpoint has already been given to the
aggregator, so it is not returned; use `Pathway.set_checkpoint/3` if you
need the measurement itself.

`checkpoint/1` defaults `pathway` to `DdDataStreams.Context.get/0` (or a new
pathway when there is none); `checkpoint/2` takes it explicitly, for a stage
that wants to continue a pathway it already holds. `tags` are the edge tags
reported for this stage, such as `["direction:in", "type:my-queue"]`.
`DdDataStreams.Tags`
decides what each key does with them: some are hashed into the node, and
others, such as `group:` or `manual_checkpoint:true`, split one node's
statistics without splitting its identity. A key the vocabulary does not know
is reported on the edge and does neither.

The checkpoint is stamped with the current time, has no payload size, and is
given to the aggregator. The caller's context is left alone: this is a
checkpoint in the middle of a pathway, not a new stage to produce from.

# `checkpoint`

```elixir
@spec checkpoint(DdDataStreams.Pathway.t(), [String.t()]) :: DdDataStreams.Pathway.t()
```

# `consume`

```elixir
@spec consume(map()) :: DdDataStreams.Pathway.t() | nil
```

Records the consume checkpoint of a message, and leaves the checkpointed
pathway in `DdDataStreams.Context`.

`message` is a Broadway message, or a map shaped like one: the body in
`:data` and the producer's own metadata map in `:metadata`. The configured
extractor names the edge from that metadata and hands back the parts a
checkpoint measures, so a pipeline that drives the library itself only has to
build the metadata the producer would have built.

Returns `nil` when the extractor finds no edge in the metadata - a message
with no subscription, no topic, nothing to report it under - and records
nothing in that case; the miss is reported as a
`[:dd_data_streams, :checkpoint, :skip]` telemetry event whose `:extractors`
metadata is the configured list that was asked. It also clears the process
context, so a later
`produce/3` cannot chain onto the previous message's pathway. An extractor
that raises is treated the same way, and
the reason is logged and reported as a `[:dd_data_streams, :handler, :error]`
telemetry event. Otherwise returns the pathway the checkpoint reached. The
checkpoint has already been given to the aggregator, so it is not returned;
use `DdDataStreams.Checkpoints.consume/3` if you need the checkpoint to
record or forward it yourself, and do not add it to this aggregator again.

# `detach`

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

Removes the telemetry handlers `attach/0` installed, so recording stops
without restarting the application.

Broadway events are no longer recorded afterwards; `consume/1`, `produce/3`
and `checkpoint/1` keep working as plain functions while the library is
enabled, and checkpoints already handed to the aggregator are still flushed.

Calling it when nothing is attached returns `:ok` too, so a test can end with
it unconditionally, and `attach/0` and `detach/0` may be interleaved freely.

# `produce`

```elixir
@spec produce(String.t(), term(), keyword()) :: [propagation()]
```

Records the produce checkpoint for `data` published to `topic` and returns the
`{key, value}` pairs to add to the outgoing message's attributes or headers.

It does not publish. The caller merges the pairs into the same metadata the
receiver will read:

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

    MyApp.PubSub.publish("orders-out", data, attributes: attributes)

The pathway key is `DdDataStreams.Config.header/0`, `"dd-pathway-ctx-base64"`
unless configured, and the topic is written under
`DdDataStreams.Propagator.topic_header/0`, `"dd-topic"`. The topic pair is
only returned when the outgoing extractor names a topic, so a consumer can
report the topic its metadata does not carry. The pathway continued is the one
in `DdDataStreams.Context`, so this is meant to be called from the stage that
consumed the message being answered: inside `handle_message/3` it chains the
produce onto that message's consume checkpoint, inside `handle_batch/4` onto
the batch's fan-in pathway. With no context the produce starts a new pathway.
This function does not change the context.

`data` is the outgoing payload. It is only measured, never inspected: a binary
is measured with `byte_size/1`, and anything else counts as zero. The size of
the pairs this function returns is added to it, because they travel with the
message and the consumer measures every attribute it receives: measure the
message without the pairs. Pass `payload_size:` when the payload's size is
known another way, or when the message's own attributes should be counted.

Call it immediately before sending: the checkpoint is recorded when this
function runs, so a send that never happens still reports a produce.
Producing several messages from one consumed message means calling this once
per message, not chaining one call's result into the next, so all of them
share one produce node.

The edge tags come from the extractor that named the consume edge of the
message being answered, which is the one `attach/0` left in the context, so a
pipeline that consumes and produces on the same technology needs nothing more.
That extractor also receives the producer info of the message being answered,
so identity kept in the producer's configuration - a Kafka cluster id, a
RabbitMQ exchange - can be reported on the produce edge; a stage that
produces without consuming passes both fields nil. A pipeline that changes
technology - consuming from Kafka and publishing to
Pub/Sub - names the outgoing one, because the topic alone does not say which
bus it goes to:

    DdDataStreams.produce("orders-out", data, extractor: MyApp.PubSubExtractor)

Options:

  * `:extractor` - the extractor that names the outgoing edge. Defaults to the
    extractor that consumed the message being answered, then to the only
    configured one; a stage that produces without consuming and has several
    extractors must pass it.
  * `:payload_size` - the size in bytes to report, a non-negative integer.
    Defaults to the size of `data`. The pairs this function returns are added
    to it, so pass the size of the message without those pairs.
  * `:service` - the service to report this produce under, overriding the
    configured `DdDataStreams.ServiceName` and `DdDataStreams.Config.service/0`.
    A consume in the same stage does not inherit it: each checkpoint asks for
    its own name.

Any other option is passed to an extractor that implements the optional
`c:DdDataStreams.Extractor.produce/3`, which is how an edge that needs more
than a topic is named - a RabbitMQ publish names its exchange and routing key.
The library neither reads nor validates those options, and an extractor that
implements only `produce/2` never sees them.

Returns `[]` when the library is disabled, and records nothing.

# `produce_batch`

```elixir
@spec produce_batch(String.t(), Enumerable.t(), keyword()) :: [propagation()]
```

Records one produce checkpoint per payload in `datas` and returns the
`{key, value}` pairs for the whole batch.

All the payloads are published to the same `topic`; the pairs are the same for
every message because the batch shares one produce node. Add them to each
message's metadata:

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

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

Each payload records its own checkpoint with its own size, but all of them
have the same hash and timestamp, so the batch is one produce edge. `datas`
is any `Enumerable`; it is materialised once. An empty enumerable returns
`[]`; a disabled library returns `[]`.

Payload size is `byte_size/1` per binary payload, or zero otherwise, plus the
size of the pairs this function returns, as `produce/3` does. There is no
`:payload_size` option here: call `produce/3` once per message when each size
needs to be measured differently.

The `:extractor` option is the same as for `produce/3`, as is any option an
extractor implementing `c:DdDataStreams.Extractor.produce/3` reads. `:service`
names the whole batch, since the batch is one produce node. Call it
immediately before the batch send, for the same reason as `produce/3`.

# `status`

```elixir
@spec status() :: %{
  enabled: boolean(),
  attached: boolean(),
  aggregator_running: boolean(),
  transport: module() | nil,
  agent_url: String.t(),
  header: String.t(),
  service: String.t(),
  env: String.t(),
  service_name: module() | nil,
  extractors: [module()]
}
```

The library's effective configuration and liveness, as one map, answered from
the local VM: no Datadog Agent, no aggregator flush, nothing that can block.

  * `:enabled` - the configured flag. `attach/0` reads it, and so does the
    default `DdDataStreams.Backend.Aggregator`; a configured backend answers
    whether a particular process records for itself.
  * `:attached` - whether the Broadway telemetry handlers are installed.
  * `:aggregator_running` - whether the process that batches checkpoints and
    posts them is running.
  * `:transport`, `:agent_url`, `:header`, `:service`, `:env`,
    `:service_name`, `:extractors` - the effective values of the same
    settings the recording itself reads, as of this call.

"Is it recording?" is all three of `:enabled`, `:attached` and
`:aggregator_running` for the default backend: the flag decides whether new
checkpoints are made, the handlers decide whether Broadway events make any,
and the aggregator decides whether what they make is ever posted. With a
configured backend, ask that backend whether the calling process records.

---

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