Add token file watcher (#979)

* support reading token from file via config

* remove empty file

* simplify token validation and add unit tests

* add unit tests for transport credentials funtions

* address PR feedback

* proto updates

* fix function name

* fix lint error: lll

* add missing PR feedback

* remove error log message

* fix unit test

* Fix apk test package naming (#961)

* modify alpine package name:
nginx-agent-3.0.0_1234 -> nginx-agent-3.0.0.1234

* protoc-gen update

* Update agent config defaults and format (#959)

* update config defaults and format

* Add config apply request queue (#949)

* add unit tests for transport credentials funtions

* fix test name

* fix error message

* fix lint error: lll

* modify error messages

* remove error logging and modify messages

* fall back to token field if error occurs when reading file

* fix bad merge

* restarting gRPC conn

* remove code from testing

* fix lint errors: whitespace, revive

* add new topic for handling Token file updates

* add CredentialWatcherService

* adding initial watcher for credential files

* trigger connection reset after credential update

* added ConnectionResetTopic and event processing

* Automatically add token-path to Credential watcher

* add function to check credential paths defined in agent config

* fix lint

* use tokenpath as config option, fixes problem with cli param parsing

* add CredentialWatcherService + tests

* fix lint errors

* updates to generated files

* Send create connection after disconnect from management plane (#967)

* correct yaml key in AuthConfig

* fix: flaky test (#968)

`TestConvertToStructs` was occasionally failing because it was expecting `range` over a `map` to be a consistent order, but per [spec]:

> The iteration order over maps is not specified and is not guaranteed to be the same from one iteration to the next.

Uses `ElementsMatch` so the test passes even when the order of elements is different.

[spec]: https://go.dev/ref/spec#RangeClause

* update tests

* update tests

* wait for create connection

* PR feedback

* fix race condition

* clean up

* clean up

* update grpc connection in command and file plugins

* move log message and fix file_plugin_test.go

* fix lint

* handle command and file service client updates after grpc reset

* update FakeCommandService

* add unit tests

* fix bad test

* formatting

* remove test

* increase timeout before checking connection after restart

* set isConnected to false when handling subscribe errors

* increase code coverage

* lint fix

* fieldalignment

* remove unused fake

* PR feedback

* update fake command service

* more PR feedback

* PR feedback: disable watcher during config apply

* don't pause credential watcher during config apply

* debug log messages

* Add mutex around updating client in the CommandService

* undo change to metric name

* update test

* lock watcher when replacing grpc connection

* add lock around subscibe client access

* fix lint error

* unlock watcher mutex in error case

* return with no blank line

* lint error ?

* handle rename cases

* remove unneccessary cases from switch

* rewrite switch as if

* handling kubernetes secret update case

* fix test

* save connection status when resetting file manager service

---------

Co-authored-by: aphralG <108004222+aphralG@users.noreply.github.com>
Co-authored-by: Donal Hurley <djhurley1990@gmail.com>
Co-authored-by: Ryan Davis <ry.davis@f5.com>
Co-authored-by: Aphral Griffin <a.griffin@f5.com>

Author: 4 people

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 service 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

@@ -98,11 +98,13 @@ func (cs *CommandService) UpdateDataPlaneStatus(
 	sendDataPlaneStatus := func() (*mpi.UpdateDataPlaneStatusResponse, error) {
 		slog.DebugContext(ctx, "Sending data plane status update request", "request", request,
 			"parent_correlation_id", correlationID)
+
+		cs.subscribeClientMutex.Lock()
 		if cs.commandServiceClient == nil {
 			return nil, errors.New("command service client is not initialized")
 		}
-
 		response, updateError := cs.commandServiceClient.UpdateDataPlaneStatus(ctx, request)
+		cs.subscribeClientMutex.Unlock()
 
 		validatedError := grpc.ValidateGrpcError(updateError)
 		if validatedError != nil {
@@ -210,6 +212,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 +234,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 +254,12 @@ func (cs *CommandService) CreateConnection(
 	return response, nil
 }
 
+func (cs *CommandService) UpdateClient(client mpi.CommandServiceClient) {
+	cs.subscribeClientMutex.Lock()
+	defer cs.subscribeClientMutex.Unlock()
+	cs.commandServiceClient = client
+}
+
 // Retry callback for sending a data plane response to the Management Plane.
 func (cs *CommandService) sendDataPlaneResponseCallback(
 	ctx context.Context,
@@ -355,11 +366,14 @@ func (cs *CommandService) dataPlaneHealthCallback(
 ) func() (*mpi.UpdateDataPlaneHealthResponse, error) {
 	return func() (*mpi.UpdateDataPlaneHealthResponse, error) {
 		slog.DebugContext(ctx, "Sending data plane health update request", "request", request)
+
+		cs.subscribeClientMutex.Lock()
 		if cs.commandServiceClient == nil {
 			return nil, errors.New("command service client is not initialized")
 		}
 
 		response, updateError := cs.commandServiceClient.UpdateDataPlaneHealth(ctx, request)
+		cs.subscribeClientMutex.Unlock()
 
 		validatedError := grpc.ValidateGrpcError(updateError)
 
@@ -427,6 +441,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)
@@ -530,7 +545,9 @@ func (cs *CommandService) connectCallback(
 	request *mpi.CreateConnectionRequest,
 ) func() (*mpi.CreateConnectionResponse, error) {
 	return func() (*mpi.CreateConnectionResponse, error) {
+		cs.subscribeClientMutex.Lock()
 		response, connectErr := cs.commandServiceClient.CreateConnection(ctx, request)
+		cs.subscribeClientMutex.Unlock()
 
 		validatedError := grpc.ValidateGrpcError(connectErr)
 		if validatedError != nil {

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("an error occurred when attempting to subscribe"),
+			"Testing handleSubscribeError"))
+}

internal/command/commandfakes/fake_command_service.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/config_test.go

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

internal/config/testdata/nginx-agent.conf

@@ -54,6 +54,7 @@ type (
 		UpdateCurrentFilesOnDisk(updateFiles map[string]*mpi.File)
 		DetermineFileActions(currentFiles, modifiedFiles map[string]*mpi.File) (map[string]*mpi.File,
 			map[string][]byte, error)
+		IsConnected() bool
 		SetIsConnected(isConnected bool)
 	}
 )
@@ -272,6 +273,10 @@ func (fms *FileManagerService) UpdateFile(
 	return err
 }
 
+func (fms *FileManagerService) IsConnected() bool {
+	return fms.isConnected.Load()
+}
+
 func (fms *FileManagerService) SetIsConnected(isConnected bool) {
 	fms.isConnected.Store(isConnected)
 }

internal/file/file_manager_service.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,24 @@ 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 {
+		var reconnect bool
+		err := fp.conn.Close(ctx)
+		if err != nil {
+			slog.ErrorContext(ctx, "File plugin: unable to close connection", "error", err)
+		}
+		fp.conn = newConnection
+
+		reconnect = fp.fileManagerService.IsConnected()
+		fp.fileManagerService = NewFileManagerService(fp.conn.FileServiceClient(), fp.config)
+		fp.fileManagerService.SetIsConnected(reconnect)
+
+		slog.DebugContext(ctx, "File plugin: client reset successfully")
+	}
+}
+
 func (fp *FilePlugin) handleConfigApplyComplete(ctx context.Context, msg *bus.Message) {
 	response, ok := msg.Data.(*mpi.DataPlaneResponse)