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

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's
  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 resource attributes as prefixed 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 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 (open, draft, implemented and passing its own
suite), 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, validation, and limits live in #2288, which namespaces every emitted
metadata key under resource_, replaces character-rewriting with validate-and-omit for invalid keys
and non-printable values, caps attributes at 32 entries / 128B key / 512B value / 4096B combined, and
warns + meters every attribute it has to drop. This PR needs no functional code change for any of
that: it calls WithResourceAttributeHeaders, so all of it happens inside pkg/chipingress.

One test-only fix is required at pin-bump time. SanitizeMetadataHeaders changed signature in
#2288, from func(map[string]string) map[string]string to
func(map[string]string) (map[string]string, []DroppedAttribute). This PR's
TestNewGRPCClient_AuthHeaderCoexistsWithResourceAttributes (pkg/beholder/client_test.go) currently
calls it as a single-return function:

want := chipingress.SanitizeMetadataHeaders(map[string]string{...})

That compiles today only because this PR still pins the pre-#2288 pkg/chipingress (see below). Once
the pin bumps to a post-#2288 commit, this call site needs updating to
want, _ := chipingress.SanitizeMetadataHeaders(...) before the module builds.

It also needs 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 + validation/limits, required)
• Server: smartcontractkit/atlas#13201
• Downstream (chainlink): bump chainlink-common for chip-ingress resource attributes chainlink#23014

Related PRs

• chipingress: send resource attributes as prefixed 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 chipingress: send resource attributes as prefixed gRPC metadata #2288's 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

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

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