fix: adding conf to metrics sender

Author: bejjrajesh

src/plugins/common.go

@@ -39,7 +39,7 @@ func LoadPlugins(commander client.Commander, binary core.NginxBinary, env core.E
 		fmt.Println(fmt.Sprintf("metrics enabled : %t", loadedConfig.IsExtensionEnabled(agent_config.FeatureMetrics)))
 		fmt.Println(fmt.Sprintf("metrics-sender enabled : %t", loadedConfig.IsExtensionEnabled(agent_config.FeatureMetricsSender)))
 		corePlugins = append(corePlugins,
-			NewMetricsSender(reporter),
+			NewMetricsSender(reporter, loadedConfig),
 		)
 	}
 

src/plugins/features.go

@@ -149,7 +149,7 @@ func (f *Features) enableMetricsFeature(_ string) []core.Plugin {
 
 		metrics := NewMetrics(f.conf, f.env, f.binary, f.processes)
 		metricsThrottle := NewMetricsThrottle(f.conf, f.env)
-		metricsSender := NewMetricsSender(f.commander)
+		metricsSender := NewMetricsSender(f.commander, conf)
 
 		return []core.Plugin{metrics, metricsThrottle, metricsSender}
 	} else {
@@ -202,7 +202,7 @@ func (f *Features) enableMetricsSenderFeature(_ string) []core.Plugin {
 		}
 		f.conf = conf
 
-		metricsSender := NewMetricsSender(f.commander)
+		metricsSender := NewMetricsSender(f.commander, conf)
 
 		return []core.Plugin{metricsSender}
 	} else {

src/plugins/metrics_sender.go

@@ -15,6 +15,7 @@ import (
 	"github.com/nginx/agent/sdk/v2/proto"
 	models "github.com/nginx/agent/sdk/v2/proto/events"
 	"github.com/nginx/agent/v2/src/core"
+	"github.com/nginx/agent/v2/src/core/config"
 	"strings"
 	"sync"
 
@@ -29,13 +30,15 @@ type MetricsSender struct {
 	started       *atomic.Bool
 	readyToSend   *atomic.Bool
 	readyToSendMu sync.RWMutex
+	conf          *config.Config
 }
 
-func NewMetricsSender(reporter client.MetricReporter) *MetricsSender {
+func NewMetricsSender(reporter client.MetricReporter, config *config.Config) *MetricsSender {
 	return &MetricsSender{
 		reporter:    reporter,
 		started:     atomic.NewBool(false),
 		readyToSend: atomic.NewBool(false),
+		conf:        config,
 	}
 }
 
@@ -62,13 +65,17 @@ func (r *MetricsSender) Info() *core.Info {
 }
 
 func (r *MetricsSender) Process(msg *core.Message) {
-	//if msg.Exact(core.AgentConnected) {
-	//	log.Debugf("MetricsSender AgentConnected Before: %v", r.readyToSend)
-	//	r.readyToSend.Store(true)
-	//	log.Debugf("MetricsSender AgentConnected After %v", r.readyToSend)
-	//	return
-	//}
-
+	if msg.Exact(core.AgentConnected) {
+		if r.conf.Features != nil && r.isFeatureEnabled(r.conf.Features) {
+			r.readyToSendMu.Lock()
+			r.readyToSend.Store(true)
+			r.readyToSendMu.Unlock()
+		} else {
+			r.readyToSendMu.Lock()
+			r.readyToSend.Store(false)
+			r.readyToSendMu.Unlock()
+		}
+	}
 	if msg.Exact(core.CommMetrics) {
 		payloads, ok := msg.Data().([]core.Payload)
 		if !ok {
@@ -107,9 +114,9 @@ func (r *MetricsSender) Process(msg *core.Message) {
 			}
 		}
 	} else if msg.Exact(core.AgentConfigChanged) {
-		switch config := msg.Data().(type) {
+		switch agentConfig := msg.Data().(type) {
 		case *proto.AgentConfig:
-			r.metricSenderBackoff(config)
+			r.metricSenderBackoff(agentConfig)
 		default:
 			log.Warnf("metrics sender expected %T type, but got: %T", &proto.AgentConfig{}, msg.Data())
 		}
@@ -119,7 +126,7 @@ func (r *MetricsSender) Process(msg *core.Message) {
 func (r *MetricsSender) metricSenderBackoff(agentConfig *proto.AgentConfig) {
 	log.Debugf("update metric reporter client configuration to %+v", agentConfig)
 	if agentConfig.Details.Features != nil {
-		if r.isFeatureEnabled(agentConfig) {
+		if r.isFeatureEnabled(agentConfig.Details.Features) {
 			r.readyToSendMu.Lock()
 			r.readyToSend.Store(true)
 			r.readyToSendMu.Unlock()
@@ -142,10 +149,10 @@ func (r *MetricsSender) Subscriptions() []string {
 	return []string{core.CommMetrics, core.AgentConnected, core.AgentConfigChanged}
 }
 
-func (r *MetricsSender) isFeatureEnabled(agentConfig *proto.AgentConfig) bool {
+func (r *MetricsSender) isFeatureEnabled(features []string) bool {
 	var isFeatureEnabled bool
-	if agentConfig.Details.Features != nil {
-		for _, feature := range agentConfig.Details.Features {
+	if features != nil {
+		for _, feature := range features {
 			if feature == agent_config.FeatureMetricsSender {
 				isFeatureEnabled = true
 				break

src/plugins/metrics_sender_test.go

@@ -10,6 +10,7 @@ package plugins
 import (
 	"context"
 	"errors"
+	"github.com/nginx/agent/v2/src/core/config"
 	"reflect"
 	"testing"
 	"time"
@@ -43,7 +44,7 @@ func TestMetricsSenderSendMetrics(t *testing.T) {
 			ctx := context.TODO()
 			mockMetricsReportClient := tutils.NewMockMetricsReportClient()
 			mockMetricsReportClient.Mock.On("Send", ctx, mock.Anything).Return(test.err)
-			pluginUnderTest := NewMetricsSender(mockMetricsReportClient)
+			pluginUnderTest := NewMetricsSender(mockMetricsReportClient, &config.Config{ClientID: "12345", Features: []string{"metrics-sender"}})
 
 			assert.False(t, pluginUnderTest.started.Load())
 			assert.False(t, pluginUnderTest.readyToSend.Load())
@@ -110,7 +111,7 @@ func TestMetricsSenderBackoff(t *testing.T) {
 		t.Run(test.name, func(_ *testing.T) {
 			ctx := context.TODO()
 			mockMetricsReportClient := tutils.NewMockMetricsReportClient()
-			pluginUnderTest := NewMetricsSender(mockMetricsReportClient)
+			pluginUnderTest := NewMetricsSender(mockMetricsReportClient, &config.Config{ClientID: "12345", Features: []string{"metrics-sender"}})
 
 			pluginUnderTest.Init(core.NewMockMessagePipe(ctx))
 			pluginUnderTest.Process(core.NewMessage(core.AgentConnected, nil))
@@ -130,6 +131,6 @@ func TestMetricsSenderBackoff(t *testing.T) {
 }
 
 func TestMetricsSenderSubscriptions(t *testing.T) {
-	pluginUnderTest := NewMetricsSender(tutils.NewMockMetricsReportClient())
+	pluginUnderTest := NewMetricsSender(tutils.NewMockMetricsReportClient(), &config.Config{ClientID: "12345"})
 	assert.Equal(t, []string{core.CommMetrics, core.AgentConnected, core.AgentConfigChanged}, pluginUnderTest.Subscriptions())
 }