DdDataStreams.Config (DD Data Streams v0.3.0)

Copy Markdown View Source

The settings that configure the library.

Every setting is read when it is asked for, never at compile time: from config :dd_data_streams, ... when that key is set, and from the process environment otherwise. Application configuration wins over the environment for every key, as it does for enabled/0, and an unset, empty or blank value counts as unset, so one key can be set in either place. A value changed at runtime - Application.put_env/3 or System.put_env/2 - is picked up by the next checkpoint or flush that reads it.

The names and defaults follow dd-trace-go v2.11.2 and dd-trace-java 1.x, with one deliberate exception: those libraries enable data streams monitoring only when told to, because they do many other things already, while this one does nothing else, so it is on once you depend on it and DD_DATA_STREAMS_ENABLED is a kill switch.

When each setting is read (see also the README table):

  • enabled?/0 - on attach, and through DdDataStreams.Backend.Aggregator on every produce, consume and checkpoint and in the Broadway handlers; once at application start for the aggregator, which then keeps that answer.
  • service/0, env/0, primary_tag/0 - per checkpoint; service/0 and env/0 again when building the payload.
  • service_name/0 - per checkpoint, when the caller passed no :service option of its own.
  • version/0 - when building the payload, on flush.
  • header/0 - on every produce and consume call.
  • extractors/0 (via DdDataStreams.extractors/0) - on attach, on every consume, and on produce only as a fallback.
  • agent_url/1, agent_host/0, agent_port/0 - on every flush.
  • flush_concurrency/0 - on every flush.
  • transport/0 - on every flush, and once at application start for the default.

The aggregator is started by DdDataStreams.Application from enabled?/0 when the application starts and keeps that answer: turning the flag on afterwards installs handlers that record checkpoints into an aggregator that is not running, and DdDataStreams.Aggregator.add/1 drops every one - the first drop in each process is logged at warning level. Turning it off afterwards stops header injection, and the handlers installed earlier stay attached but stop recording because the default backend reads the flag per event; turning it back on resumes them. The aggregator itself keeps running and flushes what it already holds either way. Restart the application to change whether the aggregator runs, or let DdDataStreams.Testing.enable!/1 flip the flag around a single test and start the aggregator for that test.

An application that does not depend on the optional Req brings no default transport: the aggregator starts with none, the start logs an error naming the missing configuration, and every flush is dropped.

agent_url/1 takes the environment as an argument because it is also asked for outside the VM's environment, from tests and from a custom transport; the application configuration is still read from the VM.

Summary

Functions

agent_url()

The base URL of the Datadog Agent, read on every flush: config :dd_data_streams, agent_url: when set, then DD_TRACE_AGENT_URL; otherwise the URL is built from agent_host:/DD_AGENT_HOST and agent_port:/DD_TRACE_AGENT_PORT, defaulting to http://localhost:8126. Unix domain sockets are not supported.

agent_url(environment)

The same as agent_url/0, resolved over the given environment and still letting the application configuration win.

enabled?()

Whether the library is enabled. On by default: depending on the library and calling DdDataStreams.attach/0 is the opt-in.

env()

The deployment environment: config :dd_data_streams, env: when set, DD_ENV otherwise. Empty when neither is.

flush_concurrency()

How many payloads are posted at once, read on every flush: config :dd_data_streams, flush_concurrency:, 8 otherwise.

header()

The attribute or header key the pathway is written under: config :dd_data_streams, header: when set, DD_DATA_STREAMS_HEADER otherwise.

primary_tag()

The primary tag: config :dd_data_streams, primary_tag: when set, DD_PRIMARY_TAG otherwise. Empty when neither is.

service()

The service the pipeline reports itself as: config :dd_data_streams, service: when set, DD_SERVICE otherwise.

service_name()

The module that names the service per checkpoint: config :dd_data_streams, service_name: when set, nil otherwise and every checkpoint is reported under service/0.

transport()

The module the payloads are posted with, read on every flush: config :dd_data_streams, transport:, otherwise DdDataStreams.Transport.Req when the optional Req dependency is available, otherwise nil and every flush is dropped.

version()

The application version: config :dd_data_streams, version: when set, DD_VERSION otherwise. Empty when neither is.

Functions

agent_url()

@spec agent_url() :: String.t()

The base URL of the Datadog Agent, read on every flush: config :dd_data_streams, agent_url: when set, then DD_TRACE_AGENT_URL; otherwise the URL is built from agent_host:/DD_AGENT_HOST and agent_port:/DD_TRACE_AGENT_PORT, defaulting to http://localhost:8126. Unix domain sockets are not supported.

The environment is taken from the VM. agent_url/1 is the same resolution over an environment passed in.

iex> DdDataStreams.Config.agent_url(%{})
"http://localhost:8126"

iex> DdDataStreams.Config.agent_url(%{"DD_AGENT_HOST" => "agent", "DD_TRACE_AGENT_PORT" => "8127"})
"http://agent:8127"

iex> DdDataStreams.Config.agent_url(%{"DD_TRACE_AGENT_URL" => "http://agent:8126"})
"http://agent:8126"

agent_url(environment)

@spec agent_url(map()) :: String.t()

The same as agent_url/0, resolved over the given environment and still letting the application configuration win.

enabled?()

@spec enabled?() :: boolean()

Whether the library is enabled. On by default: depending on the library and calling DdDataStreams.attach/0 is the opt-in.

config :dd_data_streams, enabled: true | false decides when it holds a boolean. A configured string is read the way DD_DATA_STREAMS_ENABLED is, trimmed and case-insensitive, so enabled: System.get_env("DD_DATA_STREAMS_ENABLED") needs no conversion at the call site. Otherwise DD_DATA_STREAMS_ENABLED is read, and it is a kill switch: only a value that reads as false - false, 0, no, off - turns the library off. An unset, empty or unrecognised value leaves it on; an unrecognised configuration is also reported once per process, since the flag is read per produce and a mistake there is otherwise invisible.

DdDataStreams.Application reads this when the application starts, so changing it at runtime does not start or stop the aggregator.

env()

@spec env() :: String.t()

The deployment environment: config :dd_data_streams, env: when set, DD_ENV otherwise. Empty when neither is.

flush_concurrency()

@spec flush_concurrency() :: pos_integer()

How many payloads are posted at once, read on every flush: config :dd_data_streams, flush_concurrency:, 8 otherwise.

One request goes out per service, so a pipeline that reports about eighty services posts them in batches of this size instead of one after another. An integer below 1, or a value that is not an integer, raises an ArgumentError on the flush that reads it: a typo here must not silently slow the flush down.

header()

@spec header() :: String.t()

The attribute or header key the pathway is written under: config :dd_data_streams, header: when set, DD_DATA_STREAMS_HEADER otherwise.

Defaults to "dd-pathway-ctx-base64". DdDataStreams.produce/3 returns this key and the consume checkpoint reads it, so both sides of an edge must use the same value. Changing it also stops interoperability with the official Datadog tracers, which use the default.

Read per produce and consume call, so a runtime change applies to the next one.

primary_tag()

@spec primary_tag() :: String.t()

The primary tag: config :dd_data_streams, primary_tag: when set, DD_PRIMARY_TAG otherwise. Empty when neither is.

service()

@spec service() :: String.t()

The service the pipeline reports itself as: config :dd_data_streams, service: when set, DD_SERVICE otherwise.

Defaults to "unnamed-elixir-service".

service_name()

@spec service_name() :: module() | nil

The module that names the service per checkpoint: config :dd_data_streams, service_name: when set, nil otherwise and every checkpoint is reported under service/0.

The module implements DdDataStreams.ServiceName and is asked once per checkpoint, so a pipeline can report a message under a service of its own instead of the one configured for the application. Read per checkpoint, so a runtime change applies to the next one. DdDataStreams.attach/0 validates it, so a module that cannot be called fails at start up rather than on the first message.

transport()

@spec transport() :: module() | nil

The module the payloads are posted with, read on every flush: config :dd_data_streams, transport:, otherwise DdDataStreams.Transport.Req when the optional Req dependency is available, otherwise nil and every flush is dropped.

version()

@spec version() :: String.t()

The application version: config :dd_data_streams, version: when set, DD_VERSION otherwise. Empty when neither is.