Merge branch 'add-auxiliary-command-server-to-agent-config' into add-read-only-file-plugin

Author: aphralG

internal/command/command_plugin.go

@@ -25,6 +25,17 @@ var _ bus.Plugin = (*CommandPlugin)(nil)
 
 //go:generate go run github.com/maxbrunsfeld/counterfeiter/v6@v6.8.1 -generate
 //counterfeiter:generate . commandService
+type ServerType int
+
+const (
+	Command ServerType = iota
+	Auxiliary
+)
+
+var serverType = map[ServerType]string{
+	Command:   "command",
+	Auxiliary: "auxiliary",
+}
 
 type (
 	commandService interface {
@@ -44,13 +55,13 @@ type (
 		conn              grpc.GrpcConnectionInterface
 		commandService    commandService
 		subscribeChannel  chan *mpi.ManagementPlaneRequest
-		commandServerType string
+		commandServerType ServerType
 		subscribeMutex    sync.Mutex
 	}
 )
 
 func NewCommandPlugin(agentConfig *config.Config, grpcConnection grpc.GrpcConnectionInterface,
-	commandServerType string,
+	commandServerType ServerType,
 ) *CommandPlugin {
 	return &CommandPlugin{
 		config:            agentConfig,
@@ -61,22 +72,22 @@ func NewCommandPlugin(agentConfig *config.Config, grpcConnection grpc.GrpcConnec
 }
 
 func (cp *CommandPlugin) Init(ctx context.Context, messagePipe bus.MessagePipeInterface) error {
-	slog.DebugContext(ctx, "Starting command plugin", "command_server_type", cp.commandServerType)
+	newCtx := context.WithValue(
+		ctx,
+		logger.ServerTypeContextKey, slog.Any(logger.ServerTypeKey, cp.commandServerType.String()),
+	)
+	slog.DebugContext(newCtx, "Starting command plugin", "command_server_type", cp.commandServerType.String())
 
 	cp.messagePipe = messagePipe
 	cp.commandService = NewCommandService(cp.conn.CommandServiceClient(), cp.config, cp.subscribeChannel)
 
-	newCtx := context.WithValue(
-		ctx,
-		logger.ServerTypeContextKey, slog.Any(logger.ServerTypeKey, cp.commandServerType),
-	)
 	go cp.monitorSubscribeChannel(newCtx)
 
 	return nil
 }
 
 func (cp *CommandPlugin) Close(ctx context.Context) error {
-	slog.InfoContext(ctx, "Closing command plugin", "command_server_type", cp.commandServerType)
+	slog.InfoContext(ctx, "Closing command plugin", "command_server_type", cp.commandServerType.String())
 
 	cp.subscribeMutex.Lock()
 	if cp.subscribeCancel != nil {
@@ -89,14 +100,14 @@ func (cp *CommandPlugin) Close(ctx context.Context) error {
 
 func (cp *CommandPlugin) Info() *bus.Info {
 	return &bus.Info{
-		Name: cp.commandServerType,
+		Name: cp.commandServerType.String(),
 	}
 }
 
 func (cp *CommandPlugin) Process(ctx context.Context, msg *bus.Message) {
 	slog.DebugContext(ctx, "Processing command", "command_server_type", logger.ServerType(ctx))
 
-	if logger.ServerType(ctx) == cp.commandServerType || logger.ServerType(ctx) == "" {
+	if logger.ServerType(ctx) == cp.commandServerType.String() || logger.ServerType(ctx) == "" {
 		switch msg.Topic {
 		case bus.ConnectionResetTopic:
 			cp.processConnectionReset(ctx, msg)
@@ -111,9 +122,6 @@ func (cp *CommandPlugin) Process(ctx context.Context, msg *bus.Message) {
 		default:
 			slog.DebugContext(ctx, "Command plugin received unknown topic", "topic", msg.Topic)
 		}
-	} else {
-		slog.Info("Sever type is not right ignoring message", "command_server_type",
-			logger.ServerType(ctx), "topic", msg.Topic)
 	}
 }
 
@@ -123,7 +131,7 @@ func (cp *CommandPlugin) processResourceUpdate(ctx context.Context, msg *bus.Mes
 		if !cp.commandService.IsConnected() {
 			newCtx := context.WithValue(
 				ctx,
-				logger.ServerTypeContextKey, slog.Any(logger.ServerTypeKey, cp.commandServerType),
+				logger.ServerTypeContextKey, slog.Any(logger.ServerTypeKey, cp.commandServerType.String()),
 			)
 			cp.createConnection(newCtx, resource)
 		} else {
@@ -185,7 +193,7 @@ func (cp *CommandPlugin) processInstanceHealth(ctx context.Context, msg *bus.Mes
 		err := cp.commandService.UpdateDataPlaneHealth(ctx, instances)
 		if err != nil {
 			slog.ErrorContext(ctx, "Unable to update data plane health", "error", err,
-				"command_server_type", cp.commandServerType)
+				"command_server_type", cp.commandServerType.String())
 		}
 	}
 }
@@ -248,9 +256,10 @@ func (cp *CommandPlugin) monitorSubscribeChannel(ctx context.Context) {
 				slog.InfoContext(ctx, "Received management plane config upload request")
 				cp.handleConfigUploadRequest(newCtx, message)
 			case *mpi.ManagementPlaneRequest_ConfigApplyRequest:
-				if cp.commandServerType != "command" {
+				if cp.commandServerType != Command {
 					slog.WarnContext(newCtx, "Auxiliary command server can not perform config apply",
-						"command_server_type", cp.commandServerType)
+						"command_server_type", cp.commandServerType.String())
+					cp.handleInvalidRequest(newCtx, message)
 
 					return
 				}
@@ -260,9 +269,10 @@ func (cp *CommandPlugin) monitorSubscribeChannel(ctx context.Context) {
 				slog.InfoContext(ctx, "Received management plane health request")
 				cp.handleHealthRequest(newCtx)
 			case *mpi.ManagementPlaneRequest_ActionRequest:
-				if cp.commandServerType != "command" {
+				if cp.commandServerType != Command {
 					slog.WarnContext(newCtx, "Auxiliary command server can not perform api action",
-						"command_server_type", cp.commandServerType)
+						"command_server_type", cp.commandServerType.String())
+					cp.handleInvalidRequest(newCtx, message)
 
 					return
 				}
@@ -354,6 +364,21 @@ func (cp *CommandPlugin) handleHealthRequest(newCtx context.Context) {
 	cp.messagePipe.Process(newCtx, &bus.Message{Topic: bus.DataPlaneHealthRequestTopic})
 }
 
+func (cp *CommandPlugin) handleInvalidRequest(ctx context.Context, message *mpi.ManagementPlaneRequest) {
+	err := cp.commandService.SendDataPlaneResponse(ctx, &mpi.DataPlaneResponse{
+		MessageMeta: message.GetMessageMeta(),
+		CommandResponse: &mpi.CommandResponse{
+			Status:  mpi.CommandResponse_COMMAND_STATUS_FAILURE,
+			Message: "Can not perform write action as auxiliary command server",
+			Error:   "request not allowed",
+		},
+		InstanceId: message.GetActionRequest().GetInstanceId(),
+	})
+	if err != nil {
+		slog.ErrorContext(ctx, "Unable to send data plane response", "error", err)
+	}
+}
+
 func (cp *CommandPlugin) createDataPlaneResponse(correlationID string, status mpi.CommandResponse_CommandStatus,
 	message, err string,
 ) *mpi.DataPlaneResponse {
@@ -370,3 +395,7 @@ func (cp *CommandPlugin) createDataPlaneResponse(correlationID string, status mp
 		},
 	}
 }
+
+func (s ServerType) String() string {
+	return serverType[s]
+}

internal/command/command_plugin_test.go

@@ -31,17 +31,15 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-const serverType = "command"
-
 func TestCommandPlugin_Info(t *testing.T) {
-	commandPlugin := NewCommandPlugin(types.AgentConfig(), &grpcfakes.FakeGrpcConnectionInterface{}, serverType)
+	commandPlugin := NewCommandPlugin(types.AgentConfig(), &grpcfakes.FakeGrpcConnectionInterface{}, Command)
 	info := commandPlugin.Info()
 
 	assert.Equal(t, "command", info.Name)
 }
 
 func TestCommandPlugin_Subscriptions(t *testing.T) {
-	commandPlugin := NewCommandPlugin(types.AgentConfig(), &grpcfakes.FakeGrpcConnectionInterface{}, serverType)
+	commandPlugin := NewCommandPlugin(types.AgentConfig(), &grpcfakes.FakeGrpcConnectionInterface{}, Command)
 	subscriptions := commandPlugin.Subscriptions()
 
 	assert.Equal(
@@ -62,7 +60,7 @@ func TestCommandPlugin_Init(t *testing.T) {
 	messagePipe := busfakes.NewFakeMessagePipe()
 	fakeCommandService := &commandfakes.FakeCommandService{}
 
-	commandPlugin := NewCommandPlugin(types.AgentConfig(), &grpcfakes.FakeGrpcConnectionInterface{}, serverType)
+	commandPlugin := NewCommandPlugin(types.AgentConfig(), &grpcfakes.FakeGrpcConnectionInterface{}, Command)
 	err := commandPlugin.Init(ctx, messagePipe)
 	require.NoError(t, err)
 
@@ -81,7 +79,7 @@ func TestCommandPlugin_createConnection(t *testing.T) {
 	commandService.CreateConnectionReturns(&mpi.CreateConnectionResponse{}, nil)
 	messagePipe := busfakes.NewFakeMessagePipe()
 
-	commandPlugin := NewCommandPlugin(types.AgentConfig(), &grpcfakes.FakeGrpcConnectionInterface{}, serverType)
+	commandPlugin := NewCommandPlugin(types.AgentConfig(), &grpcfakes.FakeGrpcConnectionInterface{}, Command)
 	err := commandPlugin.Init(ctx, messagePipe)
 	commandPlugin.commandService = commandService
 	require.NoError(t, err)
@@ -113,7 +111,7 @@ func TestCommandPlugin_Process(t *testing.T) {
 	messagePipe := busfakes.NewFakeMessagePipe()
 	fakeCommandService := &commandfakes.FakeCommandService{}
 
-	commandPlugin := NewCommandPlugin(types.AgentConfig(), &grpcfakes.FakeGrpcConnectionInterface{}, serverType)
+	commandPlugin := NewCommandPlugin(types.AgentConfig(), &grpcfakes.FakeGrpcConnectionInterface{}, Command)
 	err := commandPlugin.Init(ctx, messagePipe)
 	require.NoError(t, err)
 	defer commandPlugin.Close(ctx)
@@ -221,7 +219,7 @@ func TestCommandPlugin_monitorSubscribeChannel(t *testing.T) {
 
 			agentConfig := types.AgentConfig()
 			agentConfig.Features = test.configFeatures
-			commandPlugin := NewCommandPlugin(agentConfig, &grpcfakes.FakeGrpcConnectionInterface{}, serverType)
+			commandPlugin := NewCommandPlugin(agentConfig, &grpcfakes.FakeGrpcConnectionInterface{}, Command)
 			err := commandPlugin.Init(ctx, messagePipe)
 			require.NoError(tt, err)
 			defer commandPlugin.Close(ctx)
@@ -321,7 +319,7 @@ func TestCommandPlugin_FeatureDisabled(t *testing.T) {
 
 			agentConfig.Features = test.configFeatures
 
-			commandPlugin := NewCommandPlugin(agentConfig, &grpcfakes.FakeGrpcConnectionInterface{}, serverType)
+			commandPlugin := NewCommandPlugin(agentConfig, &grpcfakes.FakeGrpcConnectionInterface{}, Command)
 			err := commandPlugin.Init(ctx, messagePipe)
 			commandPlugin.commandService = fakeCommandService
 			require.NoError(tt, err)
@@ -346,7 +344,7 @@ func TestMonitorSubscribeChannel(t *testing.T) {
 	logBuf := &bytes.Buffer{}
 	stub.StubLoggerWith(logBuf)
 
-	cp := NewCommandPlugin(types.AgentConfig(), &grpcfakes.FakeGrpcConnectionInterface{}, serverType)
+	cp := NewCommandPlugin(types.AgentConfig(), &grpcfakes.FakeGrpcConnectionInterface{}, Command)
 	cp.subscribeCancel = cncl
 
 	message := protos.CreateManagementPlaneRequest()
@@ -385,7 +383,7 @@ func Test_createDataPlaneResponse(t *testing.T) {
 			Error:   "",
 		},
 	}
-	commandPlugin := NewCommandPlugin(types.AgentConfig(), &grpcfakes.FakeGrpcConnectionInterface{}, serverType)
+	commandPlugin := NewCommandPlugin(types.AgentConfig(), &grpcfakes.FakeGrpcConnectionInterface{}, Command)
 	result := commandPlugin.createDataPlaneResponse(expected.GetMessageMeta().GetCorrelationId(),
 		expected.GetCommandResponse().GetStatus(),
 		expected.GetCommandResponse().GetMessage(), expected.GetCommandResponse().GetError())

internal/command/command_service.go

@@ -261,7 +261,7 @@ func (cs *CommandService) UpdateClient(ctx context.Context, client mpi.CommandSe
 	cs.subscribeClientMutex.Unlock()
 
 	cs.isConnected.Store(false)
-	// Will this have the sever type ?
+
 	resp, err := cs.CreateConnection(ctx, cs.resource)
 	if err != nil {
 		return err

internal/logger/logger.go

@@ -75,6 +75,7 @@ func New(logPath, level string) *slog.Logger {
 		contextHandler{
 			handler, []any{
 				CorrelationIDContextKey,
+				ServerTypeContextKey,
 			},
 		})
 }

internal/plugin/plugin_manager.go

@@ -47,7 +47,7 @@ func addCommandAndFilePlugins(ctx context.Context, plugins []bus.Plugin, agentCo
 		if err != nil {
 			slog.WarnContext(ctx, "Failed to create gRPC connection for command server", "error", err)
 		} else {
-			commandPlugin := command.NewCommandPlugin(agentConfig, grpcConnection, "command")
+			commandPlugin := command.NewCommandPlugin(agentConfig, grpcConnection, command.Command)
 			plugins = append(plugins, commandPlugin)
 			filePlugin := file.NewFilePlugin(agentConfig, grpcConnection)
 			plugins = append(plugins, filePlugin)
@@ -68,13 +68,13 @@ func addAuxiliaryCommandAndFilePlugins(ctx context.Context, plugins []bus.Plugin
 		if err != nil {
 			slog.WarnContext(ctx, "Failed to create gRPC connection for auxiliary command server", "error", err)
 		} else {
-			auxCommandPlugin := command.NewCommandPlugin(agentConfig, auxGRPCConnection, "auxiliary")
+			auxCommandPlugin := command.NewCommandPlugin(agentConfig, auxGRPCConnection, command.Auxiliary)
 			plugins = append(plugins, auxCommandPlugin)
 			readFilePlugin := file.NewReadFilePlugin(agentConfig, auxGRPCConnection)
 			plugins = append(plugins, readFilePlugin)
 		}
 	} else {
-		slog.InfoContext(ctx, "Agent is not connected to an auxiliary management plane. "+
+		slog.DebugContext(ctx, "Agent is not connected to an auxiliary management plane. "+
 			"Configure a auxiliary command server to establish a connection.")
 	}