Fix remotely enabling/disabling features (#1088)

Author: dhurley

Makefile

@@ -68,12 +68,12 @@ OS                     := $(shell uname -s | tr '[:upper:]' '[:lower:]')
 uname_m    := $(shell uname -m)
 
 ifeq ($(uname_m),aarch64)
-	OSARCH = arm64
+	OSARCH ?= arm64
 else
 	ifeq ($(uname_m),x86_64)
-		OSARCH = amd64
+		OSARCH ?= amd64
 	else
-		OSARCH = $(uname_m)
+		OSARCH ?= $(uname_m)
 	endif
 endif
 

src/core/environment.go

@@ -342,7 +342,6 @@ func (env *EnvironmentType) IsContainer() bool {
 	res, err, _ := singleflightGroup.Do(IsContainerKey, func() (interface{}, error) {
 		for _, filename := range []string{dockerEnv, containerEnv, k8sServiceAcct} {
 			if _, err := os.Stat(filename); err == nil {
-				log.Debugf("Is a container because (%s) exists", filename)
 				return true, nil
 			}
 		}

src/core/metrics/collectors/nginx.go

@@ -29,6 +29,7 @@ type NginxCollector struct {
 }
 
 func NewNginxCollector(conf *config.Config, env core.Environment, collectorConf *metrics.NginxCollectorConfig, binary core.NginxBinary) *NginxCollector {
+	log.Debugf("Creating NGINX Collector")
 	host := env.NewHostInfo("agentVersion", &conf.Tags, conf.ConfigDirs, false)
 	dimensions := metrics.NewCommonDim(host, conf, collectorConf.NginxId)
 	dimensions.NginxConfPath = collectorConf.ConfPath
@@ -45,6 +46,7 @@ func NewNginxCollector(conf *config.Config, env core.Environment, collectorConf
 }
 
 func buildSources(dimensions *metrics.CommonDim, binary core.NginxBinary, collectorConf *metrics.NginxCollectorConfig, conf *config.Config, env core.Environment) []metrics.NginxSource {
+	log.Debugf("Building NGINX metric sources")
 	var nginxSources []metrics.NginxSource
 	// worker metrics
 	if len(conf.Nginx.NginxCountingSocket) > 0 && conf.IsFeatureEnabled(agent_config.FeatureNginxCounting) {
@@ -68,6 +70,7 @@ func buildSources(dimensions *metrics.CommonDim, binary core.NginxBinary, collec
 			nginxSources = append(nginxSources, sources.NewNginxStatic(dimensions, sources.OSSNamespace))
 		}
 	}
+
 	return nginxSources
 }
 

src/core/metrics/sources/nginx_plus.go

@@ -76,6 +76,7 @@ type ExtendedStats struct {
 }
 
 func NewNginxPlus(baseDimensions *metrics.CommonDim, nginxNamespace, plusNamespace, plusAPI string, clientVersion int) *NginxPlus {
+	log.Debug("Creating NGINX Plus metrics source")
 	return &NginxPlus{baseDimensions: baseDimensions, nginxNamespace: nginxNamespace, plusNamespace: plusNamespace, plusAPI: plusAPI, clientVersion: clientVersion, logger: NewMetricSourceLogger()}
 }
 

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)
 

src/plugins/agent_api.go

@@ -191,6 +191,7 @@ func (a *AgentAPI) Close() {
 	if err := a.server.Shutdown(context.Background()); err != nil {
 		log.Errorf("Agent API HTTP Server Shutdown Error: %v", err)
 	}
+	log.Info("Agent API is closed")
 }
 
 func (a *AgentAPI) Process(message *core.Message) {

src/plugins/commander.go

@@ -38,14 +38,14 @@ 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 +158,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
 }
 
@@ -45,7 +46,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 +74,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 +155,9 @@ func (r *ConfigReader) updateAgentConfig(payloadAgentConfig *proto.AgentConfig)
 		}
 
 		if synchronizeFeatures {
+			log.Info("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 +169,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("Deregistering the feature %s", feature)
 				r.deRegisterPlugin(feature)
 			}
 		}
@@ -177,16 +183,19 @@ func (r *ConfigReader) synchronizeFeatures(agtCfg *proto.AgentConfig) {
 
 func (r *ConfigReader) deRegisterPlugin(data string) {
 	if data == agent_config.FeatureFileWatcher {
-
 		err := r.messagePipeline.DeRegister([]string{agent_config.FeatureFileWatcher, agent_config.FeatureFileWatcherThrottle})
 		if err != nil {
-			log.Warnf("Error De-registering %v Plugin: %v", data, err)
+			log.Warnf("Error deregistering %v plugin: %v", data, err)
+		}
+	} else if data == agent_config.FeatureMetrics {
+		err := r.messagePipeline.DeRegister([]string{agent_config.FeatureMetrics, agent_config.FeatureMetricsThrottle, agent_config.FeatureMetricsSender})
+		if err != nil {
+			log.Warnf("Error deregistering %v plugin: %v", data, err)
 		}
-
 	} else {
 		err := r.messagePipeline.DeRegister([]string{data})
 		if err != nil {
-			log.Warnf("Error De-registering %v Plugin: %v", data, err)
+			log.Warnf("Error deregistering %v plugin: %v", data, err)
 		}
 	}
 }

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,12 +82,15 @@ 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()
 	dps.healthTicker.Stop()
 	dps.sendStatus <- true
+	log.Info("DataPlaneStatus is closed")
 }
 
 func (dps *DataPlaneStatus) Info() *core.Info {
@@ -144,8 +148,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 +190,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)