fix test and race

Author: aphralG

Makefile

@@ -167,7 +167,7 @@ official-image-integration-test: $(SELECTED_PACKAGE) build-mock-management-plane
 	TEST_ENV="Container" CONTAINER_OS_TYPE=$(CONTAINER_OS_TYPE) CONTAINER_NGINX_IMAGE_REGISTRY=${CONTAINER_NGINX_IMAGE_REGISTRY} BUILD_TARGET="install" \
 	PACKAGES_REPO=$(OSS_PACKAGES_REPO) TAG=${TAG} PACKAGE_NAME=$(PACKAGE_NAME) BASE_IMAGE=$(BASE_IMAGE) DOCKERFILE_PATH=$(OFFICIAL_IMAGE_DOCKERFILE_PATH) \
 	OS_VERSION=$(OS_VERSION) OS_RELEASE=$(OS_RELEASE) IMAGE_PATH=$(IMAGE_PATH) \
-	go test -v ./test/integration/managementplane ./test/integration/managementplane
+	go test -v ./test/integration/managementplane ./test/integration/auxiliarycommandserver
 
 performance-test:
 	@mkdir -p $(TEST_BUILD_DIR)

internal/file/file_manager_service.go

@@ -100,18 +100,20 @@ type FileManagerService struct {
 	previousManifestFiles map[string]*model.ManifestFile
 	manifestFilePath      string
 	filesMutex            sync.RWMutex
+	manifestLock          *sync.RWMutex
 }
 
