Implement Queued condition on AppWrapper Status (#101)

Author: dgrove-oss

.golangci.yml

@@ -31,7 +31,6 @@ linters:
   - errcheck
   - exportloopref
   - goconst
-  - gocyclo
   - gofmt
   - goimports
   - gosimple

api/v1beta1/appwrapper_types.go

@@ -115,6 +115,9 @@ type AppWrapperStatus struct {
 
 	// Number of transitions
 	TransitionCount int32 `json:"transitionCount,omitempty"`
+
+	// Conditions
+	Conditions []metav1.Condition `json:"conditions,omitempty"`
 }
 
 // AppWrapperState is the label for the AppWrapper status
@@ -123,6 +126,9 @@ type AppWrapperState string
 // AppWrapperState is the status of wrapped resources
 type AppWrapperStep string
 
+// AppWrapperQueuedReason is the Type for the Queued Condition
+type AppWrapperQueuedReason string
+
 const (
 	// Initial state upon creation of the AppWrapper object
 	Empty AppWrapperState = ""
@@ -150,6 +156,18 @@ const (
 
 	// MCAD is in the process of deleting the wrapped resources
 	Deleting AppWrapperStep = "deleting"
+
+	// Queued because of insufficient available resources
+	QueuedInsufficientResources = "InsufficientResources"
+
+	// Queued because of insufficient available quota
+	QueuedInsufficientQuota = "InsufficientQuota"
+
+	// Queued because it was Requeued
+	QueuedRequeue = "Requeued"
+
+	// Not Queued because it was Dispatched
+	QueuedDispatch = "Dispatched"
 )
 
 // AppWrapperService

api/v1beta1/zz_generated.deepcopy.go

@@ -95,10 +95,11 @@ func main() {
 	}
 
 	if err = (&controller.AppWrapperReconciler{
-		Client: mgr.GetClient(),
-		Scheme: mgr.GetScheme(),
-		Cache:  map[types.UID]*controller.CachedAppWrapper{}, // AppWrapper cache
-		Events: make(chan event.GenericEvent, 1),             // channel to trigger dispatch
+		Client:    mgr.GetClient(),
+		Scheme:    mgr.GetScheme(),
+		Cache:     map[types.UID]*controller.CachedAppWrapper{}, // AppWrapper cache
+		Events:    make(chan event.GenericEvent, 1),             // channel to trigger dispatch
+		Decisions: map[types.UID]*controller.QueuingDecision{},  // cache of recent queuing decisions
 	}).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "AppWrapper")
 		os.Exit(1)

cmd/main.go

@@ -596,6 +596,75 @@ spec:
           status:
             description: AppWrapperStatus defines the observed state of AppWrapper
             properties:
+              conditions:
+                description: Conditions
+                items:
+                  description: "Condition contains details for one aspect of the current
+                    state of this API Resource. --- This struct is intended for direct
+                    use as an array at the field path .status.conditions.  For example,
+                    \n type FooStatus struct{ // Represents the observations of a
+                    foo's current state. // Known .status.conditions.type are: \"Available\",
+                    \"Progressing\", and \"Degraded\" // +patchMergeKey=type // +patchStrategy=merge
+                    // +listType=map // +listMapKey=type Conditions []metav1.Condition
+                    `json:\"conditions,omitempty\" patchStrategy:\"merge\" patchMergeKey:\"type\"
+                    protobuf:\"bytes,1,rep,name=conditions\"` \n // other fields }"
+                  properties:
+                    lastTransitionTime:
+                      description: lastTransitionTime is the last time the condition
+                        transitioned from one status to another. This should be when
+                        the underlying condition changed.  If that is not known, then
+                        using the time when the API field changed is acceptable.
+                      format: date-time
+                      type: string
+                    message:
+                      description: message is a human readable message indicating
+                        details about the transition. This may be an empty string.
+                      maxLength: 32768
+                      type: string
+                    observedGeneration:
+                      description: observedGeneration represents the .metadata.generation
+                        that the condition was set based upon. For instance, if .metadata.generation
+                        is currently 12, but the .status.conditions[x].observedGeneration
+                        is 9, the condition is out of date with respect to the current
+                        state of the instance.
+                      format: int64
+                      minimum: 0
+                      type: integer
+                    reason:
+                      description: reason contains a programmatic identifier indicating
+                        the reason for the condition's last transition. Producers
+                        of specific condition types may define expected values and
+                        meanings for this field, and whether the values are considered
+                        a guaranteed API. The value should be a CamelCase string.
+                        This field may not be empty.
+                      maxLength: 1024
+                      minLength: 1
+                      pattern: ^[A-Za-z]([A-Za-z0-9_,:]*[A-Za-z0-9_])?$
+                      type: string
+                    status:
+                      description: status of the condition, one of True, False, Unknown.
+                      enum:
+                      - "True"
+                      - "False"
+                      - Unknown
+                      type: string
+                    type:
+                      description: type of condition in CamelCase or in foo.example.com/CamelCase.
+                        --- Many .condition.type values are consistent across resources
+                        like Available, but because arbitrary conditions can be useful
+                        (see .node.status.conditions), the ability to deconflict is
+                        important. The regex it matches is (dns1123SubdomainFmt/)?(qualifiedNameFmt)
+                      maxLength: 316
+                      pattern: ^([a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*/)?(([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9])$
+                      type: string
+                  required:
+                  - lastTransitionTime
+                  - message
+                  - reason
+                  - status
+                  - type
+                  type: object
+                type: array
               dispatchTimestamp:
                 description: When last dispatched
                 format: date-time

config/crd/bases/workload.codeflare.dev_appwrappers.yaml

@@ -19,11 +19,13 @@ package controller
 import (
 	"context"
 	"errors"
+	"fmt"
 	"strconv"
 	"time"
 
 	batchv1 "k8s.io/api/batch/v1"
 	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
@@ -47,6 +49,7 @@ type AppWrapperReconciler struct {
 	Events          chan event.GenericEvent         // event channel to trigger dispatch
 	ClusterCapacity Weights                         // cluster capacity available to MCAD
 	NextSync        time.Time                       // when to refresh cluster capacity
+	Decisions       map[types.UID]*QueuingDecision  // transient log of queuing decisions to enable recording in AppWrapper Status
 }
 
 const (
@@ -124,6 +127,7 @@ func (r *AppWrapperReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 		}
 		// remove AppWrapper from cache
 		r.deleteCachedAW(appWrapper)
+		delete(r.Decisions, appWrapper.UID)
 		log.FromContext(ctx).Info("Deleted")
 		return ctrl.Result{}, nil
 	}
@@ -141,8 +145,27 @@ func (r *AppWrapperReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 		return r.updateStatus(ctx, appWrapper, mcadv1beta1.Queued, mcadv1beta1.Idle)
 
 	case mcadv1beta1.Queued:
-		r.triggerDispatch()
-		return ctrl.Result{}, nil
+		// Propagate most recent queuing decision to AppWrapper's Queued Condition
+		if decision, ok := r.Decisions[appWrapper.UID]; ok {
+			meta.SetStatusCondition(&appWrapper.Status.Conditions, metav1.Condition{
+				Type:    string(mcadv1beta1.Queued),
+				Status:  metav1.ConditionTrue,
+				Reason:  string(decision.reason),
+				Message: decision.message,
+			})
+			if r.Status().Update(ctx, appWrapper) == nil {
+				// If successfully propagated, remove from in memory map
+				delete(r.Decisions, appWrapper.UID)
+			}
+		}
+
+		if meta.FindStatusCondition(appWrapper.Status.Conditions, string(mcadv1beta1.Queued)) == nil {
+			// Absence of Queued Condition strongly suggests AppWrapper is new; trigger dispatch and a short requeue
+			r.triggerDispatch()
+			return ctrl.Result{RequeueAfter: deletionDelay}, nil
+		} else {
+			return ctrl.Result{RequeueAfter: queuedDelay}, nil
+		}
 
 	case mcadv1beta1.Running:
 		switch appWrapper.Status.Step {
@@ -193,7 +216,21 @@ func (r *AppWrapperReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 			// reset status to queued/idle
 			appWrapper.Status.Restarts += 1
 			appWrapper.Status.RequeueTimestamp = metav1.Now() // overwrite requeue decision time with completion time
-			return r.updateStatus(ctx, appWrapper, mcadv1beta1.Queued, mcadv1beta1.Idle)
+			msg := "Requeued by MCAD"
+			if decision, ok := r.Decisions[appWrapper.UID]; ok && decision.reason == mcadv1beta1.QueuedRequeue {
+				msg = fmt.Sprintf("Requeued because %s", decision.message)
+			}
+			meta.SetStatusCondition(&appWrapper.Status.Conditions, metav1.Condition{
+				Type:    string(mcadv1beta1.Queued),
+				Status:  metav1.ConditionTrue,
+				Reason:  string(mcadv1beta1.QueuedRequeue),
+				Message: msg,
+			})
+			res, err := r.updateStatus(ctx, appWrapper, mcadv1beta1.Queued, mcadv1beta1.Idle)
+			if err == nil {
+				delete(r.Decisions, appWrapper.UID)
+			}
+			return res, err
 		}
 
 	case mcadv1beta1.Failed:
@@ -295,6 +332,7 @@ func (r *AppWrapperReconciler) requeueOrFail(ctx context.Context, appWrapper *mc
 	}
 	// requeue AppWrapper
 	appWrapper.Status.RequeueTimestamp = metav1.Now()
+	r.Decisions[appWrapper.UID] = &QueuingDecision{reason: mcadv1beta1.QueuedRequeue, message: reason}
 	return r.updateStatus(ctx, appWrapper, mcadv1beta1.Running, mcadv1beta1.Deleting, reason)
 }
 
@@ -332,6 +370,12 @@ func (r *AppWrapperReconciler) dispatch(ctx context.Context) (ctrl.Result, error
 		}
 		// set dispatching time and status
 		appWrapper.Status.DispatchTimestamp = metav1.Now()
+		meta.SetStatusCondition(&appWrapper.Status.Conditions, metav1.Condition{
+			Type:    string(mcadv1beta1.Queued),
+			Status:  metav1.ConditionFalse,
+			Reason:  string(mcadv1beta1.QueuedDispatch),
+			Message: "Selected for dispatch",
+		})
 		if _, err := r.updateStatus(ctx, appWrapper, mcadv1beta1.Running, mcadv1beta1.Creating); err != nil {
 			return ctrl.Result{}, err
 		}

internal/controller/appwrapper_controller.go

@@ -18,6 +18,7 @@ package controller
 
 import (
 	"context"
+	"fmt"
 	"sort"
 	"strconv"
 	"time"
@@ -31,6 +32,11 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 )
 
+type QueuingDecision struct {
+	reason  mcadv1beta1.AppWrapperQueuedReason
+	message string
+}
+
 // Compute available cluster capacity
 func (r *AppWrapperReconciler) computeCapacity(ctx context.Context) (Weights, error) {
 	capacity := Weights{}
@@ -251,8 +257,16 @@ func (r *AppWrapperReconciler) selectForDispatch(ctx context.Context) (*mcadv1be
 	// return first AppWrapper that fits if any
 	for _, appWrapper := range queue {
 		request := aggregateRequests(appWrapper)
-		if request.Fits(available[int(appWrapper.Spec.Priority)]) {
+		fits, gaps := request.Fits(available[int(appWrapper.Spec.Priority)])
+		if fits {
 			return appWrapper.DeepCopy(), nil // deep copy AppWrapper
+		} else {
+			msg := ""
+			for _, resource := range gaps {
+				msg += fmt.Sprintf("Insufficient %v; requested %v but only %v available. ", resource, request[resource], available[int(appWrapper.Spec.Priority)][resource])
+
+			}
+			r.Decisions[appWrapper.UID] = &QueuingDecision{reason: mcadv1beta1.QueuedInsufficientResources, message: msg}
 		}
 	}
 	// no queued AppWrapper fits

internal/controller/dispatch_logic.go

@@ -29,6 +29,7 @@ const (
 
 	// RequeueAfter delays
 	runDelay      = time.Minute     // how often to force check running AppWrapper health
+	queuedDelay   = 2 * time.Minute // how often to update condition information on Queued AppWrappes
 	dispatchDelay = time.Minute     // how often to force dispatch
 	deletionDelay = 5 * time.Second // how often to check deleted resources
 )

internal/controller/timings.go

@@ -100,7 +100,8 @@ func (w Weights) Max(r Weights) {
 
 // Compare receiver to argument
 // True if receiver is less than or equal to argument in every dimension
-func (w Weights) Fits(r Weights) bool {
+func (w Weights) Fits(r Weights) (bool, []v1.ResourceName) {
+	insufficient := []v1.ResourceName{}
 	zero := &inf.Dec{}    // shared zero, never mutated
 	for k, v := range w { // range over receiver not argument
 		// ignore 0 requests in case r does not contain k
@@ -109,10 +110,14 @@ func (w Weights) Fits(r Weights) bool {
 		}
 		// v > 0 so r[k] must be defined and no less than v
 		if r[k] == nil || v.Cmp(r[k]) == 1 {
-			return false
+			insufficient = append(insufficient, k)
 		}
 	}
-	return true
+	if len(insufficient) == 0 {
+		return true, nil
+	} else {
+		return false, insufficient
+	}
 }
 
 // Converts Weights to a ResourceList

internal/controller/weights.go

@@ -108,7 +108,7 @@ func createGenericAWTimeoutWithStatus(ctx context.Context, name string) *arbv1.A
 	return aw
 }
 
-func createJobAWWithInitContainer(ctx context.Context, name string, requeuingTimeInSeconds int, requeuingGrowthType string, requeuingMaxNumRequeuings int) *arbv1.AppWrapper {
+func createJobAWWithStuckInitContainer(ctx context.Context, name string, requeuingSpec arbv1.RequeuingSpec) *arbv1.AppWrapper {
 	rb := []byte(`{"apiVersion": "batch/v1",
 		"kind": "Job",
 	"metadata": {
@@ -133,7 +133,7 @@ func createJobAWWithInitContainer(ctx context.Context, name string, requeuingTim
 					{
 						"name": "job-init-container",
 						"image": "quay.io/project-codeflare/busybox:latest",
-						"command": ["sleep", "200"],
+						"command": ["sleep", "infinity"],
 						"resources": {
 							"requests": {
 								"cpu": "500m"
@@ -167,11 +167,7 @@ func createJobAWWithInitContainer(ctx context.Context, name string, requeuingTim
 		Spec: arbv1.AppWrapperSpec{
 			Scheduling: arbv1.SchedulingSpec{
 				MinAvailable: minAvailable,
-				Requeuing: arbv1.RequeuingSpec{
-					TimeInSeconds:             int64(requeuingTimeInSeconds),
-					NotImplemented_GrowthType: requeuingGrowthType,
-					MaxNumRequeuings:          int32(requeuingMaxNumRequeuings),
-				},
+				Requeuing:    requeuingSpec,
 			},
 			Resources: arbv1.AppWrapperResources{
 				GenericItems: []arbv1.GenericItem{