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,52 @@ var (
 	initGlobalsOnce sync.Once
 	prometheus      *ginprom.Prometheus
 	grpcOpts        loop.GRPCOpts
+	beholderClient  beholder.Client
 )
 
-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() == "" {
+			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(),
+			PackageName:              cfgBeholder.PackageName(),
+			ResourceAttributes:       attributes,
+			EmitterExportTimeout:     cfgBeholder.EmitterExportTimeout(),
+			TraceSampleRate:          cfgBeholder.TraceSampleRate(),
+			TraceBatchTimeout:        cfgBeholder.TraceBatchTimeout(),
+			MetricReaderInterval:     cfgBeholder.MetricReaderInterval(),
+			LogExportTimeout:         cfgBeholder.LogExportTimeout(),
+		}
+		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 +171,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 +273,7 @@ func (n ChainlinkAppFactory) NewApplication(ctx context.Context, cfg chainlink.G
 		GRPCOpts:                   grpcOpts,
 		MercuryPool:                mercuryPool,
 		CapabilitiesRegistry:       capabilitiesRegistry,
+		BeholderClient:             beholderClient,
 	})
 }
 

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,18 @@
+package config
+
+import (
+	"time"
+)
+
+type Beholder interface {
+	InsecureConnection() bool
+	CACertFile() string
+	OtelExporterGRPCEndpoint() string
+	PackageName() string
+	ResourceAttributes() map[string]string
+	EmitterExportTimeout() time.Duration
+	TraceSampleRate() float64
+	TraceBatchTimeout() time.Duration
+	MetricReaderInterval() time.Duration
+	LogExportTimeout() time.Duration
+}

core/config/docs/core.toml

@@ -651,3 +651,27 @@ 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]
+# CACertFile TODO
+CACertFile           = 'cert-file' # Example
+# Endpoint TODO
+Endpoint             = 'example.com/beholder' # Example
+# EmitterExportTimeout TODO
+EmitterExportTimeout = '30s' # Default
+# InsecureConnection TODO
+InsecureConnection   = false # Default
+# LogExportTimeout TODO
+LogExportTimeout     = '20s' # Default
+# MetricReaderInterval TODO
+MetricReaderInterval = '10s' # Default
+# PackageName TODO
+PackageName          = 'foo' # Example
+# TraceBatchTimeout TODO
+TraceBatchTimeout    = '15s' # Default
+# TraceSampleRate TODO
+TraceSampleRate      = 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 {
+	//TODO explicit enable? Or just empty endpoint?
+	CACertFile           *string
+	Endpoint             *string
+	EmitterExportTimeout *commonconfig.Duration
+	InsecureConnection   *bool
+	LogExportTimeout     *commonconfig.Duration
+	MetricReaderInterval *commonconfig.Duration
+	PackageName          *string           //TODO remove?
+	ResourceAttributes   map[string]string `toml:",omitempty"`
+	TraceBatchTimeout    *commonconfig.Duration
+	TraceSampleRate      *float64
+}
+
+func (b *Beholder) setFrom(f *Beholder) {
+	if v := f.CACertFile; v != nil {
+		b.CACertFile = v
+	}
+	if v := f.Endpoint; v != nil {
+		b.Endpoint = v
+	}
+	if v := f.EmitterExportTimeout; v != nil {
+		b.EmitterExportTimeout = v
+	}
+	if v := f.InsecureConnection; v != nil {
+		b.InsecureConnection = v
+	}
+	if v := f.LogExportTimeout; v != nil {
+		b.LogExportTimeout = v
+	}
+	if v := f.MetricReaderInterval; v != nil {
+		b.MetricReaderInterval = v
+	}
+	if v := f.PackageName; v != nil {
+		b.PackageName = v
+	}
+	if v := f.ResourceAttributes; v != nil {
+		b.ResourceAttributes = v
+	}
+	if v := f.TraceBatchTimeout; v != nil {
+		b.TraceBatchTimeout = v
+	}
+	if v := f.TraceSampleRate; v != nil {
+		b.TraceSampleRate = v
+	}
+}
+
+func (b *Beholder) ValidateConfig() (err error) {
+	//TODO implement - all required?!
+	//TODO insecure with dev mode only?
+	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)

