DdDataStreams.Payload (DD Data Streams v0.3.0)

Copy Markdown View Source

The MsgPack body the Datadog Agent forwards to the pipeline stats intake.

The wire shape matches StatsPayload in dd-trace-go v2.11.2, so the field names, their order and their types are the ones the intake already accepts from the official tracers:

  • the payload carries Env, Service, Stats, TracerVersion, Lang, Version, ProcessTags and ProductMask
  • a bucket carries Start and Duration in Unix nanoseconds, its Stats points, Backlogs, Transactions and TransactionCheckpointIds
  • a point carries EdgeTags, Hash, ParentHash, the three DDSketch distributions and TimestampType

The sketches are bin on the wire, not str: the reference decoder reads them with a bytes method and rejects a string. Integers are written in the smallest unsigned form and strings in the smallest string form, because the reference encoder does the same.

Backlogs and transactions are part of the shape but are always empty here; nothing in this library collects them.

Summary

Types

bucket()

A bucket: its start in Unix nanoseconds, its width in nanoseconds and its points. :current and :origin buckets differ only in :start and in the points' :timestamp_type.

point()

A point: the node's edge tags, its hash and parent hash, and the three distributions. Latencies are in seconds and the payload size in bytes; the distributions are the DdDataStreams.Sketch.to_proto/1 blobs.

t()

Functions

add_bucket(payload, bucket)

Appends a bucket to the payload.

encode(payload)

Encodes the payload as MsgPack iodata.

new()

Returns an empty payload for the service configured in the environment.

Types

bucket()

@type bucket() :: %{
  optional(:start) => non_neg_integer(),
  optional(:duration) => non_neg_integer(),
  optional(:stats) => [point()]
}

A bucket: its start in Unix nanoseconds, its width in nanoseconds and its points. :current and :origin buckets differ only in :start and in the points' :timestamp_type.

point()

@type point() :: %{
  optional(:edge_tags) => [String.t()],
  optional(:hash) => non_neg_integer(),
  optional(:parent_hash) => non_neg_integer(),
  optional(:pathway_latency) => binary(),
  optional(:edge_latency) => binary(),
  optional(:payload_size) => binary(),
  optional(:timestamp_type) => String.t()
}

A point: the node's edge tags, its hash and parent hash, and the three distributions. Latencies are in seconds and the payload size in bytes; the distributions are the DdDataStreams.Sketch.to_proto/1 blobs.

t()

@type t() :: %DdDataStreams.Payload{
  env: String.t(),
  lang: String.t(),
  process_tags: [String.t()],
  product_mask: non_neg_integer(),
  service: String.t(),
  stats: [bucket()],
  tracer_version: String.t(),
  version: String.t()
}

Functions

add_bucket(payload, bucket)

@spec add_bucket(t(), bucket()) :: t()

Appends a bucket to the payload.

encode(payload)

@spec encode(t()) :: iodata()

Encodes the payload as MsgPack iodata.

new()

@spec new() :: t()

Returns an empty payload for the service configured in the environment.

The identity fields are read from DdDataStreams.Config, so the aggregator can build the payload without passing them around. :stats is filled by add_bucket/2.