feat: implement drain dependency ordering in agent (#570)

* implement drain dependency ordering in agent

* fix update status

* improve update status

* register services befor delete

* linter

* prevent situation when dependent svc was detached

* update check if service exists

* typo

* improve naming

* code review

* code review

---------

Co-authored-by: michaeljguarino <[email protected]>

Author: zreigz

go.mod

@@ -37,7 +37,7 @@ require (
 	github.com/openshift/api v0.0.0-20250908150922-8634aa495a26
 	github.com/orcaman/concurrent-map/v2 v2.0.1
 	github.com/pkg/errors v0.9.1
-	github.com/pluralsh/console/go/client v1.54.0
+	github.com/pluralsh/console/go/client v1.54.1
 	github.com/pluralsh/controller-reconcile-helper v0.1.0
 	github.com/pluralsh/gophoenix v0.1.3-0.20231201014135-dff1b4309e34
 	github.com/pluralsh/polly v0.3.3

go.sum

@@ -879,8 +879,8 @@ github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 h1:GFCKgmp0tecUJ0sJuv4pzYCqS9+RGSn52M3FUwPs+uo=
 github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10/go.mod h1:t/avpk3KcrXxUnYOhZhMXJlSEyie6gQbtLq5NM3loB8=
-github.com/pluralsh/console/go/client v1.54.0 h1:kRPZvynUgDnx4KnWXOrJW0P+SYplYOvrCqA+GyF+OCA=
-github.com/pluralsh/console/go/client v1.54.0/go.mod h1:EwMrcI23s61oLY3etCVc3NE5RYxgfoiokvU0O7KDOQg=
+github.com/pluralsh/console/go/client v1.54.1 h1:+qYVU2zYP+AkU3me4PPF8Oxuck8LY9z5nzHug/MKL0I=
+github.com/pluralsh/console/go/client v1.54.1/go.mod h1:EwMrcI23s61oLY3etCVc3NE5RYxgfoiokvU0O7KDOQg=
 github.com/pluralsh/controller-reconcile-helper v0.1.0 h1:BV3dYZFH5rn8ZvZjtpkACSv/GmLEtRftNQj/Y4ddHEo=
 github.com/pluralsh/controller-reconcile-helper v0.1.0/go.mod h1:RxAbvSB4/jkvx616krCdNQXPbpGJXW3J1L3rASxeFOA=
 github.com/pluralsh/gophoenix v0.1.3-0.20231201014135-dff1b4309e34 h1:ab2PN+6if/Aq3/sJM0AVdy1SYuMAnq4g20VaKhTm/Bw=

pkg/controller/service/dependencies.go

@@ -0,0 +1,73 @@
+package service
+
+import (
+	"sync"
+
+	console "github.com/pluralsh/console/go/client"
+	"github.com/pluralsh/polly/containers"
+)
+
+var (
+	allServices    = containers.NewSet[string]()
+	servicePresent = make(map[string][]*console.ServiceDependencyFragment)
+	cacheMu        sync.RWMutex
+)
+
+func (s *ServiceReconciler) registerDependencies(svc *console.ServiceDeploymentForAgent) {
+	cacheMu.Lock()
+	defer cacheMu.Unlock()
+
+	// Update or add the service with its latest dependencies
+	servicePresent[svc.Name] = svc.Dependencies
+
+	// Sideload dependencies: ensure they exist in the map
+	for _, dep := range svc.Dependencies {
+		if _, exists := servicePresent[dep.Name]; !exists {
+			depSvc, err := s.svcCache.Get(dep.ID)
+			if err != nil && depSvc != nil && depSvc.DeletedAt == nil {
+				servicePresent[dep.Name] = depSvc.Dependencies
+			}
+		}
+	}
+}
+
+// getActiveDependents returns a list of service names that depend on the given service
+func (s *ServiceReconciler) getActiveDependents(svcName string) []string {
+	cacheMu.RLock()
+	defer cacheMu.RUnlock()
+
+	var dependents []string
+	for name, deps := range servicePresent {
+		if name == svcName {
+			continue
+		}
+		for _, d := range deps {
+			if d.Name == svcName {
+				// check if the service is still in the cache
+				// it could have been detached
+				if allServices.Has(name) {
+					dependents = append(dependents, name)
+					break
+				}
+			}
+		}
+	}
+
+	return dependents
+}
+
+// unregisterDependencies removes a service from the map
+func unregisterDependencies(svc *console.ServiceDeploymentForAgent) {
+	cacheMu.Lock()
+	defer cacheMu.Unlock()
+	delete(servicePresent, svc.Name)
+}
+
+func updateAllServices(newSet containers.Set[string]) {
+	cacheMu.Lock()
+	defer cacheMu.Unlock()
+	// Add all newly seen services
+	allServices = allServices.Union(newSet)
+	// Remove any that are no longer present
+	allServices = allServices.Intersect(newSet)
+}

pkg/controller/service/reconciler.go

@@ -7,6 +7,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/pluralsh/polly/containers"
 	"golang.org/x/time/rate"
 	"k8s.io/client-go/dynamic"
 	"k8s.io/klog/v2"
@@ -354,6 +355,8 @@ func (s *ServiceReconciler) Poll(ctx context.Context) error {
 		return nil
 	}
 
+	// Build a new set of services seen in this poll
+	currentServices := containers.NewSet[string]()
 	pager := s.ListServices(ctx)
 
 	for pager.HasNext() {
@@ -371,10 +374,13 @@ func (s *ServiceReconciler) Poll(ctx context.Context) error {
 
 			logger.V(4).Info("enqueueing update for", "service", svc.Node.ID)
 			s.svcCache.Add(svc.Node.ID, svc.Node)
+			currentServices.Add(svc.Node.Name)
 			s.svcQueue.AddAfter(svc.Node.ID, utils.Jitter(15*time.Second))
 		}
 	}
 
+	updateAllServices(currentServices)
+
 	return nil
 }
 
@@ -422,8 +428,24 @@ func (s *ServiceReconciler) Reconcile(ctx context.Context, id string) (result re
 		return
 	}
 
+	s.registerDependencies(svc)
+
 	if svc.DeletedAt != nil {
 		logger.V(2).Info("deleting service", "name", svc.Name, "namespace", svc.Namespace)
+		activeDependents := s.getActiveDependents(svc.Name)
+		if len(activeDependents) > 0 {
+			if err := s.UpdateErrors(id, &console.ServiceErrorAttributes{
+				Message: "service is being deleted, but there are active dependents: " + strings.Join(activeDependents, ", "),
+				Warning: lo.ToPtr(true),
+				Source:  "delete",
+			}); err != nil {
+				logger.Error(err, "failed to update errors")
+				return ctrl.Result{}, err
+			}
+			done = true
+			return ctrl.Result{}, nil
+		}
+
 		components, err := s.applier.Destroy(ctx, id)
 		metrics.Record().ServiceDeletion(id)
 		s.svcCache.Expire(id)
@@ -438,6 +460,10 @@ func (s *ServiceReconciler) Reconcile(ctx context.Context, id string) (result re
 			return ctrl.Result{}, err
 		}
 
+		if len(components) == 0 {
+			unregisterDependencies(svc)
+		}
+
 		// delete service when components len == 0 (no new statuses, inventory file is empty, all deleted)
 		if err := s.UpdateStatus(ctx, svc.ID, svc.Revision.ID, svc.Sha, svc.Status, lo.ToSlicePtr(components), []*console.ServiceErrorAttributes{}, nil); err != nil {
 			logger.Error(err, "Failed to update service status, ignoring for now")