Introducing FlytePropeller TURBO Mode #minor

cmd/kubectl-flyte/cmd/get.go

@@ -92,7 +92,8 @@ func (g *GetOpts) iterateOverWorkflows(f func(*v1alpha1.FlyteWorkflow) error, ba
 			return err
 		}
 		for _, w := range wList.Items {
-			if err := f(&w); err != nil {
+			_w := w
+			if err := f(&_w); err != nil {
 				return err
 			}
 			counter++
@@ -127,7 +128,8 @@ func (g *GetOpts) iterateOverQuotas(f func(quota *v12.ResourceQuota) error, batc
 			return err
 		}
 		for _, r := range rq.Items {
-			if err := f(&r); err != nil {
+			_r := r
+			if err := f(&_r); err != nil {
 				return err
 			}
 			counter++

config.yaml

@@ -32,7 +32,6 @@ tasks:
       - container
       - sidecar
       - K8S-ARRAY
-      - qubole-hive-executor
     # Uncomment to enable sagemaker plugin
     #      - sagemaker_training
     #      - sagemaker_hyperparameter_tuning
@@ -95,7 +94,7 @@ event:
   rate: 500
   capacity: 1000
 admin:
-  endpoint: localhost:30081
+  endpoint: localhost:80
   insecure: true
 catalog-cache:
   type: noop

pkg/compiler/errors/compiler_errors.go

@@ -83,7 +83,7 @@ const (
 func NewBranchNodeNotSpecified(branchNodeID string) *CompileError {
 	return newError(
 		BranchNodeIDNotFound,
-		fmt.Sprintf("BranchNode not assigned"),
+		"BranchNode not assigned",
 		branchNodeID,
 	)
 }

pkg/compiler/validators/interface.go

@@ -70,13 +70,15 @@ func ValidateUnderlyingInterface(w c.WorkflowBuilder, node c.NodeBuilder, errs e
 
 				// Compute exposed inputs as the union of all required inputs and any input overwritten by the node.
 				exposedInputs := map[string]*core.Variable{}
-				for name, p := range inputs.Parameters {
-					if p.GetRequired() {
-						exposedInputs[name] = p.Var
-					} else if containsBindingByVariableName(node.GetInputs(), name) {
-						exposedInputs[name] = p.Var
+				if inputs != nil && inputs.Parameters != nil {
+					for name, p := range inputs.Parameters {
+						if p.GetRequired() {
+							exposedInputs[name] = p.Var
+						} else if containsBindingByVariableName(node.GetInputs(), name) {
+							exposedInputs[name] = p.Var
+						}
+						// else, the param has a default value and is not being overwritten by the node
 					}
-					// else, the param has a default value and is not being overwritten by the node
 				}
 
 				iface = &core.TypedInterface{

pkg/controller/config/config.go

@@ -47,6 +47,8 @@ var (
 			MaxNodeRetriesOnSystemFailures: 3,
 			InterruptibleFailureThreshold:  1,
 		},
+		EnableTurboMode: true,
+		MaxStreakLength: 5,
 	}
 )
 
@@ -74,6 +76,8 @@ type Config struct {
 	MaxDatasetSizeBytes    int64                `json:"max-output-size-bytes" pflag:",Maximum size of outputs per task"`
 	KubeConfig             KubeClientConfig     `json:"kube-client-config" pflag:",Configuration to control the Kubernetes client"`
 	NodeConfig             NodeConfig           `json:"node-config,omitempty" pflag:",config for a workflow node"`
+	EnableTurboMode        bool                 `json:"enable-turbo-mode" pflag:",Boolean flag that enables Turbo-mode, this makes Propeller proceed to another round on successful write to etcD."`
+	MaxStreakLength        int                  `json:"max-streak-length" pflag:",Maximum number of consecutive rounds that one propeller worker can use for one workflow if turbo-mode is enabled."`
 }
 
 type KubeClientConfig struct {

pkg/controller/handler.go

@@ -32,6 +32,7 @@ type propellerMetrics struct {
 	PanicObserved            labeled.Counter
 	RoundSkipped             prometheus.Counter
 	WorkflowNotFound         prometheus.Counter
+	StreakLength             labeled.Counter
 }
 
 func newPropellerMetrics(scope promutils.Scope) *propellerMetrics {
@@ -45,6 +46,7 @@ func newPropellerMetrics(scope promutils.Scope) *propellerMetrics {
 		PanicObserved:            labeled.NewCounter("panic", "Panic during handling or aborting workflow", roundScope, labeled.EmitUnlabeledMetric),
 		RoundSkipped:             roundScope.MustNewCounter("skipped", "Round Skipped because of stale workflow"),
 		WorkflowNotFound:         roundScope.MustNewCounter("not_found", "workflow not found in the cache"),
+		StreakLength:             labeled.NewCounter("streak_length", "Number of consecutive rounds used in fast follow mode", roundScope, labeled.EmitUnlabeledMetric),
 	}
 }
 
@@ -130,6 +132,10 @@ func (p *Propeller) TryMutateWorkflow(ctx context.Context, originalW *v1alpha1.F
 	return mutableW, nil
 }
 
+func ShouldExitCurrentEvaluationLoop(turboModeEnabled bool, isWorkflowTerminated bool, newResVer, oldResVer string) bool {
+	return !(turboModeEnabled && !isWorkflowTerminated && newResVer != oldResVer)
+}
+
 // reconciler compares the actual state with the desired, and attempts to
 // converge the two. It then updates the GetExecutionStatus block of the FlyteWorkflow resource
 // with the current status of the resource.
@@ -180,40 +186,62 @@ func (p *Propeller) Handle(ctx context.Context, namespace, name string) error {
 			return nil
 		}
 	}
+	streak := 0
+	defer p.metrics.StreakLength.Add(ctx, float64(streak))
 
-	mutatedWf, err := p.TryMutateWorkflow(ctx, w)
-	if err != nil {
-		// NOTE We are overriding the deepcopy here, as we are essentially ingnoring all mutations
-		// We only want to increase failed attempts and discard any other partial changes to the CRD.
-		mutatedWf = RecordSystemError(w, err)
-		p.metrics.SystemError.Inc(ctx)
-	} else if mutatedWf == nil {
-		return nil
-	} else {
-		if !w.GetExecutionStatus().IsTerminated() {
-			// No updates in the status we detected, we will skip writing to KubeAPI
-			if mutatedWf.Status.Equals(&w.Status) {
-				logger.Info(ctx, "WF hasn't been updated in this round.")
-				return nil
+	maxLength := p.cfg.MaxStreakLength
+	if maxLength <= 0 {
+		maxLength = 1
+	}
+
+	for streak = 0; streak < maxLength; streak++ {
+		mutatedWf, err := p.TryMutateWorkflow(ctx, w)
+		if err != nil {
+			// NOTE We are overriding the deepcopy here, as we are essentially ingnoring all mutations
+			// We only want to increase failed attempts and discard any other partial changes to the CRD.
+			mutatedWf = RecordSystemError(w, err)
+			p.metrics.SystemError.Inc(ctx)
+		} else if mutatedWf == nil {
+			return nil
+		} else {
+			if !w.GetExecutionStatus().IsTerminated() {
+				// No updates in the status we detected, we will skip writing to KubeAPI
+				if mutatedWf.Status.Equals(&w.Status) {
+					logger.Info(ctx, "WF hasn't been updated in this round.")
+					return nil
+				}
+			}
+			if mutatedWf.GetExecutionStatus().IsTerminated() {
+				// If the end result is a terminated workflow, we remove the labels
+				// We add a completed label so that we can avoid polling for this workflow
+				SetCompletedLabel(mutatedWf, time.Now())
+				ResetFinalizers(mutatedWf)
 			}
 		}
-		if mutatedWf.GetExecutionStatus().IsTerminated() {
-			// If the end result is a terminated workflow, we remove the labels
-			// We add a completed label so that we can avoid polling for this workflow
-			SetCompletedLabel(mutatedWf, time.Now())
-			ResetFinalizers(mutatedWf)
+		// TODO we will need to call updatestatus when it is supported. But to preserve metadata like (label/finalizer) we will need to use update
+
+		// update the GetExecutionStatus block of the FlyteWorkflow resource. UpdateStatus will not
+		// allow changes to the Spec of the resource, which is ideal for ensuring
+		// nothing other than resource status has been updated.
+		newWf, updateErr := p.wfStore.Update(ctx, mutatedWf, workflowstore.PriorityClassCritical)
+		if updateErr != nil {
+			return updateErr
 		}
+		if err != nil {
+			// An error was encountered during the round. Let us return, so that we can back-off gracefully
+			return err
+		}
+		if ShouldExitCurrentEvaluationLoop(p.cfg.EnableTurboMode, mutatedWf.GetExecutionStatus().IsTerminated(),
+			newWf.ResourceVersion, mutatedWf.ResourceVersion) {
+			// Workflow is terminated (no need to continue) or no status was changed, we can wait
+			logger.Infof(ctx, "Will not fast follow, Reason: Enabled? %v, Wf terminated? %v, Version matched? %v",
+				p.cfg.EnableTurboMode, mutatedWf.GetExecutionStatus().IsTerminated(), newWf.ResourceVersion == mutatedWf.ResourceVersion)
+			return nil
+		}
+		logger.Infof(ctx, "FastFollow Enabled. Detected State change, we will try another round. StreakLength [%d]", streak)
+		w = newWf
 	}
-	// TODO we will need to call updatestatus when it is supported. But to preserve metadata like (label/finalizer) we will need to use update
-
-	// update the GetExecutionStatus block of the FlyteWorkflow resource. UpdateStatus will not
-	// allow changes to the Spec of the resource, which is ideal for ensuring
-	// nothing other than resource status has been updated.
-	_, updateErr := p.wfStore.Update(ctx, mutatedWf, workflowstore.PriorityClassCritical)
-	if updateErr != nil {
-		return updateErr
-	}
-	return err
+	return nil
 }
 
 func NewPropellerHandler(_ context.Context, cfg *config.Config, wfStore workflowstore.FlyteWorkflow, executor executors.Workflow, scope promutils.Scope) *Propeller {