TemporalConnection update propagation

internal/controller/clientpool/clientpool.go

@@ -22,8 +22,9 @@ import (
 )
 
 type ClientPoolKey struct {
-	HostPort  string
-	Namespace string
+	HostPort        string
+	Namespace       string
+	MutualTLSSecret string // Include secret name in key to invalidate cache when the secret name changes
 }
 
 type ClientInfo struct {
@@ -142,8 +143,9 @@ func (cp *ClientPool) UpsertClient(ctx context.Context, opts NewClientOptions) (
 	defer cp.mux.Unlock()
 
 	key := ClientPoolKey{
-		HostPort:  opts.Spec.HostPort,
-		Namespace: opts.TemporalNamespace,
+		HostPort:        opts.Spec.HostPort,
+		Namespace:       opts.TemporalNamespace,
+		MutualTLSSecret: opts.Spec.MutualTLSSecret,
 	}
 	cp.clients[key] = ClientInfo{
 		client:     c,

internal/controller/execplan.go

@@ -54,6 +54,15 @@ func (r *TemporalWorkerDeploymentReconciler) executePlan(ctx context.Context, l
 		}
 	}
 
+	// Update deployments
+	for _, d := range p.UpdateDeployments {
+		l.Info("updating deployment", "deployment", d.Name, "namespace", d.Namespace)
+		if err := r.Update(ctx, d); err != nil {
+			l.Error(err, "unable to update deployment", "deployment", d)
+			return fmt.Errorf("unable to update deployment: %w", err)
+		}
+	}
+
 	// Get deployment handler
 	deploymentHandler := temporalClient.WorkerDeploymentClient().GetHandle(p.WorkerDeploymentName)
 

internal/controller/genplan.go

@@ -27,6 +27,7 @@ type plan struct {
 	DeleteDeployments []*appsv1.Deployment
 	CreateDeployment  *appsv1.Deployment
 	ScaleDeployments  map[*corev1.ObjectReference]uint32
+	UpdateDeployments []*appsv1.Deployment
 	// Register new versions as current or with ramp
 	UpdateVersionConfig *planner.VersionConfig
 
@@ -90,6 +91,7 @@ func (r *TemporalWorkerDeploymentReconciler) generatePlan(
 		&w.Status,
 		&w.Spec,
 		temporalState,
+		connection,
 		plannerConfig,
 	)
 	if err != nil {
@@ -99,6 +101,7 @@ func (r *TemporalWorkerDeploymentReconciler) generatePlan(
 	// Convert planner result to controller plan
 	plan.DeleteDeployments = planResult.DeleteDeployments
 	plan.ScaleDeployments = planResult.ScaleDeployments
+	plan.UpdateDeployments = planResult.UpdateDeployments
 
 	// Convert version config
 	plan.UpdateVersionConfig = planResult.VersionConfig

internal/controller/worker_controller.go

@@ -21,7 +21,9 @@ import (
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 )
 
 var (
@@ -58,7 +60,6 @@ type TemporalWorkerDeploymentReconciler struct {
 //
 // For more details, check Reconcile and its Result here:
 // - https://pkg.go.dev/sigs.k8s.io/[email protected]/pkg/reconcile
-// TODO(carlydf): Add watching of temporal connection custom resource (may have issue)
 func (r *TemporalWorkerDeploymentReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
 	// TODO(Shivam): Monitor if the time taken for a successful reconciliation loop is closing in on 5 minutes. If so, we
 	// may need to increase the timeout value.
@@ -112,8 +113,9 @@ func (r *TemporalWorkerDeploymentReconciler) Reconcile(ctx context.Context, req
 
 	// Get or update temporal client for connection
 	temporalClient, ok := r.TemporalClientPool.GetSDKClient(clientpool.ClientPoolKey{
-		HostPort:  temporalConnection.Spec.HostPort,
-		Namespace: workerDeploy.Spec.WorkerOptions.TemporalNamespace,
+		HostPort:        temporalConnection.Spec.HostPort,
+		Namespace:       workerDeploy.Spec.WorkerOptions.TemporalNamespace,
+		MutualTLSSecret: temporalConnection.Spec.MutualTLSSecret,
 	}, temporalConnection.Spec.MutualTLSSecret != "")
 	if !ok {
 		c, err := r.TemporalClientPool.UpsertClient(ctx, clientpool.NewClientOptions{
@@ -212,9 +214,35 @@ func (r *TemporalWorkerDeploymentReconciler) SetupWithManager(mgr ctrl.Manager)
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&temporaliov1alpha1.TemporalWorkerDeployment{}).
 		Owns(&appsv1.Deployment{}).
+		Watches(&temporaliov1alpha1.TemporalConnection{}, handler.EnqueueRequestsFromMapFunc(r.findTWDsUsingConnection)).
 		WithOptions(controller.Options{
 			MaxConcurrentReconciles: 100,
 			RecoverPanic:            &recoverPanic,
 		}).
 		Complete(r)
 }
+
+func (r *TemporalWorkerDeploymentReconciler) findTWDsUsingConnection(ctx context.Context, tc client.Object) []reconcile.Request {
+	var requests []reconcile.Request
+
+	// Find all TWDs in same namespace that reference this TC
+	var twds temporaliov1alpha1.TemporalWorkerDeploymentList
+	if err := r.List(ctx, &twds, client.InNamespace(tc.GetNamespace())); err != nil {
+		return requests
+	}
+
+	// Filter to ones using this connection
+	for _, twd := range twds.Items {
+		if twd.Spec.WorkerOptions.TemporalConnection == tc.GetName() {
+			// Enqueue a reconcile request for this TWD
+			requests = append(requests, reconcile.Request{
+				NamespacedName: types.NamespacedName{
+					Name:      twd.Name,
+					Namespace: twd.Namespace,
+				},
+			})
+		}
+	}
+
+	return requests
+}

internal/demo/README.md

@@ -79,6 +79,9 @@ minikube kubectl -- get pods -n temporal-worker-controller -w
 # Describe the controller pod's status
 minikube kubectl -- describe pod <pod-name> -n temporal-worker-controller
 
+# Output the controller pod's logs
+minikube kubectl -- logs -n temporal-system -f pod/<pod-name>
+
 # View TemporalWorkerDeployment status
 kubectl get twd
 ```

internal/demo/helloworld/temporal_worker_deployment.yaml

@@ -18,11 +18,11 @@ spec:
     steps:
       # Increase traffic from 1% to 10% over 15 seconds
       - rampPercentage: 1
-        pauseDuration: 5s
+        pauseDuration: 30s
       - rampPercentage: 5
-        pauseDuration: 5s
+        pauseDuration: 30s
       - rampPercentage: 10
-        pauseDuration: 5s
+        pauseDuration: 30s
       # Increase traffic to 50% and wait 1 minute
       - rampPercentage: 50
         pauseDuration: 1m

internal/k8s/deployments.go

@@ -6,6 +6,8 @@ package k8s
 
 import (
 	"context"
+	"crypto/sha256"
+	"encoding/hex"
 	"fmt"
 	"regexp"
 	"sort"
@@ -25,11 +27,12 @@ import (
 const (
 	DeployOwnerKey = ".metadata.controller"
 	// BuildIDLabel is the label that identifies the build ID for a deployment
-	BuildIDLabel             = "temporal.io/build-id"
-	DeploymentNameSeparator  = "/" // TODO(carlydf): change this to "." once the server accepts `.` in deployment names
-	VersionIDSeparator       = "." // TODO(carlydf): change this to ":"
-	K8sResourceNameSeparator = "-"
-	MaxBuildIdLen            = 63
+	BuildIDLabel                 = "temporal.io/build-id"
+	DeploymentNameSeparator      = "/" // TODO(carlydf): change this to "." once the server accepts `.` in deployment names
+	VersionIDSeparator           = "." // TODO(carlydf): change this to ":"
+	K8sResourceNameSeparator     = "-"
+	MaxBuildIdLen                = 63
+	ConnectionSpecHashAnnotation = "temporal.io/connection-spec-hash"
 )
 
 // DeploymentState represents the Kubernetes state of all deployments for a temporal worker deployment
@@ -256,6 +259,12 @@ func NewDeploymentWithOwnerRef(
 		})
 	}
 
+	// Build pod annotations
+	podAnnotations := make(map[string]string)
+	for k, v := range spec.Template.Annotations {
+		podAnnotations[k] = v
+	}
+	podAnnotations[ConnectionSpecHashAnnotation] = ComputeConnectionSpecHash(connection)
 	blockOwnerDeletion := true
 
 	return &appsv1.Deployment{
@@ -284,7 +293,7 @@ func NewDeploymentWithOwnerRef(
 			Template: corev1.PodTemplateSpec{
 				ObjectMeta: metav1.ObjectMeta{
 					Labels:      podLabels,
-					Annotations: spec.Template.Annotations,
+					Annotations: podAnnotations,
 				},
 				Spec: *podSpec,
 			},
@@ -293,6 +302,21 @@ func NewDeploymentWithOwnerRef(
 	}
 }
 
+func ComputeConnectionSpecHash(connection temporaliov1alpha1.TemporalConnectionSpec) string {
+	// HostPort is required, but MutualTLSSecret can be empty for non-mTLS connections
+	if connection.HostPort == "" {
+		return ""
+	}
+
+	hasher := sha256.New()
+
+	// Hash connection spec fields in deterministic order
+	_, _ = hasher.Write([]byte(connection.HostPort))
+	_, _ = hasher.Write([]byte(connection.MutualTLSSecret))
+
+	return hex.EncodeToString(hasher.Sum(nil))
+}
+
 func NewDeploymentWithControllerRef(
 	w *temporaliov1alpha1.TemporalWorkerDeployment,
 	buildID string,

internal/k8s/deployments_test.go

@@ -451,3 +451,109 @@ func TestComputeWorkerDeploymentName_Integration_WithVersionedName(t *testing.T)
 	assert.Equal(t, expectedVersionID, versionID)
 	assert.Equal(t, "hello-world"+k8s.DeploymentNameSeparator+"demo"+k8s.VersionIDSeparator+"v1-0-0-dd84", versionID)
 }
+
+// TestNewDeploymentWithPodAnnotations tests that every new pod created has a connection spec hash annotation
+func TestNewDeploymentWithPodAnnotations(t *testing.T) {
+	connection := temporaliov1alpha1.TemporalConnectionSpec{
+		HostPort:        "localhost:7233",
+		MutualTLSSecret: "my-secret",
+	}
+
+	deployment := k8s.NewDeploymentWithOwnerRef(
+		&metav1.TypeMeta{},
+		&metav1.ObjectMeta{Name: "test", Namespace: "default"},
+		&temporaliov1alpha1.TemporalWorkerDeploymentSpec{},
+		"test-deployment",
+		"build123",
+		connection,
+	)
+
+	expectedHash := k8s.ComputeConnectionSpecHash(connection)
+	actualHash := deployment.Spec.Template.Annotations[k8s.ConnectionSpecHashAnnotation]
+
+	assert.Equal(t, expectedHash, actualHash, "Deployment should have correct connection spec hash annotation")
+}
+
+func TestComputeConnectionSpecHash(t *testing.T) {
+	t.Run("generates non-empty hash for valid connection spec", func(t *testing.T) {
+		spec := temporaliov1alpha1.TemporalConnectionSpec{
+			HostPort:        "localhost:7233",
+			MutualTLSSecret: "my-tls-secret",
+		}
+
+		result := k8s.ComputeConnectionSpecHash(spec)
+		assert.NotEmpty(t, result, "Hash should not be empty for valid spec")
+		assert.Len(t, result, 64, "SHA256 hash should be 64 characters") // hex encoded SHA256
+	})
+
+	t.Run("returns empty hash when hostport is empty", func(t *testing.T) {
+		spec := temporaliov1alpha1.TemporalConnectionSpec{
+			HostPort:        "",
+			MutualTLSSecret: "secret",
+		}
+
+		result := k8s.ComputeConnectionSpecHash(spec)
+		assert.Empty(t, result, "Hash should be empty when hostport is empty")
+	})
+
+	t.Run("is deterministic - same input produces same hash", func(t *testing.T) {
+		spec := temporaliov1alpha1.TemporalConnectionSpec{
+			HostPort:        "localhost:7233",
+			MutualTLSSecret: "my-secret",
+		}
+
+		hash1 := k8s.ComputeConnectionSpecHash(spec)
+		hash2 := k8s.ComputeConnectionSpecHash(spec)
+
+		assert.Equal(t, hash1, hash2, "Same input should produce identical hashes")
+	})
+
+	t.Run("different hostports produce different hashes", func(t *testing.T) {
+		spec1 := temporaliov1alpha1.TemporalConnectionSpec{
+			HostPort:        "localhost:7233",
+			MutualTLSSecret: "same-secret",
+		}
+		spec2 := temporaliov1alpha1.TemporalConnectionSpec{
+			HostPort:        "different-host:7233",
+			MutualTLSSecret: "same-secret",
+		}
+
+		hash1 := k8s.ComputeConnectionSpecHash(spec1)
+		hash2 := k8s.ComputeConnectionSpecHash(spec2)
+
+		assert.NotEqual(t, hash1, hash2, "Different hostports should produce different hashes")
+	})
+
+	t.Run("different mTLS secrets produce different hashes", func(t *testing.T) {
+		spec1 := temporaliov1alpha1.TemporalConnectionSpec{
+			HostPort:        "localhost:7233",
+			MutualTLSSecret: "secret1",
+		}
+		spec2 := temporaliov1alpha1.TemporalConnectionSpec{
+			HostPort:        "localhost:7233",
+			MutualTLSSecret: "secret2",
+		}
+
+		hash1 := k8s.ComputeConnectionSpecHash(spec1)
+		hash2 := k8s.ComputeConnectionSpecHash(spec2)
+
+		assert.NotEqual(t, hash1, hash2, "Different mTLS secrets should produce different hashes")
+	})
+
+	t.Run("empty mTLS secret vs non-empty produce different hashes", func(t *testing.T) {
+		spec1 := temporaliov1alpha1.TemporalConnectionSpec{
+			HostPort:        "localhost:7233",
+			MutualTLSSecret: "",
+		}
+		spec2 := temporaliov1alpha1.TemporalConnectionSpec{
+			HostPort:        "localhost:7233",
+			MutualTLSSecret: "some-secret",
+		}
+
+		hash1 := k8s.ComputeConnectionSpecHash(spec1)
+		hash2 := k8s.ComputeConnectionSpecHash(spec2)
+
+		assert.NotEqual(t, hash1, hash2, "Empty vs non-empty mTLS secret should produce different hashes")
+		assert.NotEmpty(t, hash1, "Hash should still be generated even with empty mTLS secret")
+	})
+}