beholder heartbeat

Author: pkcll

pkg/beholder/config.go

@@ -49,6 +49,9 @@ type Config struct {
 	LogRetryConfig      *RetryConfig
 	LogStreamingEnabled bool // Enable logs streaming to the OTel log exporter
 
+	// Heartbeat
+	HeartbeatEnabled bool // Enable periodic heartbeat emission
+
 	// Auth
 	AuthPublicKeyHex string
 	AuthHeaders      map[string]string
@@ -115,6 +118,7 @@ func DefaultConfig() Config {
 		LogMaxQueueSize:       2048,
 		LogBatchProcessor:     true,
 		LogStreamingEnabled:   true, // Enable logs streaming by default
+		HeartbeatEnabled:      true, // Enable heartbeat by default
 	}
 }
 

pkg/beholder/heartbeat.go

@@ -0,0 +1,277 @@
+package beholder
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+	"go.opentelemetry.io/otel/trace"
+	"google.golang.org/protobuf/proto"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/beholder/pb"
+	"github.com/smartcontractkit/chainlink-common/pkg/config/build"
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+	"github.com/smartcontractkit/chainlink-common/pkg/services"
+	"github.com/smartcontractkit/chainlink-common/pkg/timeutil"
+)
+
+// Heartbeat represents a periodic heartbeat service that emits metrics and logs
+type Heartbeat struct {
+	services.Service
+	eng *services.Engine
+
+	beat        time.Duration
+	emitter     Emitter
+	meter       metric.Meter
+	logger      logger.Logger
+	tracer      trace.Tracer
+	appID       string
+	serviceName string
+	version     string
+	commit      string
+	labels      map[string]string
+}
+
+// NewHeartbeat creates a new heartbeat service with custom configuration
+func NewHeartbeat(beat time.Duration, lggr logger.Logger, opts ...HeartbeatOpt) *Heartbeat {
+	// Setup default emitter, meter, and tracer
+	noopClient := NewNoopClient()
+
+	// Create heartbeat with defaults
+	h := &Heartbeat{
+		beat:        beat,
+		logger:      lggr,
+		emitter:     noopClient.Emitter,
+		meter:       noopClient.Meter,
+		tracer:      noopClient.Tracer,
+		appID:       "chainlink",          // Default app ID
+		serviceName: build.Program,        // Default service name
+		version:     build.Version,        // Use build version
+		commit:      build.ChecksumPrefix, // Use build commit
+		labels:      make(map[string]string),
+	}
+
+	// Apply options
+	for _, opt := range opts {
+		opt(h)
+	}
+
+	// Build labels from current values
+	h.labels = map[string]string{
+		"service": h.serviceName,
+		"version": h.version,
+		"commit":  h.commit,
+	}
+	if h.appID != "" {
+		h.labels["app_id"] = h.appID
+	}
+
+	// Create service engine
+	h.Service, h.eng = services.Config{
+		Name:  "BeholderHeartbeat",
+		Start: h.start,
+	}.NewServiceEngine(lggr)
+
+	return h
+}
+
+// HeartbeatOpt is a functional option for configuring the heartbeat
+type HeartbeatOpt func(*Heartbeat)
+
+// WithEmitter sets a custom message emitter for the heartbeat
+func WithEmitter(emitter Emitter) HeartbeatOpt {
+	return func(h *Heartbeat) {
+		h.emitter = emitter
+	}
+}
+
+// WithMeter sets a custom meter for the heartbeat
+func WithMeter(meter metric.Meter) HeartbeatOpt {
+	return func(h *Heartbeat) {
+		h.meter = meter
+	}
+}
+
+// WithTracer sets a custom tracer for the heartbeat
+func WithTracer(tracer trace.Tracer) HeartbeatOpt {
+	return func(h *Heartbeat) {
+		h.tracer = tracer
+	}
+}
+
+// WithAppID sets a custom app ID for the heartbeat
+func WithAppID(appID string) HeartbeatOpt {
+	return func(h *Heartbeat) {
+		h.appID = appID
+		if appID != "" {
+			h.labels["app_id"] = appID
+		} else {
+			delete(h.labels, "app_id")
+		}
+	}
+}
+
+// WithServiceName sets a custom service name for the heartbeat
+func WithServiceName(serviceName string) HeartbeatOpt {
+	return func(h *Heartbeat) {
+		h.serviceName = serviceName
+		h.labels["service"] = serviceName
+	}
+}
+
+// WithVersion sets a custom version for the heartbeat
+func WithVersion(version string) HeartbeatOpt {
+	return func(h *Heartbeat) {
+		h.version = version
+		h.labels["version"] = version
+	}
+}
+
+// WithCommit sets a custom commit for the heartbeat
+func WithCommit(commit string) HeartbeatOpt {
+	return func(h *Heartbeat) {
+		h.commit = commit
+		h.labels["commit"] = commit
+	}
+}
+
+// WithBeatInterval sets a custom beat interval for the heartbeat
+func WithBeatInterval(beat time.Duration) HeartbeatOpt {
+	return func(h *Heartbeat) {
+		h.beat = beat
+	}
+}
+
+// start initializes and starts the heartbeat service
+func (h *Heartbeat) start(ctx context.Context) error {
+	// Create heartbeat metrics
+	heartbeatGauge, err := h.meter.Int64Gauge("beholder_heartbeat")
+	if err != nil {
+		return fmt.Errorf("failed to create heartbeat status gauge: %w", err)
+	}
+
+	heartbeatCount, err := h.meter.Int64Counter("beholder_heartbeat_count")
+	if err != nil {
+		return fmt.Errorf("failed to create heartbeat counter: %w", err)
+	}
+
+	// Define the heartbeat function
+	beatFn := func(ctx context.Context) {
+		start := time.Now()
+
+		// Create a trace span for the heartbeat
+		ctx, span := h.tracer.Start(ctx, "beholder_heartbeat", trace.WithAttributes(
+			attribute.String("service", h.serviceName),
+			attribute.String("app_id", h.appID),
+			attribute.String("version", h.version),
+			attribute.String("commit", h.commit),
+		))
+		defer span.End()
+
+		// Record heartbeat metrics
+		heartbeatGauge.Record(ctx, 1)
+		heartbeatCount.Add(ctx, 1)
+
+		// Emit heartbeat message
+
+		payload := &pb.BaseMessage{
+			Msg:    "beholder heartbeat",
+			Labels: h.labels,
+		}
+		payloadBytes, err := proto.Marshal(payload)
+		if err != nil {
+			// log error
+			h.logger.Errorw("heartbeat marshal protobuf failed", "err", err)
+		}
+
+		err = h.emitter.Emit(ctx, payloadBytes,
+			AttrKeyDataSchema, "/beholder-base-message/versions/1", // required
+			AttrKeyDomain, "platform", // required
+			AttrKeyEntity, "BaseMessage", // required
+			"service", h.serviceName,
+			"app_id", h.appID,
+			"version", h.version,
+			"commit", h.commit,
+			"timestamp", start.Unix(),
+		)
+
+		if err != nil {
+			h.logger.Errorw("heartbeat emit failed", "err", err)
+		}
+
+		// Log heartbeat
+		h.logger.Debugw("beholder heartbeat emitted",
+			"service", h.serviceName,
+			"app_id", h.appID,
+			"version", h.version,
+			"commit", h.commit,
+			"timestamp", start.Unix(),
+		)
+	}
+
+	// Start the heartbeat ticker
+	// Execute immediately first, then continue with regular intervals
+	h.eng.Go(func(ctx context.Context) {
+		beatFn(ctx)
+	})
+	h.eng.GoTick(timeutil.NewTicker(h.GetBeat), beatFn)
+
+	h.logger.Infow("beholder heartbeat service started",
+		"service", h.serviceName,
+		"beat_interval", h.GetBeat(),
+	)
+
+	return nil
+}
+
+// GetBeat returns the heartbeat interval
+func (h *Heartbeat) GetBeat() time.Duration {
+	return h.beat
+}
+
+// GetAppID returns the app ID
+func (h *Heartbeat) GetAppID() string {
+	return h.appID
+}
+
+// GetServiceName returns the service name
+func (h *Heartbeat) GetServiceName() string {
+	return h.serviceName
+}
+
+// GetVersion returns the version
+func (h *Heartbeat) GetVersion() string {
+	return h.version
+}
+
+// GetCommit returns the commit
+func (h *Heartbeat) GetCommit() string {
+	return h.commit
+}
+
+// GetLabels returns the labels
+func (h *Heartbeat) GetLabels() map[string]string {
+	return h.labels
+}
+
+// GetEmitter returns the message emitter
+func (h *Heartbeat) GetEmitter() Emitter {
+	return h.emitter
+}
+
+// GetMeter returns the meter
+func (h *Heartbeat) GetMeter() metric.Meter {
+	return h.meter
+}
+
+// GetLogger returns the logger
+func (h *Heartbeat) GetLogger() logger.Logger {
+	return h.logger
+}
+
+// GetTracer returns the tracer
+func (h *Heartbeat) GetTracer() trace.Tracer {
+	return h.tracer
+}

