TemporalConnection update propagation (#102)

<!--- 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
- Handle updates to TemporalConnection, for TWD's that are associated
with one, so that these updates are propagated deployments.
- This was achieved by adding the hashed contents of a
TemporalConnection as pod annotations. Thus, if a TemporalConnection
were to get updated, it would result in new deployments with the updated
annotation. The new updated secret would also be mounted so that the
workers starting can be started with the latest secrets.

## Why?
- core functionality

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

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

4. How was this tested:
- Added unit tests and also tested this locally. The only thing I was
not able to do was add an integration test since I was not able to
configure our temporaltest server using mTLS.

Here's how I tested this locally:
- I ran a worker pod which was using the `temporal-cloud-mtls` secret to
connect to Temporal.
- I updated the namespace the worker was connected to by changing the
mTLS secret. This resulted in the controller and the worker pods
throwing connection errors (expected)
- I updated the local secret in the local k8s cluster (and also changed
it's name to `temporal-cloud-mtls-1` since we want our controller to
also refresh the client it's using)
- Noticed that a rolling deployment was conducted and everything
stabalized.

Proof of a rolling update:
<img width="1056" height="58" alt="Screenshot 2025-08-07 at 5 41 34 PM"
src="https://github.com/user-attachments/assets/40720118-0bb5-4f85-aa80-2c7c257b65f2"
/>



5. Any docs updates needed?
- None

Author: Shivs11

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")
+	})
+}