start work to create preliminary state

Author: carlydf

internal/controller/util.go

@@ -8,6 +8,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"github.com/temporalio/temporal-worker-controller/internal/k8s"
 	"os"
 	"strings"
 	"time"
@@ -61,7 +62,7 @@ func awaitVersionRegistrationInDeployment(
 	l logr.Logger,
 	deploymentHandler sdkclient.WorkerDeploymentHandle,
 	namespace, versionID string) error {
-	deploymentName, _, _ := strings.Cut(versionID, ".")
+	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))

internal/demo/util/worker.go

@@ -5,6 +5,7 @@
 package util
 
 import (
+	"github.com/temporalio/temporal-worker-controller/internal/k8s"
 	"net/http"
 	"time"
 
@@ -28,7 +29,7 @@ func NewVersionedWorker(opts worker.Options) (w worker.Worker, stopFunc func())
 
 	opts.DeploymentOptions = worker.DeploymentOptions{
 		UseVersioning:             true,
-		Version:                   mustGetEnv("TEMPORAL_DEPLOYMENT_NAME") + "." + mustGetEnv("WORKER_BUILD_ID"),
+		Version:                   mustGetEnv("TEMPORAL_DEPLOYMENT_NAME") + k8s.VersionIDSeparator + mustGetEnv("WORKER_BUILD_ID"),
 		DefaultVersioningBehavior: workflow.VersioningBehaviorPinned,
 	}
 

internal/k8s/deployments.go

@@ -25,7 +25,7 @@ const (
 	// BuildIDLabel is the label that identifies the build ID for a deployment
 	BuildIDLabel             = "temporal.io/build-id"
 	deploymentNameSeparator  = "/"
-	versionIDSeparator       = "."
+	VersionIDSeparator       = "." // TODO(carlydf): change this to ":"
 	K8sResourceNameSeparator = "-"
 	MaxBuildIdLen            = 63
 )
@@ -75,7 +75,7 @@ func GetDeploymentState(
 	for i := range childDeploys.Items {
 		deploy := &childDeploys.Items[i]
 		if buildID, ok := deploy.GetLabels()[BuildIDLabel]; ok {
-			versionID := workerDeploymentName + "." + buildID
+			versionID := workerDeploymentName + VersionIDSeparator + buildID
 			state.Deployments[versionID] = deploy
 			state.DeploymentsByTime = append(state.DeploymentsByTime, deploy)
 			state.DeploymentRefs[versionID] = NewObjectRef(deploy)
@@ -110,7 +110,7 @@ func NewObjectRef(obj client.Object) *corev1.ObjectReference {
 
 // ComputeVersionID generates a version ID from the worker deployment spec
 func ComputeVersionID(w *temporaliov1alpha1.TemporalWorkerDeployment) string {
-	return ComputeWorkerDeploymentName(w) + versionIDSeparator + ComputeBuildID(w)
+	return ComputeWorkerDeploymentName(w) + VersionIDSeparator + ComputeBuildID(w)
 }
 
 func ComputeBuildID(w *temporaliov1alpha1.TemporalWorkerDeployment) string {
@@ -161,12 +161,12 @@ func CleanAndTruncateString(s string, n int) string {
 	}
 	// Keep only letters, numbers, and dashes
 	re := regexp.MustCompile(`[^a-zA-Z0-9-]+`)
-	return re.ReplaceAllString(s, "-")
+	return re.ReplaceAllString(s, K8sResourceNameSeparator)
 }
 
 // SplitVersionID splits a version ID into its components
 func SplitVersionID(versionID string) (deploymentName, buildID string, err error) {
-	parts := strings.Split(versionID, ".")
+	parts := strings.Split(versionID, VersionIDSeparator)
 	if len(parts) < 2 {
 		return "", "", fmt.Errorf("invalid version ID format: %s", versionID)
 	}

internal/testhelpers/make.go

@@ -6,6 +6,11 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	"time"
+)
+
+const (
+	testTaskQueue = "hello_world"
 )
 
 func MakeTWD(
@@ -70,7 +75,7 @@ func MakePodSpec(containers []corev1.Container, labels map[string]string, taskQu
 func MakeHelloWorldPodSpec(imageName string) corev1.PodTemplateSpec {
 	return MakePodSpec([]corev1.Container{{Name: "worker", Image: imageName}},
 		map[string]string{"app": "test-worker"},
-		"hello_world")
+		testTaskQueue)
 }
 
 func MakeTWDWithImage(imageName string) *temporaliov1alpha1.TemporalWorkerDeployment {
@@ -121,6 +126,37 @@ func MakeTWDWithName(name string) *temporaliov1alpha1.TemporalWorkerDeployment {
 	return twd
 }
 
+func MakeCurrentVersion(namespace, twdName, imageName string, healthy, createDeployment bool) *temporaliov1alpha1.CurrentWorkerDeploymentVersion {
+	ret := &temporaliov1alpha1.CurrentWorkerDeploymentVersion{
+		BaseWorkerDeploymentVersion: temporaliov1alpha1.BaseWorkerDeploymentVersion{
+			VersionID:    MakeVersionId(namespace, twdName, imageName),
+			Status:       temporaliov1alpha1.VersionStatusCurrent,
+			HealthySince: nil,
+			Deployment:   nil,
+			TaskQueues: []temporaliov1alpha1.TaskQueue{
+				{Name: testTaskQueue},
+			},
+			ManagedBy: "",
+		},
+	}
+
+	if healthy {
+		h := metav1.NewTime(time.Now())
+		ret.HealthySince = &h
+	}
+
+	if createDeployment {
+		ret.Deployment = &corev1.ObjectReference{
+			Namespace: namespace,
+			Name: k8s.ComputeVersionedDeploymentName(
+				twdName,
+				MakeBuildId(twdName, imageName, nil),
+			),
+		}
+	}
+	return ret
+}
+
 func ModifyObj[T any](obj T, callback func(obj T) T) T {
 	return callback(obj)
 }

internal/tests/internal/env_helpers.go

@@ -5,11 +5,13 @@ package internal
 
 import (
 	"context"
+	"fmt"
 	"log/slog"
 	"os"
 	"os/exec"
 	"path/filepath"
 	"runtime"
+	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -38,20 +40,28 @@ func setupKubebuilderAssets() error {
 		return nil // Already set
 	}
 
-	// Try to find the assets using setup-envtest
-	cmd := exec.Command("setup-envtest", "use", "1.28.0", "--bin-dir", "bin")
+	// Get the repository root to find the setup-envtest binary
+	_, currentFile, _, ok := runtime.Caller(0)
+	if !ok {
+		return fmt.Errorf("failed to get current file path")
+	}
+	repoRoot, err := filepath.Abs(filepath.Join(filepath.Dir(currentFile), "../../.."))
+	if err != nil {
+		return fmt.Errorf("failed to get repository root: %v", err)
+	}
+
+	// Use the correct version and path that matches the Makefile
+	setupEnvtestPath := filepath.Join(repoRoot, "bin", "setup-envtest")
+	binDir := filepath.Join(repoRoot, "bin")
+	cmd := exec.Command(setupEnvtestPath, "use", "1.27.1", "--bin-dir", binDir, "-p", "path")
 	output, err := cmd.Output()
 	if err != nil {
-		return err
+		return fmt.Errorf("failed to run setup-envtest: %v", err)
 	}
 
-	// Parse the output to get the path
-	// The output format is typically: "export KUBEBUILDER_ASSETS=/path/to/assets"
-	// We need to extract just the path
-	assetsPath := string(output)
+	// The output with -p path flag is just the path, no need to parse
+	assetsPath := strings.TrimSpace(string(output))
 	if len(assetsPath) > 0 {
-		// Remove any trailing newlines
-		assetsPath = assetsPath[:len(assetsPath)-1]
 		os.Setenv("KUBEBUILDER_ASSETS", assetsPath)
 	}
 

internal/tests/internal/integration_test.go

@@ -33,6 +33,7 @@ type testCase struct {
 	// versions, so for test scenarios that start with existing deprecated version Deployments,
 	// specify the number of replicas for each deprecated build here.
 	deprecatedBuildReplicas map[string]int32
+	deprecatedBuildImages   map[string]string
 	expectedStatus          *temporaliov1alpha1.TemporalWorkerDeploymentStatus
 }
 
@@ -87,19 +88,8 @@ func TestIntegration(t *testing.T) {
 			}),
 			deprecatedBuildReplicas: nil,
 			expectedStatus: &temporaliov1alpha1.TemporalWorkerDeploymentStatus{
-				TargetVersion: nil,
-				CurrentVersion: &temporaliov1alpha1.CurrentWorkerDeploymentVersion{
-					BaseWorkerDeploymentVersion: temporaliov1alpha1.BaseWorkerDeploymentVersion{
-						VersionID: testhelpers.MakeVersionId(testNamespace.Name, "all-at-once-rollout-2-replicas", "v1"),
-						Deployment: &corev1.ObjectReference{
-							Namespace: testNamespace.Name,
-							Name: k8s.ComputeVersionedDeploymentName(
-								"all-at-once-rollout-2-replicas",
-								testhelpers.MakeBuildId("all-at-once-rollout-2-replicas", "v1", nil),
-							),
-						},
-					},
-				},
+				TargetVersion:        nil,
+				CurrentVersion:       testhelpers.MakeCurrentVersion(testNamespace.Name, "all-at-once-rollout-2-replicas", "v1", true, false),
 				RampingVersion:       nil,
 				DeprecatedVersions:   nil,
 				VersionConflictToken: nil,
@@ -122,9 +112,18 @@ func TestIntegration(t *testing.T) {
 					Namespace: testNamespace.Name,
 					Labels:    map[string]string{"app": "test-worker"},
 				}
+				obj.Status = temporaliov1alpha1.TemporalWorkerDeploymentStatus{
+					TargetVersion:        nil,
+					CurrentVersion:       testhelpers.MakeCurrentVersion(testNamespace.Name, "progressive-rollout-expect-first-step", "v0", true, true),
+					RampingVersion:       nil,
+					DeprecatedVersions:   nil,
+					VersionConflictToken: nil,
+					LastModifierIdentity: "",
+				}
 				return obj
 			}),
-			deprecatedBuildReplicas: nil,
+			deprecatedBuildReplicas: map[string]int32{testhelpers.MakeBuildId("progressive-rollout-expect-first-step", "v0", nil): 1},
+			deprecatedBuildImages:   map[string]string{testhelpers.MakeBuildId("progressive-rollout-expect-first-step", "v0", nil): "v0"},
 			expectedStatus: &temporaliov1alpha1.TemporalWorkerDeploymentStatus{
 				TargetVersion: &temporaliov1alpha1.TargetWorkerDeploymentVersion{
 					BaseWorkerDeploymentVersion: temporaliov1alpha1.BaseWorkerDeploymentVersion{
@@ -168,20 +167,129 @@ func TestIntegration(t *testing.T) {
 
 	for testName, tc := range tests {
 		t.Run(testName, func(t *testing.T) {
-			// TODO(carlydf): create starting test env
-			// - Use input.Status + deprecatedBuildReplicas to create (and maybe kill) pollers for deprecated versions in temporal
-			// - also get routing config of the deployment into the starting state before running the test
-
-			testTemporalWorkerDeploymentCreation(t, k8sClient, ts, tc.input, tc.expectedStatus)
+			ctx := context.Background()
+			// TODO(carlydf): populate all fields in tc that are set to testName, so that the user does not need to specify
+			testTemporalWorkerDeploymentCreation(ctx, t, k8sClient, ts, tc)
 		})
 
 	}
 
 }
 
+// Uses input.Status + deprecatedBuildReplicas to create (and maybe kill) pollers for deprecated versions in temporal
+// also gets routing config of the deployment into the starting state before running the test.
+// Does not set Status.VersionConflictToken, since that is only set internally by the server.
+func makePreliminaryStatusTrue(
+	ctx context.Context,
+	t *testing.T,
+	k8sClient client.Client,
+	ts *temporaltest.TestServer,
+	twd *temporaliov1alpha1.TemporalWorkerDeployment,
+	connection *temporaliov1alpha1.TemporalConnection,
+	replicas map[string]int32,
+	images map[string]string,
+) {
+	t.Logf("Creating starting test env based on input.Status")
+	for _, dv := range twd.Status.DeprecatedVersions {
+		t.Logf("Handling deprecated version %v", dv.VersionID)
+		switch dv.Status {
+		case temporaliov1alpha1.VersionStatusInactive:
+			// start a poller -- is this included in deprecated versions list?
+		case temporaliov1alpha1.VersionStatusRamping, temporaliov1alpha1.VersionStatusCurrent:
+			// these won't be in deprecated versions
+		case temporaliov1alpha1.VersionStatusDraining:
+			// TODO(carlydf): start a poller, set ramp, start a wf on that version, then unset
+		case temporaliov1alpha1.VersionStatusDrained:
+			// TODO(carlydf): start a poller, set ramp, unset, wait for drainage status visibility grace period
+		case temporaliov1alpha1.VersionStatusNotRegistered:
+			// no-op, although I think this won't occur in deprecated versions either
+		}
+	}
+	// TODO(carlydf): handle Status.LastModifierIdentity
+	if cv := twd.Status.CurrentVersion; cv != nil {
+		t.Logf("Handling current version %v", cv.VersionID)
+		if cv.Status != temporaliov1alpha1.VersionStatusCurrent {
+			t.Errorf("Current Version's status must be Current")
+		}
+		if cv.Deployment != nil {
+			t.Logf("Creating Deployment %s for Current Version", cv.Deployment.Name)
+			createWorkerDeployment(ctx, t, k8sClient, twd, cv.VersionID, connection.Spec, replicas, images)
+			expectedDeploymentName := k8s.ComputeVersionedDeploymentName(twd.Name, k8s.ComputeBuildID(twd))
+			waitForDeployment(t, k8sClient, expectedDeploymentName, twd.Namespace, 30*time.Second)
+			workerStopFuncs := applyDeployment(t, ctx, k8sClient, expectedDeploymentName, twd.Namespace)
+			defer func() {
+				for _, f := range workerStopFuncs {
+					if f != nil {
+						f()
+					}
+				}
+			}()
+		}
+	}
+
+	if rv := twd.Status.RampingVersion; rv != nil {
+		t.Logf("Handling ramping version %v", rv.VersionID)
+		if rv.Status != temporaliov1alpha1.VersionStatusRamping {
+			t.Errorf("Ramping Version's status must be Ramping")
+		}
+		if rv.Deployment != nil {
+			t.Logf("Creating Deployment %s for Ramping Version", rv.Deployment.Name)
+		}
+		// TODO(carlydf): do this
+	}
+}
+
+func createWorkerDeployment(
+	ctx context.Context,
+	t *testing.T,
+	k8sClient client.Client,
+	twd *temporaliov1alpha1.TemporalWorkerDeployment,
+	versionID string,
+	connection temporaliov1alpha1.TemporalConnectionSpec,
+	replicas map[string]int32,
+	images map[string]string,
+) {
+	t.Log("Creating a Deployment")
+	_, buildId, err := k8s.SplitVersionID(versionID)
+	if err != nil {
+		t.Error(err)
+	}
+
+	prevImageName := twd.Spec.Template.Spec.Containers[0].Image
+	prevReplicas := twd.Spec.Replicas
+	// temporarily replace it
+	twd.Spec.Template.Spec.Containers[0].Image = images[buildId]
+	newReplicas := replicas[buildId]
+	twd.Spec.Replicas = &newReplicas
+	defer func() {
+		twd.Spec.Template.Spec.Containers[0].Image = prevImageName
+		twd.Spec.Replicas = prevReplicas
+	}()
+
+	dep := k8s.NewDeploymentWithOwnerRef(
+		&twd.TypeMeta,
+		&twd.ObjectMeta,
+		&twd.Spec,
+		k8s.ComputeWorkerDeploymentName(twd),
+		buildId,
+		connection,
+	)
+
+	if err := k8sClient.Create(ctx, dep); err != nil {
+		t.Fatalf("failed to create Deployment: %v", err)
+	}
+}
+
 // testTemporalWorkerDeploymentCreation tests the creation of a TemporalWorkerDeployment and waits for the expected status
-func testTemporalWorkerDeploymentCreation(t *testing.T, k8sClient client.Client, ts *temporaltest.TestServer, twd *temporaliov1alpha1.TemporalWorkerDeployment, expectedStatus *temporaliov1alpha1.TemporalWorkerDeploymentStatus) {
-	ctx := context.Background()
+func testTemporalWorkerDeploymentCreation(
+	ctx context.Context,
+	t *testing.T,
+	k8sClient client.Client,
+	ts *temporaltest.TestServer,
+	tc testCase,
+) {
+	twd := tc.input
+	expectedStatus := tc.expectedStatus
 
 	t.Log("Creating a TemporalConnection")
 	temporalConnection := &temporaliov1alpha1.TemporalConnection{
@@ -197,6 +305,8 @@ func testTemporalWorkerDeploymentCreation(t *testing.T, k8sClient client.Client,
 		t.Fatalf("failed to create TemporalConnection: %v", err)
 	}
 
+	makePreliminaryStatusTrue(ctx, t, k8sClient, ts, twd, temporalConnection, tc.deprecatedBuildReplicas, tc.deprecatedBuildImages)
+
 	t.Log("Creating a TemporalWorkerDeployment")
 	if err := k8sClient.Create(ctx, twd); err != nil {
 		t.Fatalf("failed to create TemporalWorkerDeployment: %v", err)