The consume and produce checkpoints of a message, without knowing the broker.

consume/3 continues the pathway a received message carries in its configured pathway header, or starts a new one when the header is absent or unusable, and measures the consume edge. produce/3 measures the produce edge and returns the checkpointed pathway together with its checkpoint; it does not encode, send or mutate anything. Neither function sends anything: each returns its checkpoint for DdDataStreams.Aggregator.add/1. consume/3 reads DdDataStreams.Config for the service, environment, primary tag and header key, so the module is not strictly pure, but it performs no I/O and sends nothing, and stays callable from anywhere in a pipeline.

The edge tags are the caller's: this module records exactly the tags it is given, in the order it is given them. DdDataStreams builds them, from the configured broker type and the name the message metadata carries.

Messages

consume/3 takes the parts of an inbound message the checkpoints need, as a map:

• :data - the message body, required
• :attributes - the message attributes, as a map: nil when the broker sent none
• :headers - the message headers, as a list of {key, value} pairs, for a producer that exposes headers instead of attributes; used when :attributes is absent
• :message_attributes - the SQS message attributes, as the map BroadwaySQS exposes under message.metadata.message_attributes; the pathway is read from its _datadog JSON attribute
• :ordering_key - the ordering key, nil or "" when it is unset
• :pathway_start - the broker's publish time for a message that starts a pathway, in native time units; see Latency below
• :edge_start - the same publish time taken as the consume edge's start

These are the values BroadwayCloudPubSub v2 exposes as message.data, message.metadata.attributes and message.metadata.orderingKey, and the shape BroadwayKafka messages take once their message.metadata.headers list is passed as :headers. Extra keys are left alone. A message that carries headers is read through them, and headers take precedence over attributes; a header this library cannot read is measured as empty. Otherwise an attributes value that is neither a map nor a list is read as an empty attributes map, and a list of attributes is read as header pairs.

produce/3 takes no message: the caller measures the outgoing payload, or lets DdDataStreams.produce/3 measure its data.

Payload size

The consume size in bytes is the data, the ordering key, and every attribute's or header's key and value. nil values count as zero. It is measured before the pathway is read, so it includes the inbound header (about 49 bytes).

The produce size is the caller's: DdDataStreams.produce/3 passes the :payload_size option when given, or the size of data otherwise, and adds the size of the pairs it is about to write - the configured pathway key plus its encoded value, and the dd-topic key plus its value. Those pairs are the library's own propagation attributes, not user attributes. 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. A caller that measures its message itself measures it without those pairs.

Latency

consume/3 checkpoints at received_at, in native time units, so the consume edge latency is the time the message spent waiting in the bus rather than the time it spent queued inside Broadway. A message that carried no pathway starts at received_at, so both latencies are zero, unless it also carries the broker's publish time (:pathway_start / :edge_start): a first-hop message then starts there and reports publish to consume latency, while a message that carried a pathway header keeps the start the header encodes. A publish time before 2000-01-01 or more than a minute ahead of this node's clock is dropped rather than believed, so a garbage broker timestamp cannot report a decades-long latency or a start in the future. produce/3 checkpoints at the current time. Times are Unix milliseconds after conversion.

Pathways

A pathway with hash zero has never been checkpointed, and its first checkpoint starts it at the checkpoint time. Malformed input is not an error: an undecodable, truncated or non-base64 header is ignored and the message is consumed on a new pathway, as dd-trace-java does. A header that is present but cannot be decoded is reported as a [:dd_data_streams, :checkpoint, :skip] telemetry event with reason :malformed_header, so a consumer can tell a header that was lost or broken on the way from the ordinary start of a new pathway.

produce/3 returns the pathway the caller encodes into the message. When one consumed message produces several messages, as a batch producer does, pass the received pathway to every produce/3 call: each produce then has the consume hash as its parent and all of them share one produce node, instead of chaining produce edges onto each other.