beholder: send resource attributes to chip ingress as gRPC metadata

[pkcll]

Summary

Resource attributes set in node TOML ([Telemetry.ResourceAttributes]) reach the OTel collector
path but never the ChipIngress path, so they appear as headers on beholder__platform__messages
and not on cre. This wires them into the ChipIngress client.

They travel once per request as gRPC metadata, not stamped on every CloudEvent. Resource
attributes describe the producer rather than any individual event — the same reason OTLP factors
resource out of the payload instead of repeating it on every log record. Per-event stamping would
also repeat identical bytes for every event in a batch: ~10 attributes across a 1,000-event batch is
roughly 300 KB of duplication, which counts against maxGRPCRequestSize and therefore reduces how
many events fit per batch.

This also matches an existing chip-ingress pattern. nopInfoHeadersFromContext already reads a
connection-scoped value once and baseKafkaHeaders fans it onto every Kafka record, cached per
(domain, entity, specVersion).

Changes

• pkg/beholder/client.go — pass resource attributes to the ChipIngress client as gRPC metadata via
  chipingress.WithResourceAttributeHeaders.
• pkg/beholder/resource_attributes.go — add resourceAttributesToStringMap, the single conversion
  point, using attribute.Value.Emit for canonical stringification of any value type. (Its doc
  comment is worded to make explicit that this feeds gRPC metadata only — resource attributes are
  never stamped as CloudEvent extensions.)
• pkg/beholder/resource_attributes_test.go — unit coverage for resourceAttributesToStringMap,
  including the empty-input case.
• pkg/beholder/chip_ingress_emitter.go — doc comment on ChipIngressEmitter making the same point:
  resource attributes are not stamped on individual events here, they travel once per request as
  client-level gRPC metadata. Plus an incidental gofmt alignment fix.
• pkg/beholder/batch_emitter_service_test.go — incidental stray-blank-line gofmt fix; plus a
  mock fix required by the pin bump (see "Depends on chipingress: send whitelisted resource attributes as chainlink-* gRPC metadata #2288"): PublishBatch success mocks now
  return a response with successful results instead of (nil, nil).
• pkg/beholder/client_test.go — regression test asserting on a real gRPC connection that
  configuring AuthHeaders together with ResourceAttributes leaves the CSA node auth token intact
  and delivers the attributes alongside it.
• Root go.mod/go.sum — bump pkg/chipingress to the reworked chipingress: send whitelisted resource attributes as chainlink-* gRPC metadata #2288 head (whitelisted
  chainlink-* headers).

No exported API is added or changed in pkg/beholder.

Why the auth test

The CSA node auth token (X-Beholder-Node-Auth-Token) travels as credentials.PerRPCCredentials,
while resource attributes travel through a unary interceptor that calls
metadata.AppendToOutgoingContext. Nothing previously covered the two coexisting. Two properties
matter and both are asserted:

• the token arrives exactly once, unmodified — the interceptor appends rather than replaces, so a
  collision would send two values under one key rather than overwriting
• grpc-go validates the whole outgoing metadata map in newClientStream, so a single non-printable
  attribute value would fail the entire RPC with codes.Internal, auth included. pkg/chipingress
SanitizeMetadataHeaders prevents this upstream by omitting an attribute with an invalid key or a
  non-printable value rather than sending it malformed — see chipingress: send whitelisted resource attributes as chainlink-* gRPC metadata #2288.

Not yet visible to consumers

Chip-ingress currently reads incoming gRPC metadata only to authenticate
(metadata.FromIncomingContext appears solely in internal/auth/csa_auth.go). Kafka headers come
from CloudEvent extensions and core attributes plus a fixed set the server derives itself.

So this PR makes the attributes visible to the service but not to a Kafka consumer. Forwarding
them onto records is smartcontractkit/atlas#13201 (open, draft, implemented and passing its own
suite), which reads a closed whitelist of ten fixed chainlink-* metadata headers and translates
each to its mapped resource_<attribute key> Kafka header — keeping the namespace closed so a
resource attribute cannot shadow ce_*, table_route_name, or an identity header the server
derives from the verified auth token.

On its own, therefore, this PR does not close the header gap between cre and
beholder__platform__messages. It is one of three parts: this, #2288, and atlas#13201.

Depends on #2288

The publisher-side contract lives in #2288, reworked 2026-09-14 to a closed whitelist: ten fixed,
hand-named chainlink-* gRPC metadata headers (chainlink-resource-csa-public-key, chainlink-resource-service-name,
... — see chipingress.ResourceAttributeHeaders), matched case-insensitively against the configured
resource attributes. Attributes outside the whitelist are ignored; non-printable values are omitted,
never rewritten. Because no operator-defined key can ever become a header name, the earlier revision
charset validation, caps, drop-reporting and deny-lists were all deleted. This PR needs no
functional code change for any of that: it calls WithResourceAttributeHeaders, so all of it
happens inside pkg/chipingress.

