Update WorkloadDeploymentReconciler to leverage statful instance control for directly managing instances.

Author: joshlreese

cmd/main.go

@@ -178,6 +178,11 @@ func main() {
 		os.Exit(1)
 	}
 
+	if err = controller.AddIndexers(ctx, mgr); err != nil {
+		setupLog.Error(err, "unable to add indexers")
+		os.Exit(1)
+	}
+
 	// +kubebuilder:scaffold:builder
 
 	if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {

internal/controller/indexers.go

@@ -0,0 +1,67 @@
+package controller
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	mcmanager "sigs.k8s.io/multicluster-runtime/pkg/manager"
+
+	computev1alpha "go.datum.net/workload-operator/api/v1alpha"
+)
+
+const (
+	deploymentWorkloadUIDIndex = "deploymentWorkloadUIDIndex"
+	workloadNetworksIndex      = "workloadNetworksIndex"
+)
+
+func AddIndexers(ctx context.Context, mgr mcmanager.Manager) error {
+	return errors.Join(
+		addWorkloadDeploymentIndexers(ctx, mgr),
+		addWorkloadIndexers(ctx, mgr),
+	)
+}
+
+func addWorkloadDeploymentIndexers(ctx context.Context, mgr mcmanager.Manager) error {
+	if err := mgr.GetFieldIndexer().IndexField(ctx, &computev1alpha.WorkloadDeployment{}, deploymentWorkloadUIDIndex, deploymentWorkloadUIDIndexFunc); err != nil {
+		return fmt.Errorf("failed to add workload deployment indexer %q: %w", deploymentWorkloadUIDIndex, err)
+	}
+
+	return nil
+}
+
+func deploymentWorkloadUIDIndexFunc(o client.Object) []string {
+	return []string{
+		string(o.(*computev1alpha.WorkloadDeployment).Spec.WorkloadRef.UID),
+	}
+}
+
+func addWorkloadIndexers(ctx context.Context, mgr mcmanager.Manager) error {
+	if err := mgr.GetFieldIndexer().IndexField(ctx, &computev1alpha.Workload{}, workloadNetworksIndex, workloadNetworksIndexFunc); err != nil {
+		return fmt.Errorf("failed to add workload indexer %q: %w", workloadNetworksIndex, err)
+	}
+
+	return nil
+}
+
+func workloadNetworksIndexFunc(o client.Object) []string {
+	workload := o.(*computev1alpha.Workload)
+
+	networks := make([]string, 0, len(workload.Spec.Template.Spec.NetworkInterfaces))
+	for _, network := range workload.Spec.Template.Spec.NetworkInterfaces {
+		namespacedName := types.NamespacedName{
+			Namespace: network.Network.Namespace,
+			Name:      network.Network.Name,
+		}
+
+		if namespacedName.Namespace == "" {
+			namespacedName.Namespace = workload.GetNamespace()
+		}
+
+		networks = append(networks, namespacedName.String())
+	}
+
+	return networks
+}

internal/controller/instancecontrol/instancecontrol.go

@@ -23,13 +23,13 @@ type Strategy interface {
 	) ([]Action, error)
 }
 
