WIP

Author: aphralG

internal/resource/resource_plugin.go

@@ -10,6 +10,7 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
+	"sync"
 
 	mpi "github.com/nginx/agent/v3/api/grpc/mpi/v1"
 	"github.com/nginx/agent/v3/internal/config"
@@ -28,6 +29,7 @@ type Resource struct {
 	messagePipe     bus.MessagePipeInterface
 	resourceService resourceServiceInterface
 	agentConfig     *config.Config
+	resourceMutex   sync.Mutex
 }
 
 type errResponse struct {
@@ -46,7 +48,8 @@ var _ bus.Plugin = (*Resource)(nil)
 
 func NewResource(agentConfig *config.Config) *Resource {
 	return &Resource{
-		agentConfig: agentConfig,
+		agentConfig:   agentConfig,
+		resourceMutex: sync.Mutex{},
 	}
 }
 
@@ -94,6 +97,8 @@ func (r *Resource) Process(ctx context.Context, msg *bus.Message) {
 
 			return
 		}
+		r.resourceMutex.Lock()
+		defer r.resourceMutex.Unlock()
 		resource := r.resourceService.UpdateInstances(instanceList)
 
 		r.messagePipe.Process(ctx, &bus.Message{Topic: bus.ResourceUpdateTopic, Data: resource})

internal/resource/resource_service.go

@@ -160,6 +160,9 @@ func (r *ResourceService) DeleteInstances(instanceList []*mpi.Instance) *mpi.Res
 }
 
 func (r *ResourceService) ApplyConfig(ctx context.Context, instanceID string) error {
+	r.resourceMutex.Lock()
+	defer r.resourceMutex.Unlock()
+
 	var instance *mpi.Instance
 	operator := r.instanceOperators[instanceID]
 

internal/watcher/health/health_watcher_service.go

@@ -52,6 +52,8 @@ func NewHealthWatcherService(agentConfig *config.Config) *HealthWatcherService {
 }
 
 func (hw *HealthWatcherService) AddHealthWatcher(instances []*mpi.Instance) {
+	hw.healthWatcherMutex.Lock()
+	defer hw.healthWatcherMutex.Unlock()
 	for _, instance := range instances {
 		switch instance.GetInstanceMeta().GetInstanceType() {
 		case mpi.InstanceMeta_INSTANCE_TYPE_NGINX, mpi.InstanceMeta_INSTANCE_TYPE_NGINX_PLUS:

internal/watcher/health/nginx_health_watcher_operator.go

@@ -8,6 +8,7 @@ package health
 import (
 	"context"
 	"fmt"
+	"sync"
 
 	mpi "github.com/nginx/agent/v3/api/grpc/mpi/v1"
 	"github.com/nginx/agent/v3/internal/datasource/host/exec"
@@ -18,6 +19,7 @@ import (
 type NginxHealthWatcher struct {
 	executer        exec.ExecInterface
 	processOperator processwatcher.ProcessOperatorInterface
+	healthMutex     sync.Mutex
 }
 
 var _ healthWatcherOperator = (*NginxHealthWatcher)(nil)
@@ -26,10 +28,14 @@ func NewNginxHealthWatcher() *NginxHealthWatcher {
 	return &NginxHealthWatcher{
 		executer:        &exec.Exec{},
 		processOperator: processwatcher.NewProcessOperator(),
+		healthMutex:     sync.Mutex{},
 	}
 }
 
 func (nhw *NginxHealthWatcher) Health(ctx context.Context, instance *mpi.Instance) (*mpi.InstanceHealth, error) {
+	nhw.healthMutex.Lock()
+	defer nhw.healthMutex.Unlock()
+
 	health := &mpi.InstanceHealth{
 		InstanceId:           instance.GetInstanceMeta().GetInstanceId(),
 		InstanceHealthStatus: mpi.InstanceHealth_INSTANCE_HEALTH_STATUS_HEALTHY,

internal/watcher/watcher_plugin.go

@@ -9,6 +9,7 @@ import (
 	"context"
 	"log/slog"
 	"slices"
+	"sync"
 
 	mpi "github.com/nginx/agent/v3/api/grpc/mpi/v1"
 
@@ -39,6 +40,7 @@ type (
 		fileUpdatesChannel                 chan file.FileUpdateMessage
 		cancel                             context.CancelFunc
 		instancesWithConfigApplyInProgress []string
+		watcherMutex                       sync.Mutex
 	}
 
 	instanceWatcherServiceInterface interface {
@@ -65,6 +67,7 @@ func NewWatcher(agentConfig *config.Config) *Watcher {
 		instanceHealthChannel:              make(chan health.InstanceHealthMessage),
 		fileUpdatesChannel:                 make(chan file.FileUpdateMessage),
 		instancesWithConfigApplyInProgress: []string{},
+		watcherMutex:                       sync.Mutex{},
 	}
 }
 
@@ -148,6 +151,8 @@ func (w *Watcher) handleConfigApplyRequest(ctx context.Context, msg *bus.Message
 
 	instanceID := request.ConfigApplyRequest.GetOverview().GetConfigVersion().GetInstanceId()
 
+	w.watcherMutex.Lock()
+	defer w.watcherMutex.Unlock()
 	w.instancesWithConfigApplyInProgress = append(w.instancesWithConfigApplyInProgress, instanceID)
 	w.fileWatcherService.SetEnabled(false)
 }
@@ -163,6 +168,8 @@ func (w *Watcher) handleConfigApplySuccess(ctx context.Context, msg *bus.Message
 
 	instanceID := response.GetInstanceId()
 
+	w.watcherMutex.Lock()
+	defer w.watcherMutex.Unlock()
 	w.instancesWithConfigApplyInProgress = slices.DeleteFunc(
 		w.instancesWithConfigApplyInProgress,
 		func(element string) bool {
@@ -191,6 +198,8 @@ func (w *Watcher) handleConfigApplyComplete(ctx context.Context, msg *bus.Messag
 
 	instanceID := response.GetInstanceId()
 
+	w.watcherMutex.Lock()
+	defer w.watcherMutex.Unlock()
 	w.instancesWithConfigApplyInProgress = slices.DeleteFunc(
 		w.instancesWithConfigApplyInProgress,
 		func(element string) bool {
@@ -210,7 +219,7 @@ func (w *Watcher) monitorWatchers(ctx context.Context) {
 			w.handleInstanceUpdates(newCtx, message)
 		case message := <-w.nginxConfigContextChannel:
 			newCtx := context.WithValue(ctx, logger.CorrelationIDContextKey, message.CorrelationID)
-
+			w.watcherMutex.Lock()
 			if !slices.Contains(w.instancesWithConfigApplyInProgress, message.NginxConfigContext.InstanceID) {
 				slog.DebugContext(
 					newCtx,
@@ -229,6 +238,7 @@ func (w *Watcher) monitorWatchers(ctx context.Context) {
 					"nginx_config_context", message.NginxConfigContext,
 				)
 			}
+			w.watcherMutex.Unlock()
 		case message := <-w.instanceHealthChannel:
 			newCtx := context.WithValue(ctx, logger.CorrelationIDContextKey, message.CorrelationID)
 			w.messagePipe.Process(newCtx, &bus.Message{