integrate beholder client

.mockery.yaml

@@ -526,7 +526,6 @@ packages:
         config:
           filename: evm_mock.go
           dir: "{{ .InterfaceDir }}/rpclibmocks"
-
   github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/prices:
     config:
       dir: "{{ .InterfaceDir }}/"

core/cmd/shell.go

@@ -23,14 +23,15 @@ import (
 	"github.com/Masterminds/semver/v3"
 	"github.com/getsentry/sentry-go"
 	"github.com/gin-gonic/gin"
+	"github.com/jmoiron/sqlx"
 	"github.com/pkg/errors"
 	"github.com/urfave/cli"
+	"go.opentelemetry.io/otel/attribute"
 	"go.uber.org/multierr"
 	"go.uber.org/zap/zapcore"
 	"golang.org/x/sync/errgroup"
 
-	"github.com/jmoiron/sqlx"
-
+	"github.com/smartcontractkit/chainlink-common/pkg/beholder"
 	"github.com/smartcontractkit/chainlink-common/pkg/loop"
 	"github.com/smartcontractkit/chainlink-common/pkg/sqlutil"
 	"github.com/smartcontractkit/chainlink-common/pkg/utils/mailbox"
@@ -61,20 +62,48 @@ var (
 	initGlobalsOnce sync.Once
 	prometheus      *ginprom.Prometheus
 	grpcOpts        loop.GRPCOpts
+	beholderClient  beholder.OtelClient
 )
 
-func initGlobals(cfgProm config.Prometheus, cfgTracing config.Tracing, logger logger.Logger) error {
+func initGlobals(cfgProm config.Prometheus, cfgTracing config.Tracing, cfgBeholder config.Beholder, lggr logger.Logger) error {
 	// Avoid double initializations, but does not prevent relay methods from being called multiple times.
 	var err error
 	initGlobalsOnce.Do(func() {
 		prometheus = ginprom.New(ginprom.Namespace("service"), ginprom.Token(cfgProm.AuthToken()))
 		grpcOpts = loop.NewGRPCOpts(nil) // default prometheus.Registerer
-		err = loop.SetupTracing(loop.TracingConfig{
+
+		tracingCfg := loop.TracingConfig{
 			Enabled:         cfgTracing.Enabled(),
 			CollectorTarget: cfgTracing.CollectorTarget(),
 			NodeAttributes:  cfgTracing.Attributes(),
 			SamplingRatio:   cfgTracing.SamplingRatio(),
-			OnDialError:     func(error) { logger.Errorw("Failed to dial", "err", err) },
+			OnDialError:     func(error) { lggr.Errorw("Failed to dial", "err", err) },
+		}
+		if cfgBeholder.OtelExporterGRPCEndpoint() == "" {
+			beholderClient = beholder.NewNoopClient()
+			err = loop.SetupTracing(tracingCfg)
+			return
+		}
+
+		var attributes []attribute.KeyValue
+		for k, v := range cfgBeholder.ResourceAttributes() {
+			attributes = append(attributes, attribute.String(k, v))
+		}
+		clientCfg := beholder.Config{
+			InsecureConnection:       cfgBeholder.InsecureConnection(),
+			CACertFile:               cfgBeholder.CACertFile(),
+			OtelExporterGRPCEndpoint: cfgBeholder.OtelExporterGRPCEndpoint(),
+			ResourceAttributes:       attributes,
+			TraceSampleRatio:         cfgBeholder.TraceSampleRatio(),
+		}
+		if tracingCfg.Enabled {
+			clientCfg.TraceSpanExporter, err = tracingCfg.NewSpanExporter()
+			if err != nil {
+				return
+			}
+		}
+		beholderClient, err = beholder.NewOtelClient(clientCfg, func(err error) {
+			lggr.Errorw("Failed to emit to beholder", "err", err)
 		})
 	})
 	return err
@@ -138,7 +167,7 @@ type ChainlinkAppFactory struct{}
 
 // NewApplication returns a new instance of the node with the given config.
 func (n ChainlinkAppFactory) NewApplication(ctx context.Context, cfg chainlink.GeneralConfig, appLggr logger.Logger, db *sqlx.DB) (app chainlink.Application, err error) {
-	err = initGlobals(cfg.Prometheus(), cfg.Tracing(), appLggr)
+	err = initGlobals(cfg.Prometheus(), cfg.Tracing(), cfg.Beholder(), appLggr)
 	if err != nil {
 		appLggr.Errorf("Failed to initialize globals: %v", err)
 	}
@@ -240,6 +269,7 @@ func (n ChainlinkAppFactory) NewApplication(ctx context.Context, cfg chainlink.G
 		GRPCOpts:                   grpcOpts,
 		MercuryPool:                mercuryPool,
 		CapabilitiesRegistry:       capabilitiesRegistry,
+		BeholderClient:             beholderClient,
 	})
 }
 

core/cmd/shell_local.go

@@ -22,9 +22,8 @@ import (
 	gethCommon "github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/fatih/color"
-	"github.com/lib/pq"
-
 	"github.com/kylelemons/godebug/diff"
+	"github.com/lib/pq"
 	"github.com/pkg/errors"
 	"github.com/urfave/cli"
 	"go.uber.org/multierr"

core/config/app_config.go

@@ -56,6 +56,7 @@ type AppConfig interface {
 	Threshold() Threshold
 	WebServer() WebServer
 	Tracing() Tracing
+	Beholder() Beholder
 }
 
 type DatabaseBackupMode string

core/config/beholder_config.go

@@ -0,0 +1,9 @@
+package config
+
+type Beholder interface {
+	InsecureConnection() bool
+	CACertFile() string
+	OtelExporterGRPCEndpoint() string
+	ResourceAttributes() map[string]string
+	TraceSampleRatio() float64
+}

core/config/docs/core.toml

@@ -651,3 +651,19 @@ TransmitQueueMaxSize = 10_000 # Default
 # when sending a message to the mercury server, before aborting and considering
 # the transmission to be failed.
 TransmitTimeout = "5s" # Default
+
+[Beholder]
+# Enabled turns Beholder collection on or off.
+Enabled = false # Default
+# Endpoint TODO
+Endpoint = 'example.com/beholder' # Example
+# CACertFile TODO
+CACertFile = 'cert-file' # Example
+# InsecureConnection TODO
+InsecureConnection = false # Default
+# TraceSampleRatio TODO
+TraceSampleRatio = 0.01 # Default
+
+[Beholder.ResourceAttributes]
+# foo is an example resource attribute
+foo = "bar" # Example

core/config/toml/types.go

@@ -57,6 +57,7 @@ type Core struct {
 	Tracing          Tracing          `toml:",omitempty"`
 	Mercury          Mercury          `toml:",omitempty"`
 	Capabilities     Capabilities     `toml:",omitempty"`
+	Beholder         Beholder         `toml:",omitempty"`
 }
 
 // SetFrom updates c with any non-nil values from f. (currently TOML field only!)
@@ -93,6 +94,7 @@ func (c *Core) SetFrom(f *Core) {
 	c.Sentry.setFrom(&f.Sentry)
 	c.Insecure.setFrom(&f.Insecure)
 	c.Tracing.setFrom(&f.Tracing)
+	c.Beholder.setFrom(&f.Beholder)
 }
 
 func (c *Core) ValidateConfig() (err error) {
@@ -1481,25 +1483,25 @@ type Tracing struct {
 
 func (t *Tracing) setFrom(f *Tracing) {
 	if v := f.Enabled; v != nil {
-		t.Enabled = f.Enabled
+		t.Enabled = v
 	}
 	if v := f.CollectorTarget; v != nil {
-		t.CollectorTarget = f.CollectorTarget
+		t.CollectorTarget = v
 	}
 	if v := f.NodeID; v != nil {
-		t.NodeID = f.NodeID
+		t.NodeID = v
 	}
 	if v := f.Attributes; v != nil {
-		t.Attributes = f.Attributes
+		t.Attributes = v
 	}
 	if v := f.SamplingRatio; v != nil {
-		t.SamplingRatio = f.SamplingRatio
+		t.SamplingRatio = v
 	}
 	if v := f.Mode; v != nil {
-		t.Mode = f.Mode
+		t.Mode = v
 	}
 	if v := f.TLSCertPath; v != nil {
-		t.TLSCertPath = f.TLSCertPath
+		t.TLSCertPath = v
 	}
 }
 
@@ -1553,6 +1555,59 @@ func (t *Tracing) ValidateConfig() (err error) {
 	return err
 }
 
+type Beholder struct {
+	Enabled            *bool
+	CACertFile         *string
+	Endpoint           *string
+	InsecureConnection *bool
+	ResourceAttributes map[string]string `toml:",omitempty"`
+	TraceSampleRatio   *float64
+}
+
+func (b *Beholder) setFrom(f *Beholder) {
+	if v := f.Enabled; v != nil {
+		b.Enabled = v
+	}
+	if v := f.CACertFile; v != nil {
+		b.CACertFile = v
+	}
+	if v := f.Endpoint; v != nil {
+		b.Endpoint = v
+	}
+	if v := f.InsecureConnection; v != nil {
+		b.InsecureConnection = v
+	}
+	if v := f.ResourceAttributes; v != nil {
+		b.ResourceAttributes = v
+	}
+	if v := f.TraceSampleRatio; v != nil {
+		b.TraceSampleRatio = v
+	}
+}
+
+func (b *Beholder) ValidateConfig() (err error) {
+	if b.Enabled == nil || !*b.Enabled {
+		return nil
+	}
+	if b.Endpoint == nil || *b.Endpoint == "" {
+		err = multierr.Append(err, configutils.ErrMissing{Name: "Endpoint", Msg: "must be set when Beholder is enabled"})
+	}
+	if b.InsecureConnection != nil && *b.InsecureConnection {
+		if build.IsProd() {
+			err = multierr.Append(err, configutils.ErrInvalid{Name: "InsecureConnection", Msg: "cannot be used in production builds"})
+		}
+	} else {
+		if b.CACertFile == nil || *b.CACertFile == "" {
+			err = multierr.Append(err, configutils.ErrMissing{Name: "CACertFile", Msg: "must be set, unless InsecureConnection is used"})
+		}
+	}
+	if ratio := b.TraceSampleRatio; ratio != nil && (*ratio < 0 || *ratio > 1) {
+		err = multierr.Append(err, configutils.ErrInvalid{Name: "TraceSampleRatio", Value: *ratio, Msg: "must be between 0 and 1"})
+	}
+
+	return nil
+}
+
 var hostnameRegex = regexp.MustCompile(`^[a-zA-Z0-9-]+(\.[a-zA-Z0-9-]+)*$`)
 
 // Validates uri is valid external or local URI

core/internal/cltest/cltest.go

@@ -463,6 +463,7 @@ func NewApplicationWithConfig(t testing.TB, cfg chainlink.GeneralConfig, flagsAn
 		CapabilitiesRegistry:       capabilitiesRegistry,
 		CapabilitiesDispatcher:     dispatcher,
 		CapabilitiesPeerWrapper:    peerWrapper,
+		//TODO beholder client?
 	})
 
 	require.NoError(t, err)