DdDataStreams.Testing (DD Data Streams v0.3.0)

Testing support for a host application.

There are two ways to record with this library in a test, and the first one is the default:

Owner-scoped. Configure DdDataStreams.Testing.Backend, start it from test_helper.exs, and call start/1 in a test that wants checkpoints. Recording is per test, so tests can run with async: true, and nothing has to be flushed or waited for. This is the API to reach for.
Global. enable!/1 and capture/1 flip the application environment, the shared aggregator and the VM-wide telemetry handler id. They are the async: false tools for tests that need the full wire-format path through the aggregator and the capture transport.

Owner-scoped tests

# config/test.exs
config :dd_data_streams, backend: DdDataStreams.Testing.Backend

# test/test_helper.exs
{:ok, _} = DdDataStreams.Testing.Backend.start_link()

defmodule MyApp.OrdersTest do
  use ExUnit.Case, async: true
  use DdDataStreams.Testing

  test "the consume checkpoint links to the produce checkpoint" do
    DdDataStreams.Testing.start()

    DdDataStreams.consume(%{
      data: payload,
      metadata: %{
        subscription: "projects/demo/subscriptions/orders-sub",
        attributes: Map.new(DdDataStreams.produce("orders-out", payload))
      }
    })

    produce = DdDataStreams.Testing.assert_produced("orders-out")
    consume = DdDataStreams.Testing.assert_consumed("orders-sub")
    DdDataStreams.Testing.assert_chained(produce, consume)
  end
end

use DdDataStreams.Testing only picks private mode for an async case and shared mode otherwise. A test opts in by calling start/1; a test that never does sees produce/3 return [], records nothing, and can run alongside the others. A process a test did not start, such as a Broadway processor, is allowed with allow/2, by pid or by registered name. checkpoints/1 reads the result, and subscribe/0 turns the same points into messages when a test prefers assert_receive.

Global tests

enable!/1 and capture/1 keep their old meaning. They set :enabled and the transport, start the shared aggregator and attach the shared handler id, so a test using either runs with async: false. Use them when the test needs what only that path gives: an encoded payload through the real aggregator, or the wire-format bytes the agent would receive.

Summary

Functions

__using__(opts)

Sets up the owner-scoped mode from the ExUnit context.

allow(owner, allowed)

Allows another process to record into owner.

assert_chained(produce, consume)

Asserts that consume continues the pathway produce started.

assert_consumed(subscription, owner \\ self())

Asserts that a consume checkpoint was recorded for subscription.

assert_produced(topic, owner \\ self())

Asserts that a produce checkpoint was recorded for topic.

capture(fun)

Global, async: false mode: runs fun, flushes every bucket, and returns the points the transport captured while it ran.

checkpoints(owner \\ self())

The checkpoints recorded for owner (the calling process by default), oldest first.

decode_payload(body)

Decodes a captured transport body into a plain map.

enable!(opts \\ [])

Global, async: false mode: enables the library for one test and returns the function that undoes it.

set_from_context(context)

Picks set_private/0 for an async: true case and set_global/1 otherwise.

set_global(context)

Switches the ownership server to shared mode: every process records into the calling test process.

set_private(context \\ %{})

Switches the ownership server to private mode.

start(owner \\ self())

Makes owner (the calling process by default) an owner of the checkpoints.

subscribe()

Subscribes the calling owner to checkpoint messages.

Functions

using(opts)

(macro)

Sets up the owner-scoped mode from the ExUnit context.

Picks private mode when the test module is async: true and shared mode otherwise, so a test module only needs:

use ExUnit.Case, async: true
use DdDataStreams.Testing

In private mode each test still opts in with start/1. In shared mode every process records into the test process, so allow/2 is not needed. A module whose context already chose a mode can call set_private/0 or set_global/1 itself instead.

allow(owner, allowed)

@spec allow(pid() | atom(), pid() | atom()) :: :ok | {:error, struct()}

Allows another process to record into owner.

Broadway processor and batcher processes are started by Broadway, not by the test, so they are not in the test's $callers and need this call. Both arguments accept a pid or a registered name:

DdDataStreams.Testing.start()
DdDataStreams.Testing.allow(self(), :"my_pipeline.Broadway.Processor_0")

Returns :ok, or {:error, %NimbleOwnership.Error{}} when the ownership server refuses the allowance (for example in shared mode, where every process already records into the same owner).

assert_chained(produce, consume)

@spec assert_chained(map(), map()) :: map()

Asserts that consume continues the pathway produce started.

produce and consume are points returned by assert_produced/2 and assert_consumed/2, or read from checkpoints/1. Returns consume.

assert_consumed(subscription, owner \\ self())

@spec assert_consumed(String.t(), pid() | atom()) :: map()

Asserts that a consume checkpoint was recorded for subscription.

Returns the point, like assert_produced/2.

assert_produced(topic, owner \\ self())

@spec assert_produced(String.t(), pid() | atom()) :: map()

Asserts that a produce checkpoint was recorded for topic.

Returns the point, so a test can chain more assertions on it:

produce = DdDataStreams.Testing.assert_produced("orders-out")
assert "type:google-pubsub" in produce.edge_tags

capture(fun)

