Add chip_client metric label for Chip Ingress batch clients

go.mod

@@ -43,7 +43,7 @@ require (
 	github.com/scylladb/go-reflectx v1.0.1
 	github.com/shopspring/decimal v1.4.0
 	github.com/smartcontractkit/chain-selectors v1.0.100
-	github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260626151909-052e55e62e62
+	github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260713174528-c9ce07a477bb
 	github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20251024234028-0988426d98f4
 	github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260622152157-c8e129347b8b
 	github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20251002192024-d2ad9222409b

pkg/beholder/batch_emitter_service.go

@@ -78,6 +78,7 @@ func NewChipIngressBatchEmitterService(client chipingress.Client, cfg Config, lg
 		batch.WithShutdownTimeout(drainTimeout),
 		batch.WithMaxConcurrentSends(maxConcurrentSends),
 		batch.WithEventClone(false),
+		batch.WithChipClient(batch.ChipClientBeholder),
 	)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create batch client: %w", err)
@@ -186,6 +187,7 @@ func (e *ChipIngressBatchEmitterService) metricAttrsFor(domain, entity string) o
 	attrs := otelmetric.WithAttributeSet(attribute.NewSet(
 		attribute.String("domain", domain),
 		attribute.String("entity", entity),
+		attribute.String("chip_client", batch.ChipClientBeholder),
 	))
 	v, _ := e.metricAttrsCache.LoadOrStore(key, attrs)
 	return v.(otelmetric.MeasurementOption)

pkg/beholder/batch_emitter_service_test.go

@@ -19,6 +19,7 @@ import (
 
 	"github.com/smartcontractkit/chainlink-common/pkg/beholder"
 	"github.com/smartcontractkit/chainlink-common/pkg/chipingress"
+	"github.com/smartcontractkit/chainlink-common/pkg/chipingress/batch"
 	"github.com/smartcontractkit/chainlink-common/pkg/chipingress/mocks"
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 )
@@ -497,6 +498,7 @@ func TestChipIngressBatchEmitterService_Metrics(t *testing.T) {
 		sum, ok := metric.Data.(metricdata.Sum[int64])
 		require.True(t, ok)
 		dp := mustEmitterInt64SumPoint(t, sum, "domain", "platform", "entity", "MetricEvent")
+		assert.True(t, hasEmitterStringAttr(dp.Attributes, "chip_client", batch.ChipClientBeholder))
 		assert.GreaterOrEqual(t, dp.Value, int64(1))
 	})
 
