[Wf-Diagnostics] create messaging client only if not initialised

service/worker/diagnostics/activities.go

@@ -97,12 +97,10 @@ func (w *dw) rootCauseIssues(ctx context.Context, info rootCauseIssuesParams) ([
 }
 
 func (w *dw) emitUsageLogs(ctx context.Context, info analytics.WfDiagnosticsUsageData) error {
-	client := w.newMessagingClient()
-	return emit(ctx, info, client)
-}
-
-func (w *dw) newMessagingClient() messaging.Client {
-	return kafka.NewKafkaClient(&w.kafkaCfg, w.metricsClient, w.logger, w.tallyScope, true)
+	if w.messagingClient == nil {
+		w.messagingClient = kafka.NewKafkaClient(&w.kafkaCfg, w.metricsClient, w.logger, w.tallyScope, true)
+	}
+	return emit(ctx, info, w.messagingClient)
 }
 
 func emit(ctx context.Context, info analytics.WfDiagnosticsUsageData, client messaging.Client) error {

service/worker/diagnostics/module.go

@@ -24,6 +24,7 @@ package diagnostics
 
 import (
 	"context"
+	"github.com/uber/cadence/common/messaging"
 
 	"github.com/opentracing/opentracing-go"
 	"github.com/uber-go/tally"
@@ -46,36 +47,39 @@ type DiagnosticsWorkflow interface {
 }
 
 type dw struct {
-	svcClient     workflowserviceclient.Interface
-	clientBean    client.Bean
-	metricsClient metrics.Client
-	logger        log.Logger
-	tallyScope    tally.Scope
-	worker        worker.Worker
-	kafkaCfg      config.KafkaConfig
-	invariants    []invariant.Invariant
+	svcClient       workflowserviceclient.Interface
+	clientBean      client.Bean
+	metricsClient   metrics.Client
+	messagingClient messaging.Client
+	logger          log.Logger
+	tallyScope      tally.Scope
+	worker          worker.Worker
+	kafkaCfg        config.KafkaConfig
+	invariants      []invariant.Invariant
 }
 
 type Params struct {
-	ServiceClient workflowserviceclient.Interface
-	ClientBean    client.Bean
-	MetricsClient metrics.Client
-	Logger        log.Logger
-	TallyScope    tally.Scope
-	KafkaCfg      config.KafkaConfig
-	Invariants    []invariant.Invariant
+	ServiceClient   workflowserviceclient.Interface
+	ClientBean      client.Bean
+	MetricsClient   metrics.Client
+	MessagingClient messaging.Client
+	Logger          log.Logger
+	TallyScope      tally.Scope
+	KafkaCfg        config.KafkaConfig
+	Invariants      []invariant.Invariant
 }
 
 // New creates a new diagnostics workflow.
 func New(params Params) DiagnosticsWorkflow {
 	return &dw{
-		svcClient:     params.ServiceClient,
-		metricsClient: params.MetricsClient,
-		tallyScope:    params.TallyScope,
-		clientBean:    params.ClientBean,
-		logger:        params.Logger,
-		kafkaCfg:      params.KafkaCfg,
-		invariants:    params.Invariants,
+		svcClient:       params.ServiceClient,
+		metricsClient:   params.MetricsClient,
+		messagingClient: params.MessagingClient,
+		tallyScope:      params.TallyScope,
+		clientBean:      params.ClientBean,
+		logger:          params.Logger,
+		kafkaCfg:        params.KafkaCfg,
+		invariants:      params.Invariants,
 	}
 }
 

service/worker/service.go

@@ -343,13 +343,14 @@ func (s *Service) startFixerWorkflowWorker() {
 
 func (s *Service) startDiagnostics() {
 	params := diagnostics.Params{
-		ServiceClient: s.params.PublicClient,
-		MetricsClient: s.GetMetricsClient(),
-		TallyScope:    s.params.MetricScope,
-		ClientBean:    s.GetClientBean(),
-		Logger:        s.GetLogger(),
-		KafkaCfg:      s.params.KafkaConfig,
-		Invariants:    s.params.DiagnosticsInvariants,
+		ServiceClient:   s.params.PublicClient,
+		MetricsClient:   s.GetMetricsClient(),
+		MessagingClient: s.GetMessagingClient(),
+		TallyScope:      s.params.MetricScope,
+		ClientBean:      s.GetClientBean(),
+		Logger:          s.GetLogger(),
+		KafkaCfg:        s.params.KafkaConfig,
+		Invariants:      s.params.DiagnosticsInvariants,
 	}
 	if err := diagnostics.New(params).Start(); err != nil {
 		s.Stop()