Refactor promotel

Author: pkcll

core/cmd/shell.go

@@ -324,6 +324,7 @@ func (n ChainlinkAppFactory) NewApplication(ctx context.Context, cfg chainlink.G
 		RetirementReportCache:      retirementReportCache,
 		LLOTransmissionReaper:      lloReaper,
 		CapabilitiesRegistry:       capabilitiesRegistry,
+		Registerer:                 appRegisterer,
 	})
 }
 

core/scripts/go.mod

@@ -33,7 +33,6 @@ import (
 	"github.com/smartcontractkit/chainlink-common/pkg/utils/mailbox"
 	evmtypes "github.com/smartcontractkit/chainlink-integrations/evm/types"
 	evmutils "github.com/smartcontractkit/chainlink-integrations/evm/utils"
-	"github.com/smartcontractkit/chainlink/v2/core/services/promotel"
 	"github.com/smartcontractkit/chainlink/v2/core/bridges"
 	"github.com/smartcontractkit/chainlink/v2/core/build"
 	"github.com/smartcontractkit/chainlink/v2/core/capabilities"
@@ -203,6 +202,7 @@ type ApplicationOpts struct {
 	NewOracleFactoryFn         standardcapabilities.NewOracleFactoryFn
 	FetcherFunc                syncer.FetcherFunc
 	FetcherFactoryFn           compute.FetcherFactory
+	Registerer                 prometheus.Registerer
 }
 
 type Heartbeat struct {
@@ -280,8 +280,17 @@ func NewApplication(opts ApplicationOpts) (Application, error) {
 	restrictedHTTPClient := opts.RestrictedHTTPClient
 	unrestrictedHTTPClient := opts.UnrestrictedHTTPClient
 
-	promForwarder := promotel.NewForwarder(globalLogger, prometheus.DefaultGatherer, otel.GetMeterProvider())
-	srvcs = append(srvcs, promForwarder)
+	if beholderClient := beholder.GetClient(); beholderClient != nil {
+		forwarderOpts := promotel.DefaultOptions()
+		forwarderOpts.Endpoint = beholderClient.Config.OtelExporterGRPCEndpoint
+		forwarderOpts.TLSInsecure = beholderClient.Config.InsecureConnection
+		forwarderOpts.AuthHeaders = beholderClient.Config.AuthHeaders
+		promForwarder, err := promotel.NewForwarderService(prometheus.DefaultGatherer, opts.Registerer, globalLogger, forwarderOpts)
+		if err != nil {
+			return nil, fmt.Errorf("could not create prometheus forwarder: %w", err)
+		}
+		srvcs = append(srvcs, promForwarder)
+	}
 
 	if opts.CapabilitiesRegistry == nil {
 		// for tests only, in prod Registry should always be set at this point

core/scripts/go.sum

@@ -2,187 +2,54 @@ package promotel
 
 import (
 	"context"
-	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/client_golang/prometheus/promauto"
-	"go.opentelemetry.io/collector/consumer"
-	"go.opentelemetry.io/collector/pdata/pmetric"
-	"go.opentelemetry.io/otel/metric"
-	"go.uber.org/zap"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 	"github.com/smartcontractkit/chainlink-common/pkg/services"
-	"github.com/smartcontractkit/chainlink-common/pkg/timeutil"
 
 	promotelcommon "github.com/smartcontractkit/chainlink-common/pkg/promotel"
 )
 
 const (
-	period              = 15 * time.Second
-	heartbeatMetricName = "promotel_heartbeat"
+	name = "PromOTELForwarder"
 )
 
-type Forwarder struct {
+type Options = promotelcommon.ForwarderOptions
+type ForwarderService struct {
 	services.StateMachine
-	lggr          logger.Logger
-	gatherer      prometheus.Gatherer
-	meterProvider metric.MeterProvider
-	stopCh        services.StopChan
-	done          chan struct{}
+	lggr      logger.Logger
+	forwarder *promotelcommon.Forwarder
 }
 
-func NewForwarder(lggr logger.Logger, gatherer prometheus.Gatherer, meterProvider metric.MeterProvider) *Forwarder {
-	return &Forwarder{
-		lggr:          logger.Named(lggr, "PromOTELForwarder"),
-		gatherer:      gatherer,
-		meterProvider: meterProvider,
-		stopCh:        make(chan struct{}),
-		done:          make(chan struct{}),
-	}
-}
-
-func (f *Forwarder) HealthReport() map[string]error { return map[string]error{f.Name(): f.Healthy()} }
-
-func (f *Forwarder) Name() string { return f.lggr.Name() }
-
-func (f *Forwarder) Start(context.Context) error {
-	go reportHertbeatMetric(prometheus.DefaultRegisterer, f.lggr, 1*time.Second)
-
-	go f.run()
-	return nil
-}
-
-func (f *Forwarder) run() {
-
-	defer close(f.done)
-	ctx, cancel := f.stopCh.NewCtx()
-	defer cancel()
-	ticker := timeutil.NewTicker(func() time.Duration { return period })
-	defer ticker.Stop()
-
-	go reportHertbeatMetric(prometheus.DefaultRegisterer, f.lggr, 1*time.Second)
-
-	exporter := startExporter(ctx, f.lggr)
-
-	// Fetches metrics from in memory prometheus.Gatherer and converts to OTel format
-	receiver := f.startMetricReceiver(func(ctx context.Context, md pmetric.Metrics) error {
-		// todo: remove or make configurable
-		f.logOtelMetric(md, f.lggr)
-
-		// Exports the converted OTel metric
-		return exporter.Consumer().ConsumeMetrics(ctx, md)
-	})
-	// Close the receiver and exporter when the forwarder is closed
-	defer func() {
-		if err := receiver.Close(); err != nil {
-			f.lggr.Error("Failed to close scraper", zap.Error(err))
-		}
-		if err := exporter.Close(); err != nil {
-			f.lggr.Error("Failed to close exporter", zap.Error(err))
-		}
-	}()
-
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		case <-ticker.C:
-			f.forward(ctx)
-		}
-	}
-}
-
-func (f *Forwarder) forward(ctx context.Context) {
-	mfs, err := f.gatherer.Gather()
+func NewForwarderService(g prometheus.Gatherer, r prometheus.Registerer, lggr logger.Logger, opts Options) (*ForwarderService, error) {
+	l := logger.Named(lggr, name)
+	forwarder, err := promotelcommon.NewForwarder(g, r, l, opts)
 	if err != nil {
-		f.lggr.Errorw("Failed to gather prometheus metrics", "err", err)
-	}
-	for _, mf := range mfs {
-		for range mf.Metric {
-			if ctx.Err() != nil {
-				return
-			}
-
-			//TODO f.meterProvider.Meter()
-		}
+		return nil, err
 	}
+	return &ForwarderService{
+		lggr:      l,
+		forwarder: forwarder,
+	}, nil
 }
 
-func (f *Forwarder) Close() error {
-	close(f.stopCh)
-	<-f.done
-	return nil
+func (f *ForwarderService) HealthReport() map[string]error {
+	return map[string]error{f.Name(): f.Healthy()}
 }
 
-func reportHertbeatMetric(reg prometheus.Registerer, lggr logger.Logger, interval time.Duration) {
-	heartbeat := promauto.With(reg).NewCounter(prometheus.CounterOpts{
-		Name: heartbeatMetricName,
-		ConstLabels: prometheus.Labels{
-			"source": "promotel",
-		},
-	})
-	for {
-		heartbeat.Inc()
-		lggr.Debugw("Heartbeat promotel")
-		time.Sleep(interval)
-	}
-}
+func (f *ForwarderService) Name() string { return f.lggr.Name() }
 
-func startExporter(ctx context.Context, logger logger.Logger) promotelcommon.MetricExporter {
-	expConfig, err := promotelcommon.NewExporterConfig(map[string]any{
-		"endpoint": "localhost:4317",
-		"tls": map[string]any{
-			"insecure": true,
-		},
+func (f *ForwarderService) Start(ctx context.Context) error {
+	return f.StartOnce(name, func() error {
+		return f.forwarder.Start(ctx)
 	})
-	if err != nil {
-		logger.Fatal("Failed to create exporter config", zap.Error(err))
-	}
-	// Sends metrics data in OTLP format to otel-collector endpoint
-	exporter, err := promotelcommon.NewMetricExporter(expConfig, logger)
-	if err != nil {
-		logger.Fatal("Failed to create metric exporter", zap.Error(err))
-	}
-	err = exporter.Start(ctx)
-	if err != nil {
-		logger.Fatal("Failed to start exporter", zap.Error(err))
-	}
-	return exporter
 }
 
-func (f *Forwarder) startMetricReceiver(next consumer.ConsumeMetricsFunc) promotelcommon.Runnable {
-	f.lggr.Info("Starting promotel metric receiver")
-	config, err := promotelcommon.NewDefaultReceiverConfig()
-	if err != nil {
-		f.lggr.Fatal("Failed to create config", zap.Error(err))
-	}
-
-	// Gather metrics via promotel
-	// MetricReceiver fetches metrics from prometheus.Gatherer, then converts it to OTel format and writes formatted metrics to stdout
-	receiver, err := promotelcommon.NewMetricReceiver(config, f.gatherer, next, f.lggr)
-	if err != nil {
-		f.lggr.Fatal("Failed to create debug metric receiver", zap.Error(err))
-	}
-	// Starts the promotel
-	if err := receiver.Start(context.Background()); err != nil {
-		f.lggr.Fatal("Failed to start metric receiver", zap.Error(err))
-	}
-	return receiver
+func (f *ForwarderService) Close() error {
+	return f.StopOnce(name, f.forwarder.Close)
 }
 
-func (f *Forwarder) logOtelMetric(md pmetric.Metrics, logger logger.Logger) {
-	rms := md.ResourceMetrics()
-	for i := 0; i < rms.Len(); i++ {
-		rm := rms.At(i)
-		ilms := rm.ScopeMetrics()
-		for j := 0; j < ilms.Len(); j++ {
-			ilm := ilms.At(j)
-			metrics := ilm.Metrics()
-			for k := 0; k < metrics.Len(); k++ {
-				metric := metrics.At(k)
-				logger.Debug("Exporting OTel metric ", zap.Any("name", metric.Name()), zap.Any("value", metric.Sum().DataPoints().At(0).DoubleValue()))
-			}
-		}
-	}
+func DefaultOptions() Options {
+	return promotelcommon.DefaultForwarderOptions()
 }

core/services/chainlink/application.go

@@ -0,0 +1,76 @@
+package promotel
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+	"github.com/smartcontractkit/wsrpc/logger"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+)
+
+func TestForwarder(t *testing.T) {
+	var (
+		g              = prometheus.DefaultGatherer
+		r              = prometheus.DefaultRegisterer
+		lggr, observed = logger.TestObserved(t, zap.DebugLevel)
+		testMetricName = t.Name() + "_test_counter_metric"
+		interval       = 10 * time.Millisecond
+	)
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	go reportTestMetrics(ctx, r, testMetricName)
+
+	doneCh := make(chan struct{})
+	go func() {
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			default:
+				for _, l := range observed.All() {
+					metricName, ok := l.ContextMap()["name"].(string)
+					if ok && strings.Contains(metricName, testMetricName) {
+						doneCh <- struct{}{}
+					}
+				}
+				time.Sleep(1 * time.Second)
+			}
+		}
+	}()
+
+	forwarder, err := NewForwarderService(g, r, lggr, Options{
+		Endpoint:    "localhost:4317",
+		TLSInsecure: true,
+		Interval:    interval,
+		Verbose:     true,
+	})
+	require.NoError(t, err)
+	require.NoError(t, forwarder.Start(ctx))
+	defer forwarder.Close()
+
+	select {
+	case <-ctx.Done():
+		t.Fatal("Test timed out. Expected metric not found")
+	case <-doneCh:
+		t.Log("Found metric.")
+	}
+}
+
+func reportTestMetrics(ctx context.Context, reg prometheus.Registerer, metricName string) {
+	m := promauto.With(reg).NewCounter(prometheus.CounterOpts{Name: metricName})
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+			m.Inc()
+			time.Sleep(1 * time.Second)
+		}
+	}
+}