merge v3

api/grpc/mpi/v1/files.proto

@@ -51,8 +51,11 @@ message UpdateOverviewRequest {
     FileOverview overview = 2;
 }
 
-// Represents a the response from an UpdateOverviewRequest - intentionally left empty
-message UpdateOverviewResponse {}
+// Represents a the response from an UpdateOverviewRequest
+message UpdateOverviewResponse {
+    // The file overview with the list of files that were uploaded
+    FileOverview overview = 1;
+}
 
 // Represents a specific configuration version associated with an instance
 message ConfigVersion {
@@ -148,8 +151,8 @@ message UpdateFileResponse {
 // Define the certificate message based on https://pkg.go.dev/crypto/x509#Certificate 
 // and https://github.com/googleapis/googleapis/blob/005df4681b89bd204a90b76168a6dc9d9e7bf4fe/google/cloud/iot/v1/resources.proto#L341
 message CertificateMeta {
-    // Serial number of the certificate, usually a unique identifier, RFC5280 states the upper limit for serial number is 20 octets
-    bytes serial_number = 1 [(buf.validate.field).bytes.min_len = 0, (buf.validate.field).bytes.max_len = 21];
+    // Serial number of the certificate, usually a unique identifier, the max length is the length of an interger
+    string serial_number = 1 [(buf.validate.field).string.min_len = 0];
 
     // Issuer details (who issued the certificate)
     X509Name issuer = 2;
@@ -267,7 +270,7 @@ message X509Name {
     repeated string postal_code = 7 [(buf.validate.field).repeated.items.string.min_len = 1];
 
     // Serial Number (SN): Unique identifier or serial number.
-    string serial_number = 8 [(buf.validate.field).string.min_len = 0, (buf.validate.field).string.max_len = 21];
+    string serial_number = 8 [(buf.validate.field).string.min_len = 0];
 
     // Common Name (CN): Typically the person’s or entity's full name.
     string common_name = 9 [(buf.validate.field).string.min_len = 1];

docs/proto/protos.md

@@ -259,7 +259,7 @@ and https://github.com/googleapis/googleapis/blob/005df4681b89bd204a90b76168a6dc
 
 | Field | Type | Label | Description |
 | ----- | ---- | ----- | ----------- |
-| serial_number | [bytes](#bytes) |  | Serial number of the certificate, usually a unique identifier, RFC5280 states the upper limit for serial number is 20 octets |
+| serial_number | [string](#string) |  | Serial number of the certificate, usually a unique identifier, the max length is the length of an interger |
 | issuer | [X509Name](#mpi-v1-X509Name) |  | Issuer details (who issued the certificate) |
 | subject | [X509Name](#mpi-v1-X509Name) |  | Subject details (to whom the certificate is issued) |
 | sans | [SubjectAlternativeNames](#mpi-v1-SubjectAlternativeNames) |  | Subject Alternative Names (SAN) including DNS names and IP addresses |
@@ -484,7 +484,12 @@ Represents a list of logically grouped files that have changed e.g. configuratio
 <a name="mpi-v1-UpdateOverviewResponse"></a>
 
 ### UpdateOverviewResponse
-Represents a the response from an UpdateOverviewRequest - intentionally left empty
+Represents a the response from an UpdateOverviewRequest
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| overview | [FileOverview](#mpi-v1-FileOverview) |  | The file overview with the list of files that were uploaded |
 
 
 

internal/file/file_manager_service.go

@@ -10,7 +10,9 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
+	"maps"
 	"os"
+	"slices"
 	"sync"
 	"sync/atomic"
 
@@ -35,13 +37,15 @@ import (
 //go:generate go run github.com/maxbrunsfeld/counterfeiter/[email protected] -generate
 //counterfeiter:generate . fileManagerServiceInterface
 
+const maxAttempts = 5
+
 type (
 	fileOperator interface {
 		Write(ctx context.Context, fileContent []byte, file *mpi.FileMeta) error
 	}
 
 	fileManagerServiceInterface interface {
-		UpdateOverview(ctx context.Context, instanceID string, filesToUpdate []*mpi.File) error
+		UpdateOverview(ctx context.Context, instanceID string, filesToUpdate []*mpi.File, iteration int) error
 		ConfigApply(ctx context.Context, configApplyRequest *mpi.ConfigApplyRequest) (writeStatus model.WriteStatus,
 			err error)
 		Rollback(ctx context.Context, instanceID string) error
@@ -87,18 +91,21 @@ func (fms *FileManagerService) UpdateOverview(
 	ctx context.Context,
 	instanceID string,
 	filesToUpdate []*mpi.File,
+	iteration int,
 ) error {
 	correlationID := logger.GetCorrelationID(ctx)
-	requestCorrelationID := logger.GenerateCorrelationID()
-	newCtx := context.WithValue(ctx, logger.CorrelationIDContextKey, requestCorrelationID)
 
-	slog.InfoContext(newCtx, "Updating file overview", "instance_id", instanceID,
-		"parent_correlation_id", correlationID)
+	// error case for the UpdateOverview attempts
+	if iteration > maxAttempts {
+		return errors.New("too many UpdateOverview attempts")
+	}
+
+	newCtx, correlationID := fms.setupIdentifiers(ctx, iteration)
 
 	request := &mpi.UpdateOverviewRequest{
 		MessageMeta: &mpi.MessageMeta{
 			MessageId:     proto.GenerateMessageID(),
-			CorrelationId: requestCorrelationID.Value.String(),
+			CorrelationId: correlationID,
 			Timestamp:     timestamppb.Now(),
 		},
 		Overview: &mpi.FileOverview{
@@ -147,9 +154,57 @@ func (fms *FileManagerService) UpdateOverview(
 
 	slog.DebugContext(newCtx, "UpdateOverview response", "response", response)
 
+	if response.GetOverview() == nil {
+		slog.Debug("UpdateOverview response is empty")
+		return nil
+	}
+	delta := files.ConvertToMapOfFiles(response.GetOverview().GetFiles())
+
+	if len(delta) != 0 {
+		return fms.updateFiles(ctx, delta, instanceID, iteration)
+	}
+
 	return err
 }
 
+func (fms *FileManagerService) setupIdentifiers(ctx context.Context, iteration int) (context.Context, string) {
+	correlationID := logger.GetCorrelationID(ctx)
+	var requestCorrelationID slog.Attr
+
+	if iteration == 0 {
+		requestCorrelationID = logger.GenerateCorrelationID()
+	} else {
+		requestCorrelationID = logger.GetCorrelationIDAttr(ctx)
+	}
+
+	newCtx := context.WithValue(ctx, logger.CorrelationIDContextKey, requestCorrelationID)
+	slog.InfoContext(newCtx, "Updating file overview", "instance_id", logger.GetCorrelationIDAttr(ctx),
+		"parent_correlation_id", correlationID)
+
+	return newCtx, correlationID
+}
+
+func (fms *FileManagerService) updateFiles(
+	ctx context.Context,
+	delta map[string]*mpi.File,
+	instanceID string,
+	iteration int,
+) error {
+	diffFiles := slices.Collect(maps.Values(delta))
+
+	for _, file := range diffFiles {
+		updateErr := fms.UpdateFile(ctx, instanceID, file)
+		if updateErr != nil {
+			return updateErr
+		}
+	}
+
+	iteration++
+	slog.Debug("Updating file overview", "attempt_number", iteration)
+
+	return fms.UpdateOverview(ctx, instanceID, diffFiles, iteration)
+}
+
 func (fms *FileManagerService) UpdateFile(
 	ctx context.Context,
 	instanceID string,

internal/file/file_manager_service_test.go

@@ -29,20 +29,75 @@ import (
 func TestFileManagerService_UpdateOverview(t *testing.T) {
 	ctx := context.Background()
 
-	fileMeta := protos.FileMeta("/etc/nginx/nginx.conf", "")
+	filePath := filepath.Join(t.TempDir(), "nginx.conf")
+	fileMeta := protos.FileMeta(filePath, "")
+
+	fileContent := []byte("location /test {\n    return 200 \"Test location\\n\";\n}")
+	fileHash := files.GenerateHash(fileContent)
+
+	fileWriteErr := os.WriteFile(filePath, fileContent, 0o600)
+	require.NoError(t, fileWriteErr)
+
+	overview := protos.FileOverview(filePath, fileHash)
 
 	fakeFileServiceClient := &v1fakes.FakeFileServiceClient{}
+	fakeFileServiceClient.UpdateOverviewReturnsOnCall(0, &mpi.UpdateOverviewResponse{
+		Overview: overview,
+	}, nil)
+
+	fakeFileServiceClient.UpdateOverviewReturnsOnCall(1, &mpi.UpdateOverviewResponse{}, nil)
+
+	fakeFileServiceClient.UpdateFileReturns(&mpi.UpdateFileResponse{}, nil)
+
 	fileManagerService := NewFileManagerService(fakeFileServiceClient, types.AgentConfig())
 	fileManagerService.SetIsConnected(true)
 
 	err := fileManagerService.UpdateOverview(ctx, "123", []*mpi.File{
 		{
 			FileMeta: fileMeta,
 		},
-	})
+	}, 0)
 
 	require.NoError(t, err)
-	assert.Equal(t, 1, fakeFileServiceClient.UpdateOverviewCallCount())
+	assert.Equal(t, 2, fakeFileServiceClient.UpdateOverviewCallCount())
+}
+
+func TestFileManagerService_UpdateOverview_MaxIterations(t *testing.T) {
+	ctx := context.Background()
+
+	filePath := filepath.Join(t.TempDir(), "nginx.conf")
+	fileMeta := protos.FileMeta(filePath, "")
+
+	fileContent := []byte("location /test {\n    return 200 \"Test location\\n\";\n}")
+	fileHash := files.GenerateHash(fileContent)
+
+	fileWriteErr := os.WriteFile(filePath, fileContent, 0o600)
+	require.NoError(t, fileWriteErr)
+
+	overview := protos.FileOverview(filePath, fileHash)
+
+	fakeFileServiceClient := &v1fakes.FakeFileServiceClient{}
+
+	// do 5 iterations
+	for i := 0; i <= 5; i++ {
+		fakeFileServiceClient.UpdateOverviewReturnsOnCall(i, &mpi.UpdateOverviewResponse{
+			Overview: overview,
+		}, nil)
+	}
+
+	fakeFileServiceClient.UpdateFileReturns(&mpi.UpdateFileResponse{}, nil)
+
+	fileManagerService := NewFileManagerService(fakeFileServiceClient, types.AgentConfig())
+	fileManagerService.SetIsConnected(true)
+
+	err := fileManagerService.UpdateOverview(ctx, "123", []*mpi.File{
+		{
+			FileMeta: fileMeta,
+		},
+	}, 0)
+
+	require.Error(t, err)
+	assert.Equal(t, "too many UpdateOverview attempts", err.Error())
 }
 
 func TestFileManagerService_UpdateFile(t *testing.T) {
@@ -593,13 +648,13 @@ func TestParseX509Certificates(t *testing.T) {
 		certName       string
 		certContent    string
 		name           string
-		expectedSerial []byte
+		expectedSerial string
 	}{
 		{
 			name:           "Test 1: generated cert",
 			certName:       "public_cert",
 			certContent:    "",
-			expectedSerial: []byte{0x1, 0xe0, 0xf3},
+			expectedSerial: "123123",
 		},
 		{
 			name:     "Test 2: open ssl cert",
@@ -625,15 +680,7 @@ X/vYrzgKRoKSUWUt1ejKTntrVuaJK4NMxANOTTjIXgxyoV3YcgEmL9KzribCqILi
 p79Nno9d+kovtX5VKsJ5FCcPw9mEATgZDOQ4nLTk/HHG6bwtpubp6Zb7H1AjzBkz
 rQHX6DP4w6IwZY8JB8LS
 -----END CERTIFICATE-----`,
-			expectedSerial: []byte{
-				0x47, 0xe6, 0x6,
-				0x81, 0x11, 0xe1,
-				0x63, 0xa, 0x2d,
-				0x17, 0x20, 0x4e,
-				0xbd, 0x27, 0x35,
-				0x28, 0x3f, 0x5d,
-				0xe3, 0x99,
-			},
+			expectedSerial: "410468082718062724391949173062901619571168240537",
 		},
 	}
 

internal/file/file_plugin.go

@@ -259,7 +259,7 @@ func (fp *FilePlugin) handleNginxConfigUpdate(ctx context.Context, msg *bus.Mess
 
 	fp.fileManagerService.UpdateCurrentFilesOnDisk(files.ConvertToMapOfFiles(nginxConfigContext.Files))
 
-	err := fp.fileManagerService.UpdateOverview(ctx, nginxConfigContext.InstanceID, nginxConfigContext.Files)
+	err := fp.fileManagerService.UpdateOverview(ctx, nginxConfigContext.InstanceID, nginxConfigContext.Files, 0)
 	if err != nil {
 		slog.ErrorContext(
 			ctx,

internal/file/file_plugin_test.go

@@ -77,6 +77,10 @@ func TestFilePlugin_Process_NginxConfigUpdateTopic(t *testing.T) {
 	}
 
 	fakeFileServiceClient := &v1fakes.FakeFileServiceClient{}
+	fakeFileServiceClient.UpdateOverviewReturns(&mpi.UpdateOverviewResponse{
+		Overview: nil,
+	}, nil)
+
 	fakeGrpcConnection := &grpcfakes.FakeGrpcConnectionInterface{}
 	fakeGrpcConnection.FileServiceClientReturns(fakeFileServiceClient)
 	messagePipe := busfakes.NewFakeMessagePipe()