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.20260707100839-18a1fc4374eb
 	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("beholder"),
 	)
 	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", "beholder"),
 	))
 	v, _ := e.metricAttrsCache.LoadOrStore(key, attrs)
 	return v.(otelmetric.MeasurementOption)

pkg/beholder/batch_emitter_service_test.go

@@ -497,6 +497,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", "beholder"))
 		assert.GreaterOrEqual(t, dp.Value, int64(1))
 	})
 
@@ -540,6 +541,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", "beholder"))
 		assert.GreaterOrEqual(t, dp.Value, int64(1))
 
 		logs := observed.FilterMessage("failed to emit to chip ingress")

pkg/chipingress/batch/client.go

@@ -70,6 +70,7 @@ type Client struct {
 	batcherDone             chan struct{}
 	started                 bool
 	counters                sync.Map // map[seqnumKey]*atomic.Uint64 for per-(source,type) seqnum, cleared on Stop()
+	chipClient              string
 
 	metrics batchClientMetrics
 
@@ -88,6 +89,7 @@ type batchClientMetrics struct {
 	maxGRPCReqSizeAttr   otelmetric.MeasurementOption
 	successStatusAttr    otelmetric.MeasurementOption
 	failureStatusAttr    otelmetric.MeasurementOption
+	chipClientAttr       otelmetric.MeasurementOption
 }
 
 // Opt is a functional option for configuring the batch Client.
@@ -118,7 +120,7 @@ func NewBatchClient(client chipingress.Client, opts ...Opt) (*Client, error) {
 	}
 
 	var err error
-	c.metrics, err = newBatchClientMetrics()
+	c.metrics, err = newBatchClientMetrics(c.chipClient)
 	if err != nil {
 		return nil, err
 	}
@@ -291,7 +293,7 @@ func (b *Client) sendBatch(ctx context.Context, messages []*messageWithCallback)
 
 		splitBatches := splitMessagesByRequestSize(messages, b.effectiveMaxRequestSize, b.transactionEnabled)
 		if len(splitBatches) > 1 {
-			b.metrics.batchSplitsTotal.Add(ctx, 1)
+			b.metrics.batchSplitsTotal.Add(ctx, 1, b.metrics.chipClientAddOpts()...)
 		}
 		for _, batchMessages := range splitBatches {
 			batchReq, batchBytes := newBatchRequest(batchMessages, b.transactionEnabled)
@@ -350,7 +352,7 @@ func (b *Client) completeBatchCallbacksFromResults(messages []*messageWithCallba
 			"results", len(results),
 			"messages", len(messages),
 		)
-		b.metrics.resultsMismatchTotal.Add(context.Background(), 1)
+		b.metrics.resultsMismatchTotal.Add(context.Background(), 1, b.metrics.chipClientAddOpts()...)
 	}
 
 	b.callbackWg.Go(func() {
@@ -377,7 +379,7 @@ func (b *Client) completeBatchCallbacksFromResults(messages []*messageWithCallba
 					"expected", msg.event.Id,
 					"got", result.EventId,
 				)
-				b.metrics.resultsMismatchTotal.Add(context.Background(), 1)
+				b.metrics.resultsMismatchTotal.Add(context.Background(), 1, b.metrics.chipClientAddOpts()...)
 			}
 			if result.Error != nil {
 				msg.callback(&PublishError{
@@ -511,6 +513,13 @@ func WithLogger(log *zap.SugaredLogger) Opt {
 	}
 }
 
+// WithChipClient sets chip_client on batch client metrics. Omitted when unset.
+func WithChipClient(name string) Opt {
+	return func(c *Client) {
+		c.chipClient = name
+	}
+}
+
 // WithTransactionEnabled sets PublishOptions.transaction_enabled on every
 // batch request. The option is always emitted on the wire so client intent
 // is explicit in traces/logs; the server treats unset and explicit false
@@ -525,7 +534,14 @@ func WithTransactionEnabled(transactionEnabled bool) Opt {
 	}
 }
 
-func newBatchClientMetrics() (batchClientMetrics, error) {
+func batchMetricAttributeSet(chipClient string, kvs ...attribute.KeyValue) attribute.Set {
+	if chipClient != "" {
+		kvs = append(kvs, attribute.String("chip_client", chipClient))
+	}
+	return attribute.NewSet(kvs...)
+}
+
+func newBatchClientMetrics(chipClient string) (batchClientMetrics, error) {
 	meter := otel.Meter("chipingress/batch_client")
 	sendRequestsTotal, err := meter.Int64Counter(
 		"chip_ingress.batch.send_requests_total",
@@ -590,6 +606,13 @@ func newBatchClientMetrics() (batchClientMetrics, error) {
 		return batchClientMetrics{}, err
 	}
 
+	var chipClientAttr otelmetric.MeasurementOption
+	if chipClient != "" {
+		chipClientAttr = otelmetric.WithAttributeSet(attribute.NewSet(
+			attribute.String("chip_client", chipClient),
+		))
+	}
+
 	return batchClientMetrics{
 		sendRequestsTotal:    sendRequestsTotal,
 		requestSizeMessages:  requestSizeMessages,
@@ -598,23 +621,32 @@ func newBatchClientMetrics() (batchClientMetrics, error) {
 		configInfo:           configInfo,
 		batchSplitsTotal:     batchSplitsTotal,
 		resultsMismatchTotal: resultsMismatchTotal,
-		successStatusAttr: otelmetric.WithAttributeSet(attribute.NewSet(
+		chipClientAttr:       chipClientAttr,
+		successStatusAttr: otelmetric.WithAttributeSet(batchMetricAttributeSet(chipClient,
 			attribute.String("status", "success"),
 		)),
-		failureStatusAttr: otelmetric.WithAttributeSet(attribute.NewSet(
+		failureStatusAttr: otelmetric.WithAttributeSet(batchMetricAttributeSet(chipClient,
 			attribute.String("status", "failure"),
 		)),
 	}, nil
 }
 
+func (m *batchClientMetrics) chipClientAddOpts() []otelmetric.AddOption {
+	if m.chipClientAttr != nil {
+		return []otelmetric.AddOption{m.chipClientAttr}
+	}
+	return nil
+}
+
 func (m *batchClientMetrics) recordConfig(ctx context.Context, c *Client) {
-	m.batchSizeAttr = otelmetric.WithAttributeSet(attribute.NewSet(
+	chipClient := c.chipClient
+	m.batchSizeAttr = otelmetric.WithAttributeSet(batchMetricAttributeSet(chipClient,
 		attribute.Int("max_batch_size", c.batchSize),
 	))
-	m.maxGRPCReqSizeAttr = otelmetric.WithAttributeSet(attribute.NewSet(
+	m.maxGRPCReqSizeAttr = otelmetric.WithAttributeSet(batchMetricAttributeSet(chipClient,
 		attribute.Int("max_grpc_request_size_bytes", c.maxGRPCRequestSize),
 	))
-	m.configInfo.Record(ctx, 1, otelmetric.WithAttributes(
+	m.configInfo.Record(ctx, 1, otelmetric.WithAttributeSet(batchMetricAttributeSet(chipClient,
 		attribute.Int("max_batch_size", c.batchSize),
 		attribute.Int("message_buffer_size", cap(c.messageBuffer)),
 		attribute.Int("max_concurrent_sends", cap(c.maxConcurrentSends)),
@@ -624,7 +656,7 @@ func (m *batchClientMetrics) recordConfig(ctx context.Context, c *Client) {
 		attribute.Bool("clone_event", c.cloneEvent),
 		attribute.Bool("transaction_enabled", c.transactionEnabled),
 		attribute.Int("max_grpc_request_size_bytes", c.maxGRPCRequestSize),
-	))
+	)))
 }
 
 func (m *batchClientMetrics) recordSend(ctx context.Context, messageCount int, requestBytes int, latency time.Duration, success bool) {

pkg/chipingress/batch/client_test.go

@@ -1566,6 +1566,185 @@ func TestBatchClient_Metrics(t *testing.T) {
 	})
 }
 
+func TestBatchClient_ChipClientMetricAttribute(t *testing.T) {
+	const chipClient = "test_client"
+
+	batchMetricNames := []string{
+		"chip_ingress.batch.send_requests_total",
+		"chip_ingress.batch.request_size_messages",
+		"chip_ingress.batch.request_size_bytes",
+		"chip_ingress.batch.request_latency_ms",
+		"chip_ingress.batch.config.info",
+		"chip_ingress.batch.batch_splits_total",
+		"chip_ingress.batch.results_mismatch_total",
+	}
+
+	t.Run("with WithChipClient sets chip_client on all batch metrics", func(t *testing.T) {
+		reader, restore := useTestMeterProvider(t)
+		defer restore()
+
+		events := []*chipingress.CloudEventPb{
+			largeTestEvent("chip-client-1"),
+			largeTestEvent("chip-client-2"),
+			largeTestEvent("chip-client-3"),
+		}
+		msgs2 := []*messageWithCallback{{event: events[0]}, {event: events[1]}}
+		_, maxRequestSize := newBatchRequest(msgs2, false)
+
+		mockClient := mocks.NewClient(t)
+		done := make(chan struct{})
+		var mu sync.Mutex
+		var publishCount int
+		mockClient.
+			On("PublishBatch", mock.Anything, mock.Anything).
+			Return(&chipingress.PublishResponse{}, nil).
+			Run(func(_ mock.Arguments) {
+				mu.Lock()
+				publishCount++
+				if publishCount == 2 {
+					close(done)
+				}
+				mu.Unlock()
+			})
+
+		client, err := NewBatchClient(
+			mockClient,
+			WithChipClient(chipClient),
+			WithBatchSize(1),
+			WithBatchInterval(time.Second),
+			WithMessageBuffer(10),
+			WithMaxGRPCRequestSize(minMaxGRPCRequestSize),
+		)
+		require.NoError(t, err)
+		client.maxGRPCRequestSize = maxRequestSize
+		client.effectiveMaxRequestSize = maxRequestSize
+		client.metrics.recordConfig(t.Context(), client)
+
+		messages := make([]*messageWithCallback, 0, len(events))
+		for _, event := range events {
+			messages = append(messages, &messageWithCallback{event: event})
+		}
+		client.sendBatch(t.Context(), messages)
+
+		select {
+		case <-done:
+		case <-time.After(time.Second):
+			t.Fatal("timeout waiting for split batches")
+		}
+
+		client.completeBatchCallbacksFromResults(
+			[]*messageWithCallback{
+				{event: &chipingress.CloudEventPb{Id: "m1", Source: "s", Type: "t"}},
+				{event: &chipingress.CloudEventPb{Id: "m2", Source: "s", Type: "t"}},
+			},
+			[]*chipingress.PublishResult{{EventId: "m1"}},
+		)
+		client.completeBatchCallbacksFromResults(
+			[]*messageWithCallback{
+				{event: &chipingress.CloudEventPb{Id: "m1", Source: "s", Type: "t"}},
+			},
+			[]*chipingress.PublishResult{{EventId: "wrong-id"}},
+		)
+
+		rm := collectResourceMetrics(t, reader)
+		for _, name := range batchMetricNames {
+			metric := mustMetric(t, rm, name)
+			assertMetricHasChipClient(t, metric, chipClient)
+		}
+	})
+
+	t.Run("without WithChipClient omits chip_client", func(t *testing.T) {
+		reader, restore := useTestMeterProvider(t)
+		defer restore()
+
+		mockClient := mocks.NewClient(t)
+		mockClient.EXPECT().Close().Return(nil).Maybe()
+		done := make(chan struct{})
+		mockClient.
+			On("PublishBatch", mock.Anything, mock.Anything).
+			Return(&chipingress.PublishResponse{}, nil).
+			Run(func(_ mock.Arguments) { close(done) }).
+			Once()
+
+		client, err := NewBatchClient(
+			mockClient,
+			WithBatchSize(1),
+			WithBatchInterval(time.Second),
+			WithMessageBuffer(10),
+		)
+		require.NoError(t, err)
+		client.Start(t.Context())
+
+		require.NoError(t, client.QueueMessage(&chipingress.CloudEventPb{
+			Id: "no-chip-client", Source: "platform", Type: "Test",
+		}, nil))
+
+		select {
+		case <-done:
+		case <-time.After(time.Second):
+			t.Fatal("timeout waiting for PublishBatch")
+		}
+		client.Stop()
+
+		rm := collectResourceMetrics(t, reader)
+		for _, sm := range rm.ScopeMetrics {
+			for _, metric := range sm.Metrics {
+				if !strings.HasPrefix(metric.Name, "chip_ingress.batch.") {
+					continue
+				}
+				assertMetricOmitsChipClient(t, metric)
+			}
+		}
+	})
+}
+
+func assertMetricHasChipClient(t *testing.T, metric metricdata.Metrics, chipClient string) {
+	t.Helper()
+	forEachMetricAttrSet(t, metric, func(attrs attribute.Set) {
+		assert.True(t, hasStringAttr(attrs, "chip_client", chipClient),
+			"metric %s missing chip_client=%q", metric.Name, chipClient)
+	})
+}
+
+func assertMetricOmitsChipClient(t *testing.T, metric metricdata.Metrics) {
+	t.Helper()
+	forEachMetricAttrSet(t, metric, func(attrs attribute.Set) {
+		for _, kv := range attrs.ToSlice() {
+			assert.NotEqual(t, "chip_client", string(kv.Key), "metric %s should not have chip_client", metric.Name)
+		}
+	})
+}
+
+func forEachMetricAttrSet(t *testing.T, metric metricdata.Metrics, fn func(attribute.Set)) {
+	t.Helper()
+	var count int
+	record := func(attrs attribute.Set) {
+		count++
+		fn(attrs)
+	}
+	switch data := metric.Data.(type) {
+	case metricdata.Sum[int64]:
+		for _, dp := range data.DataPoints {
+			record(dp.Attributes)
+		}
+	case metricdata.Histogram[int64]:
+		for _, dp := range data.DataPoints {
+			record(dp.Attributes)
+		}
+	case metricdata.Histogram[float64]:
+		for _, dp := range data.DataPoints {
+			record(dp.Attributes)
+		}
+	case metricdata.Gauge[int64]:
+		for _, dp := range data.DataPoints {
+			record(dp.Attributes)
+		}
+	default:
+		t.Fatalf("metric %s has unsupported type %T", metric.Name, metric.Data)
+	}
+	require.NotZero(t, count, "metric %s has no datapoints", metric.Name)
+}
+
 func TestSplitMessagesByRequestSize(t *testing.T) {
 	t.Run("empty messages returns nil", func(t *testing.T) {
 		result := splitMessagesByRequestSize(nil, 1024, false)