pkg/beholder/heartbeat_test.go

@@ -0,0 +1,71 @@
+package beholder_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/beholder"
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+)
+
+func TestHeartbeat_NewHeartbeat(t *testing.T) {
+	lggr, err := logger.New()
+	require.NoError(t, err)
+
+	heartbeat := beholder.NewHeartbeat(
+		1*time.Second,
+		lggr,
+		beholder.WithAppID("test-app"),
+		beholder.WithServiceName("test-service"),
+		beholder.WithVersion("1.0.0"),
+		beholder.WithCommit("abc123"),
+	)
+	require.NotNil(t, heartbeat)
+
+	assert.Equal(t, "test-app", heartbeat.GetAppID())
+	assert.Equal(t, "test-service", heartbeat.GetServiceName())
+	assert.Equal(t, "1.0.0", heartbeat.GetVersion())
+	assert.Equal(t, "abc123", heartbeat.GetCommit())
+	assert.Equal(t, 1*time.Second, heartbeat.GetBeat())
+	assert.NotNil(t, heartbeat.GetEmitter())
+	assert.NotNil(t, heartbeat.GetMeter())
+}
+
+func TestHeartbeat_Start(t *testing.T) {
+	lggr, err := logger.New()
+	require.NoError(t, err)
+
+	heartbeat := beholder.NewHeartbeat(100*time.Millisecond, lggr)
+	require.NotNil(t, heartbeat)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)
+	defer cancel()
+
+	err = heartbeat.Start(ctx)
+	require.NoError(t, err)
+
+	// Wait for at least one heartbeat
+	time.Sleep(150 * time.Millisecond)
+
+	err = heartbeat.Close()
+	require.NoError(t, err)
+}
+
+func TestHeartbeat_Defaults(t *testing.T) {
+	lggr, err := logger.New()
+	require.NoError(t, err)
+
+	heartbeat := beholder.NewHeartbeat(1*time.Second, lggr)
+	require.NotNil(t, heartbeat)
+
+	// Check defaults
+	assert.Equal(t, "chainlink", heartbeat.GetAppID())
+	assert.Equal(t, "github.com/smartcontractkit/chainlink-common", heartbeat.GetServiceName())
+	assert.Equal(t, "(devel)", heartbeat.GetVersion())
+	assert.Equal(t, "unset", heartbeat.GetCommit())
+	assert.Equal(t, 1*time.Second, heartbeat.GetBeat())
+}