Check versioned and unversioned pollers in certain cases (#132)

<!--- Note to EXTERNAL Contributors -->
<!-- Thanks for opening a PR! 
If it is a significant code change, please **make sure there is an open
issue** for this.
We work best with you when we have accepted the idea first before you
code. -->

<!--- For ALL Contributors 👇 -->

## What was changed
1. Check unversioned pollers for the target version if the current
version is nil and strategy is progressive
2. Check versioned pollers for Drained versions if the controller is not
running a k8s Deployment for that version

## Why?
1. So that we can still respect Progressive rollout steps if Current
Version is nil and we confirm unversioned pollers for all Target Version
task queues
2. So that we can only block rollout of a new version if we have a
pathological # of ineligible for delete versions, meaning in most cases,
users can rely on the server to delete their old versions.

## Checklist
<!--- add/delete as needed --->

1. Closes <!-- add issue number here -->

3. How was this tested:
- Unit test for maxVersionsIneligibleForDeletion
- Integration test where Progressive strategy is respected when
unversioned pollers exist
- Integration test where deployment is at max versions ineligible and
controller doesn't roll out new deployment
- Integration test where deployment has max versions but they are
eligible for deletion so controller does roll out new deployment

4. Any docs updates needed?
<!--- update README if applicable
      or point out where to update docs.temporal.io -->

Author: carlydf

api/v1alpha1/temporalworker_webhook.go

@@ -57,11 +57,6 @@ func (s *TemporalWorkerDeploymentSpec) Default(ctx context.Context) error {
 		s.SunsetStrategy.DeleteDelay = &v1.Duration{Duration: defaults.DeleteDelay}
 	}
 
-	if s.MaxVersions == nil {
-		maxVersions := int32(defaults.MaxVersions)
-		s.MaxVersions = &maxVersions
-	}
-
 	return nil
 }
 

api/v1alpha1/temporalworker_webhook_test.go

