feat: Upgrade Decision engine incorporated into reconciliation logic

cmd/main.go

@@ -37,7 +37,6 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/webhook"
 
 	numaflowv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
-	"github.com/numaproj/numaplane/internal/common"
 	"github.com/numaproj/numaplane/internal/controller"
 	"github.com/numaproj/numaplane/internal/controller/config"
 	"github.com/numaproj/numaplane/internal/util/kubernetes"
@@ -239,7 +238,4 @@ func loadConfigs() {
 	logger.SetBaseLogger(numaLogger)
 	clog.SetLogger(*numaLogger.LogrLogger)
 
-	// feature flag
-	common.DataLossPrevention = config.DataLossPrevention
-
 }

config/install.yaml

@@ -1408,6 +1408,7 @@ metadata:
 ---
 apiVersion: v1
 data:
+  defaultUpgradeStrategy: ""
   pipelineSpecExcludedPaths: |
     - "lifecycle"
     - "limits"

config/manager/usde-config.yaml

@@ -5,6 +5,7 @@ metadata:
   labels:
     numaplane.numaproj.io/config: usde-config
 data:
+  defaultUpgradeStrategy: ""
   pipelineSpecExcludedPaths: |
     - "lifecycle"
     - "limits"

internal/common/common.go

@@ -72,7 +72,4 @@ var (
 
 	// default requeue time used by Reconcilers
 	DefaultDelayedRequeue = ctrl.Result{RequeueAfter: 20 * time.Second}
-
-	// DataLossPrevention is a feature flag used to turn on/off the automatic pause feature for pipelines based on how it's set in the Config
-	DataLossPrevention bool
 )

internal/controller/config/config.go

@@ -69,8 +69,6 @@ func (*ConfigManager) GetControllerDefinitionsMgr() *NumaflowControllerDefinitio
 type GlobalConfig struct {
 	LogLevel          int    `json:"logLevel" mapstructure:"logLevel"`
 	IncludedResources string `json:"includedResources" mapstructure:"includedResources"`
-	// Feature flag - if enabled causes pipeline(s) to be paused when pipeline, numaflow controller, or ISB Service gets updated
-	DataLossPrevention bool `json:"dataLossPrevention" mapstructure:"dataLossPrevention"`
 	// List of Numaflow Controller image names to look for
 	NumaflowControllerImageNames []string `json:"numaflowControllerImageNames" mapstructure:"numaflowControllerImageNames"`
 }

internal/controller/config/config_test.go

