fix: Runtime Operator Resiliency Pass

Signed-off-by: Lucas Fontes <[email protected]>

Author: lxfontes

runtime-operator/api/condition/conditioned_reconciler.go

@@ -166,6 +166,7 @@ type ReconcilerContext struct {
 
 const ctxKeyReconcilerContext = ctxKey("reconciler-context")
 
+// GetReconcilerContext retrieves the ReconcilerContext from the context.
 func GetReconcilerContext(ctx context.Context) *ReconcilerContext {
 	if v, ok := ctx.Value(ctxKeyReconcilerContext).(*ReconcilerContext); ok {
 		return v
@@ -179,6 +180,11 @@ func ForceStatusUpdate(ctx context.Context) {
 	GetReconcilerContext(ctx).ForceUpdate = true
 }
 
+// ForceRequeue forces an immediate requeue, regardless if status has changed.
+func ForceRequeue(ctx context.Context) {
+	GetReconcilerContext(ctx).ForceRequeue = true
+}
+
 // Reconcile reconciles the object with the given request.
 // implements the `AnyConditionedReconciler` and `reconcile.Reconciler` interfaces.
 func (r *ConditionedReconciler[T]) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
@@ -203,6 +209,8 @@ func (r *ConditionedReconciler[T]) Reconcile(ctx context.Context, req reconcile.
 		}
 	}
 
+	originalObject := obj.DeepCopyObject().(T)
+
 	reconcilerCtx := &ReconcilerContext{
 		ReconcileInterval: r.interval,
 	}
@@ -226,7 +234,6 @@ func (r *ConditionedReconciler[T]) Reconcile(ctx context.Context, req reconcile.
 
 		condErr := condEntry.fn(condCtx, obj)
 		if condErr != nil {
-			reconcilerCtx.ForceRequeue = true
 			if errors.Is(condErr, errStatusUnknown) {
 				conditions.SetConditions(UnknownCondition(condEntry.condition, "Reconcile", condErr.Error()))
 			} else if errors.Is(condErr, errNoop) {
@@ -255,7 +262,7 @@ func (r *ConditionedReconciler[T]) Reconcile(ctx context.Context, req reconcile.
 	}
 
 	if reconcilerCtx.ForceUpdate {
-		if err := r.client.Status().Update(ctx, obj); err != nil {
+		if err := r.client.Status().Patch(ctx, obj, client.MergeFrom(originalObject)); err != nil {
 			return reconcile.Result{}, err
 		}
 	}
@@ -272,7 +279,7 @@ func (r *ConditionedReconciler[T]) Reconcile(ctx context.Context, req reconcile.
 	}
 
 	if reconcilerCtx.ForceRequeue {
-		return reconcile.Result{Requeue: true}, nil
+		return reconcile.Result{RequeueAfter: time.Second}, nil
 	}
 
 	return reconcile.Result{RequeueAfter: reconcilerCtx.ReconcileInterval}, nil

runtime-operator/internal/controller/runtime/host_client.go

@@ -12,6 +12,8 @@ import (
 	"google.golang.org/protobuf/types/known/emptypb"
 )
 
+// HostRoundtripTimeout is the max timeout for host RPC calls.
+// Callers can set lower context timeouts as needed.
 const HostRoundtripTimeout = 1 * time.Minute
 
 func NewWashHostClient(bus wasmbus.Bus, hostID string) *WashHostClient {

runtime-operator/internal/controller/runtime/host_controller.go

@@ -21,6 +21,7 @@ import (
 )
 
 const (
+	hostHeartbeatTimeout  = 5 * time.Second
 	hostReconcileInterval = 1 * time.Minute
 )
 
@@ -43,7 +44,7 @@ func (r *HostReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.
 func (r *HostReconciler) reconcileReporting(ctx context.Context, host *runtimev1alpha1.Host) error {
 	client := NewWashHostClient(r.Bus, host.HostID)
 
-	ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
+	ctx, cancel := context.WithTimeout(ctx, hostHeartbeatTimeout)
 	defer cancel()
 
 	heartbeat, err := client.Heartbeat(ctx)

runtime-operator/internal/controller/runtime/workload_controller.go

@@ -18,7 +18,10 @@ import (
 )
 
 const (
-	workloadReconcileInterval     = 1 * time.Minute
+	workloadReconcileInterval     = 30 * time.Second
+	workloadStartTimeout          = 60 * time.Second
+	workloadStatusTimeout         = 5 * time.Second
+	workloadStopTimeout           = 30 * time.Second
 	workloadFinalizerName         = "runtime.wasmcloud.dev/workload-finalizer"
 	workloadSchedulableHostsIndex = "spec.isSchedulable"
 )
@@ -241,6 +244,9 @@ func (r *WorkloadReconciler) reconcilePlacement(ctx context.Context, workload *r
 	}
 
 	client := NewWashHostClient(r.Bus, workload.Status.HostID)
+	ctx, cancel := context.WithTimeout(ctx, workloadStartTimeout)
+	defer cancel()
+
 	resp, err := client.Start(ctx, req)
 	if err != nil {
 		return err
@@ -262,6 +268,9 @@ func (r *WorkloadReconciler) reconcileSync(ctx context.Context, workload *runtim
 		WorkloadId: workload.Status.WorkloadID,
 	}
 
+	ctx, cancel := context.WithTimeout(ctx, workloadStatusTimeout)
+	defer cancel()
+
 	resp, err := client.Status(ctx, req)
 	if err != nil {
 		return err
@@ -293,6 +302,9 @@ func (r *WorkloadReconciler) finalize(ctx context.Context, workload *runtimev1al
 		WorkloadId: workload.Status.WorkloadID,
 	}
 
+	ctx, cancel := context.WithTimeout(ctx, workloadStopTimeout)
+	defer cancel()
+
 	_, err := client.Stop(ctx, req)
 	if err != nil {
 		logger := ctrl.LoggerFrom(ctx)

runtime-operator/internal/controller/runtime/workload_deployment_controller.go

@@ -305,5 +305,16 @@ func resolveArtifacts(ctx context.Context, kubeClient client.Client, namespace s
 		tpl.Spec.Components[i] = comp
 	}
 
+	if tpl.Spec.Service != nil {
+		if strings.HasPrefix(tpl.Spec.Service.Image, "artifact://") {
+			artifactName := strings.TrimPrefix(tpl.Spec.Service.Image, "artifact://")
+			artifact, ok := artifactMap[artifactName]
+			if !ok {
+				return fmt.Errorf("artifact %s not found in deployment spec", artifactName)
+			}
+			tpl.Spec.Service.Image = artifact.Status.ArtifactURL
+		}
+	}
+
 	return nil
 }

runtime-operator/internal/controller/runtime/workload_replicaset_controller.go

@@ -237,13 +237,21 @@ func (r *WorkloadReplicaSetReconciler) cleanupUnhealthyWorkloads(ctx context.Con
 			continue
 		}
 
+		// catches workloads that are still syncing
+		// they might have not been placed or sync failing. in most cases host is down/gone.
+
 		syncStatus := workload.Status.GetCondition(runtimev1alpha1.WorkloadConditionSync)
-		// catches unknown & true sync status
-		if syncStatus.Status != condition.ConditionFalse {
+		if syncStatus.Status == condition.ConditionTrue {
 			continue
 		}
+
+		// this was the first failure, give it some time to recover
+		if syncStatus.LastTransitionTime.IsZero() {
+			continue
+		}
+
 		// If the workload has failed recently, give it some time to recover
-		if syncStatus.LastProbeTime.Time.Add(workloadReplicaSetReplicaGracePeriod).After(time.Now()) {
+		if syncStatus.LastTransitionTime.Time.Add(workloadReplicaSetReplicaGracePeriod).After(time.Now()) {
 			continue
 		}
 		// At this point, the workload is not ready, has failed, and has not recovered within the grace period