Skip to main content

metadata

Attaches Expanso runtime metadata — pipeline IDs, node info, runtime counters, and custom fields — to every message that flows through it. Use it to declaratively stamp events with provenance, lineage, or operational context without writing Bloblang.

It can also promote inbound message metadata into the JSON body as fields, so input-supplied metadata lands in your data lake as columns without per-pipeline Bloblang glue. See Promoting inbound metadata into the body.

pipeline:
  processors:
    - metadata:
        include: [core, orchestrator, node, pipeline]
        custom:
          pipeline_owner: [email protected]
          pii_redaction: enabled
        target: body
        format: nested
        body_key: lineage

When to Use​

Use the metadata processor when you need to:

  • Stamp events with provenance — attach run_id, pipeline_name, pipeline_version, and node_id to every message for lineage tooling or downstream auditing.
  • Tag output by node identity — write the node's region, environment, and cluster_name onto messages so downstream systems can route or partition by location.
  • Attach static custom fields — add owner emails, compliance tiers, or environment labels declaratively, instead of writing a mapping block.
  • Surface runtime counters — opt in to the runtime category to splice records_in, bytes_out, error_count, and duration_ms into the body for observability sinks.
  • Promote inbound metadata into the body — copy input-supplied metadata (Kafka headers, MQTT topic parts, file_path / file_name from tail, and anything else an input stamps) into the JSON body as columns, with optional renaming and type casting. See Promoting inbound metadata into the body.

Don't use this if:

  • You only need to reference a single key in an interpolated field — use the implicit @pipeline_id / @node_id keys directly. See the pipeline metadata guide.
  • You need conditional or computed metadata — use mapping and write meta foo = ... directly.

Configuration​

# Common config fields, showing default values
metadata:
  include: [core, orchestrator, node, pipeline]   # categories of fields to attach
  custom: {}                                      # extra static key/value pairs
  target: meta                                    # "meta" or "body"

Fields​

FieldTypeDefaultDescription
includestring list[core, orchestrator, node, pipeline]Categories of metadata fields to resolve. Valid values: core, orchestrator, node, pipeline, runtime. Note runtime is not in the default — opt in explicitly.
excludestring list[]Specific field names (or category names) to omit from the resolved set.
customstring→string map{}User-supplied key/value pairs to attach. Keys must not collide with reserved field names.
targetstringmetaWhere to write the fields. meta writes to message metadata; body splices into the JSON message body.
formatstringflatWhen target: body: flat merges fields at the JSON root; nested places them under body_key. Ignored when target: meta.
body_keystring""Required when target: body and format: nested. Must be empty otherwise.
promoteobject—Optional. Copy inbound message metadata into the JSON body as fields. Requires target: body. See Promoting inbound metadata into the body for the sub-fields and behavior.

Field Categories​

Each category in include resolves to a fixed set of keys.

core — OpenLineage-aligned identity​

