The pathway header codec.
The convenience arities, put/2 and get/1, use the configured key
DdDataStreams.Config.header/0, whose default value is
dd-pathway-ctx-base64, the name the official Datadog tracers use; the
explicit-key arities, put/3 and get/2, use the key they are given. For
backwards compatibility with older producers,
get/2 also falls back to the legacy binary dd-pathway-ctx key, decoding
its raw (non-base64) bytes, when the configured key is absent.
The payload is the 64-bit unsigned hash in little-endian byte order followed by the pathway start and the edge start as zigzag-encoded varints of Unix milliseconds, with no separator or length prefix, and the whole thing is standard base64. The timestamps are length-parsed rather than assumed to be a fixed width, so a header written by an official tracer decodes here and a header written here decodes there for every timestamp a real clock produces; the widest varint the format allows diverges, and that divergence is recorded in the interoperability fixtures. Verified against dd-trace-go v2.11.2, dd-trace-java 1.67.1 and dd-trace-py 3.17.0.
A malformed or missing header is not an error condition for a pipeline:
get/1 returns nil and the consumer starts a new pathway. fetch/1 tells
the two apart, so a consumer that wants to report a header it could not read
can.
Next to the pathway header, a producer writes the outgoing topic under
dd-topic (topic_header/0) whenever the extractor named one. It is not part
of the pathway codec: it is the tag a consumer whose metadata names only a
subscription reads to report the queue the pathway crossed.
Summary
Functions
Decodes a raw header.
Decodes a base64 header string.
Encodes a pathway as the raw header bytes.
Encodes a pathway as a base64 header string.
Reads the pathway out of a message's headers under key, like get/2, but
tells a header that is absent from one that is present and cannot be read.
Reads the pathway out of a message's headers under key. get/1 defaults
to the configured key DdDataStreams.Config.header/0, whose default value is
dd-pathway-ctx-base64.
The legacy binary header key, "dd-pathway-ctx".
Puts the pathway into a message's headers under key. put/2 defaults to
the configured key DdDataStreams.Config.header/0, whose default value is
dd-pathway-ctx-base64.
The attribute or header key the producer writes the outgoing topic under,
"dd-topic".
Functions
@spec decode(binary()) :: {:ok, DdDataStreams.Pathway.t()} | :error
Decodes a raw header.
Bytes after the two timestamps are ignored, and a varint need not be encoded in its shortest form. The ninth byte of a varint ends it even when its continuation bit is set.
A start outside the plausible window makes the whole header unusable rather
than trustworthy: get/1 then yields nil and the consumer starts a new
pathway, as it does for every other malformed header.
@spec decode_str(binary()) :: {:ok, DdDataStreams.Pathway.t()} | :error
Decodes a base64 header string.
@spec encode(DdDataStreams.Pathway.t()) :: binary()
Encodes a pathway as the raw header bytes.
@spec encode_str(DdDataStreams.Pathway.t()) :: String.t()
Encodes a pathway as a base64 header string.
@spec fetch(map() | [tuple()] | nil) :: {:ok, DdDataStreams.Pathway.t()} | :absent | {:error, :malformed}
Reads the pathway out of a message's headers under key, like get/2, but
tells a header that is absent from one that is present and cannot be read.
Answers {:ok, pathway}, :absent when neither the configured key nor the
legacy dd-pathway-ctx one is present, or {:error, :malformed} when a
header is present but is not a usable base64 pathway: not decodable,
missing the hash and two timestamps, or carrying a start outside the window
decode/1 believes. A consumer starts a new pathway either way, but a
producer writing a header this consumer cannot read would otherwise be
indistinguishable from one that sent no header at all. A header under any
other key is indistinguishable from an absent one.
@spec fetch(map() | [tuple()] | nil, String.t()) :: {:ok, DdDataStreams.Pathway.t()} | :absent | {:error, :malformed}
@spec get(map() | [tuple()] | nil) :: DdDataStreams.Pathway.t() | nil
Reads the pathway out of a message's headers under key. get/1 defaults
to the configured key DdDataStreams.Config.header/0, whose default value is
dd-pathway-ctx-base64.
The headers are an attribute map or a list of tuples whose first element is
the header key and whose value is its last element. Returns nil when the
header is absent, is not a valid base64 header, does not contain a hash and
two timestamps, carries a start outside the window decode/1 believes, or is
carried in a shape no entry can be read from.
When key is absent from the carrier, a legacy binary header stored under
dd-pathway-ctx is tried instead and decoded as raw bytes via decode/1.
The configured key always wins when it is present.
@spec get(map() | [tuple()] | nil, String.t()) :: DdDataStreams.Pathway.t() | nil
@spec legacy_header() :: String.t()
The legacy binary header key, "dd-pathway-ctx".
@spec put(map() | [tuple()], DdDataStreams.Pathway.t()) :: map() | [tuple()]
Puts the pathway into a message's headers under key. put/2 defaults to
the configured key DdDataStreams.Config.header/0, whose default value is
dd-pathway-ctx-base64.
The headers are either an attribute map or a list of tuples whose first
element is the header key, as BroadwayKafka exposes them; an existing entry
under the same key is replaced in both, keeping whatever other elements that
entry carries. A pathway that has not been checkpointed (hash zero) carries
no information and is not written. The headers are returned unchanged so a
produce path can call this unconditionally.
@spec topic_header() :: String.t()
The attribute or header key the producer writes the outgoing topic under,
"dd-topic".
It travels next to the pathway header, so a consumer whose broker metadata names only the subscription can still report the topic the message crossed.