Refactor: StatsD code removal (#48)

* removed redundant statsd code from meteor * fixed a unit test case * refactored in attempt to resolve lint error * reverting to keep a list of monitors for extensibility * formatting to keep older code as it was * recordPlugin method mock method call not needed now * removed list as only otel is being supported now --------- Co-authored-by: Sumeet Kumar Rai <[email protected]>
goto · Dec 15, 2023 · 4bf1b95 · 4bf1b95
1 parent a147080
commit 4bf1b95
Show file tree

Hide file tree

Showing 13 changed files with 38 additions and 419 deletions.
diff --git a/agent/agent.go b/agent/agent.go
@@ -27,7 +27,7 @@ type Agent struct {
 	extractorFactory *registry.ExtractorFactory
 	processorFactory *registry.ProcessorFactory
 	sinkFactory      *registry.SinkFactory
-	monitor          []Monitor
+	monitor          Monitor
 	logger           log.Logger
 	retrier          *retrier
 	stopOnSinkError  bool
@@ -279,8 +279,8 @@ func (r *Agent) setupSink(ctx context.Context, sr recipe.PluginRecipe, stream *s
 	}
 
 	retryNotification := func(e error, d time.Duration) {
-		for _, mt := range r.monitor {
-			mt.RecordSinkRetryCount(ctx, pluginInfo)
+		if r.monitor != nil {
+			r.monitor.RecordSinkRetryCount(ctx, pluginInfo)
 		}
 
 		r.logger.Warn(
@@ -302,9 +302,6 @@ func (r *Agent) setupSink(ctx context.Context, sr recipe.PluginRecipe, stream *s
 		)
 
 		pluginInfo.Success = err == nil
-		for _, mt := range r.monitor {
-			mt.RecordPlugin(ctx, pluginInfo) // this can be deleted when statsd is removed
-		}
 		if err != nil {
 			// once it reaches here, it means that the retry has been exhausted and still got error
 			r.logger.Error("error running sink", "sink", sr.Name, "error", err.Error())
@@ -328,8 +325,8 @@ func (r *Agent) setupSink(ctx context.Context, sr recipe.PluginRecipe, stream *s
 }
 
 func (r *Agent) logAndRecordMetrics(ctx context.Context, run Run) {
-	for _, monitor := range r.monitor {
-		monitor.RecordRun(ctx, run)
+	if r.monitor != nil {
+		r.monitor.RecordRun(ctx, run)
 	}
 
 	if run.Success {

diff --git a/agent/agent_test.go b/agent/agent_test.go
@@ -82,7 +82,7 @@ func TestAgentRun(t *testing.T) {
 			ProcessorFactory: pf,
 			SinkFactory:      sf,
 			Logger:           utils.Logger,
-			Monitor:          []agent.Monitor{monitor},
+			Monitor:          monitor,
 		})
 		run := r.Run(ctx, validRecipe)
 		assert.False(t, run.Success)
@@ -113,7 +113,7 @@ func TestAgentRun(t *testing.T) {
 			ProcessorFactory: registry.NewProcessorFactory(),
 			SinkFactory:      sf,
 			Logger:           utils.Logger,
-			Monitor:          []agent.Monitor{monitor},
+			Monitor:          monitor,
 		})
 		run := r.Run(ctx, validRecipe)
 		assert.False(t, run.Success)
@@ -146,7 +146,7 @@ func TestAgentRun(t *testing.T) {
 			ProcessorFactory: pf,
 			SinkFactory:      registry.NewSinkFactory(),
 			Logger:           utils.Logger,
-			Monitor:          []agent.Monitor{monitor},
+			Monitor:          monitor,
 		})
 		run := r.Run(ctx, validRecipe)
 		assert.False(t, run.Success)
@@ -183,7 +183,7 @@ func TestAgentRun(t *testing.T) {
 			ProcessorFactory: pf,
 			SinkFactory:      sf,
 			Logger:           utils.Logger,
-			Monitor:          []agent.Monitor{monitor},
+			Monitor:          monitor,
 		})
 		run := r.Run(ctx, validRecipe)
 		assert.False(t, run.Success)
@@ -222,7 +222,7 @@ func TestAgentRun(t *testing.T) {
 			ProcessorFactory: pf,
 			SinkFactory:      sf,
 			Logger:           utils.Logger,
-			Monitor:          []agent.Monitor{monitor},
+			Monitor:          monitor,
 		})
 		run := r.Run(ctx, validRecipe)
 		assert.False(t, run.Success)
@@ -263,7 +263,7 @@ func TestAgentRun(t *testing.T) {
 			ProcessorFactory: pf,
 			SinkFactory:      sf,
 			Logger:           utils.Logger,
-			Monitor:          []agent.Monitor{monitor},
+			Monitor:          monitor,
 		})
 		run := r.Run(ctx, validRecipe)
 		assert.False(t, run.Success)
@@ -305,7 +305,7 @@ func TestAgentRun(t *testing.T) {
 			ProcessorFactory: pf,
 			SinkFactory:      sf,
 			Logger:           utils.Logger,
-			Monitor:          []agent.Monitor{monitor},
+			Monitor:          monitor,
 		})
 		run := r.Run(ctx, validRecipe)
 		assert.False(t, run.Success)
@@ -346,7 +346,7 @@ func TestAgentRun(t *testing.T) {
 			ProcessorFactory: pf,
 			SinkFactory:      sf,
 			Logger:           utils.Logger,
-			Monitor:          []agent.Monitor{monitor},
+			Monitor:          monitor,
 		})
 		run := r.Run(ctx, validRecipe)
 		assert.False(t, run.Success)
@@ -394,7 +394,7 @@ func TestAgentRun(t *testing.T) {
 			ProcessorFactory: pf,
 			SinkFactory:      sf,
 			Logger:           utils.Logger,
-			Monitor:          []agent.Monitor{monitor},
+			Monitor:          monitor,
 		})
 		run := r.Run(ctx, validRecipe)
 		assert.False(t, run.Success)