-func NewFileManagerService(fileServiceClient mpi.FileServiceClient, agentConfig *config.Config) *FileManagerService {
+func NewFileManagerService(fileServiceClient mpi.FileServiceClient, agentConfig *config.Config, manifestLock *sync.RWMutex) *FileManagerService {
 	return &FileManagerService{
 		agentConfig:           agentConfig,
-		fileOperator:          NewFileOperator(),
-		fileServiceOperator:   NewFileServiceOperator(agentConfig, fileServiceClient),
+		fileOperator:          NewFileOperator(manifestLock),
+		fileServiceOperator:   NewFileServiceOperator(agentConfig, fileServiceClient, manifestLock),
 		fileActions:           make(map[string]*model.FileCache),
 		rollbackFileContents:  make(map[string][]byte),
 		currentFilesOnDisk:    make(map[string]*mpi.File),
 		previousManifestFiles: make(map[string]*model.ManifestFile),
 		manifestFilePath:      agentConfig.ManifestDir + "/manifest.json",
+		manifestLock:          manifestLock,
 	}
 }
 
@@ -423,9 +425,12 @@ func (fms *FileManagerService) UpdateCurrentFilesOnDisk(
 // seems to be a control flag, avoid control coupling
 // nolint: revive
 func (fms *FileManagerService) UpdateManifestFile(currentFiles map[string]*mpi.File, referenced bool) (err error) {
+	slog.Debug("Updating manifest file", "current_files", currentFiles, "referenced", referenced)
 	currentManifestFiles, _, readError := fms.manifestFile()
 	fms.previousManifestFiles = currentManifestFiles
 	if readError != nil && !errors.Is(readError, os.ErrNotExist) {
+		slog.Debug("Error reading manifest file", "current_manifest_files",
+			currentManifestFiles, "updated_files", currentFiles, "referenced", referenced)
 		return fmt.Errorf("unable to read manifest file: %w", readError)
 	}
 
@@ -457,6 +462,8 @@ func (fms *FileManagerService) manifestFile() (map[string]*model.ManifestFile, m
 		return nil, nil, err
 	}
 
+	fms.manifestLock.Lock()
+	defer fms.manifestLock.Unlock()
 	file, err := os.ReadFile(fms.manifestFilePath)
 	if err != nil {
 		return nil, nil, fmt.Errorf("failed to read manifest file: %w", err)
@@ -466,6 +473,9 @@ func (fms *FileManagerService) manifestFile() (map[string]*model.ManifestFile, m
 
 	err = json.Unmarshal(file, &manifestFiles)
 	if err != nil {
+		if len(file) == 0 {
+			return nil, nil, fmt.Errorf("manifest file is empty: %w", err)
+		}
 		return nil, nil, fmt.Errorf("failed to parse manifest file: %w", err)
 	}
 

internal/file/file_operator.go

@@ -14,6 +14,7 @@ import (
 	"log/slog"
 	"os"
 	"path"
+	"sync"
 
 	"github.com/nginx/agent/v3/internal/model"
 
@@ -24,14 +25,18 @@ import (
 	mpi "github.com/nginx/agent/v3/api/grpc/mpi/v1"
 )
 
-type FileOperator struct{}
+type FileOperator struct {
+	manifestLock *sync.RWMutex
+}
 
 var _ fileOperator = (*FileOperator)(nil)
 
 // FileOperator only purpose is to write files,
 
-func NewFileOperator() *FileOperator {
-	return &FileOperator{}
+func NewFileOperator(manifestLock *sync.RWMutex) *FileOperator {
+	return &FileOperator{
+		manifestLock: manifestLock,
+	}
 }
 
 func (fo *FileOperator) Write(ctx context.Context, fileContent []byte, file *mpi.FileMeta) error {
@@ -152,6 +157,8 @@ func (fo *FileOperator) WriteManifestFile(updatedFiles map[string]*model.Manifes
 		return fmt.Errorf("unable to marshal manifest file json: %w", err)
 	}
 
+	fo.manifestLock.Lock()
+	defer fo.manifestLock.Unlock()
 	// 0755 allows read/execute for all, write for owner
 	if err = os.MkdirAll(manifestDir, dirPerm); err != nil {
 		return fmt.Errorf("unable to create directory %s: %w", manifestDir, err)

internal/file/file_plugin.go

@@ -8,6 +8,7 @@ package file
 import (
 	"context"
 	"log/slog"
+	"sync"
 
 	"github.com/nginx/agent/v3/pkg/files"
 	"github.com/nginx/agent/v3/pkg/id"
@@ -32,15 +33,17 @@ type FilePlugin struct {
 	conn               grpc.GrpcConnectionInterface
 	fileManagerService fileManagerServiceInterface
 	serverType         model.ServerType
+	manifestLock       *sync.RWMutex
 }
 
 func NewFilePlugin(agentConfig *config.Config, grpcConnection grpc.GrpcConnectionInterface,
-	serverType model.ServerType,
+	serverType model.ServerType, manifestLock *sync.RWMutex,
 ) *FilePlugin {
 	return &FilePlugin{
-		config:     agentConfig,
-		conn:       grpcConnection,
-		serverType: serverType,
+		config:       agentConfig,
+		conn:         grpcConnection,
+		serverType:   serverType,
+		manifestLock: manifestLock,
 	}
 }
 
@@ -52,7 +55,7 @@ func (fp *FilePlugin) Init(ctx context.Context, messagePipe bus.MessagePipeInter
 	slog.DebugContext(ctx, "Starting file plugin")
 
 	fp.messagePipe = messagePipe
-	fp.fileManagerService = NewFileManagerService(fp.conn.FileServiceClient(), fp.config)
+	fp.fileManagerService = NewFileManagerService(fp.conn.FileServiceClient(), fp.config, fp.manifestLock)
 
 	return nil
 }
@@ -145,7 +148,7 @@ func (fp *FilePlugin) handleConnectionReset(ctx context.Context, msg *bus.Messag
 		fp.conn = newConnection
 
 		reconnect = fp.fileManagerService.IsConnected()
-		fp.fileManagerService = NewFileManagerService(fp.conn.FileServiceClient(), fp.config)
+		fp.fileManagerService = NewFileManagerService(fp.conn.FileServiceClient(), fp.config, fp.manifestLock)
 		fp.fileManagerService.SetIsConnected(reconnect)
 
 		slog.DebugContext(ctx, "File manager service client reset successfully")

internal/file/file_service_operator.go

@@ -15,6 +15,7 @@ import (
 	"math"
 	"os"
 	"slices"
+	"sync"
 	"sync/atomic"
 
 	"github.com/cenkalti/backoff/v4"
@@ -41,14 +42,14 @@ type FileServiceOperator struct {
 
 var _ fileServiceOperatorInterface = (*FileServiceOperator)(nil)
 
-func NewFileServiceOperator(agentConfig *config.Config, fileServiceClient mpi.FileServiceClient) *FileServiceOperator {
+func NewFileServiceOperator(agentConfig *config.Config, fileServiceClient mpi.FileServiceClient, manifestLock *sync.RWMutex) *FileServiceOperator {
 	isConnected := &atomic.Bool{}
 	isConnected.Store(false)
 
 	return &FileServiceOperator{
 		fileServiceClient: fileServiceClient,
 		agentConfig:       agentConfig,
-		fileOperator:      NewFileOperator(),
+		fileOperator:      NewFileOperator(manifestLock),
 		isConnected:       isConnected,
 	}
 }

internal/plugin/plugin_manager.go

@@ -8,6 +8,7 @@ package plugin
 import (
 	"context"
 	"log/slog"
+	"sync"
 
 	"github.com/nginx/agent/v3/internal/model"
 
@@ -27,9 +28,11 @@ import (
 func LoadPlugins(ctx context.Context, agentConfig *config.Config) []bus.Plugin {
 	plugins := make([]bus.Plugin, 0)
 
+	manifestLock := &sync.RWMutex{}
+
 	plugins = addResourcePlugin(plugins, agentConfig)
-	plugins = addCommandAndFilePlugins(ctx, plugins, agentConfig)
-	plugins = addAuxiliaryCommandAndFilePlugins(ctx, plugins, agentConfig)
+	plugins = addCommandAndFilePlugins(ctx, plugins, agentConfig, manifestLock)
+	plugins = addAuxiliaryCommandAndFilePlugins(ctx, plugins, agentConfig, manifestLock)
 	plugins = addCollectorPlugin(ctx, agentConfig, plugins)
 	plugins = addWatcherPlugin(plugins, agentConfig)
 
@@ -43,15 +46,15 @@ func addResourcePlugin(plugins []bus.Plugin, agentConfig *config.Config) []bus.P
 	return plugins
 }
 
-func addCommandAndFilePlugins(ctx context.Context, plugins []bus.Plugin, agentConfig *config.Config) []bus.Plugin {
+func addCommandAndFilePlugins(ctx context.Context, plugins []bus.Plugin, agentConfig *config.Config, manifestLock *sync.RWMutex) []bus.Plugin {
 	if agentConfig.IsCommandGrpcClientConfigured() {
 		grpcConnection, err := grpc.NewGrpcConnection(ctx, agentConfig, agentConfig.Command)
 		if err != nil {
 			slog.WarnContext(ctx, "Failed to create gRPC connection for command server", "error", err)
 		} else {
 			commandPlugin := command.NewCommandPlugin(agentConfig, grpcConnection, model.Command)
 			plugins = append(plugins, commandPlugin)
-			filePlugin := file.NewFilePlugin(agentConfig, grpcConnection, model.Command)
+			filePlugin := file.NewFilePlugin(agentConfig, grpcConnection, model.Command, manifestLock)
 			plugins = append(plugins, filePlugin)
 		}
 	} else {
@@ -63,7 +66,7 @@ func addCommandAndFilePlugins(ctx context.Context, plugins []bus.Plugin, agentCo
 }
 
 func addAuxiliaryCommandAndFilePlugins(ctx context.Context, plugins []bus.Plugin,
-	agentConfig *config.Config,
+	agentConfig *config.Config, manifestLock *sync.RWMutex,
 ) []bus.Plugin {
 	if agentConfig.IsAuxiliaryCommandGrpcClientConfigured() {
 		auxGRPCConnection, err := grpc.NewGrpcConnection(ctx, agentConfig, agentConfig.AuxiliaryCommand)
@@ -72,7 +75,7 @@ func addAuxiliaryCommandAndFilePlugins(ctx context.Context, plugins []bus.Plugin
 		} else {
 			auxCommandPlugin := command.NewCommandPlugin(agentConfig, auxGRPCConnection, model.Auxiliary)
 			plugins = append(plugins, auxCommandPlugin)
-			readFilePlugin := file.NewFilePlugin(agentConfig, auxGRPCConnection, model.Auxiliary)
+			readFilePlugin := file.NewFilePlugin(agentConfig, auxGRPCConnection, model.Auxiliary, manifestLock)
 			plugins = append(plugins, readFilePlugin)
 		}
 	} else {

internal/watcher/file/file_watcher_service.go

@@ -195,7 +195,7 @@ func (fws *FileWatcherService) checkForUpdates(ctx context.Context, ch chan<- Fi
 	// Check if directories no longer need to be watched
 	fws.removeWatchers(ctx)
 
-	if fws.filesChanged.Load() {
+	if fws.filesChanged.Load() && fws.enabled.Load() {
 		newCtx := context.WithValue(
 			ctx,
 			logger.CorrelationIDContextKey,

test/helpers/test_containers_utils.go

@@ -319,6 +319,8 @@ func LogAndTerminateContainers(
 	require.NoError(tb, err)
 	logs := string(buf)
 
+	assert.NotContains(tb, logs, "manifest file is empty",
+		"Error reading manifest file found in agent log")
 	tb.Log(logs)
 	if expectNoErrorsInLogs {
 		assert.NotContains(tb, logs, "level=ERROR", "agent log file contains logs at error level")

test/integration/auxiliarycommandserver/connection_test.go

@@ -8,14 +8,14 @@ package auxiliarycommandserver
 import (
 	"context"
 	"fmt"
+	"github.com/go-resty/resty/v2"
 	"net"
 	"net/http"
 	"os"
 	"sort"
 	"testing"
 	"time"
 
-	"github.com/go-resty/resty/v2"
 	mpi "github.com/nginx/agent/v3/api/grpc/mpi/v1"
 	"github.com/nginx/agent/v3/test/integration/utils"
 	"github.com/stretchr/testify/suite"
@@ -187,10 +187,6 @@ func (s *AuxiliaryTestSuite) TestAuxiliary_Test5_ConfigApply() {
 }
 
 func (s *AuxiliaryTestSuite) TestAuxiliary_Test6_ConfigApplyInvalid() {
-	// Perform config apply with aux
-	// Check new config is broken
-	// Check using hash with new API endpoint which was added to get the file overview
-
 	utils.ClearManagementPlaneResponses(s.T(), utils.MockManagementPlaneAPIAddress)
 	utils.ClearManagementPlaneResponses(s.T(), utils.AuxiliaryMockManagementPlaneAPIAddress)