@@ -25,7 +25,6 @@ func TestLoadConfigMatchValues(t *testing.T) {
 
 	assert.Nil(t, err, "Failed to load configuration")
 	assert.Equal(t, 3, config.LogLevel, "Log Level does not match")
-	assert.Equal(t, true, config.DataLossPrevention, "DataLossPrevention does not match")
 	assert.Contains(t, config.NumaflowControllerImageNames, "numaflow")
 	assert.Contains(t, config.NumaflowControllerImageNames, "numaflow-rc")
 	// now verify that if we modify the file, it will still be okay

internal/controller/config/usde_config.go

@@ -1,20 +1,10 @@
 package config
 
-import (
-	"encoding/json"
-	"fmt"
-)
-
-type USDEUserStrategy string
-
-const (
-	// ProgressiveStrategyID USDEUserStrategy = "progressive" // TODO-PROGRESSIVE: enable this
-	PPNDStrategyID USDEUserStrategy = "pause-and-drain"
-)
-
 type USDEConfig struct {
-	PipelineSpecExcludedPaths   []string `json:"pipelineSpecExcludedPaths,omitempty" yaml:"pipelineSpecExcludedPaths,omitempty"`
-	ISBServiceSpecExcludedPaths []string `json:"isbServiceSpecExcludedPaths,omitempty" yaml:"isbServiceSpecExcludedPaths,omitempty"`
+	// If user's config doesn't exist or doesn't specify strategy, this is the default
+	DefaultUpgradeStrategy      USDEUserStrategy `json:"defaultUpgradeStrategy" mapstructure:"defaultUpgradeStrategy"`
+	PipelineSpecExcludedPaths   []string         `json:"pipelineSpecExcludedPaths,omitempty" yaml:"pipelineSpecExcludedPaths,omitempty"`
+	ISBServiceSpecExcludedPaths []string         `json:"isbServiceSpecExcludedPaths,omitempty" yaml:"isbServiceSpecExcludedPaths,omitempty"`
 }
 
 func (cm *ConfigManager) UpdateUSDEConfig(config USDEConfig) {
@@ -37,23 +27,3 @@ func (cm *ConfigManager) GetUSDEConfig() USDEConfig {
 
 	return cm.usdeConfig
 }
-
-func (s *USDEUserStrategy) UnmarshalJSON(data []byte) (err error) {
-	var usdeUserStrategyStr string
-	if err := json.Unmarshal(data, &usdeUserStrategyStr); err != nil {
-		return err
-	}
-
-	// Make sure the string is one of the possible strategy values
-	if usdeUserStrategyStr != string(PPNDStrategyID) {
-		return fmt.Errorf("invalid strategy '%s' (allowed value is: %s)", usdeUserStrategyStr, PPNDStrategyID)
-	}
-	// TODO-PROGRESSIVE: replace if-statement above for if-statement below
-	// if usdeUserStrategyStr != string(ProgressiveStrategyID) && usdeUserStrategyStr != string(PPNDStrategyID) {
-	// 	return fmt.Errorf("invalid strategy '%s' (allowed values are: %s or %s)", usdeUserStrategyStr, ProgressiveStrategyID, PPNDStrategyID)
-	// }
-
-	*s = USDEUserStrategy(usdeUserStrategyStr)
-
-	return nil
-}

internal/controller/config/user_config.go

@@ -0,0 +1,47 @@
+package config
+
+import (
+	"encoding/json"
+	"fmt"
+)
+
+type USDEUserStrategy string
+
+const (
+	ProgressiveStrategyID USDEUserStrategy = "progressive"
+	PPNDStrategyID        USDEUserStrategy = "pause-and-drain"
+	NoStrategyID          USDEUserStrategy = ""
+)
+
+func (s *USDEUserStrategy) UnmarshalJSON(data []byte) (err error) {
+	var usdeUserStrategyStr string
+	if err := json.Unmarshal(data, &usdeUserStrategyStr); err != nil {
+		return err
+	}
+
+	allowedValues := map[USDEUserStrategy]struct{}{
+		ProgressiveStrategyID: {},
+		PPNDStrategyID:        {},
+		NoStrategyID:          {}} // TODO: this will no longer be allowed in the future
+
+	// Make sure the string is one of the possible strategy values
+	_, found := allowedValues[USDEUserStrategy(usdeUserStrategyStr)]
+	if !found {
+		return fmt.Errorf("invalid strategy '%s' (allowed values: %+v)", usdeUserStrategyStr, allowedValues)
+	}
+
+	*s = USDEUserStrategy(usdeUserStrategyStr)
+
+	return nil
+}
+
+func (s USDEUserStrategy) IsValid() bool {
+	switch s {
+	case ProgressiveStrategyID:
+		return true
+	case PPNDStrategyID:
+		return true
+	default:
+		return false
+	}
+}

internal/controller/in_progress_strategy.go

@@ -0,0 +1,103 @@
+package controller
+
+import (
+	"context"
+	"sync"
+
+	k8stypes "k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	apiv1 "github.com/numaproj/numaplane/pkg/apis/numaplane/v1alpha1"
+)
+
+// inProgressStrategyMgr is responsible to maintain a Rollout's inProgressStrategy
+// state is maintained both in memory as well as in the Rollout's Status
+// in memory always gives us the latest state in case the Informer cache is out of date
+// the Rollout's Status is useful as a backup mechanism in case Numaplane has just restarted
+type inProgressStrategyMgr struct {
+	getRolloutStrategy func(context.Context, client.Object) *apiv1.UpgradeStrategy
+	setRolloutStrategy func(context.Context, client.Object, apiv1.UpgradeStrategy)
+	store              *inProgressStrategyStore
+}
+
+// in memory storage of UpgradeStrategy in progress for a given Rollout
+type inProgressStrategyStore struct {
+	inProgressUpgradeStrategies map[string]apiv1.UpgradeStrategy
+	mutex                       sync.RWMutex
+}
+
+func newInProgressStrategyMgr(
+	getRolloutStrategy func(context.Context, client.Object) *apiv1.UpgradeStrategy,
+	setRolloutStrategy func(context.Context, client.Object, apiv1.UpgradeStrategy)) *inProgressStrategyMgr {
+
+	return &inProgressStrategyMgr{
+		getRolloutStrategy: getRolloutStrategy,
+		setRolloutStrategy: setRolloutStrategy,
+		store:              newInProgressStrategyStore(),
+	}
+}
+
+func newInProgressStrategyStore() *inProgressStrategyStore {
+	return &inProgressStrategyStore{
+		inProgressUpgradeStrategies: map[string]apiv1.UpgradeStrategy{},
+	}
+}
+
+func (mgr *inProgressStrategyMgr) getStrategy(ctx context.Context, rollout client.Object) apiv1.UpgradeStrategy {
+	return mgr.synchronize(ctx, rollout)
+}
+
+// make sure in-memory value and Rollout Status value are synchronized
+// if in-memory value is set, make sure Rollout Status value gets set the same
+// if in-memory value isn't set, make sure in-memory value gets set to Rollout Status value
+func (mgr *inProgressStrategyMgr) synchronize(ctx context.Context, rollout client.Object) apiv1.UpgradeStrategy {
+	namespacedName := k8stypes.NamespacedName{Namespace: rollout.GetNamespace(), Name: rollout.GetName()}
+
+	// first look for value in memory
+	foundInMemory, inMemoryStrategy := mgr.store.getStrategy(namespacedName)
+
+	// now look for the value in the Resource
+	crDefinedStrategy := mgr.getRolloutStrategy(ctx, rollout)
+
+	// if in-memory value is set, make sure Rollout Status value gets set the same
+	if foundInMemory {
+		mgr.setRolloutStrategy(ctx, rollout, inMemoryStrategy)
+		return inMemoryStrategy
+	} else {
+		// make sure in-memory value gets set to Rollout Status value
+		if crDefinedStrategy != nil {
+			mgr.store.setStrategy(namespacedName, *crDefinedStrategy)
+			return *crDefinedStrategy
+		} else {
+			return apiv1.UpgradeStrategyNoOp
+		}
+	}
+}
+
+// store in both memory and the Resource itself
+func (mgr *inProgressStrategyMgr) setStrategy(ctx context.Context, rollout client.Object, upgradeStrategy apiv1.UpgradeStrategy) {
+	namespacedName := k8stypes.NamespacedName{Namespace: rollout.GetNamespace(), Name: rollout.GetName()}
+
+	mgr.store.setStrategy(namespacedName, upgradeStrategy)
+	mgr.setRolloutStrategy(ctx, rollout, upgradeStrategy)
+}
+
+func (mgr *inProgressStrategyMgr) unsetStrategy(ctx context.Context, rollout client.Object) {
+	mgr.setStrategy(ctx, rollout, apiv1.UpgradeStrategyNoOp)
+}
+
+// return whether found, and if so, the value
+func (store *inProgressStrategyStore) getStrategy(namespacedName k8stypes.NamespacedName) (bool, apiv1.UpgradeStrategy) {
+	key := namespacedNameToKey(namespacedName)
+	store.mutex.RLock()
+	strategy, found := store.inProgressUpgradeStrategies[key]
+	store.mutex.RUnlock()
+	return found, strategy
+}
+
+func (store *inProgressStrategyStore) setStrategy(namespacedName k8stypes.NamespacedName, upgradeStrategy apiv1.UpgradeStrategy) {
+	key := namespacedNameToKey(namespacedName)
+	store.mutex.Lock()
+	store.inProgressUpgradeStrategies[key] = upgradeStrategy
+	store.mutex.Unlock()
+}

internal/controller/in_progress_strategy_test.go

@@ -0,0 +1,79 @@
+package controller
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	k8stypes "k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	apiv1 "github.com/numaproj/numaplane/pkg/apis/numaplane/v1alpha1"
+)
+
+func Test_inProgressStrategyMgr_getStrategy(t *testing.T) {
+
+	progressiveStrategy := apiv1.UpgradeStrategyProgressive
+	ppndStrategy := apiv1.UpgradeStrategyPPND
+	noStrategy := apiv1.UpgradeStrategyNoOp
+
+	testCases := []struct {
+		name                  string
+		inMemoryStrategy      *apiv1.UpgradeStrategy
+		rolloutStatusStrategy *apiv1.UpgradeStrategy
+		resultStrategy        apiv1.UpgradeStrategy
+	}{
+		{
+			name:                  "in memory and in Rollout Status (progressive result)",
+			inMemoryStrategy:      &progressiveStrategy,
+			rolloutStatusStrategy: &noStrategy,
+			resultStrategy:        progressiveStrategy,
+		},
+		{
+			name:                  "in memory and in Rollout Status (no op result)",
+			inMemoryStrategy:      &noStrategy,
+			rolloutStatusStrategy: &ppndStrategy,
+			resultStrategy:        noStrategy,
+		},
+		{
+			name:                  "in memory and not in Rollout Status",
+			inMemoryStrategy:      &progressiveStrategy,
+			rolloutStatusStrategy: nil,
+			resultStrategy:        progressiveStrategy,
+		},
+		{
+			name:                  "in Rollout Status and not in memory",
+			inMemoryStrategy:      nil,
+			rolloutStatusStrategy: &ppndStrategy,
+			resultStrategy:        ppndStrategy,
+		},
+		{
+			name:                  "neither in Rollout Status nor in memory",
+			inMemoryStrategy:      nil,
+			rolloutStatusStrategy: nil,
+			resultStrategy:        noStrategy,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			inProgressStrategyMgr := newInProgressStrategyMgr(
+				// getRolloutStrategy function:
+				func(ctx context.Context, rollout client.Object) *apiv1.UpgradeStrategy {
+					return tc.rolloutStatusStrategy
+				},
+				// setRolloutStrategy function:
+				func(ctx context.Context, rollout client.Object, strategy apiv1.UpgradeStrategy) {},
+			)
+			pipelineRollout := &apiv1.PipelineRollout{ObjectMeta: metav1.ObjectMeta{Namespace: "default", Name: "my-pipeline"}}
+			namespacedName := k8stypes.NamespacedName{Namespace: pipelineRollout.GetNamespace(), Name: pipelineRollout.GetName()}
+			if tc.inMemoryStrategy != nil {
+				inProgressStrategyMgr.store.setStrategy(namespacedName, *tc.inMemoryStrategy)
+			}
+			upgradeStrategyResult := inProgressStrategyMgr.getStrategy(context.Background(), pipelineRollout)
+			assert.Equal(t, tc.resultStrategy, upgradeStrategyResult)
+		})
+	}
+}