The pin is now bumped to the reworked #2288 branch head
(v0.0.11-0.20260915043145-8e5ac5170bc5). Until #2288 merges, the
Validate go.mod dependencies job is expected to fail, as it has throughout this stack.

One test-only fix came along with the pin bump. The new pin also pulls in the batch client
partial-delivery callback path, which dispatches per-event callbacks from PublishResponse.Results.
The batch-emitter test mocks returned (nil, nil) for PublishBatch — realistic enough for the old
client, a nil-deref / synthetic RESULTS_MISMATCH for the new one — so
pkg/beholder/batch_emitter_service_test.go now returns a response with a generous tail of
successful results, matching the real server contract of one result per event. No production code
changed.

Consumers reading platformEnv on the legacy topic will find it as resource_platformEnv on cre.
They already need changes for two other headers on these topics regardless: beholder_entity is
ce_type, and csa_public_key is ce_csapublickey — auth-derived, so more trustworthy than the
self-reported resource_csa_public_key.

Related

• Upstream (chipingress primitives): Add resource attribute support to ChipIngress client #2267 (merged), chipingress: send whitelisted resource attributes as chainlink-* gRPC metadata #2288 (prefixing + validation/limits, required)
• Server: smartcontractkit/atlas#13201
• Downstream (chainlink): bump chainlink-common and pkg/chipingress for chip-ingress resource attributes chainlink#23014

Related PRs

• chipingress: send whitelisted resource attributes as chainlink-* gRPC metadata #2288 (upstream — chipingress client
  primitives this PR depends on)
  • beholder: send resource attributes to chip ingress as gRPC metadata #2216 (this PR — beholder wiring)
  • https://github.com/smartcontractkit/atlas/pull/13201 (sibling downstream, different repo —
    chip-ingress server; forwards the metadata this PR sends onto Kafka headers; does not depend on
    this PR, only on the chipingress: send whitelisted resource attributes as chainlink-* gRPC metadata #2288 wire contract)

Test plan

go test ./pkg/beholder/...
go vet ./pkg/beholder/...

[github-actions]

✅ API Diff Results - github.com/smartcontractkit/chainlink-common/pkg/chipingress

✅ Compatible Changes (7)

./ (7)

• EventOpt — ➕ Added

• NewEventWithOpts — ➕ Added

• NewStaticHeaderProvider — ➕ Added

• SanitizeExtensionName — ➕ Added

• SanitizeMetadataHeaders — ➕ Added

• SanitizeMetadataValue — ➕ Added

• WithResourceAttributeExtensions — ➕ Added

---

📄 View full apidiff report

[github-actions]

📊 API Diff Results

No changes detected for module github.com/smartcontractkit/chainlink-common

View full report

[engnke]

overall approach lgtm. Adding the CloudEvent extensions as the guaranteed propagation path makes sense, and we're keeping the gRPC header route for observability is reasonable

[pkcll]

Split into two PRs:

• pkg/chipingress resource attributes: Add resource attribute support to ChipIngress client #2267
• pkg/beholder wiring: feat: add chip-ingress batch config keys and tune defaults #2207
• downstream chainlink bump: #23014

[Copilot]

Pull request overview

Copilot reviewed 8 out of 9 changed files in this pull request and generated no new comments.

Comments suppressed due to low confidence (1)

pkg/beholder/chip_ingress_emitter.go:64

• NewWithResourceAttributes stores the caller-provided attrs map directly on the emitter. If the caller later mutates that map (or shares it across goroutines), emits can race on concurrent map reads/writes. Since these attributes are meant to be static config, defensively clone the map before storing it.

	var resourceAttrs *resourceAttrExtensions
	if len(attrs) > 0 {
		resourceAttrs = &resourceAttrExtensions{attrs: attrs}
	}

[Copilot]

Pull request overview

Copilot reviewed 8 out of 9 changed files in this pull request and generated 1 comment.

Comments suppressed due to low confidence (1)

pkg/beholder/batch_emitter_service.go:147

• emitInternal always uses NewEventWithOpts + WithResourceAttributeExtensions, even when no resource attributes are configured (e.resourceAttrs is an empty map). This is inconsistent with the sync emitter and adds avoidable overhead / potential semantic differences when attrs are empty.

		event, err := chipingress.NewEventWithOpts(domain, entity, body, attributes, chipingress.WithResourceAttributeExtensions(e.resourceAttrs))

[Copilot]

Pull request overview

Copilot reviewed 8 out of 9 changed files in this pull request and generated 1 comment.

Comments suppressed due to low confidence (4)

pkg/beholder/batch_emitter_service.go:27