@@ -159,24 +159,6 @@ func TestTemporalWorkerDeployment_Default(t *testing.T) {
 		obj      runtime.Object
 		expected func(t *testing.T, obj *temporaliov1alpha1.TemporalWorkerDeployment)
 	}{
-		"sets default maxVersions": {
-			obj: testhelpers.MakeTWDWithName("default-max-versions", ""),
-			expected: func(t *testing.T, obj *temporaliov1alpha1.TemporalWorkerDeployment) {
-				require.NotNil(t, obj.Spec.MaxVersions)
-				assert.Equal(t, int32(75), *obj.Spec.MaxVersions)
-			},
-		},
-		"preserves existing maxVersions": {
-			obj: testhelpers.ModifyObj(testhelpers.MakeTWDWithName("preserve-max-versions", ""), func(obj *temporaliov1alpha1.TemporalWorkerDeployment) *temporaliov1alpha1.TemporalWorkerDeployment {
-				maxVersions := int32(100)
-				obj.Spec.MaxVersions = &maxVersions
-				return obj
-			}),
-			expected: func(t *testing.T, obj *temporaliov1alpha1.TemporalWorkerDeployment) {
-				require.NotNil(t, obj.Spec.MaxVersions)
-				assert.Equal(t, int32(100), *obj.Spec.MaxVersions)
-			},
-		},
 		"sets default sunset strategy delays": {
 			obj: testhelpers.MakeTWDWithName("default-sunset-delays", ""),
 			expected: func(t *testing.T, obj *temporaliov1alpha1.TemporalWorkerDeployment) {

api/v1alpha1/worker_types.go

@@ -60,18 +60,6 @@ type TemporalWorkerDeploymentSpec struct {
 
 	// TODO(jlegrone): add godoc
 	WorkerOptions WorkerOptions `json:"workerOptions"`
-
-	// MaxVersions defines the maximum number of worker deployment versions allowed.
-	// This helps prevent hitting Temporal's default limit of 100 versions per deployment.
-	// Defaults to 75. Users can override this by explicitly setting a higher value in
-	// the CRD, but should exercise caution: once the server's version limit is reached,
-	// Temporal attempts to delete an eligible version. If no version is eligible for deletion,
-	// new deployments get blocked which prevents the controller from making progress.
-	// This limit can be adjusted server-side by setting `matching.maxVersionsInDeployment`
-	// in dynamicconfig.
-	// +optional
-	// +kubebuilder:validation:Minimum=1
-	MaxVersions *int32 `json:"maxVersions,omitempty"`
 }
 
 // VersionStatus indicates the status of a version.
@@ -240,6 +228,12 @@ type DeprecatedWorkerDeploymentVersion struct {
 	// Only set when Status is VersionStatusDrained.
 	// +optional
 	DrainedSince *metav1.Time `json:"drainedSince"`
+
+	// A Version is eligible for deletion if it is drained and has no pollers on any task queue.
+	// After pollers stop polling, the server will still consider them present until `matching.PollerHistoryTTL`
+	// has passed.
+	// +optional
+	EligibleForDeletion bool `json:"eligibleForDeletion,omitempty"`
 }
 
 // DefaultVersionUpdateStrategy describes how to cut over new workflow executions
@@ -248,10 +242,20 @@ type DeprecatedWorkerDeploymentVersion struct {
 type DefaultVersionUpdateStrategy string
 
 const (
+	// UpdateManual scales worker resources up or down, but does not update the current or ramping worker deployment version.
 	UpdateManual DefaultVersionUpdateStrategy = "Manual"
 
+	// UpdateAllAtOnce starts 100% of new workflow executions on the new worker deployment version as soon as it's healthy.
 	UpdateAllAtOnce DefaultVersionUpdateStrategy = "AllAtOnce"
 
+	// UpdateProgressive ramps up the percentage of new workflow executions targeting the new worker deployment version over time.
+	//
+	// Note: If the Current Version of a Worker Deployment is nil and the controller cannot confirm that all Task Queues
+	// in the Target Version have at least one unversioned poller, the controller will immediately set the new worker
+	// deployment version to be Current and ignore the Progressive rollout steps.
+	// Sending a percentage of traffic to a "nil" version means that traffic will be sent to unversioned workers. If
+	// there are no unversioned workers, those tasks will get stuck. This behavior ensures that all traffic on the task
+	// queues in this worker deployment can be handled by an active poller.
 	UpdateProgressive DefaultVersionUpdateStrategy = "Progressive"
 )
 
@@ -263,15 +267,9 @@ type GateWorkflowConfig struct {
 type RolloutStrategy struct {
 	// Specifies how to treat concurrent executions of a Job.
 	// Valid values are:
-	// - "Manual": scale worker resources up or down, but do not update the current or ramping worker deployment version;
-	// - "AllAtOnce": start 100% of new workflow executions on the new worker deployment version as soon as it's healthy;
-	// - "Progressive": ramp up the percentage of new workflow executions targeting the new worker deployment version over time;
-	//
-	// Note: If the Current Version of a Worker Deployment is nil, the controller will ignore any Progressive Rollout
-	// Steps and immediately set the new worker deployment version to be Current.
-	// Sending a percentage of traffic to a "nil" version means that traffic will be sent to unversioned workers. If
-	// there are no unversioned workers, those tasks will get stuck. This behavior ensures that all traffic on the task
-	// queues in this worker deployment can be handled by an active poller.
+	// - "Manual"
+	// - "AllAtOnce"
+	// - "Progressive"
 	Strategy DefaultVersionUpdateStrategy `json:"strategy"`
 
 	// Gate specifies a workflow type that must run once to completion on the new worker deployment version before

api/v1alpha1/zz_generated.deepcopy.go

@@ -91,6 +91,7 @@ func main() {
 			}))),
 			mgr.GetClient(),
 		),
+		MaxDeploymentVersionsIneligibleForDeletion: controller.GetControllerMaxDeploymentVersionsIneligibleForDeletion(),
 	}).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "TemporalWorkerDeployment")
 		os.Exit(1)

