Wire up ChIP ingress client for TelemetryIngress

.changeset/tricky-actors-run.md

@@ -0,0 +1,5 @@
+---
+"chainlink": patch
+---
+
+#updated Wire up CHIP ingress client in telemetry manager

core/config/docs/core.toml

@@ -105,6 +105,8 @@ SendInterval = '500ms' # Default
 SendTimeout = '10s' # Default
 # UseBatchSend toggles sending telemetry to the ingress server using the batch client.
 UseBatchSend = true # Default
+# ChipIngressEnabled enables sending telemetry to CHIP Ingress.
+ChipIngressEnabled = false # Default
 
 [[TelemetryIngress.Endpoints]] # Example
 # Network aka EVM, Solana, Starknet

core/config/telemetry_ingress_config.go

@@ -14,6 +14,7 @@ type TelemetryIngress interface {
 	SendTimeout() time.Duration
 	UseBatchSend() bool
 	Endpoints() []TelemetryIngressEndpoint
+	ChipIngressEnabled() bool
 }
 
 type TelemetryIngressEndpoint interface {

core/config/toml/types.go

@@ -730,14 +730,15 @@ func (d *DatabaseBackup) setFrom(f *DatabaseBackup) {
 }
 
 type TelemetryIngress struct {
-	UniConn      *bool
-	Logging      *bool
-	BufferSize   *uint16
-	MaxBatchSize *uint16
-	SendInterval *commonconfig.Duration
-	SendTimeout  *commonconfig.Duration
-	UseBatchSend *bool
-	Endpoints    []TelemetryIngressEndpoint `toml:",omitempty"`
+	UniConn            *bool
+	Logging            *bool
+	BufferSize         *uint16
+	MaxBatchSize       *uint16
+	SendInterval       *commonconfig.Duration
+	SendTimeout        *commonconfig.Duration
+	UseBatchSend       *bool
+	Endpoints          []TelemetryIngressEndpoint `toml:",omitempty"`
+	ChipIngressEnabled *bool
 }
 
 type TelemetryIngressEndpoint struct {
@@ -772,6 +773,9 @@ func (t *TelemetryIngress) setFrom(f *TelemetryIngress) {
 	if v := f.Endpoints; v != nil {
 		t.Endpoints = v
 	}
+	if v := f.ChipIngressEnabled; v != nil {
+		t.ChipIngressEnabled = v
+	}
 }
 
 type AuditLogger struct {

core/services/chainlink/config_telemetry_ingress.go

@@ -56,6 +56,10 @@ func (t *telemetryIngressConfig) Endpoints() []config.TelemetryIngressEndpoint {
 	return endpoints
 }
 
+func (t *telemetryIngressConfig) ChipIngressEnabled() bool {
+	return *t.c.ChipIngressEnabled
+}
+
 func (t *telemetryIngressEndpointConfig) Network() string {
 	return *t.c.Network
 }

core/services/chainlink/config_telemetry_ingress_test.go

@@ -6,6 +6,8 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+
+	"github.com/smartcontractkit/chainlink/v2/core/config/toml"
 )
 
 func TestTelemetryIngressConfig(t *testing.T) {
@@ -32,3 +34,25 @@ func TestTelemetryIngressConfig(t *testing.T) {
 	assert.Equal(t, "prom.test", tec[0].URL().String())
 	assert.Equal(t, "test-pub-key", tec[0].ServerPubKey())
 }
+
+func TestTelemetryIngressConfig_ChipIngressEnabled(t *testing.T) {
+	t.Run("returns false when ChipIngressEnabled is explicitly false", func(t *testing.T) {
+		falseVal := false
+		config := &telemetryIngressConfig{
+			c: toml.TelemetryIngress{
+				ChipIngressEnabled: &falseVal,
+			},
+		}
+		assert.False(t, config.ChipIngressEnabled())
+	})
+
+	t.Run("returns true when ChipIngressEnabled is true", func(t *testing.T) {
+		trueVal := true
+		config := &telemetryIngressConfig{
+			c: toml.TelemetryIngress{
+				ChipIngressEnabled: &trueVal,
+			},
+		}
+		assert.True(t, config.ChipIngressEnabled())
+	})
+}

core/services/chainlink/config_test.go

@@ -315,13 +315,14 @@ func TestConfig_Marshal(t *testing.T) {
 		},
 	}
 	full.TelemetryIngress = toml.TelemetryIngress{
-		UniConn:      ptr(false),
-		Logging:      ptr(true),
-		BufferSize:   ptr[uint16](1234),
-		MaxBatchSize: ptr[uint16](4321),
-		SendInterval: commoncfg.MustNewDuration(time.Minute),
-		SendTimeout:  commoncfg.MustNewDuration(5 * time.Second),
-		UseBatchSend: ptr(true),
+		UniConn:            ptr(false),
+		Logging:            ptr(true),
+		BufferSize:         ptr[uint16](1234),
+		MaxBatchSize:       ptr[uint16](4321),
+		SendInterval:       commoncfg.MustNewDuration(time.Minute),
+		SendTimeout:        commoncfg.MustNewDuration(5 * time.Second),
+		UseBatchSend:       ptr(true),
+		ChipIngressEnabled: ptr(false),
 		Endpoints: []toml.TelemetryIngressEndpoint{{
 			Network:      ptr("EVM"),
 			ChainID:      ptr("1"),
@@ -992,6 +993,7 @@ MaxBatchSize = 4321
 SendInterval = '1m0s'
 SendTimeout = '5s'
 UseBatchSend = true
+ChipIngressEnabled = false
 
 [[TelemetryIngress.Endpoints]]
 Network = 'EVM'

core/services/chainlink/testdata/config-empty-effective.toml

@@ -43,6 +43,7 @@ MaxBatchSize = 50
 SendInterval = '500ms'
 SendTimeout = '10s'
 UseBatchSend = true
+ChipIngressEnabled = false
 
 [AuditLogger]
 Enabled = false

core/services/chainlink/testdata/config-full.toml

@@ -43,6 +43,7 @@ MaxBatchSize = 4321
 SendInterval = '1m0s'
 SendTimeout = '5s'
 UseBatchSend = true
+ChipIngressEnabled = false
 
 [[TelemetryIngress.Endpoints]]
 Network = 'EVM'

core/services/chainlink/testdata/config-multi-chain-effective.toml

@@ -43,6 +43,7 @@ MaxBatchSize = 50
 SendInterval = '500ms'
 SendTimeout = '10s'
 UseBatchSend = true
+ChipIngressEnabled = false
 
 [AuditLogger]
 Enabled = true

core/services/telemetry/manager.go

@@ -8,7 +8,8 @@ import (
 	"github.com/pkg/errors"
 	"github.com/smartcontractkit/libocr/commontypes"
 
-	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+	"github.com/smartcontractkit/chainlink-common/pkg/beholder"
+	"github.com/smartcontractkit/chainlink-common/pkg/chipingress"
 	common "github.com/smartcontractkit/chainlink-common/pkg/logger"
 	"github.com/smartcontractkit/chainlink-common/pkg/services"
 	"github.com/smartcontractkit/chainlink/v2/core/services/keystore"
@@ -32,6 +33,8 @@ type Manager struct {
 	uniConn                     bool
 	useBatchSend                bool
 	MonitoringEndpointGenerator MonitoringEndpointGenerator
+
+	chipIngressClient chipingress.Client
 }
 
 type telemetryEndpoint struct {
@@ -43,16 +46,23 @@ type telemetryEndpoint struct {
 }
 
 // NewManager create a new telemetry manager that is responsible for configuring telemetry agents and generating the defined telemetry endpoints and monitoring endpoints
-func NewManager(cfg config.TelemetryIngress, csaKeyStore keystore.CSA, lggr logger.Logger) *Manager {
+func NewManager(cfg config.TelemetryIngress, csaKeyStore keystore.CSA, lggr common.Logger) *Manager {
+	var chipIngressClient chipingress.Client
+	if cfg.ChipIngressEnabled() {
+		lggr.Info("ChIP Ingress is enabled for telemetry")
+		chipIngressClient = beholder.GetClient().Chip
+	}
+
 	m := &Manager{
-		bufferSize:   cfg.BufferSize(),
-		ks:           csaKeyStore,
-		logging:      cfg.Logging(),
-		maxBatchSize: cfg.MaxBatchSize(),
-		sendInterval: cfg.SendInterval(),
-		sendTimeout:  cfg.SendTimeout(),
-		uniConn:      cfg.UniConn(),
-		useBatchSend: cfg.UseBatchSend(),
+		bufferSize:        cfg.BufferSize(),
+		ks:                csaKeyStore,
+		logging:           cfg.Logging(),
+		maxBatchSize:      cfg.MaxBatchSize(),
+		sendInterval:      cfg.SendInterval(),
+		sendTimeout:       cfg.SendTimeout(),
+		uniConn:           cfg.UniConn(),
+		useBatchSend:      cfg.UseBatchSend(),
+		chipIngressClient: chipIngressClient,
 	}
 	m.Service, m.eng = services.Config{
 		Name: "TelemetryManager",
@@ -102,7 +112,7 @@ func (m *Manager) GenMultitypeMonitoringEndpoint(network string, chainID string,
 	return NewMultiIngressAgent(e.client, network, chainID, contractID)
 }
 
-func (m *Manager) newEndpoint(e config.TelemetryIngressEndpoint, lggr logger.Logger, cfg config.TelemetryIngress) (services.Service, error) {
+func (m *Manager) newEndpoint(e config.TelemetryIngressEndpoint, lggr common.Logger, cfg config.TelemetryIngress) (services.Service, error) {
 	if e.Network() == "" {
 		return nil, errors.New("cannot add telemetry endpoint, network cannot be empty")
 	}
@@ -123,7 +133,7 @@ func (m *Manager) newEndpoint(e config.TelemetryIngressEndpoint, lggr logger.Log
 		return nil, errors.Errorf("cannot add telemetry endpoint for network %q and chainID %q, endpoint already exists", e.Network(), e.ChainID())
 	}
 
-	lggr = logger.Sugared(lggr).Named(e.Network()).Named(e.ChainID())
+	lggr = common.Sugared(lggr).Named(e.Network()).Named(e.ChainID())
 	var tClient synchronization.TelemetryService
 	if m.useBatchSend {
 		tClient = synchronization.NewTelemetryIngressBatchClient(e.URL(), e.ServerPubKey(), m.ks, cfg.Logging(), lggr, cfg.BufferSize(), cfg.MaxBatchSize(), cfg.SendInterval(), cfg.SendTimeout(), cfg.UniConn())

core/services/telemetry/manager_test.go

@@ -23,7 +23,7 @@ import (
 	mocks2 "github.com/smartcontractkit/chainlink/v2/core/services/synchronization/mocks"
 )
 
-func setupMockConfig(t *testing.T, useBatchSend bool) *mocks.TelemetryIngress {
+func setupMockConfig(t *testing.T, useBatchSend bool, chipIngressEnabled bool) *mocks.TelemetryIngress {
 	tic := mocks.NewTelemetryIngress(t)
 	tic.On("BufferSize").Return(uint(123))
 	tic.On("Logging").Return(true)
@@ -32,12 +32,13 @@ func setupMockConfig(t *testing.T, useBatchSend bool) *mocks.TelemetryIngress {
 	tic.On("SendTimeout").Return(time.Second * 7)
 	tic.On("UniConn").Return(true)
 	tic.On("UseBatchSend").Return(useBatchSend)
+	tic.On("ChipIngressEnabled").Return(chipIngressEnabled)
 
 	return tic
 }
 
 func TestManagerAgents(t *testing.T) {
-	tic := setupMockConfig(t, true)
+	tic := setupMockConfig(t, true, false)
 	te := mocks.NewTelemetryIngressEndpoint(t)
 	te.On("Network").Return("network-1")
 	te.On("ChainID").Return("network-1-chainID-1")
@@ -55,7 +56,7 @@ func TestManagerAgents(t *testing.T) {
 	me := tm.GenMonitoringEndpoint("network-1", "network-1-chainID-1", "", "")
 	assert.Equal(t, "*telemetry.TypedIngressAgentBatch", reflect.TypeOf(me).String())
 
-	tic = setupMockConfig(t, false)
+	tic = setupMockConfig(t, false, false)
 	tic.On("Endpoints").Return([]config.TelemetryIngressEndpoint{te})
 	tm = NewManager(tic, ks, lggr)
 	require.Equal(t, "*synchronization.telemetryIngressClient", reflect.TypeOf(tm.endpoints[0].client).String())
@@ -143,7 +144,7 @@ func TestNewManager(t *testing.T) {
 		mockEndpoints = append(mockEndpoints, te)
 	}
 
-	tic := setupMockConfig(t, true)
+	tic := setupMockConfig(t, true, false)
 	tic.On("Endpoints").Return(mockEndpoints)
 
 	lggr, logObs := logger.TestLoggerObserved(t, zapcore.InfoLevel)
@@ -196,7 +197,7 @@ func TestNewManager(t *testing.T) {
 }
 
 func TestCorrectEndpointRouting(t *testing.T) {
-	tic := setupMockConfig(t, true)
+	tic := setupMockConfig(t, true, false)
 	tic.On("Endpoints").Return(nil)
 
 	lggr, obsLogs := logger.TestLoggerObserved(t, zapcore.InfoLevel)
@@ -281,3 +282,26 @@ func TestCorrectEndpointRouting(t *testing.T) {
 		require.Equal(t, []byte(e.chainID), clientSent[i].Telemetry)
 	}
 }
+
+// add test for current changes in manager
+func TestManager_ChipIngressClient(t *testing.T) {
+	t.Run("disabled chip ingress", func(t *testing.T) {
+		tic := setupMockConfig(t, true, false)
+		tic.On("Endpoints").Return(nil)
+
+		lggr, _ := logger.TestLoggerObserved(t, zapcore.InfoLevel)
+		ks := keymocks.NewCSA(t)
+		tm := NewManager(tic, ks, lggr)
+		assert.Nil(t, tm.chipIngressClient)
+	})
+
+	t.Run("enabled chip ingress", func(t *testing.T) {
+		tic := setupMockConfig(t, true, true)
+		tic.On("Endpoints").Return(nil)
+
+		lggr, _ := logger.TestLoggerObserved(t, zapcore.InfoLevel)
+		ks := keymocks.NewCSA(t)
+		tm := NewManager(tic, ks, lggr)
+		assert.NotNil(t, tm.chipIngressClient)
+	})
+}

core/web/resolver/testdata/config-empty-effective.toml

@@ -43,6 +43,7 @@ MaxBatchSize = 50
 SendInterval = '500ms'
 SendTimeout = '10s'
 UseBatchSend = true
+ChipIngressEnabled = false
 
 [AuditLogger]
 Enabled = false

core/web/resolver/testdata/config-full.toml

@@ -43,6 +43,7 @@ MaxBatchSize = 4321
 SendInterval = '1m0s'
 SendTimeout = '5s'
 UseBatchSend = true
+ChipIngressEnabled = false
 
 [[TelemetryIngress.Endpoints]]
 Network = 'EVM'

core/web/resolver/testdata/config-multi-chain-effective.toml

@@ -43,6 +43,7 @@ MaxBatchSize = 50
 SendInterval = '500ms'
 SendTimeout = '10s'
 UseBatchSend = true
+ChipIngressEnabled = false
 
 [AuditLogger]
 Enabled = true