wip instance watcher refactor

Author: aphralG

internal/plugin/plugin_manager.go

@@ -40,7 +40,7 @@ func LoadPlugins(ctx context.Context, agentConfig *config.Config) []bus.Plugin {
 }
 
 func addResourcePlugin(plugins []bus.Plugin, agentConfig *config.Config) []bus.Plugin {
-	resourcePlugin := resource.NewResource(agentConfig)
+	resourcePlugin := resource.NewNginx(agentConfig)
 	plugins = append(plugins, resourcePlugin)
 
 	return plugins

internal/plugin/plugin_manager_test.go

@@ -34,7 +34,7 @@ func TestLoadPlugins(t *testing.T) {
 			name:  "Test 1: Load plugins",
 			input: &config.Config{},
 			expected: []bus.Plugin{
-				&resource.Resource{},
+				&resource.Nginx{},
 				&watcher.Watcher{},
 			},
 		},
@@ -58,7 +58,7 @@ func TestLoadPlugins(t *testing.T) {
 				Features: config.DefaultFeatures(),
 			},
 			expected: []bus.Plugin{
-				&resource.Resource{},
+				&resource.Nginx{},
 				&command.CommandPlugin{},
 				&file.FilePlugin{},
 				&command.CommandPlugin{},
@@ -77,7 +77,7 @@ func TestLoadPlugins(t *testing.T) {
 				Features: config.DefaultFeatures(),
 			},
 			expected: []bus.Plugin{
-				&resource.Resource{},
+				&resource.Nginx{},
 				&collector.Collector{},
 				&watcher.Watcher{},
 			},
@@ -103,7 +103,7 @@ func TestLoadPlugins(t *testing.T) {
 				},
 			},
 			expected: []bus.Plugin{
-				&resource.Resource{},
+				&resource.Nginx{},
 				&command.CommandPlugin{},
 				&file.FilePlugin{},
 				&watcher.Watcher{},
@@ -134,7 +134,7 @@ func TestLoadPlugins(t *testing.T) {
 				},
 			},
 			expected: []bus.Plugin{
-				&resource.Resource{},
+				&resource.Nginx{},
 				&command.CommandPlugin{},
 				&file.FilePlugin{},
 				&collector.Collector{},

internal/resource/resource_plugin.go internal/resource/nginx_plugin.gointernal/resource/resource_plugin.go renamed to internal/resource/nginx_plugin.go

@@ -21,13 +21,13 @@ import (
 	"github.com/nginx/agent/v3/internal/bus"
 )
 
-// The resource plugin listens for a writeConfigSuccessfulTopic from the file plugin after the config apply
-// files have been written. The resource plugin then, validates the config, reloads the instance and monitors the logs.
-// This is done in the resource plugin to make the file plugin usable for every type of instance.
+// The Nginx plugin listens for a writeConfigSuccessfulTopic from the file plugin after the config apply
+// files have been written. The Nginx plugin then, validates the config, reloads the instance and monitors the logs.
+// This is done in the Nginx plugin to make the file plugin usable for every type of instance.
 
-type Resource struct {
+type Nginx struct {
 	messagePipe      bus.MessagePipeInterface
-	resourceService  resourceServiceInterface
+	nginxService     nginxServiceInterface
 	agentConfig      *config.Config
 	agentConfigMutex *sync.Mutex
 }
@@ -44,88 +44,88 @@ type plusAPIErr struct {
 	Href      string      `json:"href"`
 }
 
-var _ bus.Plugin = (*Resource)(nil)
+var _ bus.Plugin = (*Nginx)(nil)
 
-func NewResource(agentConfig *config.Config) *Resource {
-	return &Resource{
+func NewNginx(agentConfig *config.Config) *Nginx {
+	return &Nginx{
 		agentConfig:      agentConfig,
 		agentConfigMutex: &sync.Mutex{},
 	}
 }
 
-func (r *Resource) Init(ctx context.Context, messagePipe bus.MessagePipeInterface) error {
-	slog.DebugContext(ctx, "Starting resource plugin")
+func (n *Nginx) Init(ctx context.Context, messagePipe bus.MessagePipeInterface) error {
+	slog.DebugContext(ctx, "Starting nginx plugin")
 
-	r.messagePipe = messagePipe
-	r.resourceService = NewResourceService(ctx, r.agentConfig)
+	n.messagePipe = messagePipe
+	n.nginxService = NewNginxService(ctx, n.agentConfig)
 
 	return nil
 }
 
-func (*Resource) Close(ctx context.Context) error {
-	slog.InfoContext(ctx, "Closing resource plugin")
+func (*Nginx) Close(ctx context.Context) error {
+	slog.InfoContext(ctx, "Closing nginx plugin")
 	return nil
 }
 
-func (*Resource) Info() *bus.Info {
+func (*Nginx) Info() *bus.Info {
 	return &bus.Info{
-		Name: "resource",
+		Name: "nginx",
 	}
 }
 
 // cyclomatic complexity 11 max is 10
 
-func (r *Resource) Process(ctx context.Context, msg *bus.Message) {
+func (n *Nginx) Process(ctx context.Context, msg *bus.Message) {
 	switch msg.Topic {
-	case bus.UpdatedInstancesTopic:
-		slog.DebugContext(ctx, "Resource plugin received update instances message")
-		instanceList, ok := msg.Data.([]*mpi.Instance)
+	case bus.ResourceUpdateTopic:
+		resourceUpdate, ok := msg.Data.(*mpi.Resource)
+
 		if !ok {
-			slog.ErrorContext(ctx, "Unable to cast message payload to []*mpi.Instance", "payload", msg.Data)
+			slog.ErrorContext(ctx, "Unable to cast message payload to *mpi.Resource", "payload",
+				msg.Data)
 
 			return
 		}
-		resource := r.resourceService.UpdateInstances(ctx, instanceList)
-
-		r.messagePipe.Process(ctx, &bus.Message{Topic: bus.ResourceUpdateTopic, Data: resource})
+		n.nginxService.UpdateResource(ctx, resourceUpdate)
+		slog.DebugContext(ctx, "Nginx plugin received update resource message")
 
 		return
 	case bus.WriteConfigSuccessfulTopic:
-		r.handleWriteConfigSuccessful(ctx, msg)
+		n.handleWriteConfigSuccessful(ctx, msg)
 	case bus.RollbackWriteTopic:
-		r.handleRollbackWrite(ctx, msg)
+		n.handleRollbackWrite(ctx, msg)
 	case bus.APIActionRequestTopic:
-		r.handleAPIActionRequest(ctx, msg)
+		n.handleAPIActionRequest(ctx, msg)
 	case bus.AgentConfigUpdateTopic:
-		r.handleAgentConfigUpdate(ctx, msg)
+		n.handleAgentConfigUpdate(ctx, msg)
 	default:
 		slog.DebugContext(ctx, "Unknown topic", "topic", msg.Topic)
 	}
 }
 
-func (*Resource) Subscriptions() []string {
+func (*Nginx) Subscriptions() []string {
 	return []string{
-		bus.UpdatedInstancesTopic,
+		bus.ResourceUpdateTopic,
 		bus.WriteConfigSuccessfulTopic,
 		bus.RollbackWriteTopic,
 		bus.APIActionRequestTopic,
 		bus.AgentConfigUpdateTopic,
 	}
 }
 
-func (r *Resource) Reconfigure(ctx context.Context, agentConfig *config.Config) error {
-	slog.DebugContext(ctx, "Resource plugin is reconfiguring to update agent configuration")
+func (n *Nginx) Reconfigure(ctx context.Context, agentConfig *config.Config) error {
+	slog.DebugContext(ctx, "Nginx plugin is reconfiguring to update agent configuration")
 
-	r.agentConfigMutex.Lock()
-	defer r.agentConfigMutex.Unlock()
+	n.agentConfigMutex.Lock()
+	defer n.agentConfigMutex.Unlock()
 
-	r.agentConfig = agentConfig
+	n.agentConfig = agentConfig
 
 	return nil
 }
 
-func (r *Resource) handleAPIActionRequest(ctx context.Context, msg *bus.Message) {
-	slog.DebugContext(ctx, "Resource plugin received api action request message")
+func (n *Nginx) handleAPIActionRequest(ctx context.Context, msg *bus.Message) {
+	slog.DebugContext(ctx, "Nginx plugin received api action request message")
 	managementPlaneRequest, ok := msg.Data.(*mpi.ManagementPlaneRequest)
 
 	if !ok {
@@ -145,25 +145,25 @@ func (r *Resource) handleAPIActionRequest(ctx context.Context, msg *bus.Message)
 
 	switch request.ActionRequest.GetAction().(type) {
 	case *mpi.APIActionRequest_NginxPlusAction:
-		r.handleNginxPlusActionRequest(ctx, request.ActionRequest.GetNginxPlusAction(), instanceID)
+		n.handleNginxPlusActionRequest(ctx, request.ActionRequest.GetNginxPlusAction(), instanceID)
 	default:
 		slog.DebugContext(ctx, "API action request not implemented yet")
 	}
 }
 
-func (r *Resource) handleNginxPlusActionRequest(ctx context.Context, action *mpi.NGINXPlusAction, instanceID string) {
+func (n *Nginx) handleNginxPlusActionRequest(ctx context.Context, action *mpi.NGINXPlusAction, instanceID string) {
 	correlationID := logger.CorrelationID(ctx)
-	instance := r.resourceService.Instance(instanceID)
+	instance := n.nginxService.Instance(instanceID)
 	apiAction := APIAction{
-		ResourceService: r.resourceService,
+		NginxService: n.nginxService,
 	}
 	if instance == nil {
 		slog.ErrorContext(ctx, "Unable to find instance with ID", "id", instanceID)
 		resp := response.CreateDataPlaneResponse(correlationID, mpi.CommandResponse_COMMAND_STATUS_FAILURE,
 			"", instanceID, fmt.Sprintf("failed to preform API "+
 				"action, could not find instance with ID: %s", instanceID))
 
-		r.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
+		n.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
 
 		return
 	}
@@ -173,7 +173,7 @@ func (r *Resource) handleNginxPlusActionRequest(ctx context.Context, action *mpi
 		resp := response.CreateDataPlaneResponse(correlationID, mpi.CommandResponse_COMMAND_STATUS_FAILURE,
 			"", instanceID, "failed to preform API action, instance is not NGINX Plus")
 
-		r.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
+		n.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
 
 		return
 	}
@@ -182,46 +182,46 @@ func (r *Resource) handleNginxPlusActionRequest(ctx context.Context, action *mpi
 	case *mpi.NGINXPlusAction_UpdateHttpUpstreamServers:
 		slog.DebugContext(ctx, "Updating http upstream servers", "request", action.GetUpdateHttpUpstreamServers())
 		resp := apiAction.HandleUpdateHTTPUpstreamsRequest(ctx, action, instance)
-		r.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
+		n.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
 	case *mpi.NGINXPlusAction_GetHttpUpstreamServers:
 		slog.DebugContext(ctx, "Getting http upstream servers", "request", action.GetGetHttpUpstreamServers())
 		resp := apiAction.HandleGetHTTPUpstreamsServersRequest(ctx, action, instance)
-		r.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
+		n.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
 	case *mpi.NGINXPlusAction_UpdateStreamServers:
 		slog.DebugContext(ctx, "Updating stream servers", "request", action.GetUpdateStreamServers())
 		resp := apiAction.HandleUpdateStreamServersRequest(ctx, action, instance)
-		r.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
+		n.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
 	case *mpi.NGINXPlusAction_GetStreamUpstreams:
 		slog.DebugContext(ctx, "Getting stream upstreams", "request", action.GetGetStreamUpstreams())
 		resp := apiAction.HandleGetStreamUpstreamsRequest(ctx, instance)
-		r.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
+		n.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
 	case *mpi.NGINXPlusAction_GetUpstreams:
 		slog.DebugContext(ctx, "Getting upstreams", "request", action.GetGetUpstreams())
 		resp := apiAction.HandleGetUpstreamsRequest(ctx, instance)
-		r.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
+		n.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
 	default:
 		slog.DebugContext(ctx, "NGINX Plus action not implemented yet")
 	}
 }
 
-func (r *Resource) handleWriteConfigSuccessful(ctx context.Context, msg *bus.Message) {
-	slog.DebugContext(ctx, "Resource plugin received write config successful message")
+func (n *Nginx) handleWriteConfigSuccessful(ctx context.Context, msg *bus.Message) {
+	slog.DebugContext(ctx, "Nginx plugin received write config successful message")
 	data, ok := msg.Data.(*model.ConfigApplyMessage)
 	if !ok {
 		slog.ErrorContext(ctx, "Unable to cast message payload to *model.ConfigApplyMessage", "payload", msg.Data)
 
 		return
 	}
-	configContext, err := r.resourceService.ApplyConfig(ctx, data.InstanceID)
+	configContext, err := n.nginxService.ApplyConfig(ctx, data.InstanceID)
 	if err != nil {
 		data.Error = err
 		slog.ErrorContext(ctx, "errors found during config apply, "+
 			"sending error status, rolling back config", "err", err)
 		dpResponse := response.CreateDataPlaneResponse(data.CorrelationID, mpi.CommandResponse_COMMAND_STATUS_ERROR,
 			"Config apply failed, rolling back config", data.InstanceID, err.Error())
-		r.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: dpResponse})
+		n.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: dpResponse})
 
-		r.messagePipe.Process(ctx, &bus.Message{Topic: bus.ConfigApplyFailedTopic, Data: data})
+		n.messagePipe.Process(ctx, &bus.Message{Topic: bus.ConfigApplyFailedTopic, Data: data})
 
 		return
 	}
@@ -234,18 +234,18 @@ func (r *Resource) handleWriteConfigSuccessful(ctx context.Context, msg *bus.Mes
 		DataPlaneResponse: dpResponse,
 	}
 
-	r.messagePipe.Process(ctx, &bus.Message{Topic: bus.ReloadSuccessfulTopic, Data: successMessage})
+	n.messagePipe.Process(ctx, &bus.Message{Topic: bus.ReloadSuccessfulTopic, Data: successMessage})
 }
 
-func (r *Resource) handleRollbackWrite(ctx context.Context, msg *bus.Message) {
-	slog.DebugContext(ctx, "Resource plugin received rollback write message")
+func (n *Nginx) handleRollbackWrite(ctx context.Context, msg *bus.Message) {
+	slog.DebugContext(ctx, "Nginx plugin received rollback write message")
 	data, ok := msg.Data.(*model.ConfigApplyMessage)
 	if !ok {
 		slog.ErrorContext(ctx, "Unable to cast message payload to *model.ConfigApplyMessage", "payload", msg.Data)
 
 		return
 	}
-	_, err := r.resourceService.ApplyConfig(ctx, data.InstanceID)
+	_, err := n.nginxService.ApplyConfig(ctx, data.InstanceID)
 	if err != nil {
 		slog.ErrorContext(ctx, "errors found during rollback, sending failure status", "err", err)
 
@@ -256,8 +256,8 @@ func (r *Resource) handleRollbackWrite(ctx context.Context, msg *bus.Message) {
 			mpi.CommandResponse_COMMAND_STATUS_FAILURE, "Config apply failed, rollback failed",
 			data.InstanceID, data.Error.Error())
 
-		r.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: rollbackResponse})
-		r.messagePipe.Process(ctx, &bus.Message{Topic: bus.ConfigApplyCompleteTopic, Data: applyResponse})
+		n.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: rollbackResponse})
+		n.messagePipe.Process(ctx, &bus.Message{Topic: bus.ConfigApplyCompleteTopic, Data: applyResponse})
 
 		return
 	}
@@ -266,20 +266,20 @@ func (r *Resource) handleRollbackWrite(ctx context.Context, msg *bus.Message) {
 		mpi.CommandResponse_COMMAND_STATUS_FAILURE,
 		"Config apply failed, rollback successful", data.InstanceID, data.Error.Error())
 
-	r.messagePipe.Process(ctx, &bus.Message{Topic: bus.ConfigApplyCompleteTopic, Data: applyResponse})
+	n.messagePipe.Process(ctx, &bus.Message{Topic: bus.ConfigApplyCompleteTopic, Data: applyResponse})
 }
 
-func (r *Resource) handleAgentConfigUpdate(ctx context.Context, msg *bus.Message) {
-	slog.DebugContext(ctx, "Resource plugin received agent config update message")
+func (n *Nginx) handleAgentConfigUpdate(ctx context.Context, msg *bus.Message) {
+	slog.DebugContext(ctx, "Nginx plugin received agent config update message")
 
-	r.agentConfigMutex.Lock()
-	defer r.agentConfigMutex.Unlock()
+	n.agentConfigMutex.Lock()
+	defer n.agentConfigMutex.Unlock()
 
 	agentConfig, ok := msg.Data.(*config.Config)
 	if !ok {
 		slog.ErrorContext(ctx, "Unable to cast message payload to *config.Config", "payload", msg.Data)
 		return
 	}
 
-	r.agentConfig = agentConfig
+	n.agentConfig = agentConfig
 }