The behaviour that tells the library what a Broadway message is.

A Broadway producer hands its pipeline a Broadway.Message whose metadata map is whatever that producer decided to put there. Naming the edge a message was consumed on is therefore the one thing this library cannot do on its own: Pub/Sub reports the subscription it delivered from, Kafka the topic, SQS neither. An extractor answers that question, and there is one per broker.

The extractors in use are the :extractors key of the application environment, a list defaulting to [DdDataStreams.Extractor.GooglePubSub]:

config :dd_data_streams,
  extractors: [MyApp.KafkaExtractor, DdDataStreams.Extractor.GooglePubSub]

An application that talks to several technologies lists one extractor per technology, and every message is given to them in order until one recognises its metadata: the first consume/1 (or consume/2) that returns tags wins, so the more specific extractor goes first when two could claim the same message.

Writing one

consume/1 gets the message's metadata and returns the consume edge tags together with the parts of the metadata a checkpoint measures, or :error when the metadata names no edge - the library then records nothing for that message rather than inventing a node. An extractor that raises is treated like one that named no edge, and the reason is logged and reported as a [:dd_data_streams, :handler, :error] telemetry event. produce/2 gets the topic a message is published to and the producer_info/0, and returns the produce edge tags.

consume/1 is enough when the metadata alone names the edge. Some brokers keep the edge identity in the producer's configuration instead - the SQS queue, the Kafka consumer group, the RabbitMQ queue - and only Broadway knows which producer delivered a message. An extractor for such a broker implements consume/2, which receives producer_info/0 next to the metadata; the library calls consume/2 when the module exports it and consume/1 otherwise, so adding it changes nothing for existing extractors. Both callbacks return the same things, and which of the two sources names the edge - metadata or producer info - is the extractor's own decision.

Tags are key:value strings. A key DdDataStreams.Tags knows is hashed into the node or splits its statistics; one it does not know is reported on the edge and does neither. A consume edge needs direction:in and a type:; a produce edge direction:out, the topic and the same type:. The type: is the queueing technology Datadog files the stream under (google-pubsub, kafka, sqs, ...) and it is part of every node's identity, so it has to be the same value for every service in the pipeline.

An extractor implements one callback of each pair - consume/1 or consume/2, and produce/2 or produce/3 - which is why all four are optional and attach/0 is what rejects a module that implements neither of a pair.

A BroadwayKafka pipeline, for example, stamps metadata.topic, metadata.headers and metadata.ts:

defmodule MyApp.KafkaExtractor do
  @behaviour DdDataStreams.Extractor

  @impl true
  def consume(%{topic: topic} = metadata) when is_binary(topic) do
    {:ok, ["direction:in", "topic:" <> topic, "type:kafka"],
     %{
       attributes: Map.get(metadata, :headers),
       pathway_start:
         case Map.get(metadata, :ts) do
           nil -> nil
           milliseconds -> System.convert_time_unit(milliseconds, :millisecond, :native)
         end
     }}
  end

  def consume(_metadata), do: :error

  @impl true
  def produce(topic, _producer_info),
    do: ["direction:out", "topic:" <> topic, "type:kafka"]
end

The record timestamp the example reads is one of several publish times a broker stamps on a message: Pub/Sub's publishTime, Kafka's record timestamp, SQS's SentTimestamp, RabbitMQ's :timestamp. Returning one as :pathway_start makes a message that starts a pathway measure from it, so a pipeline that only ever consumes reports how long the message waited in the bus instead of zero. :received_at keeps its own meaning - when the message reached the pipeline - so a message with neither still reports zero latency, exactly as before.

A pipeline that publishes several tags per edge, such as partition: or group:, adds them here; the keys in DdDataStreams.Tags decide whether they take part in the node's identity or only split its statistics.

Naming the edge is not naming the service: the service a checkpoint is reported under comes from DdDataStreams.Config.service/0 or a configured DdDataStreams.ServiceName, so an extractor stays about one broker.

produce/2 has no message metadata to look at, so it names every edge it is asked about. It does receive the producer_info/0 of the message the stage is answering - the same one consume/2 gets - so identity kept in the producer's configuration, such as a cluster id or an exchange, can still reach the produce tags; a stage that produces without consuming gets both fields nil. A produce in a stage that consumed a message is named by the extractor that consumed it, and a stage that produces without consuming passes its extractor to DdDataStreams.produce/3.

A topic is not always enough to name an outgoing edge: a RabbitMQ publish is named by its exchange and its routing key. An extractor for such a broker implements produce/3, which receives the options the host passed to DdDataStreams.produce/3, and names the edge from whatever the host put there. The library neither reads nor validates those options, so this stays the host's vocabulary rather than a broker the core knows about.