@@ -540,6 +542,7 @@ func TestChipIngressBatchEmitterService_Metrics(t *testing.T) {
 		sum, ok := metric.Data.(metricdata.Sum[int64])
 		require.True(t, ok)
 		dp := mustEmitterInt64SumPoint(t, sum, "domain", "platform", "entity", "MetricDropEvent")
+		assert.True(t, hasEmitterStringAttr(dp.Attributes, "chip_client", batch.ChipClientBeholder))
 		assert.GreaterOrEqual(t, dp.Value, int64(1))
 
 		logs := observed.FilterMessage("failed to emit to chip ingress")

pkg/durableemitter/durable_emitter.go

@@ -18,6 +18,7 @@ import (
 	"google.golang.org/protobuf/proto"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/chipingress"
+	chipingressbatch "github.com/smartcontractkit/chainlink-common/pkg/chipingress/batch"
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 	"github.com/smartcontractkit/chainlink-common/pkg/services"
 )
@@ -239,7 +240,7 @@ func NewDurableEmitter(
 			return nil, errors.New("durable emitter metrics enabled but meter is nil")
 		}
 		var err error
-		m, err = newDurableEmitterMetrics(meter)
+		m, err = newDurableEmitterMetrics(meter, chipingressbatch.ChipClientDurableEmitter)
 		if err != nil {
 			return nil, fmt.Errorf("durable emitter metrics: %w", err)
 		}

pkg/durableemitter/durable_emitter_metrics.go

@@ -26,7 +26,7 @@ type DurableEmitterMetricsConfig struct {
 	MaxQueuePayloadBytes int64
 }
 
-// publishPhase identifies which delivery path recorded a batch publish metric.
+// publishPhase identifies the delivery path recorded by publish metrics.
 type publishPhase int
 
 const (
@@ -46,6 +46,7 @@ func (p publishPhase) String() string {
 }
 
 type durableEmitterMetrics struct {
+	chipClient         string
 	emitSuccess        metric.Int64Counter
 	emitFail           metric.Int64Counter
 	emitDuration       metric.Float64Histogram
@@ -94,11 +95,13 @@ var durationBuckets = metric.WithExplicitBucketBoundaries(
 // newDurableEmitterMetrics registers all DurableEmitter instruments on the
 // supplied meter. The caller is responsible for the meter's scope (the
 // instrument prefix below acts as the metric namespace).
-func newDurableEmitterMetrics(meter metric.Meter) (*durableEmitterMetrics, error) {
+func newDurableEmitterMetrics(meter metric.Meter, chipClient string) (*durableEmitterMetrics, error) {
 	if meter == nil {
 		return nil, fmt.Errorf("durable emitter metrics: meter is nil")
 	}
-	m := &durableEmitterMetrics{}
+	m := &durableEmitterMetrics{
+		chipClient: chipClient,
+	}
 	var err error
 	if m.emitSuccess, err = meter.Int64Counter(
 		"durable_emitter.emit.success",
@@ -147,7 +150,7 @@ func newDurableEmitterMetrics(meter metric.Meter) (*durableEmitterMetrics, error
 	if m.publishDuration, err = meter.Float64Histogram(
 		"durable_emitter.publish.duration",
 		metric.WithUnit("s"),
-		metric.WithDescription("Chip Ingress Publish RPC duration (seconds); labels: phase={batch,retransmit}, error={true,false}"),
+		metric.WithDescription("Chip Ingress Publish RPC duration; labels: phase={batch,retransmit}, error={true,false}, chip_client"),
 		durationBuckets,
 	); err != nil {
 		return nil, err
@@ -345,6 +348,7 @@ func (m *durableEmitterMetrics) recordPublish(ctx context.Context, elapsed time.
 		metric.WithAttributes(
 			attribute.String("phase", phase.String()),
 			attribute.Bool("error", err != nil),
+			attribute.String("chip_client", m.chipClient),
 		),
 	)
 }
@@ -353,7 +357,10 @@ func (m *durableEmitterMetrics) recordPublishBatchEvent(ctx context.Context, pha
 	if m == nil {
 		return
 	}
-	attrs := metric.WithAttributes(attribute.String("phase", phase.String()))
+	attrs := metric.WithAttributes(
+		attribute.String("phase", phase.String()),
+		attribute.String("chip_client", m.chipClient),
+	)
 	if err != nil {
 		m.publishBatchEvErr.Add(ctx, 1, attrs)
 	} else {

pkg/durableemitter/durable_emitter_test.go

@@ -16,6 +16,7 @@ import (
 	cepb "github.com/cloudevents/sdk-go/binding/format/protobuf/v2/pb"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
 	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
 	"go.opentelemetry.io/otel/sdk/metric/metricdata"
@@ -299,8 +300,8 @@ func TestDurableEmitter_HooksBatchPublishPath(t *testing.T) {
 	var pubCalls, markCalls atomic.Int32
 	cfg := DefaultConfig()
 	cfg.Hooks = &Hooks{
-		OnBatchPublish:       func(time.Duration, int, error) { pubCalls.Add(1) },
-		OnBatchDelete: func(time.Duration, int) { markCalls.Add(1) },
+		OnBatchPublish: func(time.Duration, int, error) { pubCalls.Add(1) },
+		OnBatchDelete:  func(time.Duration, int) { markCalls.Add(1) },
 	}
 	em, err := NewDurableEmitter(store, be, true, cfg, logger.Test(t), nil)
 	require.NoError(t, err)
@@ -320,8 +321,8 @@ func TestDurableEmitter_HooksPublishFailureSkipsMarkHook(t *testing.T) {
 	var pubCalls, markCalls atomic.Int32
 	cfg := DefaultConfig()
 	cfg.Hooks = &Hooks{
-		OnBatchPublish:       func(time.Duration, int, error) { pubCalls.Add(1) },
-		OnBatchDelete: func(time.Duration, int) { markCalls.Add(1) },
+		OnBatchPublish: func(time.Duration, int, error) { pubCalls.Add(1) },
+		OnBatchDelete:  func(time.Duration, int) { markCalls.Add(1) },
 	}
 	em, err := NewDurableEmitter(store, be, true, cfg, logger.Test(t), nil)
 	require.NoError(t, err)
@@ -815,6 +816,47 @@ func counterSumByPhase(t *testing.T, rm metricdata.ResourceMetrics, name, phase
 	return total
 }
 
+func assertMetricHasChipClientValue(t *testing.T, rm metricdata.ResourceMetrics, name, chipClient string) {
+	t.Helper()
+	for _, sm := range rm.ScopeMetrics {
+		for _, m := range sm.Metrics {
+			if m.Name != name {
+				continue
+			}
+			var count int
+			check := func(attrs attribute.Set) {
+				count++
+				assert.True(t, hasMetricStringAttr(attrs, "chip_client", chipClient),
+					"metric %s missing chip_client=%q", name, chipClient)
+			}
+			switch data := m.Data.(type) {
+			case metricdata.Sum[int64]:
+				for _, dp := range data.DataPoints {
+					check(dp.Attributes)
+				}
+			case metricdata.Histogram[float64]:
+				for _, dp := range data.DataPoints {
+					check(dp.Attributes)
+				}
+			default:
+				t.Fatalf("metric %s has unsupported type %T", name, m.Data)
+			}
+			require.NotZero(t, count, "metric %s has no datapoints", name)
+			return
+		}
+	}
+	t.Fatalf("metric %q not found", name)
+}
+
+func hasMetricStringAttr(set attribute.Set, key, want string) bool {
+	for _, kv := range set.ToSlice() {
+		if string(kv.Key) == key {
+			return kv.Value.AsString() == want
+		}
+	}
+	return false
+}
+
 func TestDurableEmitter_MetricsPublishBatchEventPhase(t *testing.T) {
 	meter, reader := newTestMeter(t)
 
@@ -864,6 +906,10 @@ func TestDurableEmitter_MetricsPublishBatchEventPhase(t *testing.T) {
 	assert.Equal(t, int64(0), counterSumByPhase(t, rm, "durable_emitter.publish.batch.events.failure", "retransmit"))
 	assert.GreaterOrEqual(t, counterSumByPhase(t, rm, "durable_emitter.publish.batch.events.success", "retransmit"), int64(1))
 	assert.Equal(t, int64(0), counterSumByPhase(t, rm, "durable_emitter.publish.batch.events.success", "batch"))
+
+	assertMetricHasChipClientValue(t, rm, "durable_emitter.publish.batch.events.failure", batch.ChipClientDurableEmitter)
+	assertMetricHasChipClientValue(t, rm, "durable_emitter.publish.batch.events.success", batch.ChipClientDurableEmitter)
+	assertMetricHasChipClientValue(t, rm, "durable_emitter.publish.duration", batch.ChipClientDurableEmitter)
 }
 
 // mockChipServer implements ChipIngressServer with controllable behaviour.

pkg/durableemitter/setup.go

@@ -114,6 +114,7 @@ func Setup(
 		chipingressbatch.WithMessageBuffer(defaultInt(cfg.MessageBufferSize, 10_000)),
 		chipingressbatch.WithMaxPublishTimeout(defaultDuration(cfg.MaxPublishTimeout, 5*time.Second)),
 		chipingressbatch.WithShutdownTimeout(defaultDuration(cfg.ShutdownTimeout, 30*time.Second)),
+		chipingressbatch.WithChipClient(chipingressbatch.ChipClientDurableEmitter),
 	)
 	if err != nil {
 		_ = batchChipClient.Close()