core/scripts/go.mod

@@ -22,7 +22,7 @@ require (
 	github.com/prometheus/client_golang v1.17.0
 	github.com/shopspring/decimal v1.4.0
 	github.com/smartcontractkit/chainlink-automation v1.0.4
-	github.com/smartcontractkit/chainlink-common v0.2.2-0.20240808143317-6b16fc28887d
+	github.com/smartcontractkit/chainlink-common v0.2.2-0.20240813195740-bdd303bb6abb
 	github.com/smartcontractkit/chainlink/v2 v2.0.0-00010101000000-000000000000
 	github.com/smartcontractkit/libocr v0.0.0-20240717100443-f6226e09bee7
 	github.com/spf13/cobra v1.8.0
@@ -119,7 +119,7 @@ require (
 	github.com/fatih/color v1.16.0 // indirect
 	github.com/fsnotify/fsnotify v1.6.0 // indirect
 	github.com/fxamacker/cbor/v2 v2.6.0 // indirect
-	github.com/gabriel-vasile/mimetype v1.4.2 // indirect
+	github.com/gabriel-vasile/mimetype v1.4.3 // indirect
 	github.com/gagliardetto/binary v0.7.7 // indirect
 	github.com/gagliardetto/solana-go v1.8.4 // indirect
 	github.com/gagliardetto/treeout v0.1.4 // indirect
@@ -149,7 +149,7 @@ require (
 	github.com/go-openapi/swag v0.22.4 // indirect
 	github.com/go-playground/locales v0.14.1 // indirect
 	github.com/go-playground/universal-translator v0.18.1 // indirect
-	github.com/go-playground/validator/v10 v10.15.5 // indirect
+	github.com/go-playground/validator/v10 v10.22.0 // indirect
 	github.com/go-webauthn/webauthn v0.9.4 // indirect
 	github.com/go-webauthn/x v0.1.5 // indirect
 	github.com/goccy/go-json v0.10.2 // indirect
@@ -178,7 +178,7 @@ require (
 	github.com/grpc-ecosystem/go-grpc-middleware/providers/prometheus v1.0.1 // indirect
 	github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.1.0 // indirect
 	github.com/grpc-ecosystem/grpc-gateway v1.16.0 // indirect
-	github.com/grpc-ecosystem/grpc-gateway/v2 v2.20.0 // indirect
+	github.com/grpc-ecosystem/grpc-gateway/v2 v2.21.0 // indirect
 	github.com/gsterjov/go-libsecret v0.0.0-20161001094733-a6f4afe4910c // indirect
 	github.com/gtank/merlin v0.1.1 // indirect
 	github.com/gtank/ristretto255 v0.1.2 // indirect
@@ -219,7 +219,7 @@ require (
 	github.com/kr/text v0.2.0 // indirect
 	github.com/kylelemons/godebug v1.1.0 // indirect
 	github.com/leanovate/gopter v0.2.10-0.20210127095200-9abe2343507a // indirect
-	github.com/leodido/go-urn v1.2.4 // indirect
+	github.com/leodido/go-urn v1.4.0 // indirect
 	github.com/lib/pq v1.10.9 // indirect
 	github.com/libp2p/go-buffer-pool v0.1.0 // indirect
 	github.com/logrusorgru/aurora v2.0.3+incompatible // indirect
@@ -343,8 +343,8 @@ require (
 	golang.org/x/tools v0.23.0 // indirect
 	gonum.org/v1/gonum v0.15.0 // indirect
 	google.golang.org/genproto v0.0.0-20240711142825-46eb208f015d // indirect
-	google.golang.org/genproto/googleapis/api v0.0.0-20240711142825-46eb208f015d // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20240711142825-46eb208f015d // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20240730163845-b1a4ccb954bf // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20240730163845-b1a4ccb954bf // indirect
 	google.golang.org/grpc v1.65.0 // indirect
 	gopkg.in/guregu/null.v4 v4.0.0 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect