Update how watchers are enabled after config apply

Author: aphralG

internal/bus/topics.go

@@ -19,7 +19,8 @@ const (
 	ConnectionResetTopic         = "connection-reset"
 	ConfigApplyRequestTopic      = "config-apply-request"
 	WriteConfigSuccessfulTopic   = "write-config-successful"
-	ConfigApplySuccessfulTopic   = "config-apply-successful"
+	ReloadSuccessfulTopic        = "reload-successful"
+	EnableWatchersTopic          = "enable-watchers"
 	ConfigApplyFailedTopic       = "config-apply-failed"
 	ConfigApplyCompleteTopic     = "config-apply-complete"
 	RollbackWriteTopic           = "rollback-write"

internal/file/file_plugin.go

@@ -106,8 +106,8 @@ func (fp *FilePlugin) Process(ctx context.Context, msg *bus.Message) {
 			fp.handleConfigApplyRequest(ctxWithMetadata, msg)
 		case bus.ConfigApplyCompleteTopic:
 			fp.handleConfigApplyComplete(ctxWithMetadata, msg)
-		case bus.ConfigApplySuccessfulTopic:
-			fp.handleConfigApplySuccess(ctxWithMetadata, msg)
+		case bus.ReloadSuccessfulTopic:
+			fp.handleReloadSuccess(ctxWithMetadata, msg)
 		case bus.ConfigApplyFailedTopic:
 			fp.handleConfigApplyFailedRequest(ctxWithMetadata, msg)
 		default:
@@ -133,11 +133,28 @@ func (fp *FilePlugin) Subscriptions() []string {
 		bus.ConfigUploadRequestTopic,
 		bus.ConfigApplyRequestTopic,
 		bus.ConfigApplyFailedTopic,
-		bus.ConfigApplySuccessfulTopic,
+		bus.ReloadSuccessfulTopic,
 		bus.ConfigApplyCompleteTopic,
 	}
 }
 
+func (fp *FilePlugin) CleanUpConfigApply(ctx context.Context,
+	configContext *model.NginxConfigContext,
+	instanceID string,
+) {
+	enableWatcher := &model.EnableWatchers{
+		ConfigContext: configContext,
+		InstanceID:    instanceID,
+	}
+
+	fp.fileManagerService.ClearCache()
+
+	fp.messagePipe.Process(ctx, &bus.Message{
+		Data:  enableWatcher,
+		Topic: bus.EnableWatchersTopic,
+	})
+}
+
 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 {
@@ -165,20 +182,21 @@ func (fp *FilePlugin) handleConfigApplyComplete(ctx context.Context, msg *bus.Me
 		return
 	}
 
-	fp.fileManagerService.ClearCache()
 	fp.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: response})
+	fp.CleanUpConfigApply(ctx, &model.NginxConfigContext{}, response.GetInstanceId())
 }
 
-func (fp *FilePlugin) handleConfigApplySuccess(ctx context.Context, msg *bus.Message) {
-	slog.DebugContext(ctx, "File plugin received config success message")
-	successMessage, ok := msg.Data.(*model.ConfigApplySuccess)
+func (fp *FilePlugin) handleReloadSuccess(ctx context.Context, msg *bus.Message) {
+	slog.InfoContext(ctx, "File plugin received reload success message", "data", msg.Data)
+
+	successMessage, ok := msg.Data.(*model.ReloadSuccess)
 
 	if !ok {
-		slog.ErrorContext(ctx, "Unable to cast message payload to *model.ConfigApplySuccess", "payload", msg.Data)
+		slog.ErrorContext(ctx, "Unable to cast message payload to *model.ReloadSuccess", "payload", msg.Data)
 		return
 	}
 
-	fp.fileManagerService.ClearCache()
+	fp.CleanUpConfigApply(ctx, successMessage.ConfigContext, successMessage.DataPlaneResponse.GetInstanceId())
 
 	if successMessage.ConfigContext.Files != nil {
 		slog.DebugContext(ctx, "Changes made during config apply, update files on disk")
@@ -191,6 +209,7 @@ func (fp *FilePlugin) handleConfigApplySuccess(ctx context.Context, msg *bus.Mes
 			slog.ErrorContext(ctx, "Unable to update current files on disk", "error", updateError)
 		}
 	}
+
 	fp.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: successMessage.DataPlaneResponse})
 }
 
@@ -264,13 +283,8 @@ func (fp *FilePlugin) handleConfigApplyRequest(ctx context.Context, msg *bus.Mes
 			"",
 		)
 
-		successMessage := &model.ConfigApplySuccess{
-			ConfigContext:     &model.NginxConfigContext{},
-			DataPlaneResponse: dpResponse,
-		}
-
 		fp.fileManagerService.ClearCache()
-		fp.messagePipe.Process(ctx, &bus.Message{Topic: bus.ConfigApplySuccessfulTopic, Data: successMessage})
+		fp.messagePipe.Process(ctx, &bus.Message{Topic: bus.ConfigApplyCompleteTopic, Data: dpResponse})
 
 		return
 	case model.Error:

internal/file/file_plugin_test.go

@@ -59,7 +59,7 @@ func TestFilePlugin_Subscriptions(t *testing.T) {
 			bus.ConfigUploadRequestTopic,
 			bus.ConfigApplyRequestTopic,
 			bus.ConfigApplyFailedTopic,
-			bus.ConfigApplySuccessfulTopic,
+			bus.ReloadSuccessfulTopic,
 			bus.ConfigApplyCompleteTopic,
 		},
 		filePlugin.Subscriptions(),
@@ -210,13 +210,13 @@ func TestFilePlugin_Process_ConfigApplyRequestTopic(t *testing.T) {
 			case test.configApplyStatus == model.NoChange:
 				assert.Len(t, messages, 1)
 
-				response, ok := messages[0].Data.(*model.ConfigApplySuccess)
+				response, ok := messages[0].Data.(*mpi.DataPlaneResponse)
 				assert.True(t, ok)
-				assert.Equal(t, bus.ConfigApplySuccessfulTopic, messages[0].Topic)
+				assert.Equal(t, bus.ConfigApplyCompleteTopic, messages[0].Topic)
 				assert.Equal(
 					t,
 					mpi.CommandResponse_COMMAND_STATUS_OK,
-					response.DataPlaneResponse.GetCommandResponse().GetStatus(),
+					response.GetCommandResponse().GetStatus(),
 				)
 			case test.message == nil:
 				assert.Empty(t, messages)
@@ -432,7 +432,7 @@ func TestFilePlugin_Process_ConfigApplyFailedTopic(t *testing.T) {
 	}
 }
 
