Remove infinite version existence check (#98)

<!--- 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
What it says in the title.
Replaced with Version Status == NotRegistered condition in planning
phase.
Also reorganized test helper functions, ensured that Deployments created
by test helpers can be found by the controller, and added a failed and
successful Gate workflow test.

## Why?
Because checking for version existence in execplan.go breaks the
abstraction of only reading server status in gen status. And we have the
information to not do another describe, so we should use it.

Changes to test files are so that helper functions and test functions
are separated and easier to read and work on.

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

1. Closes #57 

2. How was this tested:
New tests specific to Gate workflows ensure that we are not starting
those workflows on a non-existent version, and old tests also show that
SetCurrent and SetRamping calls are not erroring due to nonexistent
version.

3. Any docs updates needed?
no

Author: carlydf

internal/controller/execplan.go

@@ -58,11 +58,7 @@ func (r *TemporalWorkerDeploymentReconciler) executePlan(ctx context.Context, l
 	deploymentHandler := temporalClient.WorkerDeploymentClient().GetHandle(p.WorkerDeploymentName)
 
 	for _, wf := range p.startTestWorkflows {
-		err := awaitVersionRegistration(ctx, l, deploymentHandler, p.TemporalNamespace, wf.versionID)
-		if err != nil {
-			return fmt.Errorf("error waiting for version to register, did your pollers start successfully?: %w", err)
-		}
-		if _, err = temporalClient.ExecuteWorkflow(ctx, sdkclient.StartWorkflowOptions{
+		if _, err := temporalClient.ExecuteWorkflow(ctx, sdkclient.StartWorkflowOptions{
 			ID:                       wf.workflowID,
 			TaskQueue:                wf.taskQueue,
 			WorkflowExecutionTimeout: time.Hour,
@@ -80,13 +76,7 @@ func (r *TemporalWorkerDeploymentReconciler) executePlan(ctx context.Context, l
 	// Register current version or ramps
 	if vcfg := p.UpdateVersionConfig; vcfg != nil {
 		if vcfg.SetCurrent {
-			err := awaitVersionRegistration(ctx, l, deploymentHandler, p.TemporalNamespace, vcfg.VersionID)
-			if err != nil {
-				return fmt.Errorf("error waiting for version to register, did your pollers start successfully?: %w", err)
-			}
-
 			l.Info("registering new current version", "version", vcfg.VersionID)
-
 			if _, err := deploymentHandler.SetCurrentVersion(ctx, sdkclient.WorkerDeploymentSetCurrentVersionOptions{
 				Version:       vcfg.VersionID,
 				ConflictToken: vcfg.ConflictToken,
@@ -95,13 +85,6 @@ func (r *TemporalWorkerDeploymentReconciler) executePlan(ctx context.Context, l
 				return fmt.Errorf("unable to set current deployment version: %w", err)
 			}
 		} else {
-			if vcfg.VersionID != "" {
-				err := awaitVersionRegistration(ctx, l, deploymentHandler, p.TemporalNamespace, vcfg.VersionID)
-				if err != nil {
-					return fmt.Errorf("error waiting for version to register, did your pollers start successfully?: %w", err)
-				}
-			}
-
 			if vcfg.RampPercentage > 0 {
 				l.Info("applying ramp", "version", vcfg.VersionID, "percentage", vcfg.RampPercentage)
 			} else {

internal/controller/genplan.go

@@ -15,7 +15,6 @@ import (
 	"github.com/temporalio/temporal-worker-controller/internal/temporal"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
-	ctrl "sigs.k8s.io/controller-runtime"
 )
 
 // plan holds the actions to execute during reconciliation
@@ -133,16 +132,5 @@ func (r *TemporalWorkerDeploymentReconciler) newDeployment(
 	buildID string,
 	connection temporaliov1alpha1.TemporalConnectionSpec,
 ) (*appsv1.Deployment, error) {
-	d := k8s.NewDeploymentWithOwnerRef(
-		&w.TypeMeta,
-		&w.ObjectMeta,
-		&w.Spec,
-		k8s.ComputeWorkerDeploymentName(w),
-		buildID,
-		connection,
-	)
-	if err := ctrl.SetControllerReference(w, d, r.Scheme); err != nil {
-		return nil, err
-	}
-	return d, nil
+	return k8s.NewDeploymentWithControllerRef(w, buildID, connection, r.Scheme)
 }

internal/controller/util.go

@@ -5,17 +5,7 @@
 package controller
 
 import (
-	"context"
-	"errors"
-	"fmt"
 	"os"
-	"strings"
-	"time"
-
-	"github.com/go-logr/logr"
-	"github.com/temporalio/temporal-worker-controller/internal/k8s"
-	"go.temporal.io/api/serviceerror"
-	sdkclient "go.temporal.io/sdk/client"
 )
 
 const (
@@ -24,70 +14,6 @@ const (
 	defaultControllerIdentity = "temporal-worker-controller"
 )
 
-// TODO(carlydf): Cache describe success for versions that already exist
-// awaitVersionRegistration should be called after a poller starts polling with config of this version, since that is
-// what will register the version with the server. SetRamp and SetCurrent will fail if the version does not exist.
-func awaitVersionRegistration(
-	ctx context.Context,
-	l logr.Logger,
-	deploymentHandler sdkclient.WorkerDeploymentHandle,
-	namespace, versionID string) error {
-	ticker := time.NewTicker(1 * time.Second)
-	for {
-		l.Info(fmt.Sprintf("checking if version %s exists", versionID))
-		select {
-		case <-ctx.Done():
-			return context.Canceled
-		case <-ticker.C:
-			_, err := deploymentHandler.DescribeVersion(ctx, sdkclient.WorkerDeploymentDescribeVersionOptions{
-				Version: versionID,
-			})
-			var notFoundErr *serviceerror.NotFound
-			if err != nil {
-				if errors.As(err, &notFoundErr) {
-					continue
-				} else {
-					return fmt.Errorf("unable to describe worker deployment version %s: %w", versionID, err)
-				}
-			}
-			// After the version exists, confirm that it also exists in the worker deployment
-			// TODO(carlydf): Remove this check after next Temporal Cloud version which solves this inconsistency
-			return awaitVersionRegistrationInDeployment(ctx, l, deploymentHandler, namespace, versionID)
-		}
-	}
-}
-
-func awaitVersionRegistrationInDeployment(
-	ctx context.Context,
-	l logr.Logger,
-	deploymentHandler sdkclient.WorkerDeploymentHandle,
-	namespace, versionID string) error {
-	deploymentName, _, _ := strings.Cut(versionID, k8s.VersionIDSeparator)
-	ticker := time.NewTicker(1 * time.Second)
-	for {
-		l.Info(fmt.Sprintf("checking if version %s exists in worker deployment", versionID))
-		select {
-		case <-ctx.Done():
-			return context.Canceled
-		case <-ticker.C:
-			resp, err := deploymentHandler.Describe(ctx, sdkclient.WorkerDeploymentDescribeOptions{})
-			var notFoundErr *serviceerror.NotFound
-			if err != nil {
-				if errors.As(err, &notFoundErr) {
-					continue
-				} else {
-					return fmt.Errorf("unable to describe worker deployment %s: %w", deploymentName, err)
-				}
-			}
-			for _, vs := range resp.Info.VersionSummaries {
-				if vs.Version == versionID {
-					return nil
-				}
-			}
-		}
-	}
-}
-
 // getControllerVersion returns the version from environment variable (set by Helm from image.tag)
 func getControllerVersion() string {
 	if version := os.Getenv("CONTROLLER_VERSION"); version != "" {

internal/k8s/deployments.go

@@ -17,6 +17,8 @@ import (
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
@@ -290,3 +292,23 @@ func NewDeploymentWithOwnerRef(
 		},
 	}
 }
+
+func NewDeploymentWithControllerRef(
+	w *temporaliov1alpha1.TemporalWorkerDeployment,
+	buildID string,
+	connection temporaliov1alpha1.TemporalConnectionSpec,
+	reconcilerScheme *runtime.Scheme,
+) (*appsv1.Deployment, error) {
+	d := NewDeploymentWithOwnerRef(
+		&w.TypeMeta,
+		&w.ObjectMeta,
+		&w.Spec,
+		ComputeWorkerDeploymentName(w),
+		buildID,
+		connection,
+	)
+	if err := ctrl.SetControllerReference(w, d, reconcilerScheme); err != nil {
+		return nil, err
+	}
+	return d, nil
+}

internal/planner/planner.go

@@ -218,10 +218,12 @@ func getTestWorkflows(
 ) []WorkflowConfig {
 	var testWorkflows []WorkflowConfig
 
-	// Skip if there's no gate workflow defined or if the target version is already the current
+	// Skip if there's no gate workflow defined, if the target version is already the current, or if the target
+	// version is not yet registered in temporal
 	if config.RolloutStrategy.Gate == nil ||
 		status.CurrentVersion == nil ||
-		status.CurrentVersion.VersionID == status.TargetVersion.VersionID {
+		status.CurrentVersion.VersionID == status.TargetVersion.VersionID ||
+		status.TargetVersion.Status == temporaliov1alpha1.VersionStatusNotRegistered {
 		return nil
 	}
 
@@ -258,8 +260,9 @@ func getVersionConfigDiff(
 	strategy := config.RolloutStrategy
 	conflictToken := status.VersionConflictToken
 
-	// Do nothing if target version's deployment is not healthy yet
-	if status.TargetVersion.HealthySince == nil {
+	// Do nothing if target version's deployment is not healthy yet, or if the version is not yet registered in temporal
+	if status.TargetVersion.HealthySince == nil ||
+		status.TargetVersion.Status == temporaliov1alpha1.VersionStatusNotRegistered {
 		return nil
 	}
 

internal/testhelpers/test_builder.go

@@ -47,6 +47,16 @@ func (b *TemporalWorkerDeploymentBuilder) WithProgressiveStrategy(steps ...tempo
 	return b
 }
 
+// WithGate sets the rollout strategy have a gate workflow
+func (b *TemporalWorkerDeploymentBuilder) WithGate(expectSuccess bool) *TemporalWorkerDeploymentBuilder {
+	if expectSuccess {
+		b.twd.Spec.RolloutStrategy.Gate = &temporaliov1alpha1.GateWorkflowConfig{WorkflowType: successTestWorkflowType}
+	} else {
+		b.twd.Spec.RolloutStrategy.Gate = &temporaliov1alpha1.GateWorkflowConfig{WorkflowType: failTestWorkflowType}
+	}
+	return b
+}
+
 // WithReplicas sets the number of replicas
 func (b *TemporalWorkerDeploymentBuilder) WithReplicas(replicas int32) *TemporalWorkerDeploymentBuilder {
 	b.twd.Spec.Replicas = &replicas
@@ -157,17 +167,18 @@ func (sb *StatusBuilder) WithNamespace(k8sNamespace string) *StatusBuilder {
 }
 
 // WithCurrentVersion sets the current version in the status
-func (sb *StatusBuilder) WithCurrentVersion(imageName string, hasDeployment, createPoller bool) *StatusBuilder {
+func (sb *StatusBuilder) WithCurrentVersion(imageName string, healthy, createDeployment bool) *StatusBuilder {
 	sb.currentVersionBuilder = func(name string, namespace string) *temporaliov1alpha1.CurrentWorkerDeploymentVersion {
-		return MakeCurrentVersion(namespace, name, imageName, hasDeployment, createPoller)
+		return MakeCurrentVersion(namespace, name, imageName, healthy, createDeployment)
 	}
 	return sb
 }
 
-// WithTargetVersion sets the target version in the status
-func (sb *StatusBuilder) WithTargetVersion(imageName string, rampPercentage float32, hasDeployment bool, createPoller bool) *StatusBuilder {
+// WithTargetVersion sets the target version in the status.
+// Target Version is required.
+func (sb *StatusBuilder) WithTargetVersion(imageName string, rampPercentage float32, healthy bool, createDeployment bool) *StatusBuilder {
 	sb.targetVersionBuilder = func(name string, namespace string) temporaliov1alpha1.TargetWorkerDeploymentVersion {
-		return MakeTargetVersion(namespace, name, imageName, rampPercentage, hasDeployment, createPoller)
+		return MakeTargetVersion(namespace, name, imageName, rampPercentage, healthy, createDeployment)
 	}
 	return sb
 }
@@ -198,6 +209,8 @@ type TestCase struct {
 	deprecatedBuildReplicas map[string]int32
 	deprecatedBuildImages   map[string]string
 	expectedStatus          *temporaliov1alpha1.TemporalWorkerDeploymentStatus
+	// Time to delay before checking expected status
+	waitTime *time.Duration
 }
 
 func (tc *TestCase) GetTWD() *temporaliov1alpha1.TemporalWorkerDeployment {
@@ -216,6 +229,10 @@ func (tc *TestCase) GetExpectedStatus() *temporaliov1alpha1.TemporalWorkerDeploy
 	return tc.expectedStatus
 }
 
+func (tc *TestCase) GetWaitTime() *time.Duration {
+	return tc.waitTime
+}
+
 // TestCaseBuilder provides a fluent interface for building test cases
 type TestCaseBuilder struct {
 	name              string
@@ -225,6 +242,7 @@ type TestCaseBuilder struct {
 	twdBuilder             *TemporalWorkerDeploymentBuilder
 	expectedStatusBuilder  *StatusBuilder
 	deprecatedVersionInfos []DeprecatedVersionInfo
+	waitTime               *time.Duration
 }
 
 // NewTestCase creates a new test case builder
@@ -255,6 +273,13 @@ func (tcb *TestCaseBuilder) WithInput(twdBuilder *TemporalWorkerDeploymentBuilde
 	return tcb
 }
 
+// WithWaitTime sets the wait time. Use this if you are expecting no change to the initial status and want to ensure
+// that after some time, there is still no change.
+func (tcb *TestCaseBuilder) WithWaitTime(waitTime time.Duration) *TestCaseBuilder {
+	tcb.waitTime = &waitTime
+	return tcb
+}
+
 // DeprecatedVersionInfo defines the necessary information about a deprecated worker version, so that
 // tests can recreate state that is not visible in the TemporalWorkerDeployment status
 type DeprecatedVersionInfo struct {
@@ -284,6 +309,7 @@ func (tcb *TestCaseBuilder) WithExpectedStatus(statusBuilder *StatusBuilder) *Te
 // Build returns the constructed test case
 func (tcb *TestCaseBuilder) Build() TestCase {
 	ret := TestCase{
+		waitTime: tcb.waitTime,
 		twd: tcb.twdBuilder.
 			WithName(tcb.name).
 			WithNamespace(tcb.k8sNamespace).