KeyDescription
run_idThe unique identifier for this pipeline execution
job_nameThe pipeline name
job_namespaceThe execution namespace
event_timeCurrent time, RFC3339Nano in UTC, recomputed per message
producerURL identifying the Expanso edge agent and its version (e.g., https://expanso.io/edge/v1.2.3)

These field names are intentionally aligned with the OpenLineage spec. The edge agent emits native OpenLineage events at every pipeline lifecycle transition — see the lineage guide — and the run_id here matches the runId on those events, so per-message records correlate back to the lifecycle events emitted by the edge.

orchestrator — orchestrator-populated context​

KeyDescription
job_idThe pipeline (job) identifier
deployment_idThe deployment identifier (reserved; currently empty)
namespaceThe execution namespace
eval_idThe evaluation identifier
rollout_waveThe rollout wave for staged deployments

node — infrastructure context​

KeyDescription
node_idThe edge node running the pipeline
hostnameThe node's hostname
regionFrom the node's region label
environmentFrom the node's environment label
cluster_nameFrom the node's cluster_name label
agent_versionThe edge agent version

Only the region, environment, and cluster_name labels are promoted by this category. Arbitrary node labels remain available via the @node_label_* keys — see node labels in the metadata guide.

pipeline — pipeline provenance​

KeyDescription
pipeline_nameThe pipeline name
pipeline_versionThe pipeline version
git_commit_shaReserved — populated when the pipeline is built from a git source
git_repo_urlReserved — populated when the pipeline is built from a git source
git_branchReserved — populated when the pipeline is built from a git source

The three git_* fields are slot fields. They are emitted as empty strings until pipelines are built from a git source.

runtime — per-execution counters​

KeyDescription
start_timeWhen this pipeline execution started, RFC3339Nano in UTC
records_inMessages observed by this processor since the execution started
records_outMessages successfully written by this processor
bytes_inBytes observed by this processor
bytes_outBytes successfully written by this processor
error_countBody-write failures recorded by this processor
duration_msMilliseconds since start_time

The runtime category is not included by default — list it in include to opt in. Counter values reflect the processor's own observations of the message stream up to and including the current message.

Targets and Formats​

The target and format fields together control where resolved metadata is written:

targetformatbody_keyResult
meta(ignored)must be emptyEach resolved key becomes a Bento message metadata entry, accessible via @key_name and ${! metadata("key_name") }.
bodyflatmust be emptyResolved keys are merged at the JSON body's root. Metadata keys overwrite body keys on collision.
bodynestedrequiredResolved keys are placed under body[body_key], leaving the rest of the body untouched.

When target: body, the body must be a JSON object. Non-JSON bodies, JSON arrays, and JSON scalars are passed through unchanged; the processor logs a warning and increments error_count (visible via the runtime category).

Custom Fields​

The custom: map accepts string→string pairs that are attached alongside the resolved category fields. Custom keys must not collide with reserved field names.

metadata:
  include: [core]
  custom:
    pipeline_owner: [email protected]
    compliance_tier: pii
    cost_center: data-platform

To drop a built-in field while keeping the rest of its category, list it in exclude:

metadata:
  include: [pipeline]
  exclude: [git_commit_sha, git_repo_url, git_branch]

exclude accepts either field names or category names. Only include is checked against the closed set of categories at submission time, so unknown entries in exclude pass silently.

Promoting inbound metadata into the body​

Inputs commonly stamp messages with metadata that downstream consumers want as fields rather than out-of-band metadata — Kafka header values, MQTT topic parts, the file_path / file_name keys from the tail input, or any domain-specific keys an upstream system attaches. Without promote, every pipeline that lands these to Parquet, Delta, or any columnar store would repeat the same mapping block per field:

- mapping: |
    root.tenant       = metadata("kafka_tenant_id")
    root.event_type   = metadata("kafka_event_type")
    root.ingest_ts    = metadata("kafka_ingest_timestamp").number()
    # ...and so on for every header you want as a column

The promote block does this declaratively:

- metadata:
    target: body                # required for promote
    format: flat
    promote:
      match: "kafka_"           # prefix; or "/^kafka_.+/" for a regex
      names: [extra_header]     # explicit keys, unioned with match
      rename:                   # inbound key -> output field name
        kafka_tenant_id: tenant
        kafka_event_type: event_type
      cast:                     # inbound key -> target type
        kafka_ingest_timestamp: int
        kafka_retry_count: int
        kafka_is_internal: bool
      wrap_key: value           # wrap a non-object body under this key

promote is optional and additive: omit the whole block and the processor behaves exactly as before. When set, it requires target: body (promoting metadata into metadata is a no-op and is rejected at submission).

promote fields​

FieldTypeDefaultDescription
matchstring""Selection rule. A bare value is a literal prefix (e.g. kafka_ matches kafka_tenant_id, kafka_event_type, …). A value wrapped in /…/ is a regular expression (e.g. /^kafka_.+/). The empty regex // is rejected — it would match every metadata key and risk leaking provenance fields.
namesstring list[]Explicit inbound metadata keys to promote. Unioned with match, then deduplicated.
renamestring→string map{}Map inbound key → output field name. Always keyed by the inbound key, even when a cast for the same key is set. A rename target that collides with a reserved field name is rejected at submission.
caststring→string map{}Map inbound key → target type. Valid types: int, float, bool. A value that fails to parse — or, for float, NaN / ±Inf — falls back to the original string. Cast fallbacks do not increment error_count.
wrap_keystringvalueWhen the message body is not a JSON object (e.g. a scalar value or a JSON array), the body is wrapped as { wrap_key: <body> } before adding promoted fields. Must not collide with a reserved field name; in format: nested, must not equal body_key.

At least one of match or names must be set whenever the promote block is present.

Selection — match, names, and regex​

Selection is the union of match (prefix or regex) and names:

promote:
  match: "kafka_"               # everything starting with kafka_
  names: [trace_id, source]     # plus these explicit keys

For a regex, wrap the pattern in slashes — anything else is a literal prefix:

promote:
  match: "/^(kafka_tenant|kafka_event)/"   # regex

Selection is evaluated per message. A metadata key that exists on one message but not the next is simply skipped on the message where it's absent — the field is omitted, not set to empty. Source keys are processed in sorted order so a rename collision (two source keys → one target) resolves deterministically.

rename​

Map inbound keys to output field names. Always keyed by the inbound key, including when a cast is also configured for that key:

promote:
  match: "kafka_"
  rename:
    kafka_tenant_id:        tenant
    kafka_event_type:       event_type
    kafka_ingest_timestamp: ingest_ts
  cast:
    kafka_ingest_timestamp: int     # cast is keyed by the inbound key, not "ingest_ts"

A rename target that collides with a reserved field name (the names in Reserved Field Names below) is rejected at submission with a clear error.

cast​

Inbound metadata values arrive as strings. cast parses them into JSON-native types so downstream columnar stores see proper int/float/bool columns:

Cast typeParses withNotes
intbase-10 64-bit signed integer (-9223372036854775808 … 9223372036854775807)Overflow falls back to the original string.
float64-bit IEEE 754NaN, +Inf, -Inf, and out-of-range values fall back to the original string. This is deliberate: a non-finite cast would fail the body marshal and silently drop the entire record, so the runtime treats non-finite as a parse failure.
booltrue / false (also 1 / 0, t / f, TRUE / FALSE)Anything else falls back to the original string.

A cast fallback is not a pipeline error. Counting it via error_count would climb 1:1 with throughput on a source that always emits an un-castable value and fire false error-rate alerts. Failed casts are logged at debug only and the field is forwarded as the original string.

wrap_key — handle non-object bodies​

Some inputs emit each message as a scalar value (a number, a string) or as a JSON array, with all the structure carried in metadata. The metadata processor needs an object to splice fields into, so when promote is enabled and the body is not already a JSON object, the body is wrapped under wrap_key:

promote:
  match: "kafka_"
  wrap_key: value          # default
# scalar body 23.5
# becomes { "value": 23.5, ...promoted fields..., ...resolved Expanso fields... }

Wrapping is only done when promote is enabled. Without promote, a non-object body is passed through unchanged with a warning and an error_count increment, preserving the pre-promote behavior exactly.

If the body bytes are not valid JSON at all, they are wrapped as a string under wrap_key. A JSON array or scalar is wrapped as its decoded JSON type.

wrap_key must not collide with a reserved field name. In format: nested, it must not equal body_key — the nested metadata object would clobber the wrapped body value, and the validator rejects this at submission.

Precedence on collision​

Three sources contribute fields to the body. They layer in this order, from lowest to highest priority:

original body  <  promoted fields  <  resolved Expanso provenance

So:

  • An incoming body field is kept unless a promoted field has the same name.
  • A promoted field is kept unless a resolved Expanso provenance field has the same name (run_id, node_id, etc.).
  • Resolved Expanso provenance fields always win.

In format: nested, this layering applies inside the nested object (body_key); the rest of the body is left alone. In format: flat, it applies at the body's root.

Reserved-name guard​

Inputs and middleware can stamp arbitrary keys. The runner itself injects node_id, pipeline_id, execution_id, and per-label node_label_* keys into message metadata before the metadata processor runs — so a broad match (e.g. match: "" or a wide regex) could otherwise leak Expanso's own provenance fields or let an upstream spoof them.

The processor enforces two layers of protection:

  • Submission validation rejects static configs that would land a promoted name on a reserved field — both explicit names entries and rename targets.
  • Runtime silently drops any promoted key whose source name or renamed output is reserved or node_label_*. Renaming a reserved source (node_id → source_node_id) cannot smuggle a reserved value in either, because both the source and the output are checked.

The reserved set is the same one listed under Reserved Field Names below.

Reserved Field Names​

The following names cannot be used as custom keys:

run_id, job_name, job_namespace, event_time, producer,
job_id, deployment_id, namespace, eval_id, rollout_wave,
node_id, hostname, region, environment, cluster_name, agent_version,
pipeline_name, pipeline_version, git_commit_sha, git_repo_url, git_branch,
start_time, records_in, records_out, bytes_in, bytes_out, error_count, duration_ms,
pipeline_id, execution_id, job_version

Submitting a pipeline with a colliding custom: key is rejected with a clear error message — see Validation Errors.

The internal _execution_id field is auto-populated by the runtime at pipeline build time. User-supplied values are rejected at submission.

Validation Errors​

Pipeline submission validates the metadata processor's configuration. The errors you may see and how to fix each:

ErrorFix
unknown category "X"; valid categories: [core orchestrator node pipeline runtime]Replace X in include with one of the listed categories.
body_key cannot be set when target is "meta"Remove body_key, or change target to body.
body_key is only meaningful with format: "nested"Remove body_key for format: flat, or change format to nested.
body_key is required when target: "body" and format: "nested"Set body_key to the JSON object key the metadata should be placed under.
unknown format "X"; must be "flat" or "nested"Use flat or nested.
unknown target "X"; must be "meta" or "body"Use meta or body.
custom key "X" collides with a reserved field nameRename the key, or use exclude to drop the reserved field if you want to override its value (note: exclude removes — it does not override).
_execution_id is internal and auto-populated by the Expanso runtime; remove it from your configRemove the _execution_id entry from the processor's config.
promote requires target: "body"Set target: body, or remove the promote block. Promoting metadata into metadata is a no-op.
promote requires at least one of match or namesSet promote.match (a prefix or /regex/), promote.names (a list of inbound keys), or both.
promote.match regex "//" is empty and matches every keyUse a non-empty pattern, or drop the slashes to use a literal prefix.
invalid promote.match regex "/…/": <error>Fix the regular expression, or use a literal prefix.
cast type for "X" must be one of [int float bool]Set the cast value to int, float, or bool.
rename target "X" collides with a reserved field nameRename to a non-reserved name.
promoted name "X" collides with a reserved field namePromote a different key, or add a rename mapping it to a non-reserved name.
wrap_key "X" collides with a reserved field namePick a different wrap_key.
wrap_key "X" equals body_key in nested formatPick a different wrap_key, or switch to format: flat. The nested metadata object would otherwise clobber the wrapped body value.

Examples​

Attach the default categories to message metadata, then reference them in a downstream output:

pipeline:
  processors:
    - metadata: {}

output:
  kafka:
    addresses: [localhost:9092]
    topic: events.${! metadata("region") }
    metadata:
      exclude_prefixes: []

After the processor runs, every message carries metadata keys like @run_id, @job_name, @node_id, @region, etc. You can read them via metadata("...") in interpolation or @... in Bloblang.

  • Pipeline metadata guide — the implicit @pipeline_id, @node_id, @namespace, and @node_label_* keys available without this processor.
  • Bloblang guide — for conditional or computed metadata via mapping.
  • mapping processor — the imperative alternative for editing metadata.