Add multi management plane integration test

Makefile

@@ -161,13 +161,13 @@ integration-test: $(SELECTED_PACKAGE) build-mock-management-plane-grpc
 	TEST_ENV="Container" CONTAINER_OS_TYPE=$(CONTAINER_OS_TYPE) BUILD_TARGET="install-agent-local" CONTAINER_NGINX_IMAGE_REGISTRY=${CONTAINER_NGINX_IMAGE_REGISTRY} \
 	PACKAGES_REPO=$(OSS_PACKAGES_REPO) PACKAGE_NAME=$(PACKAGE_NAME) BASE_IMAGE=$(BASE_IMAGE) DOCKERFILE_PATH=$(DOCKERFILE_PATH) IMAGE_PATH=$(IMAGE_PATH) TAG=${IMAGE_TAG} \
 	OS_VERSION=$(OS_VERSION) OS_RELEASE=$(OS_RELEASE) \
-	go test -v ./test/integration/installuninstall ./test/integration/managementplane ./test/integration/nginxless
+	go test -v ./test/integration/installuninstall ./test/integration/managementplane ./test/integration/auxiliarycommandserver ./test/integration/nginxless
 
 official-image-integration-test: $(SELECTED_PACKAGE) build-mock-management-plane-grpc
 	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
+	go test -v ./test/integration/managementplane ./test/integration/auxiliarycommandserver
 
 performance-test:
 	@mkdir -p $(TEST_BUILD_DIR)

internal/file/file_manager_service.go

