Add token file watcher

internal/bus/topics.go

@@ -15,6 +15,8 @@ const (
 	ConfigUploadRequestTopic     = "config-upload-request"
 	DataPlaneResponseTopic       = "data-plane-response"
 	ConnectionCreatedTopic       = "connection-created"
+	CredentialUpdatedTopic       = "credential-updated"
+	ConnectionResetTopic         = "connection-reset"
 	ConfigApplyRequestTopic      = "config-apply-request"
 	WriteConfigSuccessfulTopic   = "write-config-successful"
 	ConfigApplySuccessfulTopic   = "config-apply-successful"

internal/command/command_plugin.go

@@ -31,6 +31,7 @@ type (
 		UpdateDataPlaneStatus(ctx context.Context, resource *mpi.Resource) error
 		UpdateDataPlaneHealth(ctx context.Context, instanceHealths []*mpi.InstanceHealth) error
 		SendDataPlaneResponse(ctx context.Context, response *mpi.DataPlaneResponse) error
+		UpdateClient(client mpi.CommandServiceClient)
 		Subscribe(ctx context.Context)
 		IsConnected() bool
 		CreateConnection(ctx context.Context, resource *mpi.Resource) (*mpi.CreateConnectionResponse, error)
@@ -86,6 +87,8 @@ func (cp *CommandPlugin) Info() *bus.Info {
 
 func (cp *CommandPlugin) Process(ctx context.Context, msg *bus.Message) {
 	switch msg.Topic {
+	case bus.ConnectionResetTopic:
+		cp.processConnectionReset(ctx, msg)
 	case bus.ResourceUpdateTopic:
 		cp.processResourceUpdate(ctx, msg)
 	case bus.InstanceHealthTopic:
@@ -172,8 +175,22 @@ func (cp *CommandPlugin) processDataPlaneResponse(ctx context.Context, msg *bus.
 	}
 }
 
+func (cp *CommandPlugin) processConnectionReset(ctx context.Context, msg *bus.Message) {
+	slog.DebugContext(ctx, "Command plugin received connection reset")
+	if newConnection, ok := msg.Data.(grpc.GrpcConnectionInterface); ok {
+		err := cp.conn.Close(ctx)
+		if err != nil {
+			slog.ErrorContext(ctx, "Command plugin: unable to close connection", "error", err)
+		}
+		cp.conn = newConnection
+		cp.commandService.UpdateClient(cp.conn.CommandServiceClient())
+		slog.DebugContext(ctx, "Command plugin: client reset successfully")
+	}
+}
+
 func (cp *CommandPlugin) Subscriptions() []string {
 	return []string{
+		bus.ConnectionResetTopic,
 		bus.ResourceUpdateTopic,
 		bus.InstanceHealthTopic,
 		bus.DataPlaneHealthResponseTopic,

internal/command/command_plugin_test.go

@@ -45,6 +45,7 @@ func TestCommandPlugin_Subscriptions(t *testing.T) {
 	assert.Equal(
 		t,
 		[]string{
+			bus.ConnectionResetTopic,
 			bus.ResourceUpdateTopic,
 			bus.InstanceHealthTopic,
 			bus.DataPlaneHealthResponseTopic,
@@ -142,6 +143,12 @@ func TestCommandPlugin_Process(t *testing.T) {
 	})
 	require.Equal(t, 1, fakeCommandService.UpdateDataPlaneHealthCallCount())
 	require.Equal(t, 1, fakeCommandService.SendDataPlaneResponseCallCount())
+
+	commandPlugin.Process(ctx, &bus.Message{
+		Topic: bus.ConnectionResetTopic,
+		Data:  commandPlugin.conn,
+	})
+	require.Equal(t, 1, fakeCommandService.UpdateClientCallCount())
 }
 
 func TestCommandPlugin_monitorSubscribeChannel(t *testing.T) {

internal/command/command_service.go

@@ -210,6 +210,10 @@ func (cs *CommandService) CreateConnection(
 		slog.InfoContext(ctx, "No Data Plane Instance found")
 	}
 
+	if cs.isConnected.Load() {
+		return nil, errors.New("command service already connected")
+	}
+
 	request := &mpi.CreateConnectionRequest{
 		MessageMeta: &mpi.MessageMeta{
 			MessageId:     id.GenerateMessageID(),
@@ -228,7 +232,6 @@ func (cs *CommandService) CreateConnection(
 	}
 
 	slog.DebugContext(ctx, "Sending create connection request", "request", request)
-
 	response, err := backoff.RetryWithData(
 		cs.connectCallback(ctx, request),
 		backoffHelpers.Context(ctx, commonSettings),
@@ -249,6 +252,10 @@ func (cs *CommandService) CreateConnection(
 	return response, nil
 }
 
+func (cs *CommandService) UpdateClient(client mpi.CommandServiceClient) {
+	cs.commandServiceClient = client
+}
+
 // Retry callback for sending a data plane response to the Management Plane.
 func (cs *CommandService) sendDataPlaneResponseCallback(
 	ctx context.Context,
@@ -427,6 +434,7 @@ func (cs *CommandService) handleSubscribeError(ctx context.Context, err error, e
 	codeError, ok := status.FromError(err)
 
 	if ok && codeError.Code() == codes.Unavailable {
+		cs.isConnected.Store(false)
 		slog.ErrorContext(ctx, fmt.Sprintf("Failed to %s, rpc unavailable. "+
 			"Trying create connection rpc", errorMsg), "error", err)
 		_, connectionErr := cs.CreateConnection(ctx, cs.resource)

internal/command/command_service_test.go

@@ -198,6 +198,18 @@ func TestCommandService_CreateConnection(t *testing.T) {
 	require.NoError(t, err)
 }
 
+func TestCommandService_UpdateClient(t *testing.T) {
+	commandServiceClient := &v1fakes.FakeCommandServiceClient{}
+
+	commandService := NewCommandService(
+		commandServiceClient,
+		types.AgentConfig(),
+		make(chan *mpi.ManagementPlaneRequest),
+	)
+	commandService.UpdateClient(commandServiceClient)
+	assert.NotNil(t, commandService.commandServiceClient)
+}
+
 func TestCommandService_UpdateDataPlaneHealth(t *testing.T) {
 	ctx := context.Background()
 	commandServiceClient := &v1fakes.FakeCommandServiceClient{}
@@ -501,3 +513,18 @@ func TestCommandService_isValidRequest(t *testing.T) {
 		})
 	}
 }
+
+func TestCommandService_handleSubscribeError(t *testing.T) {
+	ctx := context.Background()
+	commandServiceClient := &v1fakes.FakeCommandServiceClient{}
+
+	commandService := NewCommandService(
+		commandServiceClient,
+		types.AgentConfig(),
+		make(chan *mpi.ManagementPlaneRequest),
+	)
+	require.Error(t, commandService.handleSubscribeError(ctx, errors.New(""), ""))
+
+	err := commandService.handleSubscribeError(ctx, errors.New("blah blah blah"), "Testing handleSubscribeError")
+	require.Error(t, err)
+}

internal/config/config_test.go

@@ -901,7 +901,8 @@ func createConfig() *Config {
 				Type: Grpc,
 			},
 			Auth: &AuthConfig{
-				Token: "1234",
+				Token:     "1234",
+				TokenPath: "path/to/my_token",
 			},
 			TLS: &TLSConfig{
 				Cert:       "some.cert",

internal/config/testdata/nginx-agent.conf

@@ -54,6 +54,7 @@ command:
         type: grpc
     auth: 
         token: "1234"
+        tokenpath: "path/to/my_token"
     tls: 
         cert: "some.cert"
         key: "some.key"

internal/file/file_plugin.go

@@ -62,6 +62,8 @@ func (fp *FilePlugin) Info() *bus.Info {
 
 func (fp *FilePlugin) Process(ctx context.Context, msg *bus.Message) {
 	switch msg.Topic {
+	case bus.ConnectionResetTopic:
+		fp.handleConnectionReset(ctx, msg)
 	case bus.ConnectionCreatedTopic:
 		fp.fileManagerService.SetIsConnected(true)
 	case bus.NginxConfigUpdateTopic:
@@ -81,6 +83,7 @@ func (fp *FilePlugin) Process(ctx context.Context, msg *bus.Message) {
 
 func (fp *FilePlugin) Subscriptions() []string {
 	return []string{
+		bus.ConnectionResetTopic,
 		bus.ConnectionCreatedTopic,
 		bus.NginxConfigUpdateTopic,
 		bus.ConfigUploadRequestTopic,
@@ -91,6 +94,19 @@ func (fp *FilePlugin) Subscriptions() []string {
 	}
 }
 
+func (fp *FilePlugin) handleConnectionReset(ctx context.Context, msg *bus.Message) {
+	slog.DebugContext(ctx, "File plugin received connection reset message")
+	if newConnection, ok := msg.Data.(grpc.GrpcConnectionInterface); ok {
+		err := fp.conn.Close(ctx)
+		if err != nil {
+			slog.ErrorContext(ctx, "File plugin: unable to close connection", "error", err)
+		}
+		fp.conn = newConnection
+		fp.fileManagerService = NewFileManagerService(fp.conn.FileServiceClient(), fp.config)
+		slog.DebugContext(ctx, "File plugin: client reset successfully")
+	}
+}
+
 func (fp *FilePlugin) handleConfigApplyComplete(ctx context.Context, msg *bus.Message) {
 	response, ok := msg.Data.(*mpi.DataPlaneResponse)
 

internal/file/file_plugin_test.go

@@ -50,6 +50,7 @@ func TestFilePlugin_Subscriptions(t *testing.T) {
 	assert.Equal(
 		t,
 		[]string{
+			bus.ConnectionResetTopic,
 			bus.ConnectionCreatedTopic,
 			bus.NginxConfigUpdateTopic,
 			bus.ConfigUploadRequestTopic,

internal/grpc/grpc.go

@@ -41,7 +41,6 @@ type (
 		CommandServiceClient() mpi.CommandServiceClient
 		FileServiceClient() mpi.FileServiceClient
 		Close(ctx context.Context) error
-		Restart(ctx context.Context) (*GrpcConnection, error)
 	}
 
 	GrpcConnection struct {
@@ -68,6 +67,7 @@ var (
 	_ GrpcConnectionInterface = (*GrpcConnection)(nil)
 )
 
+// nolint: ireturn
 func NewGrpcConnection(ctx context.Context, agentConfig *config.Config) (*GrpcConnection, error) {
 	if agentConfig == nil || agentConfig.Command.Server.Type != config.Grpc {
 		return nil, errors.New("invalid command server settings")
@@ -131,22 +131,6 @@ func (gc *GrpcConnection) Close(ctx context.Context) error {
 	return nil
 }
 
-func (gc *GrpcConnection) Restart(ctx context.Context) (*GrpcConnection, error) {
-	slog.InfoContext(ctx, "Restarting grpc connection")
-	err := gc.Close(ctx)
-	if err != nil {
-		return nil, err
-	}
-
-	slog.InfoContext(ctx, "Creating grpc connection")
-	newConn, err := NewGrpcConnection(ctx, gc.config)
-	if err != nil {
-		return nil, err
-	}
-
-	return newConn, nil
-}
-
 func (w *wrappedStream) RecvMsg(message any) error {
 	err := w.ClientStream.RecvMsg(message)
 	if err == nil {