@@ -441,7 +441,7 @@ func TestAgentRun(t *testing.T) {
 			ProcessorFactory: pf,
 			SinkFactory:      sf,
 			Logger:           utils.Logger,
-			Monitor:          []agent.Monitor{monitor},
+			Monitor:          monitor,
 		})
 		run := r.Run(ctx, validRecipe)
 		assert.False(t, run.Success)
@@ -483,15 +483,15 @@ func TestAgentRun(t *testing.T) {
 
 		monitor := newMockMonitor()
 		monitor.On("RecordRun", mockCtx, mock.AnythingOfType("agent.Run")).Once()
-		monitor.On("RecordPlugin", mockCtx, mock.AnythingOfType("agent.PluginInfo"))
+		monitor.On("RecordPlugin", mockCtx, mock.AnythingOfType("agent.PluginInfo")).Maybe()
 		defer monitor.AssertExpectations(t)
 
 		r := agent.NewAgent(agent.Config{
 			ExtractorFactory: ef,
 			ProcessorFactory: pf,
 			SinkFactory:      sf,
 			Logger:           utils.Logger,
-			Monitor:          []agent.Monitor{monitor},
+			Monitor:          monitor,
 		})
 		run := r.Run(ctx, validRecipe)
 		assert.True(t, run.Success)
@@ -533,7 +533,7 @@ func TestAgentRun(t *testing.T) {
 
 		monitor := newMockMonitor()
 		monitor.On("RecordRun", mockCtx, mock.AnythingOfType("agent.Run")).Once()
-		monitor.On("RecordPlugin", mockCtx, mock.AnythingOfType("agent.PluginInfo"))
+		monitor.On("RecordPlugin", mockCtx, mock.AnythingOfType("agent.PluginInfo")).Maybe()
 		defer monitor.AssertExpectations(t)
 
 		r := agent.NewAgent(agent.Config{
@@ -542,7 +542,7 @@ func TestAgentRun(t *testing.T) {
 			SinkFactory:      sf,
 			Logger:           utils.Logger,
 			StopOnSinkError:  true,
-			Monitor:          []agent.Monitor{monitor},
+			Monitor:          monitor,
 		})
 
 		run := r.Run(ctx, validRecipe)
@@ -585,7 +585,7 @@ func TestAgentRun(t *testing.T) {
 
 		monitor := newMockMonitor()
 		monitor.On("RecordRun", mockCtx, mock.AnythingOfType("agent.Run")).Once()
-		monitor.On("RecordPlugin", mockCtx, mock.AnythingOfType("agent.PluginInfo"))
+		monitor.On("RecordPlugin", mockCtx, mock.AnythingOfType("agent.PluginInfo")).Maybe()
 		defer monitor.AssertExpectations(t)
 
 		r := agent.NewAgent(agent.Config{
@@ -594,7 +594,7 @@ func TestAgentRun(t *testing.T) {
 			SinkFactory:      sf,
 			Logger:           utils.Logger,
 			StopOnSinkError:  false,
-			Monitor:          []agent.Monitor{monitor},
+			Monitor:          monitor,
 		})
 
 		run := r.Run(ctx, validRecipe)
@@ -649,15 +649,15 @@ func TestAgentRun(t *testing.T) {
 
 		monitor := newMockMonitor()
 		monitor.On("RecordRun", mockCtx, mock.AnythingOfType("agent.Run")).Once()
-		monitor.On("RecordPlugin", mockCtx, mock.AnythingOfType("agent.PluginInfo"))
+		monitor.On("RecordPlugin", mockCtx, mock.AnythingOfType("agent.PluginInfo")).Maybe()
 		defer monitor.AssertExpectations(t)
 
 		r := agent.NewAgent(agent.Config{
 			ExtractorFactory: ef,
 			ProcessorFactory: pf,
 			SinkFactory:      sf,
 			Logger:           utils.Logger,
-			Monitor:          []agent.Monitor{monitor},
+			Monitor:          monitor,
 		})
 		run := r.Run(ctx, validRecipe)
 		assert.NoError(t, run.Error)
@@ -706,14 +706,14 @@ func TestAgentRun(t *testing.T) {
 
 		monitor := newMockMonitor()
 		monitor.On("RecordRun", mockCtx, mock.AnythingOfType("agent.Run")).Once()
-		monitor.On("RecordPlugin", mockCtx, mock.AnythingOfType("agent.PluginInfo"))
+		monitor.On("RecordPlugin", mockCtx, mock.AnythingOfType("agent.PluginInfo")).Maybe()
 		defer monitor.AssertExpectations(t)
 
 		r := agent.NewAgent(agent.Config{
 			ExtractorFactory: ef,
 			ProcessorFactory: pf,
 			SinkFactory:      sf,
-			Monitor:          []agent.Monitor{monitor},
+			Monitor:          monitor,
 			Logger:           utils.Logger,
 			TimerFn:          timerFn,
 		})
@@ -761,15 +761,15 @@ func TestAgentRun(t *testing.T) {
 
 		monitor := newMockMonitor()
 		monitor.On("RecordRun", mockCtx, mock.AnythingOfType("agent.Run")).Once()
-		monitor.On("RecordPlugin", mockCtx, mock.AnythingOfType("agent.PluginInfo"))
+		monitor.On("RecordPlugin", mockCtx, mock.AnythingOfType("agent.PluginInfo")).Maybe()
 		defer monitor.AssertExpectations(t)
 
 		r := agent.NewAgent(agent.Config{
 			ExtractorFactory:     ef,
 			ProcessorFactory:     pf,
 			SinkFactory:          sf,
 			Logger:               utils.Logger,
-			Monitor:              []agent.Monitor{monitor},
+			Monitor:              monitor,
 			MaxRetries:           2,                    // need to retry "at least" 2 times since Extractor returns RetryError twice
 			RetryInitialInterval: 1 * time.Millisecond, // this is to override default retry interval to reduce test time
 		})
@@ -815,7 +815,7 @@ func TestAgentRun(t *testing.T) {
 
 		monitor := newMockMonitor()
 		monitor.On("RecordRun", mockCtx, mock.AnythingOfType("agent.Run")).Once()
-		monitor.On("RecordPlugin", mockCtx, mock.AnythingOfType("agent.PluginInfo"))
+		monitor.On("RecordPlugin", mockCtx, mock.AnythingOfType("agent.PluginInfo")).Maybe()
 		monitor.On("RecordSinkRetryCount", mockCtx, mock.AnythingOfType("agent.PluginInfo"))
 		defer monitor.AssertExpectations(t)
 
@@ -824,7 +824,7 @@ func TestAgentRun(t *testing.T) {
 			ProcessorFactory:     pf,
 			SinkFactory:          sf,
 			Logger:               utils.Logger,
-			Monitor:              []agent.Monitor{monitor},
+			Monitor:              monitor,
 			MaxRetries:           2,                    // need to retry "at least" 2 times since Sink returns RetryError twice
 			RetryInitialInterval: 1 * time.Millisecond, // this is to override default retry interval to reduce test time
 		})
@@ -875,7 +875,7 @@ func TestAgentRun(t *testing.T) {
 
 		monitor := newMockMonitor()
 		monitor.On("RecordRun", utils.OfTypeContext(), mock.AnythingOfType("agent.Run")).Once()
-		monitor.On("RecordPlugin", utils.OfTypeContext(), mock.AnythingOfType("agent.PluginInfo"))
+		monitor.On("RecordPlugin", utils.OfTypeContext(), mock.AnythingOfType("agent.PluginInfo")).Maybe()
 		monitor.On("RecordSinkRetryCount", utils.OfTypeContext(), mock.AnythingOfType("agent.PluginInfo")).Maybe()
 		defer monitor.AssertExpectations(t)
 
@@ -884,7 +884,7 @@ func TestAgentRun(t *testing.T) {
 			ProcessorFactory:     pf,
 			SinkFactory:          sf,
 			Logger:               utils.Logger,
-			Monitor:              []agent.Monitor{monitor},
+			Monitor:              monitor,
 			MaxRetries:           5,
 			RetryInitialInterval: 10 * time.Second,
 		})
@@ -1057,15 +1057,15 @@ func TestAgentRunMultiple(t *testing.T) {
 
 		monitor := newMockMonitor()
 		monitor.On("RecordRun", mockCtx, mock.AnythingOfType("agent.Run"))
-		monitor.On("RecordPlugin", mockCtx, mock.AnythingOfType("agent.PluginInfo"))
+		monitor.On("RecordPlugin", mockCtx, mock.AnythingOfType("agent.PluginInfo")).Maybe()
 		defer monitor.AssertExpectations(t)
 
 		r := agent.NewAgent(agent.Config{
 			ExtractorFactory: ef,
 			ProcessorFactory: pf,
 			SinkFactory:      sf,
 			Logger:           utils.Logger,
-			Monitor:          []agent.Monitor{monitor},
+			Monitor:          monitor,
 		})
 		runs := r.RunMultiple(ctx, recipeList)
 
@@ -1152,7 +1152,7 @@ func TestValidate(t *testing.T) {
 			ProcessorFactory: pf,
 			SinkFactory:      sf,
 			Logger:           utils.Logger,
-			Monitor:          []agent.Monitor{newMockMonitor()},
+			Monitor:          newMockMonitor(),
 		})
 
 		var expectedErrs []error

diff --git a/agent/config.go b/agent/config.go
@@ -11,7 +11,7 @@ type Config struct {
 	ExtractorFactory     *registry.ExtractorFactory
 	ProcessorFactory     *registry.ProcessorFactory
 	SinkFactory          *registry.SinkFactory
-	Monitor              []Monitor
+	Monitor              Monitor
 	Logger               log.Logger
 	MaxRetries           int
 	RetryInitialInterval time.Duration

diff --git a/cmd/run.go b/cmd/run.go
@@ -70,16 +70,7 @@ func RunCmd() *cobra.Command {
 			ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
 			defer stop()
 
-			var mts []agent.Monitor
-			if cfg.StatsdEnabled {
-				mt, err := newStatsdMonitor(cfg)
-				if err != nil {
-					return err
-				}
-
-				mts = append(mts, mt)
-
-			}
+			var mts agent.Monitor
 
 			if cfg.OtelEnabled {
 				doneOtlp, err := metrics.InitOtel(ctx, cfg, lg, Version)
@@ -88,7 +79,7 @@ func RunCmd() *cobra.Command {
 				}
 				defer doneOtlp()
 
-				mts = append(mts, metrics.NewOtelMonitor())
+				mts = metrics.NewOtelMonitor()
 			}
 
 			runner := agent.NewAgent(agent.Config{
@@ -156,11 +147,3 @@ func RunCmd() *cobra.Command {
 
 	return cmd
 }
-
-func newStatsdMonitor(cfg config.Config) (*metrics.StatsdMonitor, error) {
-	client, err := metrics.NewStatsdClient(cfg.StatsdHost)
-	if err != nil {
-		return nil, err
-	}
-	return metrics.NewStatsdMonitor(client, cfg.AppName), nil
-}
diff --git a/config/config.go b/config/config.go
@@ -14,8 +14,6 @@ type Config struct {
 	MaxRetries                  int     `mapstructure:"MAX_RETRIES" default:"5"`
 	RetryInitialIntervalSeconds int     `mapstructure:"RETRY_INITIAL_INTERVAL_SECONDS" default:"5"`
 	StopOnSinkError             bool    `mapstructure:"STOP_ON_SINK_ERROR" default:"false"`
-	StatsdEnabled               bool    `mapstructure:"STATSD_ENABLED" default:"false"`
-	StatsdHost                  string  `mapstructure:"STATSD_HOST" default:"localhost:8125"`
 	OtelEnabled                 bool    `mapstructure:"OTEL_ENABLED" default:"false"`
 	OtelCollectorAddr           string  `mapstructure:"OTEL_COLLECTOR_ADDR" default:"localhost:4317"`
 	OtelTraceSampleProbability  float64 `mapstructure:"OTEL_TRACE_SAMPLE_PROBABILITY" default:"1"`

diff --git a/config/config_test.go b/config/config_test.go
@@ -25,8 +25,6 @@ func TestLoad(t *testing.T) {
 			expected: config.Config{
 				AppName:                     "meteor",
 				LogLevel:                    "info",
-				StatsdEnabled:               false,
-				StatsdHost:                  "localhost:8125",
 				OtelEnabled:                 false,
 				OtelCollectorAddr:           "localhost:4317",
 				OtelTraceSampleProbability:  1,
@@ -43,8 +41,6 @@ func TestLoad(t *testing.T) {
 			expected: config.Config{
 				AppName:                     "meteor",
 				LogLevel:                    "info",
-				StatsdEnabled:               false,
-				StatsdHost:                  "localhost:8125",
 				OtelEnabled:                 false,
 				OtelCollectorAddr:           "localhost:4317",
 				OtelTraceSampleProbability:  1,

diff --git a/config/meteor.yaml.sample b/config/meteor.yaml.sample
@@ -1,6 +1,4 @@
 LOG_LEVEL: info
-STATSD_ENABLED: false
-STATSD_HOST: "localhost:8125"
 MAX_RETRIES: 5
 RETRY_INITIAL_INTERVAL_SECONDS: 5
 STOP_ON_SINK_ERROR: false

diff --git a/config/testdata/invalid-config.yaml b/config/testdata/invalid-config.yaml
@@ -1 +1 @@
-STATSD_ENABLED: not-a-boolean
+OTEL_ENABLED: not-a-boolean
diff --git a/config/testdata/valid-config.yaml b/config/testdata/valid-config.yaml
@@ -1,7 +1,4 @@
 LOG_LEVEL: info
-STATSD_ENABLED: false
-STATSD_HOST: "localhost:8125"
-STATSD_PREFIX: meteor
 MAX_RETRIES: 5
 RETRY_INITIAL_INTERVAL_SECONDS: 5
 STOP_ON_SINK_ERROR: false