internal/controller/isbservicerollout_controller.go

@@ -19,6 +19,7 @@ package controller
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"time"
 
@@ -43,6 +44,8 @@ import (
 
 	numaflowv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaplane/internal/common"
+	"github.com/numaproj/numaplane/internal/controller/config"
+	"github.com/numaproj/numaplane/internal/usde"
 	"github.com/numaproj/numaplane/internal/util"
 	"github.com/numaproj/numaplane/internal/util/kubernetes"
 	"github.com/numaproj/numaplane/internal/util/logger"
@@ -290,8 +293,15 @@ func (r *ISBServiceRolloutReconciler) processExistingISBService(ctx context.Cont
 		isbServiceRollout.Status.MarkDeployed(isbServiceRollout.Generation)
 	}
 
-	if common.DataLossPrevention {
-		return processChildObjectWithoutDataLoss(ctx, isbServiceRollout, r, isbServiceNeedsUpdating, isbServiceIsUpdating, func() error {
+	// determine the Upgrade Strategy user prefers
+	upgradeStrategy, err := usde.GetUserStrategy(ctx, isbServiceRollout.Namespace)
+	if err != nil {
+		return false, err
+	}
+
+	switch upgradeStrategy {
+	case config.PPNDStrategyID:
+		return processChildObjectWithPPND(ctx, isbServiceRollout, r, isbServiceNeedsUpdating, isbServiceIsUpdating, func() error {
 			r.recorder.Eventf(isbServiceRollout, corev1.EventTypeNormal, "PipelinesPaused", "All Pipelines have paused for ISBService update")
 			err = r.updateISBService(ctx, isbServiceRollout, newISBServiceDef)
 			if err != nil {
@@ -300,13 +310,19 @@ func (r *ISBServiceRolloutReconciler) processExistingISBService(ctx context.Cont
 			r.customMetrics.ReconciliationDuration.WithLabelValues(ControllerISBSVCRollout, "update").Observe(time.Since(syncStartTime).Seconds())
 			return nil
 		})
-	} else if isbServiceNeedsUpdating {
-		// update ISBService
-		err = r.updateISBService(ctx, isbServiceRollout, newISBServiceDef)
-		if err != nil {
-			return false, err
+	case config.NoStrategyID:
+		if isbServiceNeedsUpdating {
+			// update ISBService
+			err = r.updateISBService(ctx, isbServiceRollout, newISBServiceDef)
+			if err != nil {
+				return false, err
+			}
+			r.customMetrics.ReconciliationDuration.WithLabelValues(ControllerISBSVCRollout, "update").Observe(time.Since(syncStartTime).Seconds())
 		}
-		r.customMetrics.ReconciliationDuration.WithLabelValues(ControllerISBSVCRollout, "update").Observe(time.Since(syncStartTime).Seconds())
+	case config.ProgressiveStrategyID:
+		return false, errors.New("Progressive Strategy not supported yet")
+	default:
+		return false, fmt.Errorf("%v strategy not recognized", upgradeStrategy)
 	}
 
 	return false, nil