beholder: inject meter in workflow monitoring, gateway, heartbeat (Phase 5)

core/services/chainlink/heartbeat.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"go.opentelemetry.io/otel/metric"
+	"go.opentelemetry.io/otel/trace"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/beholder"
 	"github.com/smartcontractkit/chainlink-common/pkg/custmsg"
@@ -23,6 +24,7 @@ type Heartbeat struct {
 	beat    time.Duration
 	emitter custmsg.MessageEmitter // Add emitter field
 	meter   metric.Meter           // Add meter field
+	tracer  trace.Tracer           // Add tracer field
 }
 
 type HeartbeatConfig struct {
@@ -70,12 +72,12 @@ func NewHeartbeat(cfg HeartbeatConfig, opts ...HeartbeatOpt) Heartbeat {
 		labels["csa_key"] = cfg.CSAPublicKey
 	}
 
-	cme.WithMapLabels(labels)
 	h := Heartbeat{
 		beat:    cfg.Beat,
 		opts:    cfg,
 		emitter: cme.WithMapLabels(labels),
 		meter:   beholder.GetMeter(),
+		tracer:  beholder.GetTracer(),
 	}
 
 	// Apply test options if any
@@ -105,6 +107,12 @@ func WithMeter(meter metric.Meter) HeartbeatOpt {
 	}
 }
 
