Node and Flow Metrics

This guide explains the two metric layers available for tracing signal volume through an OTel Arrow Dataflow Engine pipeline:

receiver --> processor A --> processor B --> exporter
   |            |                |           |
   +------------+----------------+-----------+-- node metrics (all nodes)
                +----------------+
                `-- flow metrics (processors only)

Use node metrics to see the messages, signal items, and logical payload size a specific receiver, processor, or exporter consumes and produces. Use flow metrics to measure a selected, contiguous processor range as one operation: how many items entered and left, how long processing took, and which decision nodes removed items.

Both metric layers are emitted through the engine's internal observability pipeline. Configure an engine.observability.pipeline to export them, or use the admin metrics endpoint while developing. See Configuration for observability-pipeline configuration.

Choose the Right Metric Layer

Question	Use
Which node changed the count of logs, metrics, or traces?	Node item metrics
What is the aggregate compute time for selected processor stages?	Flow metrics
Which decision processor dropped items within a processor range?	Flow metrics
How many items did a receiver admit or exporter emit?	Node item metrics

Node Metrics

A message is the PData batch that moves between nodes. An item is an individual log record, metric data point, or span in that batch. One message can contain multiple items. Logical size is the byte size of the current in-memory payload representation. Payload size is the encoded application payload observed at a receiver or exporter boundary.

Completion duration measures from a node boundary until the terminal ACK or NACK: from input for processors and exporters, and from output for receivers. Local duration uses the boundary defined by each node instrument, which may include encoding or backend latency. Use flow.compute.duration instead for compute time across a processor range.

none and basic enable no node metrics by default. normal adds message measurements, and detailed adds every optional measurement. A per-node policy enables its measurement at any runtime metric level.

Measurement	Engine-managed metrics	Node-implemented metrics	Default level	Per-node policy
Messages	node.input.messages, node.output.messages	receiver.received.messages, exporter.attempted.messages	normal	messages: true
Items	node.input.items, node.output.items	exporter.attempted.items	detailed	item_counts: true
Logical size	node.input.size, node.output.size	-	detailed	size: true
Payload size	-	receiver.received.payload.size, exporter.attempted.payload.size	detailed	size: true
Completion duration	node.completion.duration	-	detailed	completion_duration: true
Local duration	-	receiver.processing.duration, processor.compute.duration, exporter.attempted.duration	detailed	duration: true

The node.input.* metrics apply to processors and exporters. node.output.* metrics apply to receivers and processors. Node-implemented metrics require the implementation to use the corresponding shared instrumentation.

Message, item, and size counters have bounded signal and outcome data-point attributes. signal is one of logs, metrics, or traces; outcome is success, failure, or refused, recorded during terminal ACK/NACK unwinding. The metric-set entity attributes identify the pipeline and node, so group by those attributes when comparing nodes.

Enable Optional Measurements

To enable it for every node in a pipeline, use detailed:

policies:
  telemetry:
    runtime_metrics: detailed

To enable it only for selected nodes, opt in the relevant nodes:

nodes:
  sampler:
    type: processor:log_sampling
    policies:
      telemetry:
        item_counts: true
        size: true
    config: {}

This narrower configuration is appropriate when only a small part of a pipeline needs payload measurements. detailed enables item counts and size for every node without node-level settings.

Interpret Node Counts

For a linear topology, a node's output.items normally matches the next node's input.items for the same signal. A filtering or sampling processor can emit fewer items than it receives; a fan-out processor can emit an item on more than one output. Compare counts only along the particular edge or topology behavior being investigated.

Node metrics are the right choice when operators need to locate where a signal count changes, including receiver admission, processors, and exporter output. Use the runnable trafficgen-node-metrics.yaml example to inspect the metrics on every node, compare local duration with flow duration, or observe an individually opted-in processor.

Flow Metrics

Flow metrics are an explicit telemetry policy for a contiguous range of processor nodes. They are not enabled by runtime_metrics; declare each flow under the pipeline's policies.telemetry.flow_metrics list.

policies:
  telemetry:
    flow_metrics:
      - id: ingest_processing
        bounds:
          start_node: enrich
          end_node: filter
        duration_distribution: normal
        purpose: transform
        metrics:
          - input_messages
          - input_items
          - input_size
          - output_messages
          - output_items
          - output_size
          - compute_duration
          - dropped_items

The start_node and end_node fields name processors; the range includes both boundary processors. The engine validates that the end processor is reachable from the start processor and rejects interleaved flow ranges. Omit metrics to enable every supported flow metric. When present, it must not be empty and must not repeat a metric.

duration_distribution controls the aggregation used by compute_duration:

Value	OTLP representation	Retained data
basic	Bucketless Histogram	Count, sum, min, and max
normal	ExponentialHistogram	Normal-resolution buckets and summary statistics
detailed	ExponentialHistogram	Higher-resolution buckets and summary statistics

The setting defaults to normal. Use basic for lower aggregation cost or compatibility with consumers that do not support exponential histograms. Basic distributions do not retain buckets, so percentiles cannot be reconstructed. The setting is ignored when compute_duration is not enabled. Its distribution tier is independent of runtime_metrics, which controls whether broader metric families are enabled.

Each flow's attributes are part of its OTLP instrumentation scope, so flows using different tiers have distinct metric stream identities. Some backends flatten instrumentation scopes and require one data type per metric name. Use the same wire type across flows and deployments when exporting to such a backend: basic produces Histogram, while normal and detailed produce ExponentialHistogram.

Flow Metrics and Attributes

Flow metrics use the flow.input, flow.output, flow.compute, and flow.dropped instrumentation scopes and include these scope attributes:

Attribute	Meaning
flow.id	The configured id.
flow.node.start	The configured start processor.
flow.node.end	The configured end processor.
flow.purpose	The configured purpose, or an empty value when omitted.
flow.node.decision	The decision processor that emitted dropped.items, or an empty value for other flow metrics.

flow.purpose lets OpenTelemetry Views select a specific kind of flow when multiple flows use the shared directional scopes. For example, a view can select scope_name: flow.compute with scope_attributes: { flow.purpose: transform } to rename or route only transformation-flow metrics.

The metrics have a bounded signal data-point attribute with values logs, metrics, and traces. They intentionally do not have an outcome attribute. Flow metrics are recorded while a PData batch moves forward through the processor range, before its terminal ACK/NACK outcome is known. They describe range traversal and decision-node drops, independently of the eventual node outcome.

An end processor that ACKs without sending, such as a filter that removes every item, does not increment any flow.output metric. It still records flow.compute.duration and may record flow.dropped.items.

Configuration value	Emitted metric	Meaning
input_messages	flow.input.messages	PData messages entering the start processor.
input_items	flow.input.items	Signal items entering the start processor.
input_size	flow.input.size	Logical payload bytes entering the start processor.
output_messages	flow.output.messages	PData sends leaving the end processor.
output_items	flow.output.items	Signal items leaving the end processor.
output_size	flow.output.size	Logical payload bytes leaving the end processor.
compute_duration	flow.compute.duration	Histogram of aggregate processor compute duration in the range, in seconds.
dropped_items	flow.dropped.items	Signal items a decision processor in the range chose to drop.

For a linear flow, the sum of flow.dropped.items across flow.node.decision equals flow.input.items - flow.output.items. There is no per-decision-node kept metric: counts that survive one decision can reach a later decision, so per-node kept counts are not additive. Use the flow's flow.output.items as the flow-wide surviving count.

See trafficgen-node-metrics.yaml for a runnable comparison of node and flow metrics around a sampling processor.