Update so that telemetry can still be configured the old way

Author: george-dorin

core/config/telemetry_ingress_config.go

@@ -17,8 +17,8 @@ type TelemetryIngress interface {
 	UseBatchSend() bool
 	Endpoints() []TelemetryIngressEndpoint
 
-	ServerPubKey() string // Deprecated: Use TelemetryIngressEndpoint.ServerPubKey instead, if this field is set it will trigger an error, only used to warn NOPs of change
-	URL() *url.URL        // Deprecated: Use TelemetryIngressEndpoint.URL instead, if this field is set it will trigger an error, only used to warn NOPs of change
+	ServerPubKey() string // Deprecated: Use TelemetryIngressEndpoint.ServerPubKey instead, this field will be removed in future versions
+	URL() *url.URL        // Deprecated: Use TelemetryIngressEndpoint.URL instead, this field will be removed in future versions
 }
 
 //go:generate mockery --quiet --name TelemetryIngressEndpoint --output ./mocks/ --case=underscore --filename telemetry_ingress_endpoint.go

core/config/toml/types.go

@@ -433,8 +433,8 @@ type TelemetryIngress struct {
 	UseBatchSend *bool
 	Endpoints    []TelemetryIngressEndpoint `toml:",omitempty"`
 
-	URL          *models.URL `toml:",omitempty"` // Deprecated: Use TelemetryIngressEndpoint.URL instead, if this field is set it will trigger an error, only used to warn NOPs of change
-	ServerPubKey *string     `toml:",omitempty"` // Deprecated: Use TelemetryIngressEndpoint.ServerPubKey instead, if this field is set it will trigger an error, only used to warn NOPs of change
+	URL          *models.URL `toml:",omitempty"` // Deprecated: Use TelemetryIngressEndpoint.URL instead, this field will be removed in future versions
+	ServerPubKey *string     `toml:",omitempty"` // Deprecated: Use TelemetryIngressEndpoint.ServerPubKey instead, this field will be removed in future versions
 }
 
 type TelemetryIngressEndpoint struct {
@@ -478,30 +478,14 @@ func (t *TelemetryIngress) setFrom(f *TelemetryIngress) {
 }
 
 func (t *TelemetryIngress) ValidateConfig() (err error) {
-	if !t.URL.IsZero() && *t.ServerPubKey != "" {
+	if (!t.URL.IsZero() || *t.ServerPubKey != "") && len(t.Endpoints) > 0 {
 		return configutils.ErrInvalid{Name: "URL", Value: t.URL.String(),
-			Msg: fmt.Sprintf(`TelemetryIngress.URL and TelemetryIngress.ServerPubKey are no longer allowed. Please use TelemetryIngress.Endpoints instead:
-			[[TelemetryIngress.Endpoints]]
-			Network = '...' # e.g. EVM. Solana, Starknet, Cosmos
-			ChainID = '...' # e.g. 1, 5, devnet, mainnet-beta
-			URL = '%s'
-			ServerPubKey = '%s'`, t.URL.String(), *t.ServerPubKey)}
-	} else if !t.URL.IsZero() {
-		return configutils.ErrInvalid{Name: "URL", Value: t.URL.String(),
-			Msg: fmt.Sprintf(`TelemetryIngress.URL and TelemetryIngress.ServerPubKey are no longer allowed. Please use TelemetryIngress.Endpoints instead:
-			[[TelemetryIngress.Endpoints]]
-			Network = '...' # e.g. EVM. Solana, Starknet, Cosmos
-			ChainID = '...' # e.g. 1, 5, devnet, mainnet-beta
-			URL = '%s'
-			ServerPubKey = '...'`, t.URL.String())}
-	} else if *t.ServerPubKey != "" {
-		return configutils.ErrInvalid{Name: "ServerPubKey", Value: *t.ServerPubKey,
-			Msg: fmt.Sprintf(`TelemetryIngress.URL and TelemetryIngress.ServerPubKey are no longer allowed. Please use TelemetryIngress.Endpoints instead:
+			Msg: `Cannot set both TelemetryIngress.URL and TelemetryIngress.ServerPubKey alongside TelemetryIngress.Endpoints. Please use only TelemetryIngress.Endpoints:
 			[[TelemetryIngress.Endpoints]]
 			Network = '...' # e.g. EVM. Solana, Starknet, Cosmos
 			ChainID = '...' # e.g. 1, 5, devnet, mainnet-beta
 			URL = '...'
-			ServerPubKey = '%s'`, *t.ServerPubKey)}
+			ServerPubKey = '...'`}
 	}
 
 	return nil

core/services/chainlink/config_telemetry_ingress.go

@@ -46,12 +46,12 @@ func (t *telemetryIngressConfig) UseBatchSend() bool {
 	return *t.c.UseBatchSend
 }
 
-// Deprecated: Use TelemetryIngressEndpoint.ServerPubKey instead, if this field is set it will trigger an error, only used to warn NOPs of change
+// Deprecated: Use TelemetryIngressEndpoint.ServerPubKey, this field will be removed in future versions
 func (t *telemetryIngressConfig) ServerPubKey() string {
 	return *t.c.ServerPubKey
 }
 
-// Deprecated: Use TelemetryIngressEndpoint.URL instead, if this field is set it will trigger an error, only used to warn NOPs of change
+// Deprecated: Use TelemetryIngressEndpoint.URL instead, this field will be removed in future versions
 func (t *telemetryIngressConfig) URL() *url.URL {
 	return t.c.URL.URL()
 }

core/services/telemetry/manager.go

@@ -38,6 +38,32 @@ type Manager struct {
 	uniConn                     bool
 	useBatchSend                bool
 	MonitoringEndpointGenerator MonitoringEndpointGenerator
+
+	//legacyMode means that we are sending all telemetry to a single endpoint.
+	//In order for this to be set as true, we need to have no endpoints defined with TelemetryIngress.URL and TelemetryIngress.ServerPubKey set.
+	//This mode will be supported until we completely switch to TelemetryIngress.Endpoints in config.toml
+	legacyMode bool
+}
+
+type legacyEndpointConfig struct {
+	Url    *url.URL
+	PubKey string
+}
+
+func (l *legacyEndpointConfig) Network() string {
+	return "-"
+}
+
+func (l *legacyEndpointConfig) ChainID() string {
+	return "-"
+}
+
+func (l *legacyEndpointConfig) ServerPubKey() string {
+	return l.PubKey
+}
+
+func (l *legacyEndpointConfig) URL() *url.URL {
+	return l.Url
 }
 
 type telemetryEndpoint struct {
@@ -62,13 +88,30 @@ func NewManager(cfg config.TelemetryIngress, csaKeyStore keystore.CSA, lggr logg
 		sendTimeout:  cfg.SendTimeout(),
 		uniConn:      cfg.UniConn(),
 		useBatchSend: cfg.UseBatchSend(),
+		legacyMode:   false,
 	}
 	for _, e := range cfg.Endpoints() {
 		if err := m.addEndpoint(e); err != nil {
-			m.lggr.Error(err.Error())
+			m.lggr.Error(err)
 		}
+	}
 
+	if len(cfg.Endpoints()) == 0 && cfg.URL() != nil && cfg.ServerPubKey() != "" {
+		m.lggr.Error(`TelemetryIngress.URL and TelemetryIngress.ServerPubKey will be removed in a future version, please switch to TelemetryIngress.Endpoints:
+			[[TelemetryIngress.Endpoints]]
+			Network = '...' # e.g. EVM. Solana, Starknet, Cosmos
+			ChainID = '...' # e.g. 1, 5, devnet, mainnet-beta
+			URL = '...'
+			ServerPubKey = '...'`)
+		m.legacyMode = true
+		if err := m.addEndpoint(&legacyEndpointConfig{
+			Url:    cfg.URL(),
+			PubKey: cfg.ServerPubKey(),
+		}); err != nil {
+			m.lggr.Error(err)
+		}
 	}
+
 	return m
 }
 
@@ -124,11 +167,11 @@ func (m *Manager) GenMonitoringEndpoint(contractID string, telemType synchroniza
 }
 
 func (m *Manager) addEndpoint(e config.TelemetryIngressEndpoint) error {
-	if e.Network() == "" {
+	if e.Network() == "" && !m.legacyMode {
 		return errors.New("cannot add telemetry endpoint, network cannot be empty")
 	}
 
-	if e.ChainID() == "" {
+	if e.ChainID() == "" && !m.legacyMode {
 		return errors.New("cannot add telemetry endpoint, chainID cannot be empty")
 	}
 
@@ -164,6 +207,11 @@ func (m *Manager) addEndpoint(e config.TelemetryIngressEndpoint) error {
 }
 
 func (m *Manager) getEndpoint(network string, chainID string) (*telemetryEndpoint, bool) {
+	//in legacy mode we send telemetry to a single endpoint
+	if m.legacyMode && len(m.endpoints) == 1 {
+		return m.endpoints[0], true
+	}
+
 	for _, e := range m.endpoints {
 		if e.Network == strings.ToUpper(network) && e.ChainID == strings.ToUpper(chainID) {
 			return e, true

core/services/telemetry/manager_test.go

@@ -22,6 +22,7 @@ import (
 	mocks3 "github.com/smartcontractkit/chainlink/v2/core/services/keystore/mocks"
 	"github.com/smartcontractkit/chainlink/v2/core/services/synchronization"
 	mocks2 "github.com/smartcontractkit/chainlink/v2/core/services/synchronization/mocks"
+	"github.com/smartcontractkit/chainlink/v2/core/store/models"
 	"github.com/smartcontractkit/chainlink/v2/core/utils"
 )
 
@@ -200,6 +201,7 @@ func TestNewManager(t *testing.T) {
 func TestCorrectEndpointRouting(t *testing.T) {
 	tic := setupMockConfig(t, true)
 	tic.On("Endpoints").Return(nil)
+	tic.On("URL").Return(nil)
 
 	lggr, obsLogs := logger.TestLoggerObserved(t, zapcore.InfoLevel)
 	ks := mocks3.NewCSA(t)
@@ -286,3 +288,54 @@ func TestCorrectEndpointRouting(t *testing.T) {
 	}
 
 }
+
+func TestLegacyMode(t *testing.T) {
+	tic := setupMockConfig(t, true)
+	tic.On("Endpoints").Return(nil)
+	url, err := models.ParseURL("test.test")
+	require.NoError(t, err)
+	tic.On("URL").Return(url.URL())
+	tic.On("ServerPubKey").Return("some-pub-key")
+
+	lggr, obsLogs := logger.TestLoggerObserved(t, zapcore.InfoLevel)
+	ks := mocks3.NewCSA(t)
+
+	tm := NewManager(tic, ks, lggr)
+	require.Equal(t, true, tm.legacyMode)
+	require.Len(t, tm.endpoints, 1)
+
+	clientSent := make([]synchronization.TelemPayload, 0)
+	clientMock := mocks2.NewTelemetryService(t)
+	clientMock.On("Send", mock.Anything, mock.AnythingOfType("[]uint8"), mock.AnythingOfType("string"), mock.AnythingOfType("TelemetryType")).Return().Run(func(args mock.Arguments) {
+		clientSent = append(clientSent, synchronization.TelemPayload{
+			Telemetry:  args[1].([]byte),
+			ContractID: args[2].(string),
+			TelemType:  args[3].(synchronization.TelemetryType),
+		})
+	})
+	tm.endpoints[0].client = clientMock
+
+	e := tm.GenMonitoringEndpoint("some-contractID", "some-type", "unknown-network", "unknown-chainID")
+	require.Equal(t, "*telemetry.IngressAgentBatch", reflect.TypeOf(e).String())
+
+	e.SendLog([]byte("endpoint-1-message-1"))
+	e.SendLog([]byte("endpoint-1-message-2"))
+	e.SendLog([]byte("endpoint-1-message-3"))
+	require.Len(t, clientSent, 3)
+
+	e2 := tm.GenMonitoringEndpoint("another-contractID", "another-type", "another-unknown-network", "another-unknown-chainID")
+	require.Equal(t, "*telemetry.IngressAgentBatch", reflect.TypeOf(e).String())
+
+	e2.SendLog([]byte("endpoint-2-message-1"))
+	e2.SendLog([]byte("endpoint-2-message-2"))
+	e2.SendLog([]byte("endpoint-2-message-3"))
+	require.Len(t, clientSent, 6)
+
+	require.Equal(t, 1, obsLogs.Len()) // Deprecation warning for TelemetryIngress.URL and TelemetryIngress.ServerPubKey
+	require.Equal(t, []byte("endpoint-1-message-1"), clientSent[0].Telemetry)
+	require.Equal(t, []byte("endpoint-1-message-2"), clientSent[1].Telemetry)
+	require.Equal(t, []byte("endpoint-1-message-3"), clientSent[2].Telemetry)
+	require.Equal(t, []byte("endpoint-2-message-1"), clientSent[3].Telemetry)
+	require.Equal(t, []byte("endpoint-2-message-2"), clientSent[4].Telemetry)
+	require.Equal(t, []byte("endpoint-2-message-3"), clientSent[5].Telemetry)
+}

docs/CHANGELOG.md

@@ -20,7 +20,11 @@ ChainID = '...' # e.g. 1, 5, devnet, mainnet-beta
 URL = '...'
 ServerPubKey = '...'
 ```
-These replace `TelemetryIngress.URL` and `TelemetryIngress.ServerPubKey`, which are no longer allowed.
+These will eventually replace `TelemetryIngress.URL` and `TelemetryIngress.ServerPubKey`. Setting `TelemetryIngress.URL` and `TelemetryIngress.ServerPubKey` alongside `[[TelemetryIngress.Endpoints]]` will prevent the node from booting. Only one way of configuring telemetry endpoints is supported.
+
+### Upcoming Required Configuration Change
+
+- Starting in 2.8.0, chainlink nodes will no longer allow `TelemetryIngress.URL` and `TelemetryIngress.ServerPubKey`. Any TOML configuration that sets this fields will prevent the node from booting. These fields will be replaced by `[[TelemetryIngress.Endpoints]]`
 
 ## 2.6.0 - UNRELEASED
 
@@ -43,8 +47,6 @@ All nodes will have to remove the following secret configurations:
 
 All nodes will have to remove the following configuration field: `ExplorerURL`
 
-- Removed `TelemetryIngress.URL` and `TelemetryIngress.ServerPubKey`. They are no longer valid, and will prevent a node from booting. Use `TelemetryIngress.Endpoints` instead.
-
 ### Fixed
 
 - Unauthenticated users executing CLI commands previously generated a confusing error log, which is now removed: