WIP watcher changes

Author: aphralG

internal/bus/topics.go

@@ -6,9 +6,7 @@
 package bus
 
 const (
-	AddInstancesTopic                = "add-instances"
 	UpdatedInstancesTopic            = "updated-instances"
-	DeletedInstancesTopic            = "deleted-instances"
 	ResourceUpdateTopic              = "resource-update"
 	NginxConfigUpdateTopic           = "nginx-config-update"
 	InstanceHealthTopic              = "instance-health"

internal/resource/resource_plugin.go

@@ -77,20 +77,6 @@ func (*Resource) Info() *bus.Info {
 
 func (r *Resource) Process(ctx context.Context, msg *bus.Message) {
 	switch msg.Topic {
-	case bus.AddInstancesTopic:
-		slog.DebugContext(ctx, "Resource plugin received add instances message")
-		instanceList, ok := msg.Data.([]*mpi.Instance)
-		if !ok {
-			slog.ErrorContext(ctx, "Unable to cast message payload to []*mpi.Instance", "payload", msg.Data)
-
-			return
-		}
-
-		resource := r.resourceService.AddInstances(instanceList)
-
-		r.messagePipe.Process(ctx, &bus.Message{Topic: bus.ResourceUpdateTopic, Data: resource})
-
-		return
 	case bus.UpdatedInstancesTopic:
 		slog.DebugContext(ctx, "Resource plugin received update instances message")
 		instanceList, ok := msg.Data.([]*mpi.Instance)
@@ -103,20 +89,6 @@ func (r *Resource) Process(ctx context.Context, msg *bus.Message) {
 
 		r.messagePipe.Process(ctx, &bus.Message{Topic: bus.ResourceUpdateTopic, Data: resource})
 
-		return
-
-	case bus.DeletedInstancesTopic:
-		slog.DebugContext(ctx, "Resource plugin received delete instances message")
-		instanceList, ok := msg.Data.([]*mpi.Instance)
-		if !ok {
-			slog.ErrorContext(ctx, "Unable to cast message payload to []*mpi.Instance", "payload", msg.Data)
-
-			return
-		}
-		resource := r.resourceService.DeleteInstances(ctx, instanceList)
-
-		r.messagePipe.Process(ctx, &bus.Message{Topic: bus.ResourceUpdateTopic, Data: resource})
-
 		return
 	case bus.WriteConfigSuccessfulTopic:
 		r.handleWriteConfigSuccessful(ctx, msg)
@@ -133,9 +105,7 @@ func (r *Resource) Process(ctx context.Context, msg *bus.Message) {
 
 func (*Resource) Subscriptions() []string {
 	return []string{
-		bus.AddInstancesTopic,
 		bus.UpdatedInstancesTopic,
-		bus.DeletedInstancesTopic,
 		bus.WriteConfigSuccessfulTopic,
 		bus.RollbackWriteTopic,
 		bus.APIActionRequestTopic,

internal/resource/resource_service.go

@@ -56,9 +56,7 @@ const (
 //counterfeiter:generate . processOperator
 
 type resourceServiceInterface interface {
-	AddInstances(instanceList []*mpi.Instance) *mpi.Resource
 	UpdateInstances(ctx context.Context, instanceList []*mpi.Instance) *mpi.Resource
-	DeleteInstances(ctx context.Context, instanceList []*mpi.Instance) *mpi.Resource
 	ApplyConfig(ctx context.Context, instanceID string) (*model.NginxConfigContext, error)
 	Instance(instanceID string) *mpi.Instance
 	GetHTTPUpstreamServers(ctx context.Context, instance *mpi.Instance, upstreams string) ([]client.UpstreamServer,
@@ -91,9 +89,10 @@ type (
 
 type ResourceService struct {
 	resource          *mpi.Resource
+	napInstance       *mpi.Instance
 	nginxConfigParser parser.ConfigParser
 	agentConfig       *config.Config
-	instanceOperators map[string]instanceOperator // key is instance ID
+	instanceOperator  instanceOperator
 	info              host.InfoInterface
 	manifestFilePath  string
 	resourceMutex     sync.Mutex
@@ -106,7 +105,7 @@ func NewResourceService(ctx context.Context, agentConfig *config.Config) *Resour
 		resourceMutex:     sync.Mutex{},
 		info:              host.NewInfo(),
 		operatorsMutex:    sync.Mutex{},
-		instanceOperators: make(map[string]instanceOperator),
+		instanceOperator:  NewInstanceOperator(agentConfig),
 		nginxConfigParser: parser.NewNginxConfigParser(agentConfig),
 		agentConfig:       agentConfig,
 		manifestFilePath:  agentConfig.LibDir + "/manifest.json",
@@ -117,15 +116,6 @@ func NewResourceService(ctx context.Context, agentConfig *config.Config) *Resour
 	return resourceService
 }
 
-func (r *ResourceService) AddInstances(instanceList []*mpi.Instance) *mpi.Resource {
-	r.resourceMutex.Lock()
-	defer r.resourceMutex.Unlock()
-	r.resource.Instances = append(r.resource.GetInstances(), instanceList...)
-	r.AddOperator(instanceList)
-
-	return r.resource
-}
-
 func (r *ResourceService) Instance(instanceID string) *mpi.Instance {
 	for _, instance := range r.resource.GetInstances() {
 		if instance.GetInstanceMeta().GetInstanceId() == instanceID {
@@ -136,73 +126,25 @@ func (r *ResourceService) Instance(instanceID string) *mpi.Instance {
 	return nil
 }
 
-func (r *ResourceService) AddOperator(instanceList []*mpi.Instance) {
-	r.operatorsMutex.Lock()
-	defer r.operatorsMutex.Unlock()
-	for _, instance := range instanceList {
-		r.instanceOperators[instance.GetInstanceMeta().GetInstanceId()] = NewInstanceOperator(r.agentConfig)
-	}
-}
-
-func (r *ResourceService) RemoveOperator(instanceList []*mpi.Instance) {
-	r.operatorsMutex.Lock()
-	defer r.operatorsMutex.Unlock()
-	for _, instance := range instanceList {
-		delete(r.instanceOperators, instance.GetInstanceMeta().GetInstanceId())
-	}
-}
-
 func (r *ResourceService) UpdateInstances(ctx context.Context, instanceList []*mpi.Instance) *mpi.Resource {
 	r.resourceMutex.Lock()
 	defer r.resourceMutex.Unlock()
-
-	for _, updatedInstance := range instanceList {
-		resourceCopy, ok := proto.Clone(r.resource).(*mpi.Resource)
-		if ok {
-			for _, instance := range resourceCopy.GetInstances() {
-				if updatedInstance.GetInstanceMeta().GetInstanceId() == instance.GetInstanceMeta().GetInstanceId() {
-					instance.InstanceMeta = updatedInstance.GetInstanceMeta()
-					instance.InstanceRuntime = updatedInstance.GetInstanceRuntime()
-					instance.InstanceConfig = updatedInstance.GetInstanceConfig()
-				}
-			}
-			r.resource = resourceCopy
-		} else {
-			slog.WarnContext(ctx, "Unable to clone resource while updating instances", "resource",
-				r.resource, "instances", instanceList)
-		}
-	}
-
-	return r.resource
-}
-
-func (r *ResourceService) DeleteInstances(ctx context.Context, instanceList []*mpi.Instance) *mpi.Resource {
-	r.resourceMutex.Lock()
-	defer r.resourceMutex.Unlock()
-
-	for _, deletedInstance := range instanceList {
-		resourceCopy, ok := proto.Clone(r.resource).(*mpi.Resource)
-		if ok {
-			for index, instance := range resourceCopy.GetInstances() {
-				if deletedInstance.GetInstanceMeta().GetInstanceId() == instance.GetInstanceMeta().GetInstanceId() {
-					r.resource.Instances = append(r.resource.Instances[:index], r.resource.GetInstances()[index+1:]...)
-				}
-			}
-		} else {
-			slog.WarnContext(ctx, "Unable to clone resource while deleting instances", "resource",
-				r.resource, "instances", instanceList)
-		}
+	resourceCopy, ok := proto.Clone(r.resource).(*mpi.Resource)
+	if ok {
+		resourceCopy.Instances = instanceList
+		r.resource = resourceCopy
+	} else {
+		slog.WarnContext(ctx, "Unable to clone resource while updating instances", "resource",
+			r.resource, "instances", instanceList)
 	}
-	r.RemoveOperator(instanceList)
 
 	return r.resource
 }
 
 func (r *ResourceService) ApplyConfig(ctx context.Context, instanceID string) (*model.NginxConfigContext, error) {
 	var instance *mpi.Instance
-	operator := r.instanceOperators[instanceID]
 
-	if operator == nil {
+	if r.instanceOperator == nil {
 		return nil, fmt.Errorf("instance %s not found", instanceID)
 	}
 
@@ -224,12 +166,12 @@ func (r *ResourceService) ApplyConfig(ctx context.Context, instanceID string) (*
 
 	slog.DebugContext(ctx, "Updated Instance Runtime after parsing config", "instance", instance.GetInstanceRuntime())
 
-	valErr := operator.Validate(ctx, instance)
+	valErr := r.instanceOperator.Validate(ctx, instance)
 	if valErr != nil {
 		return nil, fmt.Errorf("failed validating config %w", valErr)
 	}
 
-	reloadErr := operator.Reload(ctx, instance)
+	reloadErr := r.instanceOperator.Reload(ctx, instance)
 	if reloadErr != nil {
 		return nil, fmt.Errorf("failed to reload NGINX %w", reloadErr)
 	}

internal/watcher/health/health_watcher_service.go

@@ -30,9 +30,9 @@ type (
 
 	HealthWatcherService struct {
 		agentConfig        *config.Config
-		cache              map[string]*mpi.InstanceHealth   // key is instanceID
-		watchers           map[string]healthWatcherOperator // key is instanceID
-		instances          map[string]*mpi.Instance         // key is instanceID
+		cache              map[string]*mpi.InstanceHealth // key is instanceID
+		watcher            healthWatcherOperator          // key is instanceID
+		instances          map[string]*mpi.Instance       // key is instanceID
 		healthWatcherMutex sync.Mutex
 	}
 
@@ -44,51 +44,21 @@ type (
 
 func NewHealthWatcherService(agentConfig *config.Config) *HealthWatcherService {
 	return &HealthWatcherService{
-		watchers:    make(map[string]healthWatcherOperator),
+		watcher:     NewNginxHealthWatcher(),
 		cache:       make(map[string]*mpi.InstanceHealth),
 		instances:   make(map[string]*mpi.Instance),
 		agentConfig: agentConfig,
 	}
 }
 
-func (hw *HealthWatcherService) AddHealthWatcher(instances []*mpi.Instance) {
-	hw.healthWatcherMutex.Lock()
-	defer hw.healthWatcherMutex.Unlock()
-
-	for _, instance := range instances {
-		switch instance.GetInstanceMeta().GetInstanceType() {
-		case mpi.InstanceMeta_INSTANCE_TYPE_NGINX, mpi.InstanceMeta_INSTANCE_TYPE_NGINX_PLUS:
-			watcher := NewNginxHealthWatcher()
-			hw.watchers[instance.GetInstanceMeta().GetInstanceId()] = watcher
-		case mpi.InstanceMeta_INSTANCE_TYPE_AGENT:
-		case mpi.InstanceMeta_INSTANCE_TYPE_UNSPECIFIED,
-			mpi.InstanceMeta_INSTANCE_TYPE_UNIT,
-			mpi.InstanceMeta_INSTANCE_TYPE_NGINX_APP_PROTECT:
-			fallthrough
-		default:
-			slog.Warn("Health watcher not implemented", "instance_type",
-				instance.GetInstanceMeta().GetInstanceType())
-		}
-		hw.instances[instance.GetInstanceMeta().GetInstanceId()] = instance
-	}
-}
-
 func (hw *HealthWatcherService) UpdateHealthWatcher(instances []*mpi.Instance) {
 	hw.healthWatcherMutex.Lock()
 	defer hw.healthWatcherMutex.Unlock()
 
-	for _, instance := range instances {
-		hw.instances[instance.GetInstanceMeta().GetInstanceId()] = instance
-	}
-}
-
-func (hw *HealthWatcherService) DeleteHealthWatcher(instances []*mpi.Instance) {
-	hw.healthWatcherMutex.Lock()
-	defer hw.healthWatcherMutex.Unlock()
+	clear(hw.instances)
 
 	for _, instance := range instances {
-		delete(hw.watchers, instance.GetInstanceMeta().GetInstanceId())
-		delete(hw.instances, instance.GetInstanceMeta().GetInstanceId())
+		hw.instances[instance.GetInstanceMeta().GetInstanceId()] = instance
 	}
 }
 
@@ -135,11 +105,11 @@ func (hw *HealthWatcherService) Watch(ctx context.Context, ch chan<- InstanceHea
 
 func (hw *HealthWatcherService) health(ctx context.Context) (updatedStatuses []*mpi.InstanceHealth, isHealthDiff bool,
 ) {
-	currentHealth := make(map[string]*mpi.InstanceHealth, len(hw.watchers))
+	currentHealth := make(map[string]*mpi.InstanceHealth, len(hw.instances))
 	allStatuses := make([]*mpi.InstanceHealth, 0)
 
-	for instanceID, watcher := range hw.watchers {
-		instanceHealth, err := watcher.Health(ctx, hw.instances[instanceID])
+	for instanceID := range hw.instances {
+		instanceHealth, err := hw.watcher.Health(ctx, hw.instances[instanceID])
 		if instanceHealth == nil {
 			instanceHealth = &mpi.InstanceHealth{
 				InstanceId:           instanceID,