Fixed race conditions found during testing

src/core/payloads/register_software_details.go

@@ -1,30 +1,29 @@
 package payloads
 
 import (
-	"sync"
-
 	"github.com/nginx/agent/sdk/v2/proto"
 )
 
 // RegisterWithDataplaneSoftwareDetailsPayload is an internal payload meant to be used as
 // part of registration when there are plugins reporting software details.
 type RegisterWithDataplaneSoftwareDetailsPayload struct {
-	dataplaneSoftwareDetails      map[string]*proto.DataplaneSoftwareDetails
-	dataplaneSoftwareDetailsMutex sync.Mutex
+	dataplaneSoftwareDetails *proto.DataplaneSoftwareDetails
+	pluginName               string
 }
 
 // NewRegisterWithDataplaneSoftwareDetailsPayload returns a pointer to an instance of a
 // RegisterWithDataplaneSoftwareDetailsPayload object.
-func NewRegisterWithDataplaneSoftwareDetailsPayload(details map[string]*proto.DataplaneSoftwareDetails) *RegisterWithDataplaneSoftwareDetailsPayload {
+func NewRegisterWithDataplaneSoftwareDetailsPayload(pluginName string, details *proto.DataplaneSoftwareDetails) *RegisterWithDataplaneSoftwareDetailsPayload {
 	return &RegisterWithDataplaneSoftwareDetailsPayload{
 		dataplaneSoftwareDetails: details,
+		pluginName:               pluginName,
 	}
 }
 
-// AddDataplaneSoftwareDetails adds the dataplane software details passed into the function to
-// the dataplane software details map object that has been sent as part of the payload.
-func (p *RegisterWithDataplaneSoftwareDetailsPayload) AddDataplaneSoftwareDetails(pluginName string, details *proto.DataplaneSoftwareDetails) {
-	p.dataplaneSoftwareDetailsMutex.Lock()
-	p.dataplaneSoftwareDetails[pluginName] = details
-	p.dataplaneSoftwareDetailsMutex.Unlock()
+func (r *RegisterWithDataplaneSoftwareDetailsPayload) GetPluginName() string {
+	return r.pluginName
+}
+
+func (r *RegisterWithDataplaneSoftwareDetailsPayload) GetDataplaneSoftwareDetails() *proto.DataplaneSoftwareDetails {
+	return r.dataplaneSoftwareDetails
 }

src/plugins/extensions_test.go

@@ -62,7 +62,8 @@ func TestExtensions_Process(t *testing.T) {
 			time.Sleep(250 * time.Millisecond)
 
 			processedMessages := messagePipe.GetProcessedMessages()
-			assert.Equal(t, 1, len(processedMessages))
+			assert.GreaterOrEqual(t, len(processedMessages), 1)
+			assert.Equal(t, core.EnableExtension, processedMessages[0].Topic())
 
 			assert.Equal(t, 2, len(messagePipe.GetPlugins()))
 			assert.Equal(t, "Extensions Plugin", messagePipe.GetPlugins()[0].Info().Name())

src/plugins/nginx_app_protect.go

@@ -66,18 +66,14 @@ func (n *NginxAppProtect) Init(pipeline core.MessagePipeInterface) {
 	ctx, cancel := context.WithCancel(n.messagePipeline.Context())
 	n.ctx = ctx
 	n.ctxCancel = cancel
+	n.addSoftwareDetailsToRegistration()
 	go n.monitor()
 }
 
-func (n *NginxAppProtect) Process(msg *core.Message) {
-	switch {
-	case msg.Exact(core.RegisterWithDataplaneSoftwareDetails):
-		n.addSoftwareDetailsToRegistration(msg)
-	}
-}
+func (n *NginxAppProtect) Process(msg *core.Message) {}
 
 func (n *NginxAppProtect) Subscriptions() []string {
-	return []string{core.RegisterWithDataplaneSoftwareDetails}
+	return []string{}
 }
 
 func (n *NginxAppProtect) Close() {
@@ -88,18 +84,14 @@ func (n *NginxAppProtect) Close() {
 // addSoftwareDetailsToRegistration adds the dataplane software details produced by the
 // NAP plugin to the OneTimeRegistration dataplane software details map that has been sent
 // as part of the message.
-func (n *NginxAppProtect) addSoftwareDetailsToRegistration(msg *core.Message) {
-	switch commandData := msg.Data().(type) {
-	case *payloads.RegisterWithDataplaneSoftwareDetailsPayload:
-		log.Debugf("%s is adding software details to registration", napPluginName)
-		napReport := n.generateNAPDetailsProtoCommand()
-		napSoftwareDetails := &proto.DataplaneSoftwareDetails{
-			Data: napReport,
-		}
-		commandData.AddDataplaneSoftwareDetails(napPluginName, napSoftwareDetails)
-	default:
-		log.Errorf("Expected the type %T but got %T", payloads.RegisterWithDataplaneSoftwareDetailsPayload{}, commandData)
+func (n *NginxAppProtect) addSoftwareDetailsToRegistration() {
+	log.Debugf("%s is adding software details to registration", napPluginName)
+	napReport := n.generateNAPDetailsProtoCommand()
+	napSoftwareDetails := &proto.DataplaneSoftwareDetails{
+		Data: napReport,
 	}
+
+	n.messagePipeline.Process(core.NewMessage(core.RegisterWithDataplaneSoftwareDetails, payloads.NewRegisterWithDataplaneSoftwareDetailsPayload(napPluginName, napSoftwareDetails)))
 }
 
 // monitor Monitors the system for any changes related to NAP, if any changes are detected

src/plugins/nginx_app_protect_test.go

@@ -10,6 +10,7 @@ import (
 	"github.com/nginx/agent/sdk/v2/proto"
 	"github.com/nginx/agent/v2/src/core"
 	"github.com/nginx/agent/v2/src/core/config"
+	"github.com/nginx/agent/v2/src/core/payloads"
 	"github.com/nginx/agent/v2/src/extensions/nginx-app-protect/nap"
 	tutils "github.com/nginx/agent/v2/test/utils"
 )
@@ -97,4 +98,10 @@ func TestNginxAppProtect(t *testing.T) {
 		currentNAPPluginDetails = napPlugin.generateNAPDetailsProtoCommand()
 		assert.Equal(t, testNAPDetailsDegraded, currentNAPPluginDetails)
 	})
+
+	t.Run("software details are sent on startup", func(t *testing.T) {
+		assert.Equal(t, core.RegisterWithDataplaneSoftwareDetails, messagePipe.GetProcessedMessages()[0].Topic())
+		assert.Equal(t, "Nginx App Protect", messagePipe.GetProcessedMessages()[0].Data().(*payloads.RegisterWithDataplaneSoftwareDetailsPayload).GetPluginName())
+		assert.NotNil(t, messagePipe.GetProcessedMessages()[0].Data().(*payloads.RegisterWithDataplaneSoftwareDetailsPayload).GetDataplaneSoftwareDetails())
+	})
 }

src/plugins/registration.go

@@ -3,6 +3,7 @@ package plugins
 import (
 	"context"
 	"fmt"
+	"sync"
 	"time"
 
 	"github.com/gogo/protobuf/types"
@@ -23,15 +24,16 @@ const (
 )
 
 type OneTimeRegistration struct {
-	agentVersion             string
-	tags                     *[]string
-	meta                     *proto.Metadata
-	config                   *config.Config
-	env                      core.Environment
-	host                     *proto.HostInfo
-	binary                   core.NginxBinary
-	dataplaneSoftwareDetails map[string]*proto.DataplaneSoftwareDetails
-	pipeline                 core.MessagePipeInterface
+	agentVersion                  string
+	tags                          *[]string
+	meta                          *proto.Metadata
+	config                        *config.Config
+	env                           core.Environment
+	host                          *proto.HostInfo
+	binary                        core.NginxBinary
+	dataplaneSoftwareDetails      map[string]*proto.DataplaneSoftwareDetails
+	pipeline                      core.MessagePipeInterface
+	dataplaneSoftwareDetailsMutex sync.Mutex
 }
 
 func NewOneTimeRegistration(
@@ -44,14 +46,15 @@ func NewOneTimeRegistration(
 	// this might be slow so do on startup
 	host := env.NewHostInfo(version, &config.Tags, config.ConfigDirs, true)
 	return &OneTimeRegistration{
-		tags:                     &config.Tags,
-		agentVersion:             version,
-		meta:                     meta,
-		config:                   config,
-		env:                      env,
-		host:                     host,
-		binary:                   binary,
-		dataplaneSoftwareDetails: make(map[string]*proto.DataplaneSoftwareDetails),
+		tags:                          &config.Tags,
+		agentVersion:                  version,
+		meta:                          meta,
+		config:                        config,
+		env:                           env,
+		host:                          host,
+		binary:                        binary,
+		dataplaneSoftwareDetails:      make(map[string]*proto.DataplaneSoftwareDetails),
+		dataplaneSoftwareDetailsMutex: sync.Mutex{},
 	}
 }
 
@@ -73,18 +76,33 @@ func (r *OneTimeRegistration) Process(msg *core.Message) {
 	switch {
 	case msg.Exact(core.RegistrationCompletedTopic):
 		log.Info("OneTimeRegistration completed")
+	case msg.Exact(core.RegisterWithDataplaneSoftwareDetails):
+		switch data := msg.Data().(type) {
+		case *payloads.RegisterWithDataplaneSoftwareDetailsPayload:
+			r.dataplaneSoftwareDetailsMutex.Lock()
+			defer r.dataplaneSoftwareDetailsMutex.Unlock()
+			r.dataplaneSoftwareDetails[data.GetPluginName()] = data.GetDataplaneSoftwareDetails()
+		}
 	}
 }
 
 func (r *OneTimeRegistration) Subscriptions() []string {
-	return []string{core.RegistrationCompletedTopic}
+	return []string{
+		core.RegistrationCompletedTopic,
+		core.RegisterWithDataplaneSoftwareDetails,
+	}
 }
 
 func (r *OneTimeRegistration) startRegistration() {
 	// Check if there are any plugins that will report dataplane software details upon registration and
 	// if they have already reported their details or not.
 	if pluginsReportingDataplaneSoftwareDetails(*r.config) && r.dataplaneSoftwareDetailsMissing() {
-		r.registerWithDataplaneSoftwareDetails()
+		r.dataplaneSoftwareDetailsMutex.Lock()
+		defer r.dataplaneSoftwareDetailsMutex.Unlock()
+		for _, plugin := range getPluginsReportingDataplaneSoftwareDetails(*r.config) {
+			r.dataplaneSoftwareDetails[plugin] = nil
+		}
+		go r.waitAndRegister()
 		return
 	}
 	r.registerAgent()
@@ -152,19 +170,6 @@ func (r *OneTimeRegistration) registerAgent() {
 	)
 }
 
-// registerWithDataplaneSoftwareDetails Attempts to ensure that the plugins enabled that transmit dataplane
-// software details have transmitted their details to OneTimeRegistration then registers.
-func (r *OneTimeRegistration) registerWithDataplaneSoftwareDetails() {
-	for _, plugin := range getPluginsReportingDataplaneSoftwareDetails(*r.config) {
-		r.dataplaneSoftwareDetails[plugin] = nil
-	}
-
-	registrationPayload := payloads.NewRegisterWithDataplaneSoftwareDetailsPayload(r.dataplaneSoftwareDetails)
-	r.pipeline.Process(core.NewMessage(core.RegisterWithDataplaneSoftwareDetails, registrationPayload))
-
-	go r.waitAndRegister()
-}
-
 // waitAndRegister checks in a retry loop if the plugins enabled that transmit dataplane
 // software details have transmitted their details to OneTimeRegistration then registers.
 // If the plugins do not successfully transmit their details before the max retries is
@@ -190,6 +195,8 @@ func (r *OneTimeRegistration) waitAndRegister() {
 func (r *OneTimeRegistration) dataplaneSoftwareDetailsReady() error {
 	pluginsMissingDetails := []string{}
 
+	r.dataplaneSoftwareDetailsMutex.Lock()
+	defer r.dataplaneSoftwareDetailsMutex.Unlock()
 	for pluginName, detailsReported := range r.dataplaneSoftwareDetails {
 		if detailsReported == nil {
 			pluginsMissingDetails = append(pluginsMissingDetails, pluginName)
@@ -209,6 +216,8 @@ func (r *OneTimeRegistration) dataplaneSoftwareDetailsReady() error {
 // transmit dataplane software details have already transmitted their details to
 // OneTimeRegistration. If they have then false is returned, if not then true is returned.
 func (r *OneTimeRegistration) dataplaneSoftwareDetailsMissing() bool {
+	r.dataplaneSoftwareDetailsMutex.Lock()
+	defer r.dataplaneSoftwareDetailsMutex.Unlock()
 	for _, plugin := range getPluginsReportingDataplaneSoftwareDetails(*r.config) {
 		if _, ok := r.dataplaneSoftwareDetails[plugin]; !ok {
 			return true
@@ -222,6 +231,8 @@ func (r *OneTimeRegistration) dataplaneSoftwareDetailsMissing() bool {
 func (r *OneTimeRegistration) dataplaneSoftwareDetailsSlice() []*proto.DataplaneSoftwareDetails {
 	allDetails := []*proto.DataplaneSoftwareDetails{}
 
+	r.dataplaneSoftwareDetailsMutex.Lock()
+	defer r.dataplaneSoftwareDetailsMutex.Unlock()
 	for _, details := range r.dataplaneSoftwareDetails {
 		if details != nil {
 			allDetails = append(allDetails, details)

src/plugins/registration_test.go

@@ -61,29 +61,6 @@ func TestRegistration_Process(t *testing.T) {
 	}
 }
 
-func TestRegistration_RegisterWithDataplaneSoftwareDetails(t *testing.T) {
-	cfg := &config.Config{
-		NginxAppProtect: config.NginxAppProtect{
-			ReportInterval: time.Duration(1) * time.Millisecond,
-		},
-	}
-
-	pluginUnderTest := NewOneTimeRegistration(cfg, tutils.GetMockNginxBinary(), tutils.GetMockEnvWithHostAndProcess(), nil, "")
-	messagePipe := core.SetupMockMessagePipe(t, context.TODO(), pluginUnderTest)
-
-	expectedMessages := []string{core.RegisterWithDataplaneSoftwareDetails}
-
-	messagePipe.Run()
-
-	messages := messagePipe.GetProcessedMessages()
-	assert.Len(t, messages, len(expectedMessages))
-
-	for idx, message := range messages {
-		assert.EqualValues(t, expectedMessages[idx], message.Topic())
-	}
-	messagePipe.ClearMessages()
-}
-
 func TestRegistration_DataplaneReady(t *testing.T) {
 	conf := tutils.GetMockAgentConfig()
 	conf.NginxAppProtect = config.NginxAppProtect{ReportInterval: time.Duration(15) * time.Second}
@@ -96,7 +73,7 @@ func TestRegistration_DataplaneReady(t *testing.T) {
 func TestRegistration_Subscriptions(t *testing.T) {
 	pluginUnderTest := NewOneTimeRegistration(tutils.GetMockAgentConfig(), nil, tutils.GetMockEnv(), nil, "")
 
-	assert.Equal(t, []string{core.RegistrationCompletedTopic}, pluginUnderTest.Subscriptions())
+	assert.Equal(t, []string{core.RegistrationCompletedTopic, core.RegisterWithDataplaneSoftwareDetails}, pluginUnderTest.Subscriptions())
 }
 
 func TestRegistration_Info(t *testing.T) {