cmd/main.go

@@ -46,10 +46,6 @@ spec:
             type: object
           spec:
             properties:
-              maxVersions:
-                format: int32
-                minimum: 1
-                type: integer
               minReadySeconds:
                 format: int32
                 type: integer
@@ -3876,6 +3872,8 @@ spec:
                     drainedSince:
                       format: date-time
                       type: string
+                    eligibleForDeletion:
+                      type: boolean
                     healthySince:
                       format: date-time
                       type: string

helm/temporal-worker-controller/templates/crds/temporal.io_temporalworkerdeployments.yaml

@@ -119,8 +119,8 @@ func (r *TemporalWorkerDeploymentReconciler) executePlan(ctx context.Context, l
 			},
 			MetadataUpdate: sdkclient.WorkerDeploymentMetadataUpdate{
 				UpsertEntries: map[string]interface{}{
-					controllerIdentityKey: getControllerIdentity(),
-					controllerVersionKey:  getControllerVersion(),
+					controllerIdentityMetadataKey: getControllerIdentity(),
+					controllerVersionMetadataKey:  getControllerVersion(),
 				},
 			},
 		}); err != nil { // would be cool to do this atomically with the update

internal/controller/execplan.go

@@ -100,6 +100,7 @@ func (r *TemporalWorkerDeploymentReconciler) generatePlan(
 		connection,
 		plannerConfig,
 		workerDeploymentName,
+		r.MaxDeploymentVersionsIneligibleForDeletion,
 	)
 	if err != nil {
 		return nil, fmt.Errorf("error generating plan: %w", err)

internal/controller/genplan.go

@@ -6,7 +6,6 @@ package controller
 
 import (
 	"context"
-	"fmt"
 
 	"github.com/go-logr/logr"
 	temporaliov1alpha1 "github.com/temporalio/temporal-worker-controller/api/v1alpha1"
@@ -23,22 +22,11 @@ func (r *TemporalWorkerDeploymentReconciler) generateStatus(
 	req ctrl.Request,
 	workerDeploy *temporaliov1alpha1.TemporalWorkerDeployment,
 	temporalState *temporal.TemporalWorkerState,
+	k8sState *k8s.DeploymentState,
 ) (*temporaliov1alpha1.TemporalWorkerDeploymentStatus, error) {
 	workerDeploymentName := k8s.ComputeWorkerDeploymentName(workerDeploy)
 	targetBuildID := k8s.ComputeBuildID(workerDeploy)
 
-	// Fetch Kubernetes deployment state
-	k8sState, err := k8s.GetDeploymentState(
-		ctx,
-		r.Client,
-		req.Namespace,
-		req.Name,
-		workerDeploymentName,
-	)
-	if err != nil {
-		return nil, fmt.Errorf("unable to get Kubernetes deployment state: %w", err)
-	}
-
 	// Fetch test workflow status for the desired version
 	if targetBuildID != temporalState.CurrentBuildID {
 		testWorkflows, err := temporal.GetTestWorkflowStatus(

internal/controller/genstatus.go

@@ -160,11 +160,17 @@ func (m *stateMapper) mapDeprecatedWorkerDeploymentVersionByBuildID(buildID stri
 		return nil
 	}
 
+	eligibleForDeletion := false
+	if vInfo, exists := m.temporalState.Versions[buildID]; exists {
+		eligibleForDeletion = vInfo.Status == v1alpha1.VersionStatusDrained && vInfo.NoTaskQueuesHaveVersionedPoller
+	}
+
 	version := &v1alpha1.DeprecatedWorkerDeploymentVersion{
 		BaseWorkerDeploymentVersion: v1alpha1.BaseWorkerDeploymentVersion{
 			BuildID: buildID,
 			Status:  v1alpha1.VersionStatusNotRegistered,
 		},
+		EligibleForDeletion: eligibleForDeletion,
 	}
 
 	// Set deployment reference if it exists