• Adding a map field makes ChipIngressBatchEmitterService no longer comparable. Comparability is part of the exported type’s API surface (and is enforced by api-diff in this repo), similar to the rationale noted in chip_ingress_emitter.go. To preserve comparability, store resource attributes behind a pointer wrapper (nil meaning “not configured”).

	batchClient   *batch.Client
	resourceAttrs map[string]string

pkg/beholder/batch_emitter_service.go:99

• Resource attributes are currently stored as a (possibly empty) map and always passed through WithResourceAttributeExtensions. With an empty map, this still allocates/sorts on every emit (sanitizeResourceAttributeKeys). If no resource attrs are configured, skip the EventOpt entirely by keeping resourceAttrs nil and falling back to chipingress.NewEvent.

	e := &ChipIngressBatchEmitterService{
		batchClient:   batchClient,
		resourceAttrs: resourceAttributesToStringMap(cfg.ResourceAttributes),
		metrics:       metrics,
	}

pkg/beholder/batch_emitter_service.go:151

• emitInternal unconditionally applies WithResourceAttributeExtensions, which does extra work even when no resource attributes are configured. Once resourceAttrs is stored as nil when unset, this can branch to NewEvent (no opts) to avoid per-event sanitization/sorting overhead.

		attributes := newAttributes(attrKVs...)

		event, err := chipingress.NewEventWithOpts(domain, entity, body, attributes, chipingress.WithResourceAttributeExtensions(e.resourceAttrs))
		if err != nil {
			return fmt.Errorf("failed to create CloudEvent: %w", err)
		}

pkg/beholder/chip_ingress_emitter.go:64

• NewWithResourceAttributes stores the provided attrs map directly. Since this is an exported API, callers could mutate the map after construction, which can cause data races or panics when emitting (concurrent map read/write). Consider defensively cloning the map before storing it.

	var resourceAttrs *resourceAttrExtensions
	if len(attrs) > 0 {
		resourceAttrs = &resourceAttrExtensions{attrs: attrs}
	}

[pkcll]

Done in a4260cd — resourceAttributesToStringMap now returns nil for empty input, so the no-attributes case (the common one) allocates nothing. The caller gates on len(resourceAttrs) > 0, which holds for a nil map. Test updated to assert nil for both nil and empty-slice input.

[Copilot]

Pull request overview

Copilot reviewed 7 out of 8 changed files in this pull request and generated no new comments.

Comments suppressed due to low confidence (1)

pkg/beholder/resource_attributes.go:8

• The doc comment claims this helper is the single source of truth for both gRPC metadata headers and CloudEvent extension keys/values, but in the current codebase it’s only used to derive gRPC metadata headers (the only call site is in client.go). This is misleading for future maintainers; either wire the CloudEvent extension propagation here as well or adjust the comment to match current behavior.

// resourceAttributesToStringMap converts OTel resource attributes into a plain string map,
// using attribute.Value.Emit for canonical stringification of any value type. This is the
// single source of truth used to derive both the gRPC metadata headers and the CloudEvent
// extension keys/values sent to ChipIngress, so both mechanisms stay consistent.

[github-actions]

This PR is stale because it has been open 30 days with no activity.
Remove the stale label or comment or this will be closed in 7 days.

[pkcll]

Data flow (cross-posted from the server-side design doc in smartcontractkit/atlas#13201)

For easier review — what this PR wires up, end to end:

flowchart LR
    subgraph node["Chainlink node"]
        toml["TOML config<br/>Telemetry.ResourceAttributes"]
        beholder["beholder.Config.ResourceAttributes<br/>pkg/beholder"]
        client["chipingress client<br/>SanitizeMetadataHeaders:<br/>whitelist projection, once per client<br/>pkg/chipingress"]
        toml --> beholder --> client
    end

    subgraph ingress["chip-ingress"]
        auth["auth interceptor<br/>verified CSA token"]
        fwd["resourceHeadersFromContext<br/>reads the 10 fixed keys only,<br/>first value wins, sorted"]
        base["baseKafkaHeaders<br/>built once per request,<br/>reused by every record"]
        auth -->|"server-derived:<br/>ce_csapublickey<br/>ce_nodeoperatorname"| base
        fwd -->|"client-asserted:<br/>resource_*"| base
    end

    kafka[("Kafka records<br/>cre topic")]

    client -->|"gRPC metadata, once per request:<br/>chainlink-resource-service-name: chainlink<br/>chainlink-resource-csa-public-key: abc123<br/>(non-whitelisted attributes dropped client-side)"| ingress
    base -->|"fanned out onto every record:<br/>resource_service.name: chainlink<br/>resource_csa_public_key: abc123"| kafka

Loading

This PR is the pkg/beholder box: it converts beholder.Config.ResourceAttributes to a string map and hands it to chipingress.WithResourceAttributeHeaders; the whitelist projection, header naming, and server fan-out all live in #2288 (merged) and smartcontractkit/atlas#13201.