core/services/promotel: add Forwarder service

Author: jmank88

core/cmd/shell.go

@@ -61,7 +61,7 @@ import (
 
 var (
 	initGlobalsOnce sync.Once
-	prometheus      *ginprom.Prometheus
+	ginPrometheus   *ginprom.Prometheus
 	grpcOpts        loop.GRPCOpts
 )
 
@@ -70,7 +70,7 @@ func initGlobals(cfgProm config.Prometheus, cfgTracing config.Tracing, cfgTeleme
 	var err error
 	initGlobalsOnce.Do(func() {
 		err = func() error {
-			prometheus = ginprom.New(ginprom.Namespace("service"), ginprom.Token(cfgProm.AuthToken()))
+			ginPrometheus = ginprom.New(ginprom.Namespace("service"), ginprom.Token(cfgProm.AuthToken()))
 			grpcOpts = loop.NewGRPCOpts(nil) // default prometheus.Registerer
 
 			otel.SetErrorHandler(otel.ErrorHandlerFunc(func(err error) {
@@ -400,7 +400,7 @@ func (n ChainlinkRunner) Run(ctx context.Context, app chainlink.Application) err
 		return errors.New("You must specify at least one port to listen on")
 	}
 
-	handler, err := web.NewRouter(app, prometheus)
+	handler, err := web.NewRouter(app, ginPrometheus)
 	if err != nil {
 		return errors.Wrap(err, "failed to create web router")
 	}

core/services/chainlink/application.go

@@ -14,6 +14,7 @@ import (
 	"github.com/grafana/pyroscope-go"
 	"github.com/jonboulle/clockwork"
 	"github.com/pkg/errors"
+	"github.com/prometheus/client_golang/prometheus"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/trace"
@@ -26,6 +27,7 @@ import (
 	"github.com/smartcontractkit/chainlink-common/pkg/utils"
 	"github.com/smartcontractkit/chainlink-common/pkg/utils/jsonserializable"
 	"github.com/smartcontractkit/chainlink-common/pkg/utils/mailbox"
+	"github.com/smartcontractkit/chainlink/v2/core/services/promotel"
 
 	"github.com/smartcontractkit/chainlink/v2/core/bridges"
 	"github.com/smartcontractkit/chainlink/v2/core/build"
@@ -205,6 +207,9 @@ 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 opts.CapabilitiesRegistry == nil {
 		// for tests only, in prod Registry should always be set at this point
 		opts.CapabilitiesRegistry = capabilities.NewRegistry(globalLogger)

core/services/promotel/promotel.go

@@ -0,0 +1,81 @@
+package promotel
+
+import (
+	"context"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"go.opentelemetry.io/otel/metric"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+	"github.com/smartcontractkit/chainlink-common/pkg/services"
+	"github.com/smartcontractkit/chainlink-common/pkg/timeutil"
+)
+
+const period = 15 * time.Second
+
+type Forwarder struct {
+	services.StateMachine
+	lggr          logger.Logger
+	gatherer      prometheus.Gatherer
+	meterProvider metric.MeterProvider
+	stopCh        services.StopChan
+	done          chan struct{}
+}
+
+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 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()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			f.forward(ctx)
+		}
+	}
+}
+
+func (f *Forwarder) forward(ctx context.Context) {
+	mfs, err := f.gatherer.Gather()
+	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()
+		}
+	}
+}
+
+func (f *Forwarder) Close() error {
+	close(f.stopCh)
+	<-f.done
+	return nil
+}