-func TestFilePlugin_Process_ConfigApplyRollbackCompleteTopic(t *testing.T) {
+func TestFilePlugin_Process_ConfigApplyReloadSuccessTopic(t *testing.T) {
 	ctx := context.Background()
 	instance := protos.NginxOssInstance([]string{})
 	mockFileManager := &filefakes.FakeFileManagerServiceInterface{}
@@ -460,14 +460,22 @@ func TestFilePlugin_Process_ConfigApplyRollbackCompleteTopic(t *testing.T) {
 		InstanceId: instance.GetInstanceMeta().GetInstanceId(),
 	}
 
-	filePlugin.Process(ctx, &bus.Message{Topic: bus.ConfigApplySuccessfulTopic, Data: &model.ConfigApplySuccess{
+	filePlugin.Process(ctx, &bus.Message{Topic: bus.ReloadSuccessfulTopic, Data: &model.ReloadSuccess{
 		ConfigContext:     &model.NginxConfigContext{},
 		DataPlaneResponse: expectedResponse,
 	}})
 
 	messages := messagePipe.Messages()
-	response, ok := messages[0].Data.(*mpi.DataPlaneResponse)
+
+	watchers, ok := messages[0].Data.(*model.EnableWatchers)
+	assert.True(t, ok)
+	assert.Equal(t, bus.EnableWatchersTopic, messages[0].Topic)
+	assert.Equal(t, &model.NginxConfigContext{}, watchers.ConfigContext)
+	assert.Equal(t, instance.GetInstanceMeta().GetInstanceId(), watchers.InstanceID)
+
+	response, ok := messages[1].Data.(*mpi.DataPlaneResponse)
 	assert.True(t, ok)
+	assert.Equal(t, bus.DataPlaneResponseTopic, messages[1].Topic)
 
 	assert.Equal(t, expectedResponse.GetCommandResponse().GetStatus(), response.GetCommandResponse().GetStatus())
 	assert.Equal(t, expectedResponse.GetCommandResponse().GetMessage(), response.GetCommandResponse().GetMessage())

internal/model/config.go

@@ -77,11 +77,16 @@ const (
 	OK
 )
 
-type ConfigApplySuccess struct {
+type ReloadSuccess struct {
 	ConfigContext     *NginxConfigContext
 	DataPlaneResponse *v1.DataPlaneResponse
 }
 
+type EnableWatchers struct {
+	ConfigContext *NginxConfigContext
+	InstanceID    string
+}
+
 //nolint:revive,cyclop // cyclomatic complexity is 16
 func (ncc *NginxConfigContext) Equal(otherNginxConfigContext *NginxConfigContext) bool {
 	if ncc.StubStatus != nil && otherNginxConfigContext.StubStatus != nil {

internal/resource/resource_plugin.go

@@ -242,12 +242,12 @@ func (r *Resource) handleWriteConfigSuccessful(ctx context.Context, msg *bus.Mes
 	dpResponse := response.CreateDataPlaneResponse(data.CorrelationID, mpi.CommandResponse_COMMAND_STATUS_OK,
 		"Config apply successful", data.InstanceID, "")
 
-	successMessage := &model.ConfigApplySuccess{
+	successMessage := &model.ReloadSuccess{
 		ConfigContext:     configContext,
 		DataPlaneResponse: dpResponse,
 	}
 
-	r.messagePipe.Process(ctx, &bus.Message{Topic: bus.ConfigApplySuccessfulTopic, Data: successMessage})
+	r.messagePipe.Process(ctx, &bus.Message{Topic: bus.ReloadSuccessfulTopic, Data: successMessage})
 }
 
 func (r *Resource) handleRollbackWrite(ctx context.Context, msg *bus.Message) {

internal/resource/resource_plugin_test.go

@@ -142,7 +142,7 @@ func TestResource_Process_Apply(t *testing.T) {
 				},
 			},
 			applyErr: nil,
-			topic:    []string{bus.ConfigApplySuccessfulTopic},
+			topic:    []string{bus.ReloadSuccessfulTopic},
 		},
 		{
 			name: "Test 2: Write Config Successful Topic - Fail Status",

internal/watcher/watcher_plugin.go

@@ -140,12 +140,10 @@ func (w *Watcher) Process(ctx context.Context, msg *bus.Message) {
 	switch msg.Topic {
 	case bus.ConfigApplyRequestTopic:
 		w.handleConfigApplyRequest(ctx, msg)
-	case bus.ConfigApplySuccessfulTopic:
-		w.handleConfigApplySuccess(ctx, msg)
-	case bus.ConfigApplyCompleteTopic:
-		w.handleConfigApplyComplete(ctx, msg)
 	case bus.DataPlaneHealthRequestTopic:
 		w.handleHealthRequest(ctx)
+	case bus.EnableWatchersTopic:
+		w.handleEnableWatchers(ctx, msg)
 	default:
 		slog.DebugContext(ctx, "Watcher plugin unknown topic", "topic", msg.Topic)
 	}
@@ -154,12 +152,43 @@ func (w *Watcher) Process(ctx context.Context, msg *bus.Message) {
 func (*Watcher) Subscriptions() []string {
 	return []string{
 		bus.ConfigApplyRequestTopic,
-		bus.ConfigApplySuccessfulTopic,
-		bus.ConfigApplyCompleteTopic,
 		bus.DataPlaneHealthRequestTopic,
+		bus.EnableWatchersTopic,
 	}
 }
 
+func (w *Watcher) handleEnableWatchers(ctx context.Context, msg *bus.Message) {
+	slog.DebugContext(ctx, "Watcher plugin received enable watchers message")
+	enableWatchersMessage, ok := msg.Data.(*model.EnableWatchers)
+	if !ok {
+		slog.ErrorContext(ctx, "Unable to cast message payload to *model.EnableWatchers", "payload",
+			msg.Data, "topic", msg.Topic)
+
+		return
+	}
+
+	instanceID := enableWatchersMessage.InstanceID
+	configContext := enableWatchersMessage.ConfigContext
+
+	// if config apply ended in a reload there is no need to reparse the config so an empty config context is sent
+	// from the file plugin
+	if configContext.InstanceID != "" {
+		w.instanceWatcherService.HandleNginxConfigContextUpdate(ctx, instanceID, configContext)
+	}
+
+	w.watcherMutex.Lock()
+	w.instancesWithConfigApplyInProgress = slices.DeleteFunc(
+		w.instancesWithConfigApplyInProgress,
+		func(element string) bool {
+			return element == instanceID
+		},
+	)
+
+	w.fileWatcherService.SetEnabled(true)
+	w.instanceWatcherService.SetEnabled(true)
+	w.watcherMutex.Unlock()
+}
+
 func (w *Watcher) handleConfigApplyRequest(ctx context.Context, msg *bus.Message) {
 	slog.DebugContext(ctx, "Watcher plugin received config apply request message")
 	managementPlaneRequest, ok := msg.Data.(*mpi.ManagementPlaneRequest)
@@ -188,69 +217,13 @@ func (w *Watcher) handleConfigApplyRequest(ctx context.Context, msg *bus.Message
 	w.instanceWatcherService.SetEnabled(false)
 }
 
-func (w *Watcher) handleConfigApplySuccess(ctx context.Context, msg *bus.Message) {
-	slog.DebugContext(ctx, "Watcher plugin received config apply success message")
-	successMessage, ok := msg.Data.(*model.ConfigApplySuccess)
-	if !ok {
-		slog.ErrorContext(ctx, "Unable to cast message payload to *model.ConfigApplySuccess", "payload",
-			msg.Data, "topic", msg.Topic)
-
-		return
-	}
-
-	instanceID := successMessage.DataPlaneResponse.GetInstanceId()
-
-	// If the config apply had no changes to any files, it is results in a ConfigApplySuccessfulTopic with an empty
-	// configContext being sent, there is no need to reparse the config as no change has occurred.
-	if successMessage.ConfigContext.InstanceID != "" {
-		w.instanceWatcherService.HandleNginxConfigContextUpdate(ctx, instanceID, successMessage.ConfigContext)
-	}
-
-	w.watcherMutex.Lock()
-	w.instancesWithConfigApplyInProgress = slices.DeleteFunc(
-		w.instancesWithConfigApplyInProgress,
-		func(element string) bool {
-			return element == instanceID
-		},
-	)
-
-	w.fileWatcherService.SetEnabled(true)
-	w.instanceWatcherService.SetEnabled(true)
-	w.watcherMutex.Unlock()
-}
-
 func (w *Watcher) handleHealthRequest(ctx context.Context) {
 	slog.DebugContext(ctx, "Watcher plugin received health request message")
 	w.messagePipe.Process(ctx, &bus.Message{
 		Topic: bus.DataPlaneHealthResponseTopic, Data: w.healthWatcherService.InstancesHealth(),
 	})
 }
 
-func (w *Watcher) handleConfigApplyComplete(ctx context.Context, msg *bus.Message) {
-	slog.DebugContext(ctx, "Watcher plugin received config apply complete message")
-	response, ok := msg.Data.(*mpi.DataPlaneResponse)
-	if !ok {
-		slog.ErrorContext(ctx, "Unable to cast message payload to *mpi.DataPlaneResponse", "payload",
-			msg.Data, "topic", msg.Topic)
-
-		return
-	}
-
-	instanceID := response.GetInstanceId()
-
-	w.watcherMutex.Lock()
-	defer w.watcherMutex.Unlock()
-	w.instancesWithConfigApplyInProgress = slices.DeleteFunc(
-		w.instancesWithConfigApplyInProgress,
-		func(element string) bool {
-			return element == instanceID
-		},
-	)
-
-	w.instanceWatcherService.SetEnabled(true)
-	w.fileWatcherService.SetEnabled(true)
-}
-
 func (w *Watcher) monitorWatchers(ctx context.Context) {
 	for {
 		select {