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.
• 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 — bump pkg/chipingress.

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. Values are
  sanitized upstream in pkg/chipingress to prevent this.

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 the CloudEvent's 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, which requires a resource_ prefix inbound and
emits the key unchanged — 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 prefixing lives in #2288, which namespaces every emitted metadata key under
resource_. This PR needs no code change for it: it calls WithResourceAttributeHeaders, so
prefixing happens inside pkg/chipingress, and its test asserts through SanitizeMetadataHeaders
rather than hardcoded key names. Verified by running this suite against #2288's branch with a local
replace.

It does need the pkg/chipingress pin bumped once #2288 merges. Until then the
Validate go.mod dependencies job is expected to fail, as it has throughout this stack.

Consumers reading platformEnv today should move to resource_platformenv. 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 value on the legacy
topic.

Related

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

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

[patrickhuie19]

any way to avoid the local replace?

[pkcll]

For sure, will clean this up.

[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 (2)

pkg/beholder/batch_emitter_service.go:26

• ChipIngressBatchEmitterService is an exported struct type; adding a map[string]string field makes the type non-comparable, which can be an API-breaking change (similar to the concern noted in chip_ingress_emitter.go). Consider storing resource attributes behind a pointer wrapper (or another comparable container) to preserve type comparability, and only dereference it when stamping extensions.

type ChipIngressBatchEmitterService struct {
	services.Service
	eng *services.Engine

	batchClient   *batch.Client
	resourceAttrs map[string]string

pkg/beholder/chip_ingress_emitter.go:64

• NewWithResourceAttributes stores the caller-provided attrs map directly. Because maps are mutable and not concurrency-safe, a caller mutating the map after construction (or concurrently with Emit) can cause data races or panics. Defensive-copy the map on entry before storing it on the emitter.

	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 no new comments.

Comments suppressed due to low confidence (2)

pkg/beholder/chip_ingress_emitter.go:64

• NewWithResourceAttributes stores the passed-in attrs map directly. Because maps are reference types, callers can mutate it after construction (or concurrently), which can change emitted events and potentially introduce data races. Consider defensively copying the map before storing it on the emitter.

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

pkg/beholder/batch_emitter_service.go:150

• emitInternal always uses NewEventWithOpts + WithResourceAttributeExtensions, even when no resource attributes are configured (resourceAttrs is an empty map). This adds overhead on every event and may inadvertently change behavior vs the previous NewEvent path. Consider only applying the opt when resourceAttrs is non-empty (mirroring ChipIngressEmitter.Emit).

		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)
		}

[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}
	}

[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.