[BCF-2728] Add telemetry to LOOPP deps (#205)

Author: cedric-cordenier

pkg/loop/internal/pb/reporting_plugin_service.pb.go

@@ -20,7 +20,8 @@ message NewReportingPluginFactoryRequest {
   uint32 providerID = 1;
   uint32 errorLogID = 2;
   uint32 pipelineRunnerID = 3;
-  ReportingPluginServiceConfig ReportingPluginServiceConfig = 4;
+  uint32 telemetryID = 4;
+  ReportingPluginServiceConfig ReportingPluginServiceConfig = 5;
 }
 
 // NewReportingPluginFactoryReply has return arguments for [github.com/smartcontractkit/chainlink-relay/pkg/loop/reporting_plugins/LOOPPService.NewReportingPluginFactory].

pkg/loop/internal/pb/reporting_plugin_service.proto

@@ -26,29 +26,44 @@ func NewReportingPluginServiceClient(broker Broker, brokerCfg BrokerConfig, conn
 	return &ReportingPluginServiceClient{pluginClient: pc, reportingPluginService: pb.NewReportingPluginServiceClient(pc), serviceClient: newServiceClient(pc.brokerExt, pc)}
 }
 
-func (m *ReportingPluginServiceClient) NewReportingPluginFactory(ctx context.Context, config types.ReportingPluginServiceConfig, grpcProvider grpc.ClientConnInterface, pipelineRunner types.PipelineRunnerService, errorLog types.ErrorLog) (types.ReportingPluginFactory, error) {
+func (m *ReportingPluginServiceClient) NewReportingPluginFactory(
+	ctx context.Context,
+	config types.ReportingPluginServiceConfig,
+	grpcProvider grpc.ClientConnInterface,
+	pipelineRunner types.PipelineRunnerService,
+	telemetry types.TelemetryClient,
+	errorLog types.ErrorLog,
+) (types.ReportingPluginFactory, error) {
 	cc := m.newClientConn("ReportingPluginServiceFactory", func(ctx context.Context) (id uint32, deps resources, err error) {
 		providerID, providerRes, err := m.serve("PluginProvider", proxy.NewProxy(grpcProvider))
 		if err != nil {
 			return 0, nil, err
 		}
 		deps.Add(providerRes)
 
-		errorLogID, errorLogRes, err := m.serveNew("ErrorLog", func(s *grpc.Server) {
-			pb.RegisterErrorLogServer(s, &errorLogServer{impl: errorLog})
+		pipelineRunnerID, pipelineRunnerRes, err := m.serveNew("PipelineRunner", func(s *grpc.Server) {
+			pb.RegisterPipelineRunnerServiceServer(s, &pipelineRunnerServiceServer{impl: pipelineRunner})
 		})
 		if err != nil {
 			return 0, nil, err
 		}
-		deps.Add(errorLogRes)
+		deps.Add(pipelineRunnerRes)
 
-		pipelineRunnerID, pipelineRunnerRes, err := m.serveNew("PipelineRunner", func(s *grpc.Server) {
-			pb.RegisterPipelineRunnerServiceServer(s, &pipelineRunnerServiceServer{impl: pipelineRunner})
+		telemetryID, telemetryRes, err := m.serveNew("Telemetry", func(s *grpc.Server) {
+			pb.RegisterTelemetryServer(s, NewTelemetryServer(telemetry))
 		})
 		if err != nil {
 			return 0, nil, err
 		}
-		deps.Add(pipelineRunnerRes)
+		deps.Add(telemetryRes)
+
+		errorLogID, errorLogRes, err := m.serveNew("ErrorLog", func(s *grpc.Server) {
+			pb.RegisterErrorLogServer(s, &errorLogServer{impl: errorLog})
+		})
+		if err != nil {
+			return 0, nil, err
+		}
+		deps.Add(errorLogRes)
 
 		reply, err := m.reportingPluginService.NewReportingPluginFactory(ctx, &pb.NewReportingPluginFactoryRequest{
 			ReportingPluginServiceConfig: &pb.ReportingPluginServiceConfig{
@@ -60,6 +75,7 @@ func (m *ReportingPluginServiceClient) NewReportingPluginFactory(ctx context.Con
 			ProviderID:       providerID,
 			ErrorLogID:       errorLogID,
 			PipelineRunnerID: pipelineRunnerID,
+			TelemetryID:      telemetryID,
 		})
 		if err != nil {
 			return 0, nil, err
@@ -110,23 +126,31 @@ func (m *reportingPluginServiceServer) NewReportingPluginFactory(ctx context.Con
 	pipelineRunnerRes := resource{pipelineRunnerConn, "PipelineRunner"}
 	pipelineRunner := newPipelineRunnerClient(pipelineRunnerConn)
 
+	telemetryConn, err := m.dial(request.TelemetryID)
+	if err != nil {
+		m.closeAll(errorLogRes, providerRes, pipelineRunnerRes)
+		return nil, ErrConnDial{Name: "Telemetry", ID: request.TelemetryID, Err: err}
+	}
+	telemetryRes := resource{telemetryConn, "Telemetry"}
+	telemetry := NewTelemetryClient(telemetryConn)
+
 	config := types.ReportingPluginServiceConfig{
 		ProviderType: request.ReportingPluginServiceConfig.ProviderType,
 		PluginConfig: request.ReportingPluginServiceConfig.PluginConfig,
 		PluginName:   request.ReportingPluginServiceConfig.PluginName,
 		Command:      request.ReportingPluginServiceConfig.Command,
 	}
 
-	factory, err := m.impl.NewReportingPluginFactory(ctx, config, providerConn, pipelineRunner, errorLog)
+	factory, err := m.impl.NewReportingPluginFactory(ctx, config, providerConn, pipelineRunner, telemetry, errorLog)
 	if err != nil {
-		m.closeAll(providerRes, errorLogRes, pipelineRunnerRes)
+		m.closeAll(providerRes, errorLogRes, pipelineRunnerRes, telemetryRes)
 		return nil, err
 	}
 
 	id, _, err := m.serveNew("ReportingPluginProvider", func(s *grpc.Server) {
 		pb.RegisterServiceServer(s, &serviceServer{srv: factory})
 		pb.RegisterReportingPluginFactoryServer(s, newReportingPluginFactoryServer(factory, m.brokerExt))
-	}, providerRes, errorLogRes, pipelineRunnerRes)
+	}, providerRes, errorLogRes, pipelineRunnerRes, telemetryRes)
 	if err != nil {
 		return nil, err
 	}

pkg/loop/internal/pb/telemetry.pb.go

@@ -3,70 +3,56 @@ package internal
 import (
 	"context"
 	"errors"
-	"fmt"
 
-	"github.com/smartcontractkit/libocr/commontypes"
 	"google.golang.org/grpc"
 	"google.golang.org/protobuf/types/known/emptypb"
 
-	"github.com/smartcontractkit/chainlink-relay/pkg/logger"
 	"github.com/smartcontractkit/chainlink-relay/pkg/loop/internal/pb"
 	"github.com/smartcontractkit/chainlink-relay/pkg/types"
 )
 
-var _ types.Telemetry = (*telemetryClient)(nil)
-var _ types.MonitoringEndpointGenerator = (*telemetryClient)(nil)
-var _ commontypes.MonitoringEndpoint = (*telemetryEndpoint)(nil)
+var _ types.TelemetryClient = (*telemetryClient)(nil)
 
 type TelemetryClient struct {
 	*telemetryClient
 }
 
 type telemetryClient struct {
 	grpc pb.TelemetryClient
-
-	lggr logger.Logger
 }
 
 type telemetryEndpoint struct {
-	lggr logger.Logger
-
 	grpc          pb.TelemetryClient
 	relayID       pb.RelayID
 	contractID    string
 	telemetryType string
 }
 
-func (t *telemetryEndpoint) SendLog(log []byte) {
-	_, err := t.grpc.Send(context.Background(), &pb.TelemetryMessage{
+func (t *telemetryEndpoint) SendLog(ctx context.Context, log []byte) error {
+	_, err := t.grpc.Send(ctx, &pb.TelemetryMessage{
 		RelayID:       &t.relayID,
 		ContractID:    t.contractID,
 		TelemetryType: t.telemetryType,
 		Payload:       log,
 	})
-	if err != nil {
-		t.lggr.Errorw("cannot send telemetry", "err", err)
-	}
+	return err
 }
 
-// GenMonitoringEndpoint generates a new monitoring endpoint, returns nil if one cannot be generated
-func (t *telemetryClient) GenMonitoringEndpoint(network string, chainID string, contractID string, telemetryType string) commontypes.MonitoringEndpoint {
+// NewEndpoint generates a new monitoring endpoint, returns nil if one cannot be generated
+func (t *telemetryClient) NewEndpoint(ctx context.Context, network string, chainID string, contractID string, telemetryType string) (types.TelemetryClientEndpoint, error) {
 	if contractID == "" {
-		t.lggr.Errorw("cannot generate monitoring endpoint, contractID is empty", "contractID", contractID, "telemetryType", telemetryType, "network", network, "chainID", chainID)
-		return nil
+		return nil, errors.New("contractID cannot be empty")
 	}
 	if telemetryType == "" {
-		t.lggr.Errorw("cannot generate monitoring endpoint, telemetryType is empty", "contractID", contractID, "telemetryType", telemetryType, "network", network, "chainID", chainID)
-		return nil
+		return nil, errors.New("telemetryType cannot be empty")
 	}
 	if network == "" {
-		t.lggr.Errorw("cannot generate monitoring endpoint, network is empty", "contractID", contractID, "telemetryType", telemetryType, "network", network, "chainID", chainID)
-		return nil
+		return nil, errors.New("network cannot be empty")
 	}
 	if chainID == "" {
-		t.lggr.Errorw("cannot generate monitoring endpoint, chainID is empty", "contractID", contractID, "telemetryType", telemetryType, "network", network, "chainID", chainID)
-		return nil
+		return nil, errors.New("chainId cannot be empty")
 	}
+
 	return &telemetryEndpoint{
 		grpc: t.grpc,
 		relayID: pb.RelayID{
@@ -75,8 +61,7 @@ func (t *telemetryClient) GenMonitoringEndpoint(network string, chainID string,
 		},
 		contractID:    contractID,
 		telemetryType: telemetryType,
-		lggr:          t.lggr,
-	}
+	}, nil
 }
 
 // Send sends payload to the desired endpoint based on network and chainID
@@ -111,58 +96,28 @@ func (t *telemetryClient) Send(ctx context.Context, network string, chainID stri
 	return nil
 }
 
-func NewTelemetryClient(cc grpc.ClientConnInterface, lggr logger.Logger) *telemetryClient {
-	return &telemetryClient{grpc: pb.NewTelemetryClient(cc), lggr: lggr}
+func NewTelemetryClient(cc grpc.ClientConnInterface) *telemetryClient {
+	return &telemetryClient{grpc: pb.NewTelemetryClient(cc)}
 }
 
 var _ pb.TelemetryServer = (*telemetryServer)(nil)
 
 type telemetryServer struct {
 	pb.UnimplementedTelemetryServer
 
-	impl      types.MonitoringEndpointGenerator
-	endpoints map[string]commontypes.MonitoringEndpoint
+	impl types.TelemetryService
 }
 
 func (t *telemetryServer) Send(ctx context.Context, message *pb.TelemetryMessage) (*emptypb.Empty, error) {
-	e, err := t.getOrCreateEndpoint(message)
-	if err != nil {
-		return nil, err
-	}
-	e.SendLog(message.Payload)
-
-	return nil, nil
-}
-
-func (t *telemetryServer) getOrCreateEndpoint(m *pb.TelemetryMessage) (commontypes.MonitoringEndpoint, error) {
-	if m.ContractID == "" {
-		return nil, errors.New("contractID cannot be empty")
-	}
-	if m.TelemetryType == "" {
-		return nil, errors.New("telemetryType cannot be empty")
-	}
-	if m.RelayID == nil {
-		return nil, errors.New("RelayID cannot be nil")
+	var network, chainID string
+	if message.RelayID != nil {
+		network = message.RelayID.Network
+		chainID = message.RelayID.ChainId
 	}
-	if m.RelayID.Network == "" {
-		return nil, errors.New("RelayID.Network cannot be empty")
-	}
-	if m.RelayID.ChainId == "" {
-		return nil, errors.New("RelayID.ChainId cannot be empty")
-	}
-
-	key := makeKey(m)
-	e, ok := t.endpoints[key]
-	if !ok {
-		e = t.impl.GenMonitoringEndpoint(m.RelayID.Network, m.RelayID.ChainId, m.ContractID, m.TelemetryType)
-	}
-	return e, nil
-}
-
-func makeKey(m *pb.TelemetryMessage) string {
-	return fmt.Sprintf("%s_%s_%s_%s", m.RelayID.Network, m.RelayID.ChainId, m.ContractID, m.TelemetryType)
+	err := t.impl.Send(ctx, network, chainID, message.ContractID, message.TelemetryType, message.Payload)
+	return &emptypb.Empty{}, err
 }
 
-func NewTelemetryServer(impl types.MonitoringEndpointGenerator) *telemetryServer {
+func NewTelemetryServer(impl types.TelemetryService) *telemetryServer {
 	return &telemetryServer{impl: impl}
 }