+func WithTracer(tracer trace.Tracer) HeartbeatOpt {
+	return func(h *Heartbeat) {
+		h.tracer = tracer
+	}
+}
+
 // Update the start method to use the injected emitter if provided
 func (h *Heartbeat) start(_ context.Context) error {
 	// Setup the heartbeat gauge and count
@@ -120,8 +128,7 @@ func (h *Heartbeat) start(_ context.Context) error {
 
 	// Define tick functions
 	beatFn := func(ctx context.Context) {
-		// TODO allow override of tracer provider into engine for beholder
-		_, innerSpan := beholder.GetTracer().Start(ctx, "heartbeat.beat")
+		_, innerSpan := h.tracer.Start(ctx, "heartbeat.beat")
 		defer innerSpan.End()
 
 		gauge.Record(ctx, 1)

core/services/chainlink/heartbeat_test.go

@@ -1,9 +1,7 @@
 package chainlink_test
 
 import (
-	"bytes"
 	"context"
-	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -14,7 +12,8 @@ import (
 	"go.opentelemetry.io/otel/metric/embedded"
 	"go.opentelemetry.io/otel/metric/noop"
 
-	"github.com/smartcontractkit/chainlink-common/pkg/beholder"
+	"github.com/smartcontractkit/chainlink-common/pkg/beholder/beholdertest"
+	"github.com/smartcontractkit/chainlink-common/pkg/custmsg"
 	"github.com/smartcontractkit/chainlink/v2/core/logger"
 	"github.com/smartcontractkit/chainlink/v2/core/services/chainlink"
 )
@@ -55,28 +54,29 @@ func TestHeartbeat_MeterEvents(t *testing.T) {
 	if testing.Short() {
 		t.Skip("too slow for testing.Short")
 	}
+	t.Parallel()
 
-	lggr := logger.TestLogger(t)
-
-	// Use a thread-safe byte collector
-	collector := &byteCollector{}
-	client, err := beholder.NewWriterClient(collector)
-	require.NoError(t, err)
-
-	// Set the global beholder client
-	beholder.SetClient(client)
-	defer beholder.SetClient(beholder.NewNoopClient()) // Reset to prevent test side effects
+	// Inject a per-test recorder instead of swapping the global beholder client
+	rec := beholdertest.NewRecorder()
+	labels := map[string]string{
+		"system":  "Application",
+		"peer_id": "peer-id",
+		"appID":   "app-id",
+	}
 
 	// Create custom metric counter to track gauge calls and set up heartbeat service
 	var heartbeatCounter, heartbeatCountCounter int32
 	mockMeter := newCountingMeter(t, &heartbeatCounter, &heartbeatCountCounter)
 	c := chainlink.HeartbeatConfig{
 		Beat:  50 * time.Millisecond,
-		Lggr:  lggr,
+		Lggr:  logger.TestLogger(t),
 		P2P:   "peer-id",
 		AppID: "app-id",
 	}
-	heartbeat := chainlink.NewHeartbeat(c, chainlink.WithMeter(mockMeter))
+	heartbeat := chainlink.NewHeartbeat(c,
+		chainlink.WithMeter(mockMeter),
+		chainlink.WithEmitter(custmsg.NewLabelerWithEmitter(rec).WithMapLabels(labels)),
+	)
 	require.NoError(t, heartbeat.Start(t.Context()))
 
 	// Wait for ~10 heartbeats
@@ -90,9 +90,17 @@ func TestHeartbeat_MeterEvents(t *testing.T) {
 	assert.InDelta(t, expectedCalls, hb, 2, "Expected ~%d heartbeat gauge calls", expectedCalls)
 	assert.InDelta(t, expectedCalls, hbCount, 2, "Expected ~%d heartbeat count gauge calls", expectedCalls)
 
-	// Check the output buffer for heartbeat messages
-	outputStr := collector.String()
-	assert.Contains(t, outputStr, "heartbeat", "Output should contain heartbeat messages")
+	// Each tick emits one heartbeat custom message through the injected
+	// emitter, so the message count tracks the gauge counts
+	msgs, err := rec.BaseMessagesForLabels(t, labels)
+	require.NoError(t, err)
+	heartbeatMsgs := 0
+	for _, m := range msgs {
+		if m.Msg == "heartbeat" {
+			heartbeatMsgs++
+		}
+	}
+	assert.InDelta(t, expectedCalls, heartbeatMsgs, 2, "Expected ~%d heartbeat messages", expectedCalls)
 }
 
 // mockMeter is a custom implementation of metric.Meter that counts gauge creation specifically for heartbeat metrics.
@@ -138,21 +146,3 @@ func (g *countingGauge) Record(ctx context.Context, value int64, options ...metr
 func (g *countingGauge) Enabled(ctx context.Context) bool {
 	return true
 }
-
-// byteCollector collects all bytes written to it in a thread-safe manner
-type byteCollector struct {
-	mu     sync.Mutex
-	buffer bytes.Buffer
-}
-
-func (bc *byteCollector) Write(p []byte) (n int, err error) {
-	bc.mu.Lock()
-	defer bc.mu.Unlock()
-	return bc.buffer.Write(p)
-}
-
-func (bc *byteCollector) String() string {
-	bc.mu.Lock()
-	defer bc.mu.Unlock()
-	return bc.buffer.String()
-}

core/services/gateway/monitoring/metrics_test.go

@@ -7,24 +7,16 @@ import (
 	"github.com/stretchr/testify/require"
 	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
 	"go.opentelemetry.io/otel/sdk/metric/metricdata"
-
-	"github.com/smartcontractkit/chainlink-common/pkg/beholder"
-	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 )
 
-func TestGatewayMetrics_RecordReadiness(t *testing.T) { //nolint:paralleltest // The test replaces the process-global Beholder client.
+func TestGatewayMetrics_RecordReadiness(t *testing.T) {
+	t.Parallel()
+
 	reader := sdkmetric.NewManualReader()
 	meterProvider := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader))
 	t.Cleanup(func() { require.NoError(t, meterProvider.Shutdown(context.Background())) })
 
-	previousClient := beholder.GetClient()
-	t.Cleanup(func() { beholder.SetClient(previousClient) })
-	client := beholder.NoopClientConfig{Lggr: logger.Test(t)}.New()
-	client.Meter = meterProvider.Meter("gateway-test")
-	client.MeterProvider = meterProvider
-	beholder.SetClient(client)
-
-	metrics, err := NewGatewayMetrics(beholder.GetMeter())
+	metrics, err := NewGatewayMetrics(meterProvider.Meter("gateway-test"))
 	require.NoError(t, err)
 	metrics.RecordDONConnectionState(t.Context(), "workflow_1_zone-a", 5, 7, 10)
 	metrics.RecordUserReady(t.Context(), true)