-type ActionType int
+type ActionType string
 
 const (
-	ActionTypeCreate ActionType = iota
-	ActionTypeUpdate
-	ActionTypeDelete
-	ActionTypeWait
+	ActionTypeCreate ActionType = "Create"
+	ActionTypeUpdate ActionType = "Update"
+	ActionTypeDelete ActionType = "Delete"
+	ActionTypeWait   ActionType = "Wait"
 )
 
 type Action struct {

internal/controller/instancecontrol/stateful/stateful_control.go

@@ -20,7 +20,7 @@ import (
 type statefulControl struct {
 }
 
-func NewStatefulControl() instancecontrol.Strategy {
+func New() instancecontrol.Strategy {
 	return &statefulControl{}
 }
 
@@ -80,11 +80,7 @@ func (c *statefulControl) GetActions(
 				},
 			}
 
-			if desiredInstances[i].Labels == nil {
-				desiredInstances[i].Labels = map[string]string{}
-			}
-
-			desiredInstances[i].Labels[v1alpha.InstanceIndexLabel] = strconv.Itoa(int(i))
+			addInstanceControllerLabels(desiredInstances[i], getInstanceOrdinal(desiredInstances[i].Name), deployment)
 
 			if err := controllerutil.SetControllerReference(deployment, desiredInstances[i], scheme); err != nil {
 				return nil, fmt.Errorf("failed to set controller reference: %w", err)
@@ -111,12 +107,7 @@ func (c *statefulControl) GetActions(
 				updatedInstance.Annotations = deployment.Spec.Template.Annotations
 				updatedInstance.Labels = deployment.Spec.Template.Labels
 
-				if updatedInstance.Labels == nil {
-					updatedInstance.Labels = map[string]string{}
-				}
-
-				// Shouldn't get removed, but just in case.
-				updatedInstance.Labels[v1alpha.InstanceIndexLabel] = strconv.Itoa(getInstanceOrdinal(instance.Name))
+				addInstanceControllerLabels(updatedInstance, getInstanceOrdinal(updatedInstance.Name), deployment)
 
 				updatedInstance.Spec = deployment.Spec.Template.Spec
 				updateActions = append(updateActions, instancecontrol.NewUpdateAction(updatedInstance))
@@ -156,3 +147,13 @@ func (c *statefulControl) GetActions(
 
 	return actions, nil
 }
+
+func addInstanceControllerLabels(instance *v1alpha.Instance, index int, deployment *v1alpha.WorkloadDeployment) {
+	if instance.Labels == nil {
+		instance.Labels = map[string]string{}
+	}
+
+	instance.Labels[v1alpha.InstanceIndexLabel] = strconv.Itoa(index)
+	instance.Labels[v1alpha.WorkloadUIDLabel] = string(deployment.Spec.WorkloadRef.UID)
+	instance.Labels[v1alpha.WorkloadDeploymentUIDLabel] = string(deployment.GetUID())
+}

internal/controller/instancecontrol/stateful/stateful_control_test.go

@@ -27,7 +27,7 @@ func init() {
 
 func TestFreshDeployment(t *testing.T) {
 	ctx := context.Background()
-	control := NewStatefulControl()
+	control := New()
 
 	deployment := getWorkloadDeployment("test-fresh-deploy", 2)
 
@@ -49,7 +49,7 @@ func TestFreshDeployment(t *testing.T) {
 
 func TestUpdateWithAllReadyInstances(t *testing.T) {
 	ctx := context.Background()
-	control := NewStatefulControl()
+	control := New()
 
 	deployment := getWorkloadDeployment("test-deploy", 2)
 
@@ -75,7 +75,7 @@ func TestUpdateWithAllReadyInstances(t *testing.T) {
 
 func TestScaleUpWithNotReadyInstance(t *testing.T) {
 	ctx := context.Background()
-	control := NewStatefulControl()
+	control := New()
 
 	deployment := getWorkloadDeployment("test-deploy", 3)
 
@@ -105,7 +105,7 @@ func TestScaleUpWithNotReadyInstance(t *testing.T) {
 
 func TestScaleUpWithDeletingReadyInstance(t *testing.T) {
 	ctx := context.Background()
-	control := NewStatefulControl()
+	control := New()
 
 	deployment := getWorkloadDeployment("test-deploy", 3)
 
@@ -132,7 +132,7 @@ func TestScaleUpWithDeletingReadyInstance(t *testing.T) {
 
 func TestScaleDownWithAllReadyInstances(t *testing.T) {
 	ctx := context.Background()
-	control := NewStatefulControl()
+	control := New()
 
 	deployment := getWorkloadDeployment("test-deploy", 1)
 

internal/controller/workload_controller.go

@@ -12,13 +12,17 @@ import (
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	apimeta "k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 	kerrors "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/apimachinery/pkg/util/sets"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/cluster"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 	"sigs.k8s.io/controller-runtime/pkg/finalizer"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	mcbuilder "sigs.k8s.io/multicluster-runtime/pkg/builder"
 	mccontext "sigs.k8s.io/multicluster-runtime/pkg/context"
 	mcmanager "sigs.k8s.io/multicluster-runtime/pkg/manager"
@@ -29,7 +33,6 @@ import (
 )
 
 const workloadControllerFinalizer = "compute.datumapis.com/workload-controller"
-const deploymentWorkloadUID = "spec.workloadRef.uid"
 
 // WorkloadReconciler reconciles a Workload object
 type WorkloadReconciler struct {
@@ -296,7 +299,7 @@ func (r *WorkloadReconciler) reconcileWorkloadStatus(
 
 	workload.Status = *newWorkloadStatus
 	if err := upstreamClient.Status().Update(ctx, workload); err != nil {
-		return fmt.Errorf("failed updating workload status")
+		return fmt.Errorf("failed updating workload status: %w", err)
 	}
 
 	return nil
@@ -317,7 +320,7 @@ func (r *WorkloadReconciler) Finalize(ctx context.Context, obj client.Object) (f
 	}
 
 	listOpts := client.MatchingFields{
-		deploymentWorkloadUID: string(obj.GetUID()),
+		deploymentWorkloadUIDIndex: string(obj.GetUID()),
 	}
 	var deployments computev1alpha.WorkloadDeploymentList
 	if err := cl.GetClient().List(ctx, &deployments, listOpts); err != nil {
@@ -368,7 +371,7 @@ func (r *WorkloadReconciler) getDeploymentsForWorkload(
 ) (desired []computev1alpha.WorkloadDeployment, orphaned []computev1alpha.WorkloadDeployment, err error) {
 
 	listOpts := client.MatchingFields{
-		deploymentWorkloadUID: string(workload.UID),
+		deploymentWorkloadUIDIndex: string(workload.UID),
 	}
 	var deployments computev1alpha.WorkloadDeploymentList
 	if err := upstreamClient.List(ctx, &deployments, listOpts); err != nil {
@@ -462,22 +465,46 @@ func (r *WorkloadReconciler) SetupWithManager(mgr mcmanager.Manager) error {
 		return fmt.Errorf("failed to register finalizer: %w", err)
 	}
 
-	// TODO(jreese) move to indexer package
-
-	err := mgr.GetFieldIndexer().IndexField(context.Background(), &computev1alpha.WorkloadDeployment{}, deploymentWorkloadUID, func(o client.Object) []string {
-		return []string{
-			string(o.(*computev1alpha.WorkloadDeployment).Spec.WorkloadRef.UID),
-		}
-	})
-	if err != nil {
-		return fmt.Errorf("failed to add workload deployment field indexer: %w", err)
-	}
-
-	// TODO(jreese) add watch against networks that triggers a reconcile for
-	// workloads that are attached and are in an error state for networks not
-	// existing.
 	return mcbuilder.ControllerManagedBy(mgr).
 		For(&computev1alpha.Workload{}, mcbuilder.WithEngageWithLocalCluster(false)).
 		Owns(&computev1alpha.WorkloadDeployment{}, mcbuilder.WithEngageWithLocalCluster(false)).
+		Watches(&networkingv1alpha.Network{}, func(clusterName string, cl cluster.Cluster) handler.TypedEventHandler[client.Object, mcreconcile.Request] {
+			return handler.TypedEnqueueRequestsFromMapFunc(func(ctx context.Context, network client.Object) []mcreconcile.Request {
+				logger := log.FromContext(ctx)
+
+				cluster, err := mgr.GetCluster(ctx, clusterName)
+				if err != nil {
+					logger.Error(err, "failed to get cluster")
+					return nil
+				}
+				clusterClient := cluster.GetClient()
+
+				networkName := client.ObjectKeyFromObject(network).String()
+				listOpts := client.MatchingFields{
+					workloadNetworksIndex: networkName,
+				}
+
+				var workloads computev1alpha.WorkloadList
+				if err := clusterClient.List(ctx, &workloads, listOpts); err != nil {
+					logger.Error(err, "failed to list workloads")
+					return nil
+				}
+
+				var requests []mcreconcile.Request
+				for _, workload := range workloads.Items {
+					requests = append(requests, mcreconcile.Request{
+						Request: reconcile.Request{
+							NamespacedName: types.NamespacedName{
+								Namespace: workload.Namespace,
+								Name:      workload.Name,
+							},
+						},
+						ClusterName: clusterName,
+					})
+				}
+
+				return requests
+			})
+		}).
 		Complete(r)
 }

internal/controller/workloaddeployment_controller.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	apimeta "k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -19,6 +20,8 @@ import (
 
 	networkingv1alpha "go.datum.net/network-services-operator/api/v1alpha"
 	computev1alpha "go.datum.net/workload-operator/api/v1alpha"
+
+	instancecontrolstateful "go.datum.net/workload-operator/internal/controller/instancecontrol/stateful"
 )
 
 // WorkloadDeploymentReconciler reconciles a WorkloadDeployment object
@@ -62,9 +65,7 @@ func (r *WorkloadDeploymentReconciler) Reconcile(ctx context.Context, req mcreco
 		return ctrl.Result{}, nil
 	}
 
-	// TODO(jreese) shortcut work on a status condition for network bindings
-	// being ready
-
+	allNetworkBindingsReady := true
 	for i, networkInterface := range deployment.Spec.Template.Spec.NetworkInterfaces {
 		var networkBinding networkingv1alpha.NetworkBinding
 		networkBindingObjectKey := client.ObjectKey{
@@ -97,6 +98,46 @@ func (r *WorkloadDeploymentReconciler) Reconcile(ctx context.Context, req mcreco
 			}
 		}
 
+		if !apimeta.IsStatusConditionTrue(networkBinding.Status.Conditions, networkingv1alpha.NetworkBindingReady) {
+			allNetworkBindingsReady = false
+		}
+	}
+
+	if !allNetworkBindingsReady {
+		logger.Info("waiting for network bindings to be ready")
+		return ctrl.Result{}, nil
+	}
+
+	// Collect all instances for this deployment
+	listOpts := client.MatchingLabels{
+		computev1alpha.WorkloadDeploymentUIDLabel: string(deployment.GetUID()),
+	}
+
+	var instances computev1alpha.InstanceList
+	if err := cl.GetClient().List(ctx, &instances, listOpts); err != nil {
+		return ctrl.Result{}, fmt.Errorf("failed listing instances: %w", err)
+	}
+
+	instanceControl := instancecontrolstateful.New()
+
+	actions, err := instanceControl.GetActions(ctx, cl.GetScheme(), &deployment, instances.Items)
+	if err != nil {
+		return ctrl.Result{}, fmt.Errorf("failed getting instance control actions: %w", err)
+	}
+
+	logger.Info("collected instance control actions", "count", len(actions))
+
+	for _, action := range actions {
+		// We don't need to actually check this, but it'll reduce log noise.
+		if action.IsSkipped() {
+			continue
+		}
+
+		logger.Info("instance control action", "instance", action.Object.GetName(), "action", action.ActionType())
+
+		if err := action.Execute(ctx, cl.GetClient()); err != nil {
+			return ctrl.Result{}, fmt.Errorf("failed executing instance control action: %w", err)
+		}
 	}
 
 	return ctrl.Result{}, nil
@@ -108,5 +149,7 @@ func (r *WorkloadDeploymentReconciler) SetupWithManager(mgr mcmanager.Manager) e
 	// TODO(jreese) finalizers
 	return mcbuilder.ControllerManagedBy(mgr).
 		For(&computev1alpha.WorkloadDeployment{}, mcbuilder.WithEngageWithLocalCluster(false)).
+		Owns(&computev1alpha.Instance{}).
+		Owns(&networkingv1alpha.NetworkBinding{}).
 		Complete(r)
 }