@@ -88,6 +88,7 @@ type (
 )
 
 type FileManagerService struct {
+	manifestLock        *sync.RWMutex
 	agentConfig         *config.Config
 	fileOperator        fileOperator
 	fileServiceOperator fileServiceOperatorInterface
@@ -102,16 +103,19 @@ type FileManagerService struct {
 	filesMutex            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 +427,13 @@ 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 +465,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 +476,10 @@ 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_manager_service_test.go

@@ -11,6 +11,7 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"sync"
 	"testing"
 
 	"github.com/nginx/agent/v3/internal/model"
@@ -54,7 +55,7 @@ func TestFileManagerService_ConfigApply_Add(t *testing.T) {
 	agentConfig := types.AgentConfig()
 	agentConfig.AllowedDirectories = []string{tempDir}
 
-	fileManagerService := NewFileManagerService(fakeFileServiceClient, agentConfig)
+	fileManagerService := NewFileManagerService(fakeFileServiceClient, agentConfig, &sync.RWMutex{})
 	fileManagerService.agentConfig.ManifestDir = manifestDirPath
 	fileManagerService.manifestFilePath = manifestFilePath
 
@@ -101,7 +102,7 @@ func TestFileManagerService_ConfigApply_Add_LargeFile(t *testing.T) {
 	fakeFileServiceClient.GetFileStreamReturns(fakeServerStreamingClient, nil)
 	agentConfig := types.AgentConfig()
 	agentConfig.AllowedDirectories = []string{tempDir}
-	fileManagerService := NewFileManagerService(fakeFileServiceClient, agentConfig)
+	fileManagerService := NewFileManagerService(fakeFileServiceClient, agentConfig, &sync.RWMutex{})
 	fileManagerService.agentConfig.ManifestDir = manifestDirPath
 	fileManagerService.manifestFilePath = manifestFilePath
 
@@ -160,7 +161,7 @@ func TestFileManagerService_ConfigApply_Update(t *testing.T) {
 	agentConfig := types.AgentConfig()
 	agentConfig.AllowedDirectories = []string{tempDir}
 
-	fileManagerService := NewFileManagerService(fakeFileServiceClient, agentConfig)
+	fileManagerService := NewFileManagerService(fakeFileServiceClient, agentConfig, &sync.RWMutex{})
 	fileManagerService.agentConfig.ManifestDir = manifestDirPath
 	fileManagerService.manifestFilePath = manifestFilePath
 	err := fileManagerService.UpdateCurrentFilesOnDisk(ctx, filesOnDisk, false)
@@ -209,7 +210,7 @@ func TestFileManagerService_ConfigApply_Delete(t *testing.T) {
 	agentConfig := types.AgentConfig()
 	agentConfig.AllowedDirectories = []string{tempDir}
 
-	fileManagerService := NewFileManagerService(fakeFileServiceClient, agentConfig)
+	fileManagerService := NewFileManagerService(fakeFileServiceClient, agentConfig, &sync.RWMutex{})
 	fileManagerService.agentConfig.ManifestDir = manifestDirPath
 	fileManagerService.manifestFilePath = manifestFilePath
 	err := fileManagerService.UpdateCurrentFilesOnDisk(ctx, filesOnDisk, false)
@@ -247,7 +248,7 @@ func TestFileManagerService_ConfigApply_Delete(t *testing.T) {
 
 func TestFileManagerService_checkAllowedDirectory(t *testing.T) {
 	fakeFileServiceClient := &v1fakes.FakeFileServiceClient{}
-	fileManagerService := NewFileManagerService(fakeFileServiceClient, types.AgentConfig())
+	fileManagerService := NewFileManagerService(fakeFileServiceClient, types.AgentConfig(), &sync.RWMutex{})
 
 	allowedFiles := []*mpi.File{
 		{
@@ -281,7 +282,7 @@ func TestFileManagerService_checkAllowedDirectory(t *testing.T) {
 
 func TestFileManagerService_ClearCache(t *testing.T) {
 	fakeFileServiceClient := &v1fakes.FakeFileServiceClient{}
-	fileManagerService := NewFileManagerService(fakeFileServiceClient, types.AgentConfig())
+	fileManagerService := NewFileManagerService(fakeFileServiceClient, types.AgentConfig(), &sync.RWMutex{})
 
 	filesCache := map[string]*model.FileCache{
 		"file/path/test.conf": {
@@ -394,7 +395,7 @@ func TestFileManagerService_Rollback(t *testing.T) {
 
 	instanceID := protos.NginxOssInstance([]string{}).GetInstanceMeta().GetInstanceId()
 	fakeFileServiceClient := &v1fakes.FakeFileServiceClient{}
-	fileManagerService := NewFileManagerService(fakeFileServiceClient, types.AgentConfig())
+	fileManagerService := NewFileManagerService(fakeFileServiceClient, types.AgentConfig(), &sync.RWMutex{})
 	fileManagerService.rollbackFileContents = fileContentCache
 	fileManagerService.fileActions = filesCache
 	fileManagerService.agentConfig.ManifestDir = manifestDirPath
@@ -576,7 +577,7 @@ func TestFileManagerService_DetermineFileActions(t *testing.T) {
 			manifestFilePath := manifestFile.Name()
 
 			fakeFileServiceClient := &v1fakes.FakeFileServiceClient{}
-			fileManagerService := NewFileManagerService(fakeFileServiceClient, types.AgentConfig())
+			fileManagerService := NewFileManagerService(fakeFileServiceClient, types.AgentConfig(), &sync.RWMutex{})
 			fileManagerService.agentConfig.ManifestDir = manifestDirPath
 			fileManagerService.manifestFilePath = manifestFilePath
 
@@ -597,7 +598,7 @@ func TestFileManagerService_DetermineFileActions(t *testing.T) {
 func CreateTestManifestFile(t testing.TB, tempDir string, currentFiles map[string]*mpi.File) *os.File {
 	t.Helper()
 	fakeFileServiceClient := &v1fakes.FakeFileServiceClient{}
-	fileManagerService := NewFileManagerService(fakeFileServiceClient, types.AgentConfig())
+	fileManagerService := NewFileManagerService(fakeFileServiceClient, types.AgentConfig(), &sync.RWMutex{})
 	manifestFiles := fileManagerService.convertToManifestFileMap(currentFiles, true)
 	manifestJSON, err := json.MarshalIndent(manifestFiles, "", "  ")
 	require.NoError(t, err)
@@ -685,7 +686,7 @@ func TestFileManagerService_fileActions(t *testing.T) {
 			Contents: newFileContent,
 		},
 	}, nil)
-	fileManagerService := NewFileManagerService(fakeFileServiceClient, types.AgentConfig())
+	fileManagerService := NewFileManagerService(fakeFileServiceClient, types.AgentConfig(), &sync.RWMutex{})
 
 	fileManagerService.fileActions = filesCache
 

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_operator_test.go

@@ -9,6 +9,7 @@ import (
 	"context"
 	"os"
 	"path/filepath"
+	"sync"
 	"testing"
 
 	"github.com/nginx/agent/v3/pkg/files"
@@ -28,7 +29,7 @@ func TestFileOperator_Write(t *testing.T) {
 	fileContent, err := os.ReadFile("../../test/config/nginx/nginx.conf")
 	require.NoError(t, err)
 	defer helpers.RemoveFileWithErrorCheck(t, filePath)
-	fileOp := NewFileOperator()
+	fileOp := NewFileOperator(&sync.RWMutex{})
 
 	fileMeta := protos.FileMeta(filePath, files.GenerateHash(fileContent))
 

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"
@@ -27,6 +28,7 @@ var _ bus.Plugin = (*FilePlugin)(nil)
 // the file plugin does not care about the instance type
 
 type FilePlugin struct {
+	manifestLock       *sync.RWMutex
 	messagePipe        bus.MessagePipeInterface
 	config             *config.Config
 	conn               grpc.GrpcConnectionInterface
@@ -35,12 +37,13 @@ type FilePlugin struct {
 }
 
 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_plugin_test.go

@@ -9,6 +9,7 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"sync"
 	"testing"
 	"time"
 
@@ -31,22 +32,24 @@ import (
 )
 
 func TestFilePlugin_Info(t *testing.T) {
-	filePlugin := NewFilePlugin(types.AgentConfig(), &grpcfakes.FakeGrpcConnectionInterface{}, model.Command)
+	filePlugin := NewFilePlugin(types.AgentConfig(), &grpcfakes.FakeGrpcConnectionInterface{},
+		model.Command, &sync.RWMutex{})
 	assert.Equal(t, "file", filePlugin.Info().Name)
 }
 
 func TestFilePlugin_Close(t *testing.T) {
 	ctx := context.Background()
 	fakeGrpcConnection := &grpcfakes.FakeGrpcConnectionInterface{}
 
-	filePlugin := NewFilePlugin(types.AgentConfig(), fakeGrpcConnection, model.Command)
+	filePlugin := NewFilePlugin(types.AgentConfig(), fakeGrpcConnection, model.Command, &sync.RWMutex{})
 	filePlugin.Close(ctx)
 
 	assert.Equal(t, 1, fakeGrpcConnection.CloseCallCount())
 }
 
 func TestFilePlugin_Subscriptions(t *testing.T) {
-	filePlugin := NewFilePlugin(types.AgentConfig(), &grpcfakes.FakeGrpcConnectionInterface{}, model.Command)
+	filePlugin := NewFilePlugin(types.AgentConfig(), &grpcfakes.FakeGrpcConnectionInterface{},
+		model.Command, &sync.RWMutex{})
 	assert.Equal(
 		t,
 		[]string{
@@ -62,7 +65,8 @@ func TestFilePlugin_Subscriptions(t *testing.T) {
 		filePlugin.Subscriptions(),
 	)
 
-	readOnlyFilePlugin := NewFilePlugin(types.AgentConfig(), &grpcfakes.FakeGrpcConnectionInterface{}, model.Auxiliary)
+	readOnlyFilePlugin := NewFilePlugin(types.AgentConfig(), &grpcfakes.FakeGrpcConnectionInterface{},
+		model.Auxiliary, &sync.RWMutex{})
 	assert.Equal(t, []string{
 		bus.ConnectionResetTopic,
 		bus.ConnectionCreatedTopic,
@@ -93,7 +97,7 @@ func TestFilePlugin_Process_NginxConfigUpdateTopic(t *testing.T) {
 	fakeGrpcConnection.FileServiceClientReturns(fakeFileServiceClient)
 	messagePipe := busfakes.NewFakeMessagePipe()
 
-	filePlugin := NewFilePlugin(types.AgentConfig(), fakeGrpcConnection, model.Command)
+	filePlugin := NewFilePlugin(types.AgentConfig(), fakeGrpcConnection, model.Command, &sync.RWMutex{})
 	err := filePlugin.Init(ctx, messagePipe)
 	require.NoError(t, err)
 
@@ -168,7 +172,7 @@ func TestFilePlugin_Process_ConfigApplyRequestTopic(t *testing.T) {
 			fakeFileManagerService := &filefakes.FakeFileManagerServiceInterface{}
 			fakeFileManagerService.ConfigApplyReturns(test.configApplyStatus, test.configApplyReturnsErr)
 			messagePipe := busfakes.NewFakeMessagePipe()
-			filePlugin := NewFilePlugin(agentConfig, fakeGrpcConnection, model.Command)
+			filePlugin := NewFilePlugin(agentConfig, fakeGrpcConnection, model.Command, &sync.RWMutex{})
 			err := filePlugin.Init(ctx, messagePipe)
 			filePlugin.fileManagerService = fakeFileManagerService
 			require.NoError(t, err)
@@ -266,7 +270,7 @@ func TestFilePlugin_Process_ConfigUploadRequestTopic(t *testing.T) {
 	fakeGrpcConnection.FileServiceClientReturns(fakeFileServiceClient)
 	messagePipe := busfakes.NewFakeMessagePipe()
 
-	filePlugin := NewFilePlugin(types.AgentConfig(), fakeGrpcConnection, model.Command)
+	filePlugin := NewFilePlugin(types.AgentConfig(), fakeGrpcConnection, model.Command, &sync.RWMutex{})
 	err := filePlugin.Init(ctx, messagePipe)
 	require.NoError(t, err)
 
@@ -321,7 +325,7 @@ func TestFilePlugin_Process_ConfigUploadRequestTopic_Failure(t *testing.T) {
 	fakeGrpcConnection.FileServiceClientReturns(fakeFileServiceClient)
 	messagePipe := busfakes.NewFakeMessagePipe()
 
-	filePlugin := NewFilePlugin(types.AgentConfig(), fakeGrpcConnection, model.Command)
+	filePlugin := NewFilePlugin(types.AgentConfig(), fakeGrpcConnection, model.Command, &sync.RWMutex{})
 	err := filePlugin.Init(ctx, messagePipe)
 	require.NoError(t, err)
 
@@ -389,7 +393,7 @@ func TestFilePlugin_Process_ConfigApplyFailedTopic(t *testing.T) {
 
 			messagePipe := busfakes.NewFakeMessagePipe()
 			agentConfig := types.AgentConfig()
-			filePlugin := NewFilePlugin(agentConfig, fakeGrpcConnection, model.Command)
+			filePlugin := NewFilePlugin(agentConfig, fakeGrpcConnection, model.Command, &sync.RWMutex{})
 
 			err := filePlugin.Init(ctx, messagePipe)
 			require.NoError(t, err)
@@ -436,7 +440,7 @@ func TestFilePlugin_Process_ConfigApplyRollbackCompleteTopic(t *testing.T) {
 	messagePipe := busfakes.NewFakeMessagePipe()
 	agentConfig := types.AgentConfig()
 	fakeGrpcConnection := &grpcfakes.FakeGrpcConnectionInterface{}
-	filePlugin := NewFilePlugin(agentConfig, fakeGrpcConnection, model.Command)
+	filePlugin := NewFilePlugin(agentConfig, fakeGrpcConnection, model.Command, &sync.RWMutex{})
 
 	err := filePlugin.Init(ctx, messagePipe)
 	require.NoError(t, err)
@@ -481,7 +485,7 @@ func TestFilePlugin_Process_ConfigApplyCompleteTopic(t *testing.T) {
 	messagePipe := busfakes.NewFakeMessagePipe()
 	agentConfig := types.AgentConfig()
 	fakeGrpcConnection := &grpcfakes.FakeGrpcConnectionInterface{}
-	filePlugin := NewFilePlugin(agentConfig, fakeGrpcConnection, model.Command)
+	filePlugin := NewFilePlugin(agentConfig, fakeGrpcConnection, model.Command, &sync.RWMutex{})
 
 	err := filePlugin.Init(ctx, messagePipe)
 	require.NoError(t, err)