Fix panic when updating config concurrently

src/plugins/common.go

@@ -36,7 +36,7 @@ func LoadPlugins(commander client.Commander, binary core.NginxBinary, env core.E
 
 	if (loadedConfig.IsFeatureEnabled(agent_config.FeatureMetrics) || loadedConfig.IsFeatureEnabled(agent_config.FeatureMetricsSender)) && reporter != nil {
 		corePlugins = append(corePlugins,
-			NewMetricsSender(reporter),
+			NewMetricsSender(reporter, loadedConfig),
 		)
 	}
 

src/plugins/config_reader.go

@@ -152,7 +152,9 @@ func (r *ConfigReader) updateAgentConfig(payloadAgentConfig *proto.AgentConfig)
 		}
 
 		if synchronizeFeatures {
+			log.Debugf("agent config  features changed, synchronizing features")
 			r.synchronizeFeatures(payloadAgentConfig)
+			r.config.Features = payloadAgentConfig.Details.Features
 		}
 
 		r.messagePipeline.Process(core.NewMessage(core.AgentConfigChanged, payloadAgentConfig))
@@ -164,6 +166,7 @@ func (r *ConfigReader) synchronizeFeatures(agtCfg *proto.AgentConfig) {
 		r.detailsMu.RLock()
 		for _, feature := range r.config.Features {
 			if feature != agent_config.FeatureRegistration && feature != agent_config.FeatureNginxConfigAsync {
+				log.Debugf("config_reader: deregistering the feature %s", feature)
 				r.deRegisterPlugin(feature)
 			}
 		}

src/plugins/dataplane_status.go

@@ -23,25 +23,26 @@ import (
 )
 
 type DataPlaneStatus struct {
-	messagePipeline             core.MessagePipeInterface
-	ctx                         context.Context
-	sendStatus                  chan bool
-	healthTicker                *time.Ticker
-	interval                    time.Duration
-	meta                        *proto.Metadata
-	binary                      core.NginxBinary
-	env                         core.Environment
-	version                     string
-	tags                        *[]string
-	configDirs                  string
-	lastSendDetails             time.Time
-	envHostInfo                 *proto.HostInfo
-	reportInterval              time.Duration
-	softwareDetails             map[string]*proto.DataplaneSoftwareDetails
-	nginxConfigActivityStatuses map[string]*proto.AgentActivityStatus
-	softwareDetailsMutex        sync.RWMutex
-	structMu                    sync.RWMutex
-	processes                   []*core.Process
+	messagePipeline                  core.MessagePipeInterface
+	ctx                              context.Context
+	sendStatus                       chan bool
+	healthTicker                     *time.Ticker
+	interval                         time.Duration
+	meta                             *proto.Metadata
+	binary                           core.NginxBinary
+	env                              core.Environment
+	version                          string
+	tags                             *[]string
+	configDirs                       string
+	lastSendDetails                  time.Time
+	envHostInfo                      *proto.HostInfo
+	reportInterval                   time.Duration
+	softwareDetails                  map[string]*proto.DataplaneSoftwareDetails
+	nginxConfigActivityStatuses      map[string]*proto.AgentActivityStatus
+	nginxConfigActivityStatusesMutex sync.RWMutex
+	softwareDetailsMutex             sync.RWMutex
+	structMu                         sync.RWMutex
+	processes                        []*core.Process
 }
 
 const (
@@ -81,7 +82,9 @@ func (dps *DataPlaneStatus) Init(pipeline core.MessagePipeInterface) {
 
 func (dps *DataPlaneStatus) Close() {
 	log.Info("DataPlaneStatus is wrapping up")
+	dps.nginxConfigActivityStatusesMutex.Lock()
 	dps.nginxConfigActivityStatuses = nil
+	dps.nginxConfigActivityStatusesMutex.Unlock()
 	dps.softwareDetailsMutex.Lock()
 	dps.softwareDetails = nil
 	dps.softwareDetailsMutex.Unlock()
@@ -144,8 +147,10 @@ func (dps *DataPlaneStatus) Subscriptions() []string {
 
 func (dps *DataPlaneStatus) updateNginxConfigActivityStatuses(newAgentActivityStatus *proto.AgentActivityStatus) {
 	log.Tracef("DataplaneStatus: Updating nginxConfigActivityStatuses with %v", newAgentActivityStatus)
-	if _, ok := newAgentActivityStatus.GetStatus().(*proto.AgentActivityStatus_NginxConfigStatus); ok {
+	if _, ok := newAgentActivityStatus.GetStatus().(*proto.AgentActivityStatus_NginxConfigStatus); dps.nginxConfigActivityStatuses != nil && ok {
+		dps.nginxConfigActivityStatusesMutex.Lock()
 		dps.nginxConfigActivityStatuses[newAgentActivityStatus.GetNginxConfigStatus().GetNginxId()] = newAgentActivityStatus
+		dps.nginxConfigActivityStatusesMutex.Unlock()
 	}
 }
 
@@ -184,6 +189,8 @@ func (dps *DataPlaneStatus) healthGoRoutine(pipeline core.MessagePipeInterface)
 func (dps *DataPlaneStatus) dataplaneStatus(forceDetails bool) *proto.DataplaneStatus {
 	forceDetails = forceDetails || time.Now().UTC().Add(-dps.reportInterval).After(dps.lastSendDetails)
 
+	dps.nginxConfigActivityStatusesMutex.Lock()
+	defer dps.nginxConfigActivityStatusesMutex.Unlock()
 	agentActivityStatuses := []*proto.AgentActivityStatus{}
 	for _, nginxConfigActivityStatus := range dps.nginxConfigActivityStatuses {
 		agentActivityStatuses = append(agentActivityStatuses, nginxConfigActivityStatus)

src/plugins/features.go

@@ -135,7 +135,7 @@ func (f *Features) Process(msg *core.Message) {
 
 func (f *Features) enableMetricsFeature(_ string) []core.Plugin {
 	if !f.pipeline.IsPluginAlreadyRegistered(agent_config.FeatureMetrics) {
-
+		log.Debugf("features.go: enabling metrics feature")
 		conf, err := config.GetConfig(f.conf.ClientID)
 		if err != nil {
 			log.Warnf("Unable to get agent config, %v", err)
@@ -144,7 +144,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}
 	}
@@ -188,14 +188,14 @@ func (f *Features) enableMetricsThrottleFeature(_ string) []core.Plugin {
 func (f *Features) enableMetricsSenderFeature(_ string) []core.Plugin {
 	if !f.pipeline.IsPluginAlreadyRegistered(agent_config.FeatureMetrics) &&
 		!f.pipeline.IsPluginAlreadyRegistered(agent_config.FeatureMetricsSender) {
-
+		log.Debugf("features.go: enabling metrics_sender")
 		conf, err := config.GetConfig(f.conf.ClientID)
 		if err != nil {
 			log.Warnf("Unable to get agent config, %v", err)
 		}
 		f.conf = conf
 
-		metricsSender := NewMetricsSender(f.commander)
+		metricsSender := NewMetricsSender(f.commander, conf)
 
 		return []core.Plugin{metricsSender}
 	}

src/plugins/metrics_sender.go

@@ -9,32 +9,36 @@ package plugins
 
 import (
 	"context"
-	"strings"
-
 	"github.com/nginx/agent/sdk/v2"
 	agent_config "github.com/nginx/agent/sdk/v2/agent/config"
 	"github.com/nginx/agent/sdk/v2/client"
 	"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"
 
 	log "github.com/sirupsen/logrus"
 	"go.uber.org/atomic"
 )
 
 type MetricsSender struct {
-	reporter    client.MetricReporter
-	pipeline    core.MessagePipeInterface
-	ctx         context.Context
-	started     *atomic.Bool
-	readyToSend *atomic.Bool
+	reporter      client.MetricReporter
+	pipeline      core.MessagePipeInterface
+	ctx           context.Context
+	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,
 	}
 }
 
@@ -50,8 +54,10 @@ func (r *MetricsSender) Init(pipeline core.MessagePipeInterface) {
 
 func (r *MetricsSender) Close() {
 	log.Info("MetricsSender is wrapping up")
+	r.readyToSendMu.Lock()
 	r.started.Store(false)
 	r.readyToSend.Store(false)
+	defer r.readyToSendMu.Unlock()
 }
 
 func (r *MetricsSender) Info() *core.Info {
@@ -60,10 +66,16 @@ func (r *MetricsSender) Info() *core.Info {
 
 func (r *MetricsSender) Process(msg *core.Message) {
 	if msg.Exact(core.AgentConnected) {
-		r.readyToSend.Toggle()
-		return
+		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 {
@@ -72,6 +84,7 @@ func (r *MetricsSender) Process(msg *core.Message) {
 		}
 		for _, p := range payloads {
 			if !r.readyToSend.Load() {
+				log.Debugf("metrics_sender is not ready to send the metrics")
 				continue
 			}
 
@@ -99,9 +112,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())
 		}
@@ -110,7 +123,17 @@ 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.Details.Features) {
+			r.readyToSendMu.Lock()
+			r.readyToSend.Store(true)
+			r.readyToSendMu.Unlock()
+		} else {
+			r.readyToSendMu.Lock()
+			r.readyToSend.Store(false)
+			r.readyToSendMu.Unlock()
+		}
+	}
 	if agentConfig.GetDetails() == nil || agentConfig.GetDetails().GetServer() == nil || agentConfig.GetDetails().GetServer().GetBackoff() == nil {
 		log.Debug("not updating metric reporter client configuration as new Agent backoff settings is nil")
 		return
@@ -123,3 +146,16 @@ func (r *MetricsSender) metricSenderBackoff(agentConfig *proto.AgentConfig) {
 func (r *MetricsSender) Subscriptions() []string {
 	return []string{core.CommMetrics, core.AgentConnected, core.AgentConfigChanged}
 }
+
+func (r *MetricsSender) isFeatureEnabled(features []string) bool {
+	var isFeatureEnabled bool
+	if features != nil {
+		for _, feature := range features {
+			if feature == agent_config.FeatureMetricsSender {
+				isFeatureEnabled = true
+				break
+			}
+		}
+	}
+	return isFeatureEnabled
+}

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())
 }