fix resource race condition

Author: aphralG

internal/resource/resource_plugin.go

@@ -9,14 +9,12 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"log/slog"
-	"sync"
-
 	mpi "github.com/nginx/agent/v3/api/grpc/mpi/v1"
 	"github.com/nginx/agent/v3/internal/config"
 	response "github.com/nginx/agent/v3/internal/datasource/proto"
 	"github.com/nginx/agent/v3/internal/logger"
 	"github.com/nginx/agent/v3/internal/model"
+	"log/slog"
 
 	"github.com/nginx/agent/v3/internal/bus"
 )
@@ -29,7 +27,6 @@ type Resource struct {
 	messagePipe     bus.MessagePipeInterface
 	resourceService resourceServiceInterface
 	agentConfig     *config.Config
-	resourceMutex   sync.Mutex
 }
 
 type errResponse struct {
@@ -48,8 +45,7 @@ var _ bus.Plugin = (*Resource)(nil)
 
 func NewResource(agentConfig *config.Config) *Resource {
 	return &Resource{
-		agentConfig:   agentConfig,
-		resourceMutex: sync.Mutex{},
+		agentConfig: agentConfig,
 	}
 }
 
@@ -97,8 +93,6 @@ 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

@@ -10,6 +10,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"google.golang.org/protobuf/proto"
 	"log/slog"
 	"net"
 	"net/http"
@@ -131,13 +132,20 @@ func (r *ResourceService) UpdateInstances(instanceList []*mpi.Instance) *mpi.Res
 	defer r.resourceMutex.Unlock()
 
 	for _, updatedInstance := range instanceList {
-		for _, instance := range r.resource.GetInstances() {
-			if updatedInstance.GetInstanceMeta().GetInstanceId() == instance.GetInstanceMeta().GetInstanceId() {
-				instance.InstanceMeta = updatedInstance.GetInstanceMeta()
-				instance.InstanceRuntime = updatedInstance.GetInstanceRuntime()
-				instance.InstanceConfig = updatedInstance.GetInstanceConfig()
+		resourceCopy, ok := proto.Clone(r.resource).(*mpi.Resource)
+		if ok {
+			for _, instance := range resourceCopy.GetInstances() {
+				if updatedInstance.GetInstanceMeta().GetInstanceId() == instance.GetInstanceMeta().GetInstanceId() {
+					instance.InstanceMeta = updatedInstance.GetInstanceMeta()
+					instance.InstanceRuntime = updatedInstance.GetInstanceRuntime()
+					instance.InstanceConfig = updatedInstance.GetInstanceConfig()
+				}
 			}
+			r.resource = resourceCopy
+		} else {
+			slog.Warn("Error updating resource instances", "instances", instanceList)
 		}
+
 	}
 
 	return r.resource
@@ -148,21 +156,24 @@ func (r *ResourceService) DeleteInstances(instanceList []*mpi.Instance) *mpi.Res
 	defer r.resourceMutex.Unlock()
 
 	for _, deletedInstance := range instanceList {
-		for index, instance := range r.resource.GetInstances() {
-			if deletedInstance.GetInstanceMeta().GetInstanceId() == instance.GetInstanceMeta().GetInstanceId() {
-				r.resource.Instances = append(r.resource.Instances[:index], r.resource.GetInstances()[index+1:]...)
+		resourceCopy, ok := proto.Clone(r.resource).(*mpi.Resource)
+		if ok {
+			for index, instance := range resourceCopy.GetInstances() {
+				if deletedInstance.GetInstanceMeta().GetInstanceId() == instance.GetInstanceMeta().GetInstanceId() {
+					r.resource.Instances = append(r.resource.Instances[:index], r.resource.GetInstances()[index+1:]...)
+				}
 			}
+		} else {
+			slog.Warn("Error deleting instances from resource", "instances", instanceList)
 		}
+
 	}
 	r.RemoveOperator(instanceList)
 
 	return r.resource
 }
 
 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/nginx_health_watcher_operator.go

@@ -8,8 +8,6 @@ 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"
 	processwatcher "github.com/nginx/agent/v3/internal/watcher/process"
@@ -19,7 +17,6 @@ import (
 type NginxHealthWatcher struct {
 	executer        exec.ExecInterface
 	processOperator processwatcher.ProcessOperatorInterface
-	healthMutex     sync.Mutex
 }
 
 var _ healthWatcherOperator = (*NginxHealthWatcher)(nil)
@@ -28,13 +25,10 @@ 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(),

internal/watcher/instance/instance_watcher_service.go

@@ -167,8 +167,6 @@ func (iw *InstanceWatcherService) ReparseConfig(ctx context.Context, instanceID
 func (iw *InstanceWatcherService) checkForUpdates(
 	ctx context.Context,
 ) {
-	iw.cacheMutex.Lock()
-	defer iw.cacheMutex.Unlock()
 
 	var instancesToParse []*mpi.Instance
 	correlationID := logger.GenerateCorrelationID()
@@ -245,6 +243,8 @@ func (iw *InstanceWatcherService) instanceUpdates(ctx context.Context) (
 	instanceUpdates InstanceUpdates,
 	err error,
 ) {
+	iw.cacheMutex.Lock()
+	defer iw.cacheMutex.Unlock()
 	processes, err := iw.processOperator.Processes(ctx)
 	if err != nil {
 		return instanceUpdates, err