Wire resource attributes into Beholder ChipIngress emitters

[pkcll]

Summary

Wire node resource attributes from beholder.Config.ResourceAttributes into the ChipIngress batch and sync emitters via CloudEvent extensions, so they reach the cre topic as ce_<sanitized> Kafka headers.

Changes

• pkg/beholder/batch_emitter_service.go: store resource attributes and stamp them on each emitted event via chipingress.WithResourceAttributeExtensions.
• pkg/beholder/chip_ingress_emitter.go: same for the sync emitter.
• pkg/beholder/client.go: pass resource attributes into the ChipIngress client header provider.
• pkg/beholder/resource_attributes.go: add resourceAttributesToStringMap helper.
• Root go.mod: bump pkg/chipingress to the resource-attributes branch (Add resource attribute support to ChipIngress client #2267).

Related PRs

• Upstream (chipingress client): Add resource attribute support to ChipIngress client #2267
• Downstream (chainlink): bump chainlink-common for chip-ingress resource attributes chainlink#23014

Test plan

• go test ./pkg/beholder/...

Notes

• The previous Co-authored-by: Cursor attribution has been removed from the commit history.
• gRPC metadata → Kafka header forwarding is unconfirmed on the chip-ingress server side; the CE-extension path guarantees ce_<name> headers immediately.
• pkg/durableemitter remains out of scope.

[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 - github.com/smartcontractkit/chainlink-common

✅ Compatible Changes (1)

pkg/beholder.ChipIngressEmitterConfig (1)

• NewWithResourceAttributes — ➕ Added

---

📄 View full apidiff 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 3 comments.

Comments suppressed due to low confidence (1)

pkg/beholder/chip_ingress_emitter.go:64

• NewWithResourceAttributes stores the caller-provided map[string]string directly. Because maps are reference types, a caller could mutate the map after construction (or share it across goroutines), leading to data races or non-deterministic extensions. Copy the map before storing it internally.

	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 (1)

pkg/beholder/chip_ingress_emitter.go:64

• NewWithResourceAttributes stores the provided attrs map directly. Because Emit can be called concurrently, any external mutation of the map after construction could cause a data race (and surprising behavior). Consider defensively cloning the map before storing it in 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/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}
	}