Fix the metrics feature so that it can be enabled and disabled remotely

Author: dhurley

src/core/pipe.go

@@ -71,7 +71,7 @@ func (p *MessagePipe) Register(size int, plugins []Plugin, extensionPlugins []Ex
 	pluginsRegistered := []string{}
 	extensionPluginsRegistered := []string{}
 
-	for _, plugin := range p.plugins {
+	for _, plugin := range plugins {
 		for _, subscription := range plugin.Subscriptions() {
 			p.regMu.Lock()
 			err := p.bus.Subscribe(subscription, plugin.Process)
@@ -83,7 +83,7 @@ func (p *MessagePipe) Register(size int, plugins []Plugin, extensionPlugins []Ex
 		pluginsRegistered = append(pluginsRegistered, *plugin.Info().name)
 	}
 
-	for _, plugin := range p.extensionPlugins {
+	for _, plugin := range extensionPlugins {
 		for _, subscription := range plugin.Subscriptions() {
 			p.regMu.Lock()
 			err := p.bus.Subscribe(subscription, plugin.Process)
@@ -94,6 +94,7 @@ func (p *MessagePipe) Register(size int, plugins []Plugin, extensionPlugins []Ex
 		}
 		extensionPluginsRegistered = append(extensionPluginsRegistered, *plugin.Info().name)
 	}
+
 	log.Infof("The following core plugins have been registered: %q", pluginsRegistered)
 	log.Infof("The following extension plugins have been registered: %q", extensionPluginsRegistered)
 
@@ -127,6 +128,8 @@ func (p *MessagePipe) DeRegister(pluginNames []string) error {
 					return err
 				}
 			}
+
+			plugin = nil
 		}
 	}
 

src/plugins/commander.go

@@ -38,14 +38,15 @@ func NewCommander(cmdr client.Commander, config *config.Config) *Commander {
 }
 
 func (c *Commander) Init(pipeline core.MessagePipeInterface) {
+	log.Info("Commander initializing")
 	c.pipeline = pipeline
 	c.ctx = pipeline.Context()
-	log.Info("Commander initializing")
 	go c.dispatchLoop()
 }
 
 func (c *Commander) Close() {
 	log.Info("Commander is wrapping up")
+	log.Info("Commander is closed")
 }
 
 func (c *Commander) Info() *core.Info {
@@ -158,7 +159,6 @@ func (c *Commander) dispatchLoop() {
 		case *proto.Command_AgentConnectRequest, *proto.Command_AgentConnectResponse:
 			topic = core.AgentConnected
 		case *proto.Command_AgentConfigRequest, *proto.Command_AgentConfig:
-			log.Debugf("agent config %T command data type received and ignored", cmd.GetData())
 			topic = core.AgentConfig
 		case *proto.Command_CmdStatus:
 			data := cmd.GetData().(*proto.Command_CmdStatus)

src/plugins/common.go

@@ -6,6 +6,7 @@ import (
 	"github.com/nginx/agent/v2/src/core/config"
 	"github.com/nginx/agent/v2/src/extensions"
 	log "github.com/sirupsen/logrus"
+	"go.uber.org/atomic"
 
 	agent_config "github.com/nginx/agent/sdk/v2/agent/config"
 	"github.com/nginx/agent/sdk/v2/agent/events"
@@ -36,15 +37,15 @@ 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, atomic.NewBool(false)),
 		)
 	}
 
 	corePlugins = append(corePlugins,
 		NewConfigReader(loadedConfig),
 		NewNginx(commander, binary, env, loadedConfig, processes),
 		NewExtensions(loadedConfig, env),
-		NewFeatures(commander, loadedConfig, env, binary, loadedConfig.Version, processes, agentEventsMeta),
+		NewFeatures(commander, reporter, loadedConfig, env, binary, loadedConfig.Version, processes, agentEventsMeta),
 	)
 
 	if loadedConfig.IsFeatureEnabled(agent_config.FeatureRegistration) {

src/plugins/config_reader.go

@@ -37,6 +37,7 @@ func NewConfigReader(config *config.Config) *ConfigReader {
 }
 
 func (r *ConfigReader) Init(pipeline core.MessagePipeInterface) {
+	log.Info("ConfigReader initializing")
 	r.messagePipeline = pipeline
 }
 
@@ -46,6 +47,7 @@ func (r *ConfigReader) Info() *core.Info {
 
 func (r *ConfigReader) Close() {
 	log.Info("ConfigReader is wrapping up")
+	log.Info("ConfigReader is closed")
 }
 
 func (r *ConfigReader) Process(msg *core.Message) {
@@ -73,8 +75,10 @@ func (r *ConfigReader) Process(msg *core.Message) {
 			// Update the agent config on disk
 			switch commandData := cmd.Data.(type) {
 			case *proto.Command_AgentConfig:
+				log.Debugf("Config reader: AgentConfig message recevied: %v, topic: %v", commandData, msg.Topic())
 				r.updateAgentConfig(commandData.AgentConfig)
 			case *proto.Command_AgentConnectResponse:
+				log.Debugf("Config reader: AgentConnectResponse message recevied: %v, topic: %v", commandData, msg.Topic())
 				r.updateAgentConfig(commandData.AgentConnectResponse.AgentConfig)
 			}
 		}
@@ -152,7 +156,7 @@ func (r *ConfigReader) updateAgentConfig(payloadAgentConfig *proto.AgentConfig)
 		}
 
 		if synchronizeFeatures {
-			log.Debugf("Agent config features changed, synchronizing features")
+			log.Info("Agent config features changed, synchronizing features")
 			r.synchronizeFeatures(payloadAgentConfig)
 			r.config.Features = payloadAgentConfig.Details.Features
 		}

src/plugins/dataplane_status.go

@@ -90,6 +90,7 @@ func (dps *DataPlaneStatus) Close() {
 	dps.softwareDetailsMutex.Unlock()
 	dps.healthTicker.Stop()
 	dps.sendStatus <- true
+	log.Info("DataPlaneStatus is closed")
 }
 
 func (dps *DataPlaneStatus) Info() *core.Info {

src/plugins/events.go

@@ -64,6 +64,7 @@ func (a *Events) Init(pipeline core.MessagePipeInterface) {
 
 func (a *Events) Close() {
 	log.Info("Events is wrapping up")
+	log.Info("Events is closed")
 }
 
 func (a *Events) Process(msg *core.Message) {

src/plugins/extensions.go

@@ -37,6 +37,7 @@ func (e *Extensions) Init(pipeline core.MessagePipeInterface) {
 
 func (e *Extensions) Close() {
 	log.Info("Extensions is wrapping up")
+	log.Info("Extensions is closed")
 }
 
 func (e *Extensions) Process(msg *core.Message) {

src/plugins/features.go

@@ -14,13 +14,15 @@ import (
 	sdkGRPC "github.com/nginx/agent/sdk/v2/grpc"
 	"github.com/nginx/agent/v2/src/core"
 	"github.com/nginx/agent/v2/src/core/config"
+	"go.uber.org/atomic"
 
 	"github.com/google/uuid"
 	log "github.com/sirupsen/logrus"
 )
 
 type Features struct {
 	commander       client.Commander
+	metricReporter  client.MetricReporter
 	pipeline        core.MessagePipeInterface
 	conf            *config.Config
 	env             core.Environment
@@ -33,6 +35,7 @@ type Features struct {
 
 func NewFeatures(
 	commander client.Commander,
+	metricReporter client.MetricReporter,
 	conf *config.Config,
 	env core.Environment,
 	binary core.NginxBinary,
@@ -42,6 +45,7 @@ func NewFeatures(
 ) *Features {
 	return &Features{
 		commander:       commander,
+		metricReporter:  metricReporter,
 		conf:            conf,
 		env:             env,
 		binary:          binary,
@@ -94,6 +98,7 @@ func (f *Features) Init(pipeline core.MessagePipeInterface) {
 
 func (f *Features) Close() {
 	log.Info("Features is wrapping up")
+	log.Info("Features is closed")
 }
 
 func (f *Features) Info() *core.Info {
@@ -135,7 +140,7 @@ func (f *Features) Process(msg *core.Message) {
 
 func (f *Features) enableMetricsFeature(_ string) []core.Plugin {
 	if !f.pipeline.IsPluginAlreadyRegistered(agent_config.FeatureMetrics) {
-		log.Debugf("Enabling metrics feature")
+		log.Debug("Enabling metrics feature")
 		conf, err := config.GetConfig(f.conf.ClientID)
 		if err != nil {
 			log.Warnf("Unable to get agent config, %v", err)
@@ -144,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.metricReporter, atomic.NewBool(true))
 
 		return []core.Plugin{metrics, metricsThrottle, metricsSender}
 	}
@@ -154,7 +159,7 @@ func (f *Features) enableMetricsFeature(_ string) []core.Plugin {
 func (f *Features) enableMetricsCollectionFeature(_ string) []core.Plugin {
 	if !f.pipeline.IsPluginAlreadyRegistered(agent_config.FeatureMetrics) &&
 		!f.pipeline.IsPluginAlreadyRegistered(agent_config.FeatureMetricsCollection) {
-		log.Debugf("Enabling metrics-collection feature")
+		log.Debug("Enabling metrics-collection feature")
 		conf, err := config.GetConfig(f.conf.ClientID)
 		if err != nil {
 			log.Warnf("Unable to get agent config, %v", err)
@@ -171,7 +176,7 @@ func (f *Features) enableMetricsCollectionFeature(_ string) []core.Plugin {
 func (f *Features) enableMetricsThrottleFeature(_ string) []core.Plugin {
 	if !f.pipeline.IsPluginAlreadyRegistered(agent_config.FeatureMetrics) &&
 		!f.pipeline.IsPluginAlreadyRegistered(agent_config.FeatureMetricsThrottle) {
-		log.Debugf("Enabling metrics-throttle feature")
+		log.Debug("Enabling metrics-throttle feature")
 		conf, err := config.GetConfig(f.conf.ClientID)
 		if err != nil {
 			log.Warnf("Unable to get agent config, %v", err)
@@ -188,14 +193,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("Enabling metrics-sender feature")
+		log.Debug("Enabling metrics-sender feature")
 		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.metricReporter, atomic.NewBool(true))
 
 		return []core.Plugin{metricsSender}
 	}
@@ -204,8 +209,8 @@ func (f *Features) enableMetricsSenderFeature(_ string) []core.Plugin {
 
 func (f *Features) enableAgentAPIFeature(_ string) []core.Plugin {
 	if !f.pipeline.IsPluginAlreadyRegistered(agent_config.FeatureAgentAPI) {
+		log.Debug("Enabling agent-api feature")
 		conf, err := config.GetConfig(f.conf.ClientID)
-		log.Debugf("Enabling agent-api feature")
 		if err != nil {
 			log.Warnf("Unable to get agent config, %v", err)
 		}
@@ -220,7 +225,7 @@ func (f *Features) enableAgentAPIFeature(_ string) []core.Plugin {
 
 func (f *Features) enableRegistrationFeature(_ string) []core.Plugin {
 	if !f.pipeline.IsPluginAlreadyRegistered(agent_config.FeatureRegistration) {
-		log.Debugf("Enabling registration feature")
+		log.Debug("Enabling registration feature")
 		conf, err := config.GetConfig(f.conf.ClientID)
 		if err != nil {
 			log.Warnf("Unable to get agent config, %v", err)
@@ -236,7 +241,7 @@ func (f *Features) enableRegistrationFeature(_ string) []core.Plugin {
 
 func (f *Features) enableDataPlaneStatusFeature(_ string) []core.Plugin {
 	if !f.pipeline.IsPluginAlreadyRegistered(agent_config.FeatureDataPlaneStatus) {
-		log.Debugf("Enabling dataplane-status feature")
+		log.Debug("Enabling dataplane-status feature")
 		conf, err := config.GetConfig(f.conf.ClientID)
 		if err != nil {
 			log.Warnf("Unable to get agent config, %v", err)
@@ -252,7 +257,7 @@ func (f *Features) enableDataPlaneStatusFeature(_ string) []core.Plugin {
 
 func (f *Features) enableProcessWatcherFeature(_ string) []core.Plugin {
 	if !f.pipeline.IsPluginAlreadyRegistered(agent_config.FeatureProcessWatcher) {
-		log.Debugf("Enabling process-watcher feature")
+		log.Debug("Enabling process-watcher feature")
 		conf, err := config.GetConfig(f.conf.ClientID)
 		if err != nil {
 			log.Warnf("Unable to get agent config, %v", err)
@@ -268,23 +273,23 @@ func (f *Features) enableProcessWatcherFeature(_ string) []core.Plugin {
 
 func (f *Features) enableActivityEventsFeature(_ string) []core.Plugin {
 	if !f.pipeline.IsPluginAlreadyRegistered(agent_config.FeatureActivityEvents) {
-		log.Debugf("Enabling activity-events feature")
+		log.Debug("Enabling activity-events feature")
 		conf, err := config.GetConfig(f.conf.ClientID)
 		if err != nil {
 			log.Warnf("Unable to get agent config, %v", err)
 		}
 		f.conf = conf
 
-		events := NewEvents(f.conf, f.env, sdkGRPC.NewMessageMeta(uuid.NewString()), f.binary, f.agentEventsMeta)
+		eventsPlugin := NewEvents(f.conf, f.env, sdkGRPC.NewMessageMeta(uuid.NewString()), f.binary, f.agentEventsMeta)
 
-		return []core.Plugin{events}
+		return []core.Plugin{eventsPlugin}
 	}
 	return []core.Plugin{}
 }
 
 func (f *Features) enableFileWatcherFeature(_ string) []core.Plugin {
 	if !f.pipeline.IsPluginAlreadyRegistered(agent_config.FeatureFileWatcher) {
-		log.Debugf("Enabling file-watcher feature")
+		log.Debug("Enabling file-watcher feature")
 		conf, err := config.GetConfig(f.conf.ClientID)
 		if err != nil {
 			log.Warnf("Unable to get agent config, %v", err)
@@ -303,7 +308,7 @@ func (f *Features) enableNginxCountingFeature(_ string) []core.Plugin {
 	countingPlugins := []core.Plugin{}
 	if len(f.conf.Nginx.NginxCountingSocket) > 0 {
 		if !f.pipeline.IsPluginAlreadyRegistered(agent_config.FeatureNginxCounting) {
-			log.Debugf("Enabling nginx-counting feature")
+			log.Debug("Enabling nginx-counting feature")
 			conf, err := config.GetConfig(f.conf.ClientID)
 			if err != nil {
 				log.Warnf("Unable to get agent config, %v", err)

src/plugins/features_test.go

@@ -93,10 +93,11 @@ func TestFeatures_Process(t *testing.T) {
 	binary.On("UpdateNginxDetailsFromProcesses", mock.Anything).Return()
 
 	cmdr := tutils.NewMockCommandClient()
+	reporter := tutils.NewMockMetricsReportClient()
 
 	configuration, _ := config.GetConfig("1234")
 
-	pluginUnderTest := NewFeatures(cmdr, configuration, env, binary, "agentVersion", processes, events.NewAgentEventMeta(
+	pluginUnderTest := NewFeatures(cmdr, reporter, configuration, env, binary, "agentVersion", processes, events.NewAgentEventMeta(
 		config.MODULE,
 		"v0.0.1",
 		"75231",

src/plugins/file_watcher.go

@@ -104,6 +104,7 @@ func (fw *FileWatcher) Close() {
 	fw.enabled = false
 	fw.cancelFunction()
 	fw.watcher.Close()
+	log.Info("File Watcher is closed")
 }
 
 func (fw *FileWatcher) Process(message *core.Message) {