A pathway is the chain of nodes a payload passes through: the producer, the broker edge, the consumer, and so on.
hash identifies the current node in that chain. It is a 64-bit FNV-1 hash of
the node's own identity (node_hash/4) chained onto the hash of the node that
sent the payload (pathway_hash/2), so the payload carries the same value on
to the next service. pathway_start and edge_start are the Unix times in
milliseconds at which the pathway and the current edge started; the
difference between them and the time of a checkpoint is the latency reported
for the edge. A pathway_start of 0 is the sentinel for an unstarted
pathway: its first checkpoint starts it at the checkpoint time.
Only tags that identify a node take part in the hash. Tags that vary per
message, such as partition, are reported but never hashed, because hashing
them would split one node into a node per partition and stop the downstream
produce edges from joining up.
tags are the edge tags of the node hash identifies, so the pathway a
consume left behind carries the consume edge's tags and a produce in that
stage can read them. They stay in the VM: DdDataStreams.Propagator writes
the hash and the two starts on the wire, not the tags, so a consumer cannot
read its producer's tags from the header it receives.
Summary
Types
The measurement a checkpoint produces. pathway_latency and edge_latency
are in milliseconds and measure back from the checkpoint's timestamp.
payload_size is the byte size of the message the checkpoint was recorded
for; DdDataStreams.Checkpoints.consume/3 and produce/3 add it for
aggregation, while set_checkpoint/3 leaves it out and the aggregator
treats an absent value as zero. service is added the same way, and is the
service the checkpoint is reported under.
Functions
Samples one pathway from a fan-in, so a batch that consumed several pathways continues exactly one of them. An empty list yields an unstarted pathway.
Builds a pathway from the given fields, defaulting missing ones to zero.
Returns the hash of the node identified by service, env, primary_tag and
tags.
Chains a node hash onto the hash of the node that sent the payload.
Sets a checkpoint for the node reached by tags at :now (milliseconds) and
returns the checkpointed pathway together with the measurement.
Types
@type checkpoint() :: %{ :hash => non_neg_integer(), :parent_hash => non_neg_integer(), :tags => [String.t()], :timestamp => integer(), :pathway_latency => non_neg_integer(), :edge_latency => non_neg_integer(), optional(:payload_size) => non_neg_integer(), optional(:service) => String.t() }
The measurement a checkpoint produces. pathway_latency and edge_latency
are in milliseconds and measure back from the checkpoint's timestamp.
payload_size is the byte size of the message the checkpoint was recorded
for; DdDataStreams.Checkpoints.consume/3 and produce/3 add it for
aggregation, while set_checkpoint/3 leaves it out and the aggregator
treats an absent value as zero. service is added the same way, and is the
service the checkpoint is reported under.
@type t() :: %DdDataStreams.Pathway{ closest_opposite_direction_edge_start: term(), closest_opposite_direction_hash: term(), edge_start: non_neg_integer(), extractor: module() | nil, hash: non_neg_integer(), pathway_start: non_neg_integer(), previous_direction: term(), producer_info: DdDataStreams.Extractor.producer_info() | nil, tags: [String.t()] }
Functions
Samples one pathway from a fan-in, so a batch that consumed several pathways continues exactly one of them. An empty list yields an unstarted pathway.
Builds a pathway from the given fields, defaulting missing ones to zero.
iex> DdDataStreams.Pathway.new(hash: 234)
%DdDataStreams.Pathway{hash: 234, pathway_start: 0, edge_start: 0, extractor: nil}
iex> DdDataStreams.Pathway.new(edge_start: 1700003600000)
%DdDataStreams.Pathway{hash: 0, pathway_start: 0, edge_start: 1700003600000, extractor: nil}@spec node_hash(String.t(), String.t(), String.t() | nil, [String.t()]) :: non_neg_integer()
Returns the hash of the node identified by service, env, primary_tag and
tags.
The fields are concatenated without a separator: the base hash over service,
environment and primary tag, followed by the hashable edge tags in byte order.
Tags outside the hashable vocabulary are dropped; a tag repeated in tags
takes part as many times as it appears.
@spec pathway_hash(non_neg_integer(), non_neg_integer()) :: non_neg_integer()
Chains a node hash onto the hash of the node that sent the payload.
The two hashes are hashed as 64-bit unsigned little-endian integers, node first, so a downstream node can be computed from the parent it received.
@spec set_checkpoint(t(), [String.t()], keyword()) :: {t(), checkpoint()}
Sets a checkpoint for the node reached by tags at :now (milliseconds) and
returns the checkpointed pathway together with the measurement.
The first checkpoint of a pathway starts it at :now, or at the :pathway_start
override. Both starts can be overridden per checkpoint: the consume side uses
that to measure from the time the broker received the message rather than the
time the consumer picked it up.
Options:
:now- checkpoint time in milliseconds, defaults to the current time:service,:env,:primary_tag- identity of the node; seenode_hash/4:pathway_start,:edge_start- override the respective start time
The returned pathway carries tags as the tags of the node it reached, so a
produce in the stage that consumed a message reads the consume edge's tags
from the pathway in DdDataStreams.Context.