@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.

@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.

@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.

@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.

@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 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.

@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 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.

@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.