@spec capture((-> any())) :: [map()]

Global, async: false mode: runs fun, flushes every bucket, and returns the points the transport captured while it ran.

What it flushes is the one shared aggregator, so a test using it must not run next to another test that records. For the owner-scoped checkpoints a pipeline records, use checkpoints/1 instead.

Attaches a handler of its own under a unique id to [:dd_data_streams, :transport, :captured], so it sees exactly what is flushed between the start of the call and the flush it makes itself, and detaches it again whether fun returns or raises. The transport must report on that event, which DdDataStreams.Transport.Capture does - pass it to enable!/1 - and the aggregator must be running, or the flush has nothing to send and the result is [].

The flush happens after fun returns, so a checkpoint recorded a moment earlier is assertable without waiting for the ten-second bucket window, the way DdDataStreams.Aggregator.flush(:all) sends it. Every point of every payload flushed during the call is returned as one flat list of maps with :edge_tags, :hash, :parent_hash, the three distributions and :timestamp_type, in the shape decode_payload/1 gives each point.

checkpoints(owner \\ self())

@spec checkpoints(pid() | atom()) :: [map()]

The checkpoints recorded for owner (the calling process by default), oldest first.

Each point is a plain map shaped like the points decode_payload/1 returns: :edge_tags, :hash, :parent_hash, :pathway_latency, :edge_latency, :payload_size and :timestamp_type. The latencies and the size are the integers the checkpoint measured, not the flushed distributions, and :timestamp_type is "current" because this is the checkpoint itself rather than an aggregator bucket.

Reading is synchronous: record/1 in the backend has already stored the point by the time the call that produced it returns, so no flush or wait is needed. A test that never called start/1 gets []. In shared mode the shared owner's points are returned.

decode_payload(body)

@spec decode_payload(binary()) :: map()

Decodes a captured transport body into a plain map.

The body is the gzip-compressed MsgPack payload the agent would have received, as [:dd_data_streams, :transport, :captured] reports it; it is gunzipped and read back into the same shape the library builds it in: atom keys in this library's vocabulary at every level.

iex> payload = %DdDataStreams.Payload{env: "prod", service: "checkout"}
iex> body = payload |> DdDataStreams.Payload.encode() |> IO.iodata_to_binary() |> :zlib.gzip()
iex> decoded = DdDataStreams.Testing.decode_payload(body)
iex> {decoded.env, decoded.service, decoded.stats}
{"prod", "checkout", []}

A payload carries :env, :service, :stats, :tracer_version, :lang, :version, :process_tags and :product_mask; each bucket in :stats carries :start, :duration and its own :stats points, and each point carries :edge_tags, :hash, :parent_hash, :pathway_latency, :edge_latency, :payload_size and :timestamp_type. The three distributions stay binaries, exactly as they went on the wire; they are DdDataStreams.Sketch protos, not numbers. A key the payload vocabulary does not know keeps the name it arrived with, so nothing is dropped.

It reads what DdDataStreams.Payload encodes; anything else raises.

enable!(opts \\ [])

@spec enable!(keyword()) :: (-> :ok)

Global, async: false mode: enables the library for one test and returns the function that undoes it.

This is the aggregator-and-wire-format path. For tests that only want to see the checkpoints a pipeline records, prefer the owner-scoped API (start/1 and checkpoints/1), which leaves the shared application environment and aggregator alone and can run with async: true.

Sets enabled: true and any of the options :transport, :service and :env: that are given, attaches the handlers DdDataStreams.attach/0 installs, and starts the aggregator if it is not already running, so checkpoints recorded during the test are aggregated instead of dropped. Any other option raises an ArgumentError, so a typo does not silently configure nothing.

The returned zero-arity function stops the aggregator this call started, detaches the handlers this call attached and restores the application environment to what it was before the call; anything the call found already running or already attached it leaves alone. It is safe to call twice, and the aggregator is linked to the calling process, so a test that never runs the cleanup still cannot leak one. Pass it to ExUnit.Callbacks.on_exit/1:

on_exit(DdDataStreams.Testing.enable!())

set_from_context(context)

@spec set_from_context(map()) :: :ok

Picks set_private/0 for an async: true case and set_global/1 otherwise.

set_global(context)

@spec set_global(map()) :: :ok

Switches the ownership server to shared mode: every process records into the calling test process.

Raises when the ExUnit context is async: true, because a shared owner would receive another test's checkpoints.

set_private(context \\ %{})

@spec set_private(map()) :: :ok

Switches the ownership server to private mode.

This is the default mode and the one set_from_context/1 picks for an async: true case.

start(owner \\ self())

@spec start(pid() | atom()) :: :ok

Makes owner (the calling process by default) an owner of the checkpoints.

After this call, produce/3, produce_batch/3, consume/1, checkpoint/1 and the Broadway handlers record into owner, and checkpoints/1 reads what they recorded. Calling it again is harmless. A Task records into the owner that spawned it without this call, because it inherits $callers; any other process needs allow/2.

Returns :ok. Raises when the ownership server is not running, because a test that thinks it is recording and is not is worse than a failed test.

In shared mode the shared owner receives the checkpoints, so start/1 is not needed there; calling it with the shared owner is harmless.