From 22db854e36e09a2ea2a9c093f04b1183700ee3ad Mon Sep 17 00:00:00 2001 From: Niyomukiza Mechack Date: Mon, 24 Aug 2026 12:57:31 -0700 Subject: [PATCH 1/5] FEATURE: Add ClusterConnection cluster-scoped resource --- api/v1alpha1/clusterconnection_types.go | 36 ++ api/v1alpha1/workerdeployment_types.go | 19 +- api/v1alpha1/zz_generated.deepcopy.go | 59 +++ .../temporal.io_clusterconnections.yaml | 80 ++++ .../temporal.io_workerdeployments.yaml | 6 + .../templates/rbac.yaml | 4 + internal/controller/clusterconnection_test.go | 424 ++++++++++++++++++ internal/controller/worker_controller.go | 158 +++++-- 8 files changed, 738 insertions(+), 48 deletions(-) create mode 100644 api/v1alpha1/clusterconnection_types.go create mode 100644 helm/temporal-worker-controller-crds/templates/temporal.io_clusterconnections.yaml create mode 100644 internal/controller/clusterconnection_test.go diff --git a/api/v1alpha1/clusterconnection_types.go b/api/v1alpha1/clusterconnection_types.go new file mode 100644 index 00000000..1868a540 --- /dev/null +++ b/api/v1alpha1/clusterconnection_types.go @@ -0,0 +1,36 @@ +// Unless explicitly stated otherwise all files in this repository are licensed under the MIT License. +// +// This product includes software developed at Datadog (https://www.datadoghq.com/). Copyright 2024 Datadog, Inc. + +package v1alpha1 + +import metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + +//+kubebuilder:object:root=true +//+kubebuilder:resource:scope=Cluster +//+kubebuilder:subresource:status +//+kubebuilder:printcolumn:name="Host",type="string",JSONPath=".spec.hostPort",description="Temporal server endpoint" +//+kubebuilder:printcolumn:name="Age",type="date",JSONPath=".metadata.creationTimestamp",description="Age" + +// ClusterConnection is the cluster-scoped analog of Connection: +// a single ClusterConnection can be referenced by WorkerDeployments in any namespace. +type ClusterConnection struct { + metav1.TypeMeta `json:",inline"` + metav1.ObjectMeta `json:"metadata,omitempty"` + + Spec ConnectionSpec `json:"spec,omitempty"` + Status ConnectionStatus `json:"status,omitempty"` +} + +//+kubebuilder:object:root=true + +// ClusterConnectionList contains a list of ClusterConnection +type ClusterConnectionList struct { + metav1.TypeMeta `json:",inline"` + metav1.ListMeta `json:"metadata,omitempty"` + Items []ClusterConnection `json:"items"` +} + +func init() { + SchemeBuilder.Register(&ClusterConnection{}, &ClusterConnectionList{}) +} diff --git a/api/v1alpha1/workerdeployment_types.go b/api/v1alpha1/workerdeployment_types.go index f0aaf615..923544ed 100644 --- a/api/v1alpha1/workerdeployment_types.go +++ b/api/v1alpha1/workerdeployment_types.go @@ -12,17 +12,30 @@ import ( // NOTE: json tags are required. Any new fields you add must have json tags for the fields to be serialized. -// ConnectionReference contains the name of a Connection resource -// in the same namespace as the WorkerDeployment. +// ConnectionReference identifies a connection resource to use. By default it +// refers to a namespaced Connection in the same namespace as the +// WorkerDeployment. When Kind is "ClusterConnection", it refers to a +// cluster-scoped ClusterConnection instead; The authentication +// secret is still resolved from the WorkerDeployment's own namespace. type ConnectionReference struct { // Name of the Connection resource. // +kubebuilder:validation:Required // +kubebuilder:validation:Pattern=`^[a-z0-9]([-a-z0-9]*[a-z0-9])?$` Name string `json:"name"` + // Kind is the type of connection resource. + // ClusterConnection selects a cluster-scoped ClusterConnection. + // Omitting this field preserves the pre-existing behavior of referencing a + // namespaced Connection. + // +optional + // +kubebuilder:default=Connection + // +kubebuilder:validation:Enum=Connection;ClusterConnection + Kind string `json:"kind,omitempty"` } type WorkerOptions struct { - // The name of a Connection in the same namespace as the WorkerDeployment. + // ConnectionRef selects the connection resource for this worker. By default + // it names a Connection in the same namespace; set connectionRef.kind to + // "ClusterConnection" to reference a cluster-scoped ClusterConnection. ConnectionRef ConnectionReference `json:"connectionRef"` // The Temporal namespace for the worker to connect to. // +kubebuilder:validation:MinLength=1 diff --git a/api/v1alpha1/zz_generated.deepcopy.go b/api/v1alpha1/zz_generated.deepcopy.go index 056a0b95..109324af 100644 --- a/api/v1alpha1/zz_generated.deepcopy.go +++ b/api/v1alpha1/zz_generated.deepcopy.go @@ -55,6 +55,65 @@ func (in *BaseWorkerDeploymentVersion) DeepCopy() *BaseWorkerDeploymentVersion { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *ClusterConnection) DeepCopyInto(out *ClusterConnection) { + *out = *in + out.TypeMeta = in.TypeMeta + in.ObjectMeta.DeepCopyInto(&out.ObjectMeta) + in.Spec.DeepCopyInto(&out.Spec) + out.Status = in.Status +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ClusterConnection. +func (in *ClusterConnection) DeepCopy() *ClusterConnection { + if in == nil { + return nil + } + out := new(ClusterConnection) + in.DeepCopyInto(out) + return out +} + +// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object. +func (in *ClusterConnection) DeepCopyObject() runtime.Object { + if c := in.DeepCopy(); c != nil { + return c + } + return nil +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *ClusterConnectionList) DeepCopyInto(out *ClusterConnectionList) { + *out = *in + out.TypeMeta = in.TypeMeta + in.ListMeta.DeepCopyInto(&out.ListMeta) + if in.Items != nil { + in, out := &in.Items, &out.Items + *out = make([]ClusterConnection, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ClusterConnectionList. +func (in *ClusterConnectionList) DeepCopy() *ClusterConnectionList { + if in == nil { + return nil + } + out := new(ClusterConnectionList) + in.DeepCopyInto(out) + return out +} + +// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object. +func (in *ClusterConnectionList) DeepCopyObject() runtime.Object { + if c := in.DeepCopy(); c != nil { + return c + } + return nil +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *Connection) DeepCopyInto(out *Connection) { *out = *in diff --git a/helm/temporal-worker-controller-crds/templates/temporal.io_clusterconnections.yaml b/helm/temporal-worker-controller-crds/templates/temporal.io_clusterconnections.yaml new file mode 100644 index 00000000..e9f0ce9f --- /dev/null +++ b/helm/temporal-worker-controller-crds/templates/temporal.io_clusterconnections.yaml @@ -0,0 +1,80 @@ +--- +apiVersion: apiextensions.k8s.io/v1 +kind: CustomResourceDefinition +metadata: + annotations: + controller-gen.kubebuilder.io/version: v0.19.0 + name: clusterconnections.temporal.io +spec: + group: temporal.io + names: + kind: ClusterConnection + listKind: ClusterConnectionList + plural: clusterconnections + singular: clusterconnection + scope: Cluster + versions: + - additionalPrinterColumns: + - description: Temporal server endpoint + jsonPath: .spec.hostPort + name: Host + type: string + - description: Age + jsonPath: .metadata.creationTimestamp + name: Age + type: date + name: v1alpha1 + schema: + openAPIV3Schema: + properties: + apiVersion: + type: string + kind: + type: string + metadata: + type: object + spec: + properties: + apiKeySecretRef: + properties: + key: + type: string + name: + default: "" + type: string + optional: + type: boolean + required: + - key + type: object + x-kubernetes-map-type: atomic + hostPort: + pattern: ^[a-zA-Z0-9.-]+:[0-9]+$ + type: string + mutualTLSSecretRef: + properties: + name: + pattern: ^[a-z0-9]([-a-z0-9]*[a-z0-9])?$ + type: string + required: + - name + type: object + tls: + properties: + serverName: + pattern: ^[a-zA-Z0-9.-]+$ + type: string + type: object + required: + - hostPort + type: object + x-kubernetes-validations: + - message: Only one of mutualTLSSecretRef or apiKeySecretRef may be set + rule: '!(has(self.mutualTLSSecretRef) && has(self.apiKeySecretRef))' + status: + type: object + type: object + served: true + storage: true + subresources: + status: {} diff --git a/helm/temporal-worker-controller-crds/templates/temporal.io_workerdeployments.yaml b/helm/temporal-worker-controller-crds/templates/temporal.io_workerdeployments.yaml index 98bfe3e5..88dab589 100644 --- a/helm/temporal-worker-controller-crds/templates/temporal.io_workerdeployments.yaml +++ b/helm/temporal-worker-controller-crds/templates/temporal.io_workerdeployments.yaml @@ -3980,6 +3980,12 @@ spec: properties: connectionRef: properties: + kind: + default: Connection + enum: + - Connection + - ClusterConnection + type: string name: pattern: ^[a-z0-9]([-a-z0-9]*[a-z0-9])?$ type: string diff --git a/helm/temporal-worker-controller/templates/rbac.yaml b/helm/temporal-worker-controller/templates/rbac.yaml index 756bec18..513bba1d 100644 --- a/helm/temporal-worker-controller/templates/rbac.yaml +++ b/helm/temporal-worker-controller/templates/rbac.yaml @@ -110,6 +110,7 @@ rules: - apiGroups: - temporal.io resources: + - clusterconnections - connections - temporalconnections - workerresourcetemplates @@ -122,6 +123,7 @@ rules: - apiGroups: - temporal.io resources: + - clusterconnections/finalizers - connections/finalizers - temporalconnections/finalizers - temporalworkerdeployments/finalizers @@ -228,6 +230,7 @@ rules: - apiGroups: - temporal.io resources: + - clusterconnections - connections - temporalconnections - workerresourcetemplates @@ -240,6 +243,7 @@ rules: - apiGroups: - temporal.io resources: + - clusterconnections/finalizers - connections/finalizers - temporalconnections/finalizers - temporalworkerdeployments/finalizers diff --git a/internal/controller/clusterconnection_test.go b/internal/controller/clusterconnection_test.go new file mode 100644 index 00000000..7cc65d52 --- /dev/null +++ b/internal/controller/clusterconnection_test.go @@ -0,0 +1,424 @@ +package controller + +import ( + "context" + "testing" + + "github.com/go-logr/logr" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + temporaliov1alpha1 "github.com/temporalio/temporal-worker-controller/api/v1alpha1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" + "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" + "sigs.k8s.io/controller-runtime/pkg/reconcile" +) + +// makeClusterConnection creates a minimal cluster-scoped ClusterConnection. +func makeClusterConnection(name, hostPort string) *temporaliov1alpha1.ClusterConnection { + return &temporaliov1alpha1.ClusterConnection{ + TypeMeta: metav1.TypeMeta{ + APIVersion: temporaliov1alpha1.GroupVersion.String(), + Kind: "ClusterConnection", + }, + ObjectMeta: metav1.ObjectMeta{ + Name: name, // intentionally no Namespace + }, + Spec: temporaliov1alpha1.ConnectionSpec{ + HostPort: hostPort, + }, + } +} + +// makeWDWithKind builds a WorkerDeployment whose connectionRef carries an +// explicit Kind ("", "Connection", or "ClusterConnection"). +func makeWDWithKind(name, namespace, connName, kind string) *temporaliov1alpha1.WorkerDeployment { + wd := makeWD(name, namespace, connName) + wd.Spec.WorkerOptions.ConnectionRef.Kind = kind + return wd +} + +// hasFinalizer re-Gets obj from the client and reports whether it still carries our finalizer +func hasFinalizer(t *testing.T, c client.Client, obj client.Object, key types.NamespacedName) bool { + t.Helper() + require.NoError(t, c.Get(context.Background(), key, obj)) + return controllerutil.ContainsFinalizer(obj, finalizerName) +} + +func TestResolveConnection(t *testing.T) { + ctx := context.Background() + + t.Run("KindConnection_fetchesNamespaced", func(t *testing.T) { + conn := makeNoCredsConnection("conn", "default", "h:7233") + wd := makeWDWithKind("wd", "default", "conn", "Connection") + r, _ := newTestReconciler([]client.Object{conn, wd}) + + spec, obj, err := r.resolveConnection(ctx, wd) + require.NoError(t, err) + assert.Equal(t, "h:7233", spec.HostPort) + _, ok := obj.(*temporaliov1alpha1.Connection) + assert.True(t, ok, "expected a *Connection object") + }) + + t.Run("KindEmpty_identicalToConnection", func(t *testing.T) { + conn := makeNoCredsConnection("conn", "default", "h:7233") + wd := makeWDWithKind("wd", "default", "conn", "") + r, _ := newTestReconciler([]client.Object{conn, wd}) + + spec, obj, err := r.resolveConnection(ctx, wd) + require.NoError(t, err) + assert.Equal(t, "h:7233", spec.HostPort) + _, ok := obj.(*temporaliov1alpha1.Connection) + assert.True(t, ok, "empty kind must resolve to a namespaced Connection") + }) + + t.Run("KindClusterConnection_fetchesClusterScoped", func(t *testing.T) { + cc := makeClusterConnection("conn", "h:7233") + wd := makeWDWithKind("wd", "default", "conn", "ClusterConnection") + r, _ := newTestReconciler([]client.Object{cc, wd}) + + spec, obj, err := r.resolveConnection(ctx, wd) + require.NoError(t, err) + assert.Equal(t, "h:7233", spec.HostPort) + got, ok := obj.(*temporaliov1alpha1.ClusterConnection) + require.True(t, ok, "expected a *ClusterConnection object") + assert.Empty(t, got.Namespace, "cluster-scoped object must have no namespace") + }) + + t.Run("SameName_bothKinds_resolveDistinctObjects", func(t *testing.T) { + conn := makeNoCredsConnection("foo", "default", "ns-conn:7233") + cc := makeClusterConnection("foo", "cluster-conn:7233") + wdNS := makeWDWithKind("wd-ns", "default", "foo", "Connection") + wdCC := makeWDWithKind("wd-cc", "default", "foo", "ClusterConnection") + r, _ := newTestReconciler([]client.Object{conn, cc, wdNS, wdCC}) + + specNS, objNS, err := r.resolveConnection(ctx, wdNS) + require.NoError(t, err) + assert.Equal(t, "ns-conn:7233", specNS.HostPort) + _, ok := objNS.(*temporaliov1alpha1.Connection) + assert.True(t, ok) + + specCC, objCC, err := r.resolveConnection(ctx, wdCC) + require.NoError(t, err) + assert.Equal(t, "cluster-conn:7233", specCC.HostPort) + _, ok = objCC.(*temporaliov1alpha1.ClusterConnection) + assert.True(t, ok) + }) + + t.Run("ClusterExists_butNamespacedRequested_notFound", func(t *testing.T) { + cc := makeClusterConnection("foo", "cluster-conn:7233") + wd := makeWDWithKind("wd", "default", "foo", "Connection") + r, _ := newTestReconciler([]client.Object{cc, wd}) + + _, obj, err := r.resolveConnection(ctx, wd) + require.Error(t, err) + assert.True(t, apierrors.IsNotFound(err), "must be NotFound, not a stray cluster resolve") + assert.Nil(t, obj) + }) + + t.Run("NotFound_namespaced", func(t *testing.T) { + wd := makeWDWithKind("wd", "default", "missing", "Connection") + r, _ := newTestReconciler([]client.Object{wd}) + + _, obj, err := r.resolveConnection(ctx, wd) + require.Error(t, err) + assert.True(t, apierrors.IsNotFound(err)) + assert.Nil(t, obj) + }) + + t.Run("NotFound_cluster", func(t *testing.T) { + wd := makeWDWithKind("wd", "default", "missing", "ClusterConnection") + r, _ := newTestReconciler([]client.Object{wd}) + + _, obj, err := r.resolveConnection(ctx, wd) + require.Error(t, err) + assert.True(t, apierrors.IsNotFound(err)) + assert.Nil(t, obj) + }) +} + +func TestEnsureConnectionFinalizer(t *testing.T) { + ctx := context.Background() + newConn := func() *temporaliov1alpha1.Connection { + return makeNoCredsConnection("conn", "default", "h:7233") + } + newCC := func() *temporaliov1alpha1.ClusterConnection { + return makeClusterConnection("conn", "h:7233") + } + tests := []struct { + name string + seed client.Object + key types.NamespacedName + refetch client.Object + wantUpdateCount int + }{ + { + name: "Connection_addsFinalizer", + seed: newConn(), + key: types.NamespacedName{Name: "conn", Namespace: "default"}, + refetch: &temporaliov1alpha1.Connection{}, + wantUpdateCount: 1, + }, + { + name: "ClusterConnection_addsFinalizer", + seed: newCC(), + key: types.NamespacedName{Name: "conn"}, // cluster-scoped: no namespace + refetch: &temporaliov1alpha1.ClusterConnection{}, + wantUpdateCount: 1, + }, + { + name: "Connection_idempotent_noUpdate", + seed: func() client.Object { + c := newConn() + c.Finalizers = []string{finalizerName} + return c + }(), + key: types.NamespacedName{Name: "conn", Namespace: "default"}, + refetch: &temporaliov1alpha1.Connection{}, + wantUpdateCount: 0, + }, + { + name: "ClusterConnection_idempotent_noUpdate", + seed: func() client.Object { + c := newCC() + c.Finalizers = []string{finalizerName} + return c + }(), + key: types.NamespacedName{Name: "conn"}, + refetch: &temporaliov1alpha1.ClusterConnection{}, + wantUpdateCount: 0, + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + var updateCount int + funcs := interceptor.Funcs{ + Update: func(ctx context.Context, c client.WithWatch, obj client.Object, opts ...client.UpdateOption) error { + updateCount++ + return c.Update(ctx, obj, opts...) + }, + } + r, _ := newTestReconcilerWithInterceptors([]client.Object{tc.seed}, funcs) + err := r.ensureConnectionFinalizer(ctx, logr.Discard(), tc.seed) + require.NoError(t, err) + assert.Equal(t, tc.wantUpdateCount, updateCount, "unexpected number of Update calls") + assert.True(t, hasFinalizer(t, r.Client, tc.refetch, tc.key), "finalizer must be present afterwards") + }) + } +} + +func TestRemoveConnectionFinalizerIfUnused(t *testing.T) { + ctx := context.Background() + + // connection object that already has the finalizer + connWithFinalizer := func(name, ns, hostPort string) *temporaliov1alpha1.Connection { + c := makeNoCredsConnection(name, ns, hostPort) + c.Finalizers = []string{finalizerName} + return c + } + ccWithFinalizer := func(name, hostPort string) *temporaliov1alpha1.ClusterConnection { + c := makeClusterConnection(name, hostPort) + c.Finalizers = []string{finalizerName} + return c + } + + t.Run("namespaced_unused_removes", func(t *testing.T) { + conn := connWithFinalizer("conn", "default", "h:7233") + del := makeWDWithKind("del", "default", "conn", "Connection") + r, _ := newTestReconciler([]client.Object{conn, del}) + + err := r.removeConnectionFinalizerIfUnused(ctx, logr.Discard(), del) + require.NoError(t, err) + assert.False(t, hasFinalizer(t, r.Client, &temporaliov1alpha1.Connection{}, + types.NamespacedName{Name: "conn", Namespace: "default"})) + }) + + t.Run("namespaced_stillUsed_keeps", func(t *testing.T) { + conn := connWithFinalizer("conn", "default", "h:7233") + del := makeWDWithKind("del", "default", "conn", "Connection") + other := makeWDWithKind("other", "default", "conn", "Connection") + r, _ := newTestReconciler([]client.Object{conn, del, other}) + + err := r.removeConnectionFinalizerIfUnused(ctx, logr.Discard(), del) + require.NoError(t, err) + assert.True(t, hasFinalizer(t, r.Client, &temporaliov1alpha1.Connection{}, + types.NamespacedName{Name: "conn", Namespace: "default"})) + }) + + t.Run("namespaced_otherNamespace_ignored_removes", func(t *testing.T) { + conn := connWithFinalizer("conn", "default", "h:7233") + del := makeWDWithKind("del", "default", "conn", "Connection") + // same-named WD in a DIFFERENT namespace — must not count as a referrer + otherNS := makeWDWithKind("other", "ns-b", "conn", "Connection") + r, _ := newTestReconciler([]client.Object{conn, del, otherNS}) + + err := r.removeConnectionFinalizerIfUnused(ctx, logr.Discard(), del) + require.NoError(t, err) + assert.False(t, hasFinalizer(t, r.Client, &temporaliov1alpha1.Connection{}, + types.NamespacedName{Name: "conn", Namespace: "default"})) + }) + + t.Run("cluster_unused_removes", func(t *testing.T) { + cc := ccWithFinalizer("conn", "h:7233") + del := makeWDWithKind("del", "ns-a", "conn", "ClusterConnection") + r, _ := newTestReconciler([]client.Object{cc, del}) + + err := r.removeConnectionFinalizerIfUnused(ctx, logr.Discard(), del) + require.NoError(t, err) + assert.False(t, hasFinalizer(t, r.Client, &temporaliov1alpha1.ClusterConnection{}, + types.NamespacedName{Name: "conn"})) + }) + + // A ClusterConnection referenced by a WD in another namespace must keep its + // finalizer when one referrer is deleted. If InNamespace is ever reintroduced + // on the cluster path, this test fails while everything else passes. + t.Run("cluster_stillUsedFromAnotherNamespace_keeps", func(t *testing.T) { + cc := ccWithFinalizer("conn", "h:7233") + del := makeWDWithKind("del", "ns-a", "conn", "ClusterConnection") + otherNS := makeWDWithKind("other", "ns-b", "conn", "ClusterConnection") + r, _ := newTestReconciler([]client.Object{cc, del, otherNS}) + + err := r.removeConnectionFinalizerIfUnused(ctx, logr.Discard(), del) + require.NoError(t, err) + assert.True(t, hasFinalizer(t, r.Client, &temporaliov1alpha1.ClusterConnection{}, + types.NamespacedName{Name: "conn"}), + "ClusterConnection must NOT be released while another namespace still references it") + }) + + t.Run("kindDisambiguation_clusterReleased_namespacedUntouched", func(t *testing.T) { + // Connection "foo" and ClusterConnection "foo" coexist. + conn := connWithFinalizer("foo", "default", "ns-conn:7233") + cc := ccWithFinalizer("foo", "cluster-conn:7233") + // deleting the WD that used the ClusterConnection + del := makeWDWithKind("del", "default", "foo", "ClusterConnection") + // a WD still using the NAMESPACED Connection "foo" + nsUser := makeWDWithKind("ns-user", "default", "foo", "Connection") + r, _ := newTestReconciler([]client.Object{conn, cc, del, nsUser}) + + err := r.removeConnectionFinalizerIfUnused(ctx, logr.Discard(), del) + require.NoError(t, err) + // ClusterConnection "foo" released (no other cluster referrer) + assert.False(t, hasFinalizer(t, r.Client, &temporaliov1alpha1.ClusterConnection{}, + types.NamespacedName{Name: "foo"})) + // namespaced Connection "foo" untouched (still used by ns-user) + assert.True(t, hasFinalizer(t, r.Client, &temporaliov1alpha1.Connection{}, + types.NamespacedName{Name: "foo", Namespace: "default"})) + }) + + t.Run("skipSelf_sameNameDifferentNamespace_keeps", func(t *testing.T) { + cc := ccWithFinalizer("conn", "h:7233") + // two WDs with the SAME name in different namespaces + del := makeWDWithKind("samename", "ns-a", "conn", "ClusterConnection") + other := makeWDWithKind("samename", "ns-b", "conn", "ClusterConnection") + r, _ := newTestReconciler([]client.Object{cc, del, other}) + + err := r.removeConnectionFinalizerIfUnused(ctx, logr.Discard(), del) + require.NoError(t, err) + // the other "samename" in ns-b is a real referrer, not "self" — keep finalizer + assert.True(t, hasFinalizer(t, r.Client, &temporaliov1alpha1.ClusterConnection{}, + types.NamespacedName{Name: "conn"}), + "same-name WD in another namespace must be counted as a referrer, not skipped as self") + }) + + t.Run("alreadyGone_noError", func(t *testing.T) { + // connection object does not exist when removal runs + del := makeWDWithKind("del", "default", "missing", "Connection") + r, _ := newTestReconciler([]client.Object{del}) + + err := r.removeConnectionFinalizerIfUnused(ctx, logr.Discard(), del) + require.NoError(t, err, "missing connection must be treated as already released") + }) +} + +func TestFindTWDsUsingConnection(t *testing.T) { + ctx := context.Background() + + t.Run("namespacedConnection_enqueuesMatchingWDs", func(t *testing.T) { + conn := makeNoCredsConnection("conn", "default", "h:7233") + wd := makeWDWithKind("wd", "default", "conn", "Connection") + r, _ := newTestReconciler([]client.Object{conn, wd}) + + reqs := r.findTWDsUsingConnection(ctx, conn) + assert.Contains(t, reqs, reconcile.Request{ + NamespacedName: types.NamespacedName{Name: "wd", Namespace: "default"}, + }) + }) + + t.Run("clusterConnectionRefOfSameName_notEnqueued", func(t *testing.T) { + conn := makeNoCredsConnection("foo", "default", "h:7233") + // this WD points at a ClusterConnection named "foo", not the namespaced one + wd := makeWDWithKind("wd", "default", "foo", "ClusterConnection") + r, _ := newTestReconciler([]client.Object{conn, wd}) + + reqs := r.findTWDsUsingConnection(ctx, conn) + assert.Empty(t, reqs, "cluster-ref WD must not be enqueued by the namespaced Connection mapper") + }) + + t.Run("noMatches_emptySlice", func(t *testing.T) { + conn := makeNoCredsConnection("conn", "default", "h:7233") + wd := makeWDWithKind("wd", "default", "other-conn", "Connection") + r, _ := newTestReconciler([]client.Object{conn, wd}) + + reqs := r.findTWDsUsingConnection(ctx, conn) + assert.Empty(t, reqs) + }) +} + +func TestFindTWDsUsingClusterConnection(t *testing.T) { + ctx := context.Background() + + t.Run("multipleNamespaces_allEnqueued", func(t *testing.T) { + cc := makeClusterConnection("conn", "h:7233") + wdA := makeWDWithKind("wd-a", "ns-a", "conn", "ClusterConnection") + wdB := makeWDWithKind("wd-b", "ns-b", "conn", "ClusterConnection") + r, _ := newTestReconciler([]client.Object{cc, wdA, wdB}) + + reqs := r.findTWDsUsingClusterConnection(ctx, cc) + assert.Contains(t, reqs, reconcile.Request{ + NamespacedName: types.NamespacedName{Name: "wd-a", Namespace: "ns-a"}, + }) + assert.Contains(t, reqs, reconcile.Request{ + NamespacedName: types.NamespacedName{Name: "wd-b", Namespace: "ns-b"}, + }) + assert.Len(t, reqs, 2) + }) + + t.Run("namespacedConnectionRefOfSameName_notEnqueued", func(t *testing.T) { + cc := makeClusterConnection("foo", "h:7233") + // this WD points at a namespaced Connection named "foo", not the cluster one + wd := makeWDWithKind("wd", "default", "foo", "Connection") + r, _ := newTestReconciler([]client.Object{cc, wd}) + + reqs := r.findTWDsUsingClusterConnection(ctx, cc) + assert.Empty(t, reqs, "namespaced-ref WD must not be enqueued by the cluster mapper") + }) + + t.Run("sameNameDifferentNamespaces_bothEnqueued", func(t *testing.T) { + cc := makeClusterConnection("conn", "h:7233") + // two WDs with the same NAME in different namespaces + wdA := makeWDWithKind("samename", "ns-a", "conn", "ClusterConnection") + wdB := makeWDWithKind("samename", "ns-b", "conn", "ClusterConnection") + r, _ := newTestReconciler([]client.Object{cc, wdA, wdB}) + + reqs := r.findTWDsUsingClusterConnection(ctx, cc) + assert.Contains(t, reqs, reconcile.Request{ + NamespacedName: types.NamespacedName{Name: "samename", Namespace: "ns-a"}, + }) + assert.Contains(t, reqs, reconcile.Request{ + NamespacedName: types.NamespacedName{Name: "samename", Namespace: "ns-b"}, + }) + assert.Len(t, reqs, 2, "same-name WDs in different namespaces must both be distinct requests") + }) + + t.Run("noMatches_emptySlice", func(t *testing.T) { + cc := makeClusterConnection("conn", "h:7233") + wd := makeWDWithKind("wd", "default", "other-conn", "ClusterConnection") + r, _ := newTestReconciler([]client.Object{cc, wd}) + + reqs := r.findTWDsUsingClusterConnection(ctx, cc) + assert.Empty(t, reqs) + }) +} diff --git a/internal/controller/worker_controller.go b/internal/controller/worker_controller.go index 099074e2..bba04442 100644 --- a/internal/controller/worker_controller.go +++ b/internal/controller/worker_controller.go @@ -88,6 +88,8 @@ type WorkerDeploymentReconciler struct { // +kubebuilder:rbac:groups=temporal.io,resources=temporalworkerdeployments/finalizers,verbs=update // +kubebuilder:rbac:groups=temporal.io,resources=connections,verbs=get;list;watch;update;patch // +kubebuilder:rbac:groups=temporal.io,resources=connections/finalizers,verbs=update +// +kubebuilder:rbac:groups=temporal.io,resources=clusterconnections,verbs=get;list;watch;update;patch +// +kubebuilder:rbac:groups=temporal.io,resources=clusterconnections/finalizers,verbs=update // +kubebuilder:rbac:groups=temporal.io,resources=workerdeployments,verbs=get;list;watch;create;update;patch;delete // +kubebuilder:rbac:groups=temporal.io,resources=workerdeployments/status,verbs=get;update;patch // +kubebuilder:rbac:groups=temporal.io,resources=workerdeployments/finalizers,verbs=update @@ -220,11 +222,8 @@ func (r *WorkerDeploymentReconciler) Reconcile(ctx context.Context, req ctrl.Req // Note: ConnectionRef.Name is validated by webhook due to +kubebuilder:validation:Required // Fetch the connection parameters - var connection temporaliov1alpha1.Connection - if err := r.Get(ctx, types.NamespacedName{ - Name: workerDeploy.Spec.WorkerOptions.ConnectionRef.Name, - Namespace: workerDeploy.Namespace, - }, &connection); err != nil { + connSpec, connObj, err := r.resolveConnection(ctx, &workerDeploy) + if err != nil { l.Error(err, "unable to fetch Connection") r.recordWarningAndSetBlocked(ctx, &workerDeploy, temporaliov1alpha1.ReasonConnectionNotFound, @@ -233,10 +232,11 @@ func (r *WorkerDeploymentReconciler) Reconcile(ctx context.Context, req ctrl.Req return ctrl.Result{}, err } - // Ensure our finalizer is on the Connection so it cannot be deleted - // while this WD still references it. This guarantees the connection is available - // during WD deletion cleanup. - if err := r.ensureConnectionFinalizer(ctx, l, &connection); err != nil { + connection := temporaliov1alpha1.Connection{Spec: connSpec} + + // Ensure the finalizer is on the connection object so it cannot be deleted + // while this WD still references it. + if err := r.ensureConnectionFinalizer(ctx, l, connObj); err != nil { return ctrl.Result{}, err } @@ -248,7 +248,7 @@ func (r *WorkerDeploymentReconciler) Reconcile(ctx context.Context, req ctrl.Req // validation failures. r.recordWarningAndSetBlocked(ctx, &workerDeploy, temporaliov1alpha1.ReasonAuthSecretInvalid, - fmt.Sprintf("Unable to resolve auth secret from Connection %q: %v", connection.Name, err), + fmt.Sprintf("Unable to resolve auth secret from Connection %q: %v", connObj.GetName(), err), fmt.Sprintf("Unable to resolve auth secret: %v", err)) return ctrl.Result{}, err } @@ -563,13 +563,11 @@ func (r *WorkerDeploymentReconciler) handleDeletion( // Resolve Connection. // The Connection is guaranteed to exist because we hold a finalizer on it // that prevents deletion while any WD references it. - var connection temporaliov1alpha1.Connection - if err := r.Get(ctx, types.NamespacedName{ - Name: workerDeploy.Spec.WorkerOptions.ConnectionRef.Name, - Namespace: workerDeploy.Namespace, - }, &connection); err != nil { + connSpec, _, err := r.resolveConnection(ctx, workerDeploy) + if err != nil { return fmt.Errorf("unable to fetch Connection: %w", err) } + connection := temporaliov1alpha1.Connection{Spec: connSpec} if err := connection.Spec.Validate(); err != nil { // TODO(jaypipes): As of TWC release <=v1.8.1, the only validation @@ -808,18 +806,51 @@ func (r *WorkerDeploymentReconciler) recordWarningAndSetBlocked( _ = r.Status().Update(ctx, workerDeploy) } +// connectionRefIsCluster reports whether a connectionRef targets a +// cluster-scoped ClusterConnection. +func connectionRefIsCluster(ref temporaliov1alpha1.ConnectionReference) bool { + return ref.Kind == "ClusterConnection" +} + +// resolveConnection fetches the connection resource referenced by the +// WorkerDeployment's connectionRef and returns its spec and the underlying object. +// The object is returned as a client.Object so callers can +// manage the finalizer on it regardless of whether it is a namespaced +// Connection or a cluster-scoped ClusterConnection. +func (r *WorkerDeploymentReconciler) resolveConnection( + ctx context.Context, + workerDeploy *temporaliov1alpha1.WorkerDeployment, +) (temporaliov1alpha1.ConnectionSpec, client.Object, error) { + connName := workerDeploy.Spec.WorkerOptions.ConnectionRef.Name + if workerDeploy.Spec.WorkerOptions.ConnectionRef.Kind == "ClusterConnection" { + var cc temporaliov1alpha1.ClusterConnection + if err := r.Get(ctx, types.NamespacedName{Name: connName}, &cc); err != nil { + return temporaliov1alpha1.ConnectionSpec{}, nil, err + } + return cc.Spec, &cc, nil + } + var conn temporaliov1alpha1.Connection + if err := r.Get(ctx, types.NamespacedName{ + Name: connName, + Namespace: workerDeploy.Namespace, + }, &conn); err != nil { + return temporaliov1alpha1.ConnectionSpec{}, nil, err + } + return conn.Spec, &conn, nil +} + // ensureConnectionFinalizer adds our finalizer to the Connection so it // cannot be deleted while this WD still needs it for cleanup. func (r *WorkerDeploymentReconciler) ensureConnectionFinalizer( ctx context.Context, l logr.Logger, - tc *temporaliov1alpha1.Connection, + conn client.Object, ) error { - if !controllerutil.ContainsFinalizer(tc, finalizerName) { - l.Info("Adding finalizer to Connection", "connection", tc.Name) - controllerutil.AddFinalizer(tc, finalizerName) - if err := r.Update(ctx, tc); err != nil { - return fmt.Errorf("unable to add finalizer to Connection %q: %w", tc.Name, err) + if !controllerutil.ContainsFinalizer(conn, finalizerName) { + l.Info("Adding finalizer to connection", "connection", conn.GetName()) + controllerutil.AddFinalizer(conn, finalizerName) + if err := r.Update(ctx, conn); err != nil { + return fmt.Errorf("unable to add finalizer to connection %q: %w", conn.GetName(), err) } } return nil @@ -832,44 +863,56 @@ func (r *WorkerDeploymentReconciler) removeConnectionFinalizerIfUnused( l logr.Logger, deletingWD *temporaliov1alpha1.WorkerDeployment, ) error { - connectionName := deletingWD.Spec.WorkerOptions.ConnectionRef.Name + ref := deletingWD.Spec.WorkerOptions.ConnectionRef + isCluster := connectionRefIsCluster(ref) + + // Scope the "is it still used?" query correctly for the kind: + // - Namespaced Connection: only WDs in its own namespace can reference it, + // so restrict the list to deletingWD.Namespace. + // - ClusterConnection: a WD in ANY namespace can reference it, so we must + // list across all namespaces. + var listOpts []client.ListOption + if !isCluster { + listOpts = append(listOpts, client.InNamespace(deletingWD.Namespace)) + } - // List all WDs in the same namespace var wds temporaliov1alpha1.WorkerDeploymentList - if err := r.List(ctx, &wds, client.InNamespace(deletingWD.Namespace)); err != nil { - return fmt.Errorf("unable to list WDs: %w", err) + if err := r.List(ctx, &wds, listOpts...); err != nil { + return fmt.Errorf("unable to list WorkerDeployments: %w", err) } - // Check if any other WD (not the one being deleted) references this connection for i := range wds.Items { wd := &wds.Items[i] - if wd.Name == deletingWD.Name { + // Skip self by namespace and name: under a cluster-wide list, two WDs in + // different namespaces can share the same name, so name alone is not a + // unique identity. + if wd.Namespace == deletingWD.Namespace && wd.Name == deletingWD.Name { continue } - if wd.Spec.WorkerOptions.ConnectionRef.Name == connectionName { - l.Info("Connection still referenced by another WD, keeping finalizer", - "connection", connectionName, "referencedBy", wd.Name) + otherRef := wd.Spec.WorkerOptions.ConnectionRef + // Same target only if BOTH name and (normalized) kind match, so a + // namespaced Connection "foo" and a ClusterConnection "foo" are distinct. + if otherRef.Name == ref.Name && connectionRefIsCluster(otherRef) == isCluster { + l.Info("Connection still referenced by another WorkerDeployment, keeping finalizer", + "connection", ref.Name, "kind", ref.Kind, + "referencedBy", wd.Name, "referencedByNamespace", wd.Namespace) return nil } } - // No other WDs reference this connection, remove the finalizer - var tc temporaliov1alpha1.Connection - if err := r.Get(ctx, types.NamespacedName{ - Name: connectionName, - Namespace: deletingWD.Namespace, - }, &tc); err != nil { + _, connObj, err := r.resolveConnection(ctx, deletingWD) + if err != nil { if apierrors.IsNotFound(err) { - return nil // already gone + return nil } - return fmt.Errorf("unable to fetch Connection %q: %w", connectionName, err) + return fmt.Errorf("unable to fetch connection %q: %w", ref.Name, err) } - if controllerutil.ContainsFinalizer(&tc, finalizerName) { - l.Info("Removing finalizer from Connection", "connection", connectionName) - controllerutil.RemoveFinalizer(&tc, finalizerName) - if err := r.Update(ctx, &tc); err != nil { - return fmt.Errorf("unable to remove finalizer from Connection %q: %w", connectionName, err) + if controllerutil.ContainsFinalizer(connObj, finalizerName) { + l.Info("Removing finalizer from connection", "connection", ref.Name, "kind", ref.Kind) + controllerutil.RemoveFinalizer(connObj, finalizerName) + if err := r.Update(ctx, connObj); err != nil { + return fmt.Errorf("unable to remove finalizer from connection %q: %w", ref.Name, err) } } @@ -915,6 +958,7 @@ func (r *WorkerDeploymentReconciler) SetupWithManager(mgr ctrl.Manager) error { For(&temporaliov1alpha1.WorkerDeployment{}). Owns(&appsv1.Deployment{}). Watches(&temporaliov1alpha1.Connection{}, handler.EnqueueRequestsFromMapFunc(r.findTWDsUsingConnection)). + Watches(&temporaliov1alpha1.ClusterConnection{}, handler.EnqueueRequestsFromMapFunc(r.findTWDsUsingClusterConnection)). Watches(&temporaliov1alpha1.WorkerResourceTemplate{}, handler.EnqueueRequestsFromMapFunc(r.reconcileRequestForWRT)) if !r.DisableDeprecatedTWD { // Watch deprecated TemporalWorkerDeployments so that any modification to an existing TWD @@ -960,8 +1004,9 @@ func (r *WorkerDeploymentReconciler) findTWDsUsingConnection(ctx context.Context // Filter to ones using this connection for _, twd := range twds.Items { - if twd.Spec.WorkerOptions.ConnectionRef.Name == tc.GetName() { - // Enqueue a reconcile request for this TWD + ref := twd.Spec.WorkerOptions.ConnectionRef + // Only namespaced Connection refs are driven by this (Connection) watch. + if !connectionRefIsCluster(ref) && ref.Name == tc.GetName() { requests = append(requests, reconcile.Request{ NamespacedName: types.NamespacedName{ Name: twd.Name, @@ -974,6 +1019,29 @@ func (r *WorkerDeploymentReconciler) findTWDsUsingConnection(ctx context.Context return requests } +func (r *WorkerDeploymentReconciler) findTWDsUsingClusterConnection( + ctx context.Context, + cc client.Object, +) []reconcile.Request { + var requests []reconcile.Request + var twds temporaliov1alpha1.WorkerDeploymentList + if err := r.List(ctx, &twds); err != nil { + return requests + } + for _, twd := range twds.Items { + ref := twd.Spec.WorkerOptions.ConnectionRef + if connectionRefIsCluster(ref) && ref.Name == cc.GetName() { + requests = append(requests, reconcile.Request{ + NamespacedName: types.NamespacedName{ + Name: twd.Name, + Namespace: twd.Namespace, + }, + }) + } + } + return requests +} + func isAccessDeniedErr(err error) bool { var permDenied *serviceerror.PermissionDenied if errors.As(err, &permDenied) { From ce8db11e6f0bc09418750adc085bdb1e4604d64e Mon Sep 17 00:00:00 2001 From: Niyomukiza Mechack Date: Tue, 25 Aug 2026 12:21:19 -0700 Subject: [PATCH 2/5] feat: release old connection finalizer on connectionRef change --- api/v1alpha1/workerdeployment_types.go | 6 ++ api/v1alpha1/zz_generated.deepcopy.go | 5 + .../temporal.io_workerdeployments.yaml | 14 +++ internal/controller/clusterconnection_test.go | 87 ++++++++++++++++++ internal/controller/worker_controller.go | 91 ++++++++++++++----- 5 files changed, 182 insertions(+), 21 deletions(-) diff --git a/api/v1alpha1/workerdeployment_types.go b/api/v1alpha1/workerdeployment_types.go index 923544ed..ee42c3d8 100644 --- a/api/v1alpha1/workerdeployment_types.go +++ b/api/v1alpha1/workerdeployment_types.go @@ -242,6 +242,12 @@ type WorkerDeploymentStatus struct { // Conditions represent the latest available observations of the WorkerDeployment's current state. // +optional Conditions []metav1.Condition `json:"conditions,omitempty"` + + // ObservedConnectionRef records the connectionRef the controller last + // finalized, so the reconciler can detect when connectionRef changes + // (name or kind) and release the finalizer from the previously-referenced + // connection. + ObservedConnectionRef *ConnectionReference `json:"observedConnectionRef,omitempty"` } // WorkflowExecutionStatus describes the current state of a workflow. diff --git a/api/v1alpha1/zz_generated.deepcopy.go b/api/v1alpha1/zz_generated.deepcopy.go index 109324af..333b4482 100644 --- a/api/v1alpha1/zz_generated.deepcopy.go +++ b/api/v1alpha1/zz_generated.deepcopy.go @@ -888,6 +888,11 @@ func (in *WorkerDeploymentStatus) DeepCopyInto(out *WorkerDeploymentStatus) { (*in)[i].DeepCopyInto(&(*out)[i]) } } + if in.ObservedConnectionRef != nil { + in, out := &in.ObservedConnectionRef, &out.ObservedConnectionRef + *out = new(ConnectionReference) + **out = **in + } } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new WorkerDeploymentStatus. diff --git a/helm/temporal-worker-controller-crds/templates/temporal.io_workerdeployments.yaml b/helm/temporal-worker-controller-crds/templates/temporal.io_workerdeployments.yaml index 88dab589..d50aa4b8 100644 --- a/helm/temporal-worker-controller-crds/templates/temporal.io_workerdeployments.yaml +++ b/helm/temporal-worker-controller-crds/templates/temporal.io_workerdeployments.yaml @@ -4142,6 +4142,20 @@ spec: type: string managerIdentity: type: string + observedConnectionRef: + properties: + kind: + default: Connection + enum: + - Connection + - ClusterConnection + type: string + name: + pattern: ^[a-z0-9]([-a-z0-9]*[a-z0-9])?$ + type: string + required: + - name + type: object targetVersion: properties: buildID: diff --git a/internal/controller/clusterconnection_test.go b/internal/controller/clusterconnection_test.go index 7cc65d52..e41a26e1 100644 --- a/internal/controller/clusterconnection_test.go +++ b/internal/controller/clusterconnection_test.go @@ -422,3 +422,90 @@ func TestFindTWDsUsingClusterConnection(t *testing.T) { assert.Empty(t, reqs) }) } + +//releaseConnectionFinalizerIfUnused: migration path + +// After a WD switches to a different connection, +// releasing the OLD (now-unused) connection's finalizer must succeed. +func TestReleaseConnectionFinalizerIfUnused_ReleasesUnused(t *testing.T) { + ctx := context.Background() + // old-conn still carries the finalizer from before the switch. + oldConn := makeNoCredsConnection("old-conn", "default", "h:7233") + oldConn.Finalizers = []string{finalizerName} + // The WD ("w") now points at a DIFFERENT connection, so nothing references + // old-conn anymore. + self := makeWDWithKind("w", "default", "new-conn", "Connection") + r, _ := newTestReconciler([]client.Object{oldConn, self}) + + oldRef := temporaliov1alpha1.ConnectionReference{Name: "old-conn", Kind: "Connection"} + require.NoError(t, r.releaseConnectionFinalizerIfUnused(ctx, logr.Discard(), oldRef, "default", "w")) + + assert.False(t, hasFinalizer(t, r.Client, &temporaliov1alpha1.Connection{}, + types.NamespacedName{Name: "old-conn", Namespace: "default"}), + "old connection's finalizer must be released once no WD references it") +} + +// Migrating away from a shared ClusterConnection must NOT +// release its finalizer while a WD in another namespace still references it. +func TestReleaseConnectionFinalizerIfUnused_KeepsSharedStillUsed(t *testing.T) { + ctx := context.Background() + shared := makeClusterConnection("shared", "h:7233") + shared.Finalizers = []string{finalizerName} + // A WD in ns-b still references the shared ClusterConnection. + wdB := makeWDWithKind("wb", "ns-b", "shared", "ClusterConnection") + r, _ := newTestReconciler([]client.Object{shared, wdB}) + + // Simulate the WD "w" in ns-a migrating away from "shared". + sharedRef := temporaliov1alpha1.ConnectionReference{Name: "shared", Kind: "ClusterConnection"} + require.NoError(t, r.releaseConnectionFinalizerIfUnused(ctx, logr.Discard(), sharedRef, "ns-a", "w")) + + assert.True(t, hasFinalizer(t, r.Client, &temporaliov1alpha1.ClusterConnection{}, + types.NamespacedName{Name: "shared"}), + "shared ClusterConnection finalizer must be KEPT while a WD in another namespace references it") +} + +// A connectionRef whose Kind was defaulted from +// "" to "Connection" must NOT be seen as a change, or every pre-existing WD would +// try to release its own connection on the first reconcile after upgrade. +func TestSameConnectionRef(t *testing.T) { + ref := func(name, kind string) temporaliov1alpha1.ConnectionReference { + return temporaliov1alpha1.ConnectionReference{Name: name, Kind: kind} + } + + tests := []struct { + name string + a, b temporaliov1alpha1.ConnectionReference + want bool + }{ + { + name: "empty kind equals Connection (normalization)", + a: ref("c", ""), + b: ref("c", "Connection"), + want: true, + }, + { + name: "Connection differs from ClusterConnection", + a: ref("c", "Connection"), + b: ref("c", "ClusterConnection"), + want: false, + }, + { + name: "different names differ", + a: ref("a", "Connection"), + b: ref("b", "Connection"), + want: false, + }, + { + name: "same cluster ref equals itself", + a: ref("c", "ClusterConnection"), + b: ref("c", "ClusterConnection"), + want: true, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + assert.Equal(t, tc.want, sameConnectionRef(tc.a, tc.b)) + }) + } +} diff --git a/internal/controller/worker_controller.go b/internal/controller/worker_controller.go index bba04442..11e5e183 100644 --- a/internal/controller/worker_controller.go +++ b/internal/controller/worker_controller.go @@ -240,6 +240,17 @@ func (r *WorkerDeploymentReconciler) Reconcile(ctx context.Context, req ctrl.Req return ctrl.Result{}, err } + // If connectionRef changed since we last finalized (name or kind), release the + // finalizer from the previously-referenced connection if no other WD uses it. + // The new connection is already protected by ensureConnectionFinalizer above, + // so the WD is never left unprotected. + current := workerDeploy.Spec.WorkerOptions.ConnectionRef + if observed := workerDeploy.Status.ObservedConnectionRef; observed != nil && !sameConnectionRef(*observed, current) { + if err := r.releaseConnectionFinalizerIfUnused(ctx, l, *observed, workerDeploy.Namespace, workerDeploy.Name); err != nil { + return ctrl.Result{}, err + } + } + if err := connection.Spec.Validate(); err != nil { l.Error(err, "connection spec not valid") // TODO(jaypipes): As of TWC release <=v1.8.1, the only validation @@ -348,6 +359,7 @@ func (r *WorkerDeploymentReconciler) Reconcile(ctx context.Context, req ctrl.Req } // Preserve conditions that were set during this reconciliation status.Conditions = workerDeploy.Status.Conditions + status.ObservedConnectionRef = workerDeploy.Spec.WorkerOptions.ConnectionRef.DeepCopy() workerDeploy.Status = *status // TODO(jlegrone): Set defaults via webhook rather than manually @@ -812,33 +824,49 @@ func connectionRefIsCluster(ref temporaliov1alpha1.ConnectionReference) bool { return ref.Kind == "ClusterConnection" } -// resolveConnection fetches the connection resource referenced by the -// WorkerDeployment's connectionRef and returns its spec and the underlying object. -// The object is returned as a client.Object so callers can -// manage the finalizer on it regardless of whether it is a namespaced -// Connection or a cluster-scoped ClusterConnection. -func (r *WorkerDeploymentReconciler) resolveConnection( +// sameConnectionRef reports whether two connectionRefs point at the same +// "" and "Connection" are treated as equal so a connectionRef whose Kind was +// defaulted from "" to "Connection" must not be seen as a change. +func sameConnectionRef(a, b temporaliov1alpha1.ConnectionReference) bool { + return a.Name == b.Name && connectionRefIsCluster(a) == connectionRefIsCluster(b) +} + +// getConnectionByRef fetches the connection resource identified by ref and +// returns its spec and the underlying object. namespace is used only for a +// namespaced Connection; it is ignored for a cluster-scoped ClusterConnection. +// The object is returned as a client.Object so callers can manage the finalizer +// on it regardless of kind. +func (r *WorkerDeploymentReconciler) getConnectionByRef( ctx context.Context, - workerDeploy *temporaliov1alpha1.WorkerDeployment, + ref temporaliov1alpha1.ConnectionReference, + namespace string, ) (temporaliov1alpha1.ConnectionSpec, client.Object, error) { - connName := workerDeploy.Spec.WorkerOptions.ConnectionRef.Name - if workerDeploy.Spec.WorkerOptions.ConnectionRef.Kind == "ClusterConnection" { + if connectionRefIsCluster(ref) { var cc temporaliov1alpha1.ClusterConnection - if err := r.Get(ctx, types.NamespacedName{Name: connName}, &cc); err != nil { + if err := r.Get(ctx, types.NamespacedName{Name: ref.Name}, &cc); err != nil { return temporaliov1alpha1.ConnectionSpec{}, nil, err } return cc.Spec, &cc, nil } var conn temporaliov1alpha1.Connection if err := r.Get(ctx, types.NamespacedName{ - Name: connName, - Namespace: workerDeploy.Namespace, + Name: ref.Name, + Namespace: namespace, }, &conn); err != nil { return temporaliov1alpha1.ConnectionSpec{}, nil, err } return conn.Spec, &conn, nil } +// resolveConnection fetches the connection resource referenced by the +// WorkerDeployment's current connectionRef. +func (r *WorkerDeploymentReconciler) resolveConnection( + ctx context.Context, + workerDeploy *temporaliov1alpha1.WorkerDeployment, +) (temporaliov1alpha1.ConnectionSpec, client.Object, error) { + return r.getConnectionByRef(ctx, workerDeploy.Spec.WorkerOptions.ConnectionRef, workerDeploy.Namespace) +} + // ensureConnectionFinalizer adds our finalizer to the Connection so it // cannot be deleted while this WD still needs it for cleanup. func (r *WorkerDeploymentReconciler) ensureConnectionFinalizer( @@ -856,24 +884,27 @@ func (r *WorkerDeploymentReconciler) ensureConnectionFinalizer( return nil } -// removeConnectionFinalizerIfUnused removes our finalizer from the Connection -// if no other WDs (besides the one being deleted) still reference it. -func (r *WorkerDeploymentReconciler) removeConnectionFinalizerIfUnused( +// releaseConnectionFinalizerIfUnused removes our finalizer from the connection +// identified by ref, unless some other WorkerDeployment (other than +// selfNamespace/selfName) still references it. It is used both when a WD is +// deleted (ref = current connectionRef) and when a WD's connectionRef changes +// (ref = previously-observed connectionRef). +func (r *WorkerDeploymentReconciler) releaseConnectionFinalizerIfUnused( ctx context.Context, l logr.Logger, - deletingWD *temporaliov1alpha1.WorkerDeployment, + ref temporaliov1alpha1.ConnectionReference, + selfNamespace, selfName string, ) error { - ref := deletingWD.Spec.WorkerOptions.ConnectionRef isCluster := connectionRefIsCluster(ref) // Scope the "is it still used?" query correctly for the kind: // - Namespaced Connection: only WDs in its own namespace can reference it, - // so restrict the list to deletingWD.Namespace. + // so restrict the list to selfNamespace. // - ClusterConnection: a WD in ANY namespace can reference it, so we must // list across all namespaces. var listOpts []client.ListOption if !isCluster { - listOpts = append(listOpts, client.InNamespace(deletingWD.Namespace)) + listOpts = append(listOpts, client.InNamespace(selfNamespace)) } var wds temporaliov1alpha1.WorkerDeploymentList @@ -886,7 +917,7 @@ func (r *WorkerDeploymentReconciler) removeConnectionFinalizerIfUnused( // Skip self by namespace and name: under a cluster-wide list, two WDs in // different namespaces can share the same name, so name alone is not a // unique identity. - if wd.Namespace == deletingWD.Namespace && wd.Name == deletingWD.Name { + if wd.Namespace == selfNamespace && wd.Name == selfName { continue } otherRef := wd.Spec.WorkerOptions.ConnectionRef @@ -900,7 +931,10 @@ func (r *WorkerDeploymentReconciler) removeConnectionFinalizerIfUnused( } } - _, connObj, err := r.resolveConnection(ctx, deletingWD) + // Fetch by the passed ref, NOT resolveConnection(deletingWD): during a + // connectionRef change the WD's current ref points at the NEW connection, so + // resolveConnection would strip the finalizer off the wrong (new) object. + _, connObj, err := r.getConnectionByRef(ctx, ref, selfNamespace) if err != nil { if apierrors.IsNotFound(err) { return nil @@ -919,6 +953,21 @@ func (r *WorkerDeploymentReconciler) removeConnectionFinalizerIfUnused( return nil } +// removeConnectionFinalizerIfUnused releases the finalizer from the WD's current +// connection when no other WD references it. Used by the deletion path. +func (r *WorkerDeploymentReconciler) removeConnectionFinalizerIfUnused( + ctx context.Context, + l logr.Logger, + deletingWD *temporaliov1alpha1.WorkerDeployment, +) error { + return r.releaseConnectionFinalizerIfUnused( + ctx, l, + deletingWD.Spec.WorkerOptions.ConnectionRef, + deletingWD.Namespace, + deletingWD.Name, + ) +} + // SetupWithManager sets up the controller with the Manager. func (r *WorkerDeploymentReconciler) SetupWithManager(mgr ctrl.Manager) error { if err := mgr.GetFieldIndexer().IndexField(context.Background(), &appsv1.Deployment{}, deployOwnerKey, func(rawObj client.Object) []string { From 7557b09ab0d46f89730325441715c22af641b94c Mon Sep 17 00:00:00 2001 From: Niyomukiza Mechack Date: Tue, 25 Aug 2026 12:38:59 -0700 Subject: [PATCH 3/5] fixing linting --- internal/controller/clusterconnection_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal/controller/clusterconnection_test.go b/internal/controller/clusterconnection_test.go index e41a26e1..c800e701 100644 --- a/internal/controller/clusterconnection_test.go +++ b/internal/controller/clusterconnection_test.go @@ -423,7 +423,7 @@ func TestFindTWDsUsingClusterConnection(t *testing.T) { }) } -//releaseConnectionFinalizerIfUnused: migration path +// ReleaseConnectionFinalizerIfUnused: migration path // After a WD switches to a different connection, // releasing the OLD (now-unused) connection's finalizer must succeed. From 0a5f83b62ab2efd43eb36565da13d43078ec99d5 Mon Sep 17 00:00:00 2001 From: Niyomukiza Mechack Date: Wed, 26 Aug 2026 15:47:17 -0700 Subject: [PATCH 4/5] refactor: address review feedback (Jay, Tomba) --- .../workerdeployment_cel_validation_test.go | 93 ++++++++++++++++++ api/v1alpha1/workerdeployment_types.go | 47 +++++---- api/v1alpha1/zz_generated.deepcopy.go | 11 ++- hack/sync-rbac-rules.py | 32 ++++-- .../temporal.io_workerdeployments.yaml | 65 ++++++++++--- .../templates/rbac.yaml | 2 - internal/controller/clusterconnection_test.go | 97 ++++++++++++------- internal/controller/worker_controller.go | 87 +++++++---------- 8 files changed, 301 insertions(+), 133 deletions(-) diff --git a/api/v1alpha1/workerdeployment_cel_validation_test.go b/api/v1alpha1/workerdeployment_cel_validation_test.go index 3b56cc90..db4209d8 100644 --- a/api/v1alpha1/workerdeployment_cel_validation_test.go +++ b/api/v1alpha1/workerdeployment_cel_validation_test.go @@ -18,6 +18,8 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) +func ptr[T any](v T) *T { return &v } + var _ = Describe("WorkerDeployment CRD CEL validation", func() { var ns string @@ -212,4 +214,95 @@ var _ = Describe("WorkerDeployment CRD CEL validation", func() { Expect(err).To(HaveOccurred()) Expect(err.Error()).To(ContainSubstring(messageTypeNotAllowedErr)) }) + + It("accepts connectionRef with objectRef (Connection kind)", func() { + twd := baseTWD("objref-connection") + twd.Spec.WorkerOptions.ConnectionRef = ConnectionReference{ + ObjectRef: &corev1.TypedObjectReference{ + APIGroup: ptr("temporal.io"), + Kind: "Connection", + Name: "my-connection", + }, + } + Expect(k8sClient.Create(ctx, twd)).To(Succeed()) + }) + + It("accepts connectionRef with objectRef (ClusterConnection kind)", func() { + twd := baseTWD("objref-cluster") + twd.Spec.WorkerOptions.ConnectionRef = ConnectionReference{ + ObjectRef: &corev1.TypedObjectReference{ + APIGroup: ptr("temporal.io"), + Kind: "ClusterConnection", + Name: "shared-connection", + }, + } + Expect(k8sClient.Create(ctx, twd)).To(Succeed()) + }) + + It("rejects connectionRef with both name and objectRef set", func() { + twd := baseTWD("both-set") + twd.Spec.WorkerOptions.ConnectionRef = ConnectionReference{ + Name: "my-connection", + ObjectRef: &corev1.TypedObjectReference{ + APIGroup: ptr("temporal.io"), + Kind: "Connection", + Name: "my-connection", + }, + } + err := k8sClient.Create(ctx, twd) + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("exactly one of name or objectRef")) + }) + + It("rejects connectionRef with neither name nor objectRef set", func() { + twd := baseTWD("neither-set") + twd.Spec.WorkerOptions.ConnectionRef = ConnectionReference{} + err := k8sClient.Create(ctx, twd) + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("exactly one of name or objectRef")) + }) + + It("rejects objectRef with an invalid kind", func() { + twd := baseTWD("bad-kind") + twd.Spec.WorkerOptions.ConnectionRef = ConnectionReference{ + ObjectRef: &corev1.TypedObjectReference{ + APIGroup: ptr("temporal.io"), + Kind: "SomethingRandom", + Name: "my-connection", + }, + } + err := k8sClient.Create(ctx, twd) + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("objectRef.kind must be Connection or ClusterConnection")) + }) + + It("rejects objectRef with a non-temporal.io apiGroup", func() { + twd := baseTWD("bad-apigroup") + twd.Spec.WorkerOptions.ConnectionRef = ConnectionReference{ + ObjectRef: &corev1.TypedObjectReference{ + APIGroup: ptr("example.com"), + Kind: "Connection", + Name: "my-connection", + }, + } + err := k8sClient.Create(ctx, twd) + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("objectRef.apiGroup must be temporal.io")) + }) + + It("rejects objectRef with a populated namespace (cross-namespace not supported yet)", func() { + twd := baseTWD("with-namespace") + twd.Spec.WorkerOptions.ConnectionRef = ConnectionReference{ + ObjectRef: &corev1.TypedObjectReference{ + APIGroup: ptr("temporal.io"), + Kind: "Connection", + Name: "my-connection", + Namespace: ptr("other-namespace"), + }, + } + err := k8sClient.Create(ctx, twd) + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("objectRef.namespace is not supported")) + }) + }) diff --git a/api/v1alpha1/workerdeployment_types.go b/api/v1alpha1/workerdeployment_types.go index ee42c3d8..3d1ba04a 100644 --- a/api/v1alpha1/workerdeployment_types.go +++ b/api/v1alpha1/workerdeployment_types.go @@ -10,26 +10,33 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) -// NOTE: json tags are required. Any new fields you add must have json tags for the fields to be serialized. - -// ConnectionReference identifies a connection resource to use. By default it -// refers to a namespaced Connection in the same namespace as the -// WorkerDeployment. When Kind is "ClusterConnection", it refers to a -// cluster-scoped ClusterConnection instead; The authentication -// secret is still resolved from the WorkerDeployment's own namespace. +// ConnectionReference identifies a connection resource to use. Exactly one of +// Name or ObjectRef must be set. +// +// Name is a shorthand that selects a namespaced Connection in the +// WorkerDeployment's own namespace. ObjectRef is the general form carrying full +// type information (apiGroup + kind + name) and selects either a namespaced +// Connection or a cluster-scoped ClusterConnection. +// +// The authentication secret is always in the WorkerDeployment's own +// namespace regardless of which connection is referenced. Cross-namespace +// connection is not yet implemented, so objectRef.namespace must NOT +// be set. +// +kubebuilder:validation:XValidation:rule="has(self.name) != has(self.objectRef)",message="exactly one of name or objectRef must be set" +// +kubebuilder:validation:XValidation:rule="!has(self.objectRef) || self.objectRef.kind in ['Connection','ClusterConnection']",message="objectRef.kind must be Connection or ClusterConnection" +// +kubebuilder:validation:XValidation:rule="!has(self.objectRef) || self.objectRef.apiGroup == 'temporal.io'",message="objectRef.apiGroup must be temporal.io" +// +kubebuilder:validation:XValidation:rule="!has(self.objectRef) || !has(self.objectRef.__namespace__)",message="objectRef.namespace is not supported yet" type ConnectionReference struct { - // Name of the Connection resource. - // +kubebuilder:validation:Required + // Name of a namespaced Connection in the WorkerDeployment's namespace. + // Shorthand for objectRef: {apiGroup: temporal.io, kind: Connection, name: }. + // +optional // +kubebuilder:validation:Pattern=`^[a-z0-9]([-a-z0-9]*[a-z0-9])?$` - Name string `json:"name"` - // Kind is the type of connection resource. - // ClusterConnection selects a cluster-scoped ClusterConnection. - // Omitting this field preserves the pre-existing behavior of referencing a - // namespaced Connection. + Name string `json:"name,omitempty"` + // ObjectRef references the connection resource by full type information. + // kind must be "Connection" or "ClusterConnection", apiGroup must be + // "temporal.io", and namespace must not be set (not yet supported). // +optional - // +kubebuilder:default=Connection - // +kubebuilder:validation:Enum=Connection;ClusterConnection - Kind string `json:"kind,omitempty"` + ObjectRef *corev1.TypedObjectReference `json:"objectRef,omitempty"` } type WorkerOptions struct { @@ -248,6 +255,12 @@ type WorkerDeploymentStatus struct { // (name or kind) and release the finalizer from the previously-referenced // connection. ObservedConnectionRef *ConnectionReference `json:"observedConnectionRef,omitempty"` + + // ObservedGeneration is the .metadata.generation the controller last + // reconciled. Compare against .metadata.generation to tell whether the + // controller has caught up with the latest spec change. + // +optional + ObservedGeneration int64 `json:"observedGeneration,omitempty"` } // WorkflowExecutionStatus describes the current state of a workflow. diff --git a/api/v1alpha1/zz_generated.deepcopy.go b/api/v1alpha1/zz_generated.deepcopy.go index 333b4482..5446601e 100644 --- a/api/v1alpha1/zz_generated.deepcopy.go +++ b/api/v1alpha1/zz_generated.deepcopy.go @@ -176,6 +176,11 @@ func (in *ConnectionList) DeepCopyObject() runtime.Object { // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *ConnectionReference) DeepCopyInto(out *ConnectionReference) { *out = *in + if in.ObjectRef != nil { + in, out := &in.ObjectRef, &out.ObjectRef + *out = new(v1.TypedObjectReference) + (*in).DeepCopyInto(*out) + } } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ConnectionReference. @@ -843,7 +848,7 @@ func (in *WorkerDeploymentSpec) DeepCopyInto(out *WorkerDeploymentSpec) { } in.RolloutStrategy.DeepCopyInto(&out.RolloutStrategy) in.SunsetStrategy.DeepCopyInto(&out.SunsetStrategy) - out.WorkerOptions = in.WorkerOptions + in.WorkerOptions.DeepCopyInto(&out.WorkerOptions) } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new WorkerDeploymentSpec. @@ -891,7 +896,7 @@ func (in *WorkerDeploymentStatus) DeepCopyInto(out *WorkerDeploymentStatus) { if in.ObservedConnectionRef != nil { in, out := &in.ObservedConnectionRef, &out.ObservedConnectionRef *out = new(ConnectionReference) - **out = **in + (*in).DeepCopyInto(*out) } } @@ -908,7 +913,7 @@ func (in *WorkerDeploymentStatus) DeepCopy() *WorkerDeploymentStatus { // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *WorkerOptions) DeepCopyInto(out *WorkerOptions) { *out = *in - out.ConnectionRef = in.ConnectionRef + in.ConnectionRef.DeepCopyInto(&out.ConnectionRef) } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new WorkerOptions. diff --git a/hack/sync-rbac-rules.py b/hack/sync-rbac-rules.py index 2e7643c9..dfcea9af 100644 --- a/hack/sync-rbac-rules.py +++ b/hack/sync-rbac-rules.py @@ -24,7 +24,12 @@ BEGIN_MARKER_NS = " # GENERATED RULES (NAMESPACED) BEGIN" END_MARKER_NS = " # GENERATED RULES (NAMESPACED) END" -CLUSTER_SCOPED_RESOURCES = {"namespaces", "subjectaccessreviews"} +CLUSTER_SCOPED_RESOURCES = { + "namespaces", + "subjectaccessreviews", + "clusterconnections", + "clusterconnections/finalizers", +} def extract_rules_text(path): @@ -65,20 +70,29 @@ def filter_namespaced(rules_text): if current: blocks.append("".join(current)) - filtered = [] + filtered = [] for block in blocks: - resources = set() + out_lines = [] in_resources = False - for line in block.splitlines(): + kept_resources = 0 + for line in block.splitlines(keepends=True): stripped = line.strip() if stripped == "resources:": in_resources = True - elif in_resources and stripped.startswith("- "): - resources.add(stripped[2:].strip()) - elif in_resources and not stripped.startswith("- "): + out_lines.append(line) + continue + if in_resources and stripped.startswith("- "): + resource = stripped[2:].strip() + if resource in CLUSTER_SCOPED_RESOURCES: + continue + kept_resources += 1 + out_lines.append(line) + continue + if in_resources and not stripped.startswith("- "): in_resources = False - if not CLUSTER_SCOPED_RESOURCES.intersection(resources): - filtered.append(block) + out_lines.append(line) + if kept_resources > 0: + filtered.append("".join(out_lines)) return "".join(filtered) diff --git a/helm/temporal-worker-controller-crds/templates/temporal.io_workerdeployments.yaml b/helm/temporal-worker-controller-crds/templates/temporal.io_workerdeployments.yaml index d50aa4b8..62a34f1a 100644 --- a/helm/temporal-worker-controller-crds/templates/temporal.io_workerdeployments.yaml +++ b/helm/temporal-worker-controller-crds/templates/temporal.io_workerdeployments.yaml @@ -3980,18 +3980,33 @@ spec: properties: connectionRef: properties: - kind: - default: Connection - enum: - - Connection - - ClusterConnection - type: string name: pattern: ^[a-z0-9]([-a-z0-9]*[a-z0-9])?$ type: string - required: - - name + objectRef: + properties: + apiGroup: + type: string + kind: + type: string + name: + type: string + namespace: + type: string + required: + - kind + - name + type: object type: object + x-kubernetes-validations: + - message: exactly one of name or objectRef must be set + rule: has(self.name) != has(self.objectRef) + - message: objectRef.kind must be Connection or ClusterConnection + rule: '!has(self.objectRef) || self.objectRef.kind in [''Connection'',''ClusterConnection'']' + - message: objectRef.apiGroup must be temporal.io + rule: '!has(self.objectRef) || self.objectRef.apiGroup == ''temporal.io''' + - message: objectRef.namespace is not supported yet + rule: '!has(self.objectRef) || !has(self.objectRef.__namespace__)' temporalNamespace: minLength: 1 type: string @@ -4144,18 +4159,36 @@ spec: type: string observedConnectionRef: properties: - kind: - default: Connection - enum: - - Connection - - ClusterConnection - type: string name: pattern: ^[a-z0-9]([-a-z0-9]*[a-z0-9])?$ type: string - required: - - name + objectRef: + properties: + apiGroup: + type: string + kind: + type: string + name: + type: string + namespace: + type: string + required: + - kind + - name + type: object type: object + x-kubernetes-validations: + - message: exactly one of name or objectRef must be set + rule: has(self.name) != has(self.objectRef) + - message: objectRef.kind must be Connection or ClusterConnection + rule: '!has(self.objectRef) || self.objectRef.kind in [''Connection'',''ClusterConnection'']' + - message: objectRef.apiGroup must be temporal.io + rule: '!has(self.objectRef) || self.objectRef.apiGroup == ''temporal.io''' + - message: objectRef.namespace is not supported yet + rule: '!has(self.objectRef) || !has(self.objectRef.__namespace__)' + observedGeneration: + format: int64 + type: integer targetVersion: properties: buildID: diff --git a/helm/temporal-worker-controller/templates/rbac.yaml b/helm/temporal-worker-controller/templates/rbac.yaml index 513bba1d..7a6fa6e8 100644 --- a/helm/temporal-worker-controller/templates/rbac.yaml +++ b/helm/temporal-worker-controller/templates/rbac.yaml @@ -110,7 +110,6 @@ rules: - apiGroups: - temporal.io resources: - - clusterconnections - connections - temporalconnections - workerresourcetemplates @@ -123,7 +122,6 @@ rules: - apiGroups: - temporal.io resources: - - clusterconnections/finalizers - connections/finalizers - temporalconnections/finalizers - temporalworkerdeployments/finalizers diff --git a/internal/controller/clusterconnection_test.go b/internal/controller/clusterconnection_test.go index c800e701..b2ad7e61 100644 --- a/internal/controller/clusterconnection_test.go +++ b/internal/controller/clusterconnection_test.go @@ -8,6 +8,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" temporaliov1alpha1 "github.com/temporalio/temporal-worker-controller/api/v1alpha1" + corev1 "k8s.io/api/core/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" @@ -33,11 +34,20 @@ func makeClusterConnection(name, hostPort string) *temporaliov1alpha1.ClusterCon } } -// makeWDWithKind builds a WorkerDeployment whose connectionRef carries an -// explicit Kind ("", "Connection", or "ClusterConnection"). +// makeWDWithKind builds a WorkerDeployment whose connectionRef targets a +// connection of the given kind ("", "Connection", or "ClusterConnection"). +// "" and "Connection" use the Name shorthand; "ClusterConnection" uses ObjectRef. func makeWDWithKind(name, namespace, connName, kind string) *temporaliov1alpha1.WorkerDeployment { wd := makeWD(name, namespace, connName) - wd.Spec.WorkerOptions.ConnectionRef.Kind = kind + if kind == "ClusterConnection" { + wd.Spec.WorkerOptions.ConnectionRef = temporaliov1alpha1.ConnectionReference{ + ObjectRef: &corev1.TypedObjectReference{ + APIGroup: ptr("temporal.io"), + Kind: "ClusterConnection", + Name: connName, + }, + } + } return wd } @@ -56,7 +66,7 @@ func TestResolveConnection(t *testing.T) { wd := makeWDWithKind("wd", "default", "conn", "Connection") r, _ := newTestReconciler([]client.Object{conn, wd}) - spec, obj, err := r.resolveConnection(ctx, wd) + spec, obj, err := r.getConnectionByRef(ctx, wd.Spec.WorkerOptions.ConnectionRef, wd.Namespace) require.NoError(t, err) assert.Equal(t, "h:7233", spec.HostPort) _, ok := obj.(*temporaliov1alpha1.Connection) @@ -68,7 +78,7 @@ func TestResolveConnection(t *testing.T) { wd := makeWDWithKind("wd", "default", "conn", "") r, _ := newTestReconciler([]client.Object{conn, wd}) - spec, obj, err := r.resolveConnection(ctx, wd) + spec, obj, err := r.getConnectionByRef(ctx, wd.Spec.WorkerOptions.ConnectionRef, wd.Namespace) require.NoError(t, err) assert.Equal(t, "h:7233", spec.HostPort) _, ok := obj.(*temporaliov1alpha1.Connection) @@ -80,7 +90,7 @@ func TestResolveConnection(t *testing.T) { wd := makeWDWithKind("wd", "default", "conn", "ClusterConnection") r, _ := newTestReconciler([]client.Object{cc, wd}) - spec, obj, err := r.resolveConnection(ctx, wd) + spec, obj, err := r.getConnectionByRef(ctx, wd.Spec.WorkerOptions.ConnectionRef, wd.Namespace) require.NoError(t, err) assert.Equal(t, "h:7233", spec.HostPort) got, ok := obj.(*temporaliov1alpha1.ClusterConnection) @@ -95,13 +105,13 @@ func TestResolveConnection(t *testing.T) { wdCC := makeWDWithKind("wd-cc", "default", "foo", "ClusterConnection") r, _ := newTestReconciler([]client.Object{conn, cc, wdNS, wdCC}) - specNS, objNS, err := r.resolveConnection(ctx, wdNS) + specNS, objNS, err := r.getConnectionByRef(ctx, wdNS.Spec.WorkerOptions.ConnectionRef, wdNS.Namespace) require.NoError(t, err) assert.Equal(t, "ns-conn:7233", specNS.HostPort) _, ok := objNS.(*temporaliov1alpha1.Connection) assert.True(t, ok) - specCC, objCC, err := r.resolveConnection(ctx, wdCC) + specCC, objCC, err := r.getConnectionByRef(ctx, wdCC.Spec.WorkerOptions.ConnectionRef, wdCC.Namespace) require.NoError(t, err) assert.Equal(t, "cluster-conn:7233", specCC.HostPort) _, ok = objCC.(*temporaliov1alpha1.ClusterConnection) @@ -113,7 +123,7 @@ func TestResolveConnection(t *testing.T) { wd := makeWDWithKind("wd", "default", "foo", "Connection") r, _ := newTestReconciler([]client.Object{cc, wd}) - _, obj, err := r.resolveConnection(ctx, wd) + _, obj, err := r.getConnectionByRef(ctx, wd.Spec.WorkerOptions.ConnectionRef, wd.Namespace) require.Error(t, err) assert.True(t, apierrors.IsNotFound(err), "must be NotFound, not a stray cluster resolve") assert.Nil(t, obj) @@ -123,7 +133,7 @@ func TestResolveConnection(t *testing.T) { wd := makeWDWithKind("wd", "default", "missing", "Connection") r, _ := newTestReconciler([]client.Object{wd}) - _, obj, err := r.resolveConnection(ctx, wd) + _, obj, err := r.getConnectionByRef(ctx, wd.Spec.WorkerOptions.ConnectionRef, wd.Namespace) require.Error(t, err) assert.True(t, apierrors.IsNotFound(err)) assert.Nil(t, obj) @@ -133,7 +143,7 @@ func TestResolveConnection(t *testing.T) { wd := makeWDWithKind("wd", "default", "missing", "ClusterConnection") r, _ := newTestReconciler([]client.Object{wd}) - _, obj, err := r.resolveConnection(ctx, wd) + _, obj, err := r.getConnectionByRef(ctx, wd.Spec.WorkerOptions.ConnectionRef, wd.Namespace) require.Error(t, err) assert.True(t, apierrors.IsNotFound(err)) assert.Nil(t, obj) @@ -230,7 +240,7 @@ func TestRemoveConnectionFinalizerIfUnused(t *testing.T) { del := makeWDWithKind("del", "default", "conn", "Connection") r, _ := newTestReconciler([]client.Object{conn, del}) - err := r.removeConnectionFinalizerIfUnused(ctx, logr.Discard(), del) + err := r.releaseConnectionFinalizerIfUnused(ctx, logr.Discard(), del.Spec.WorkerOptions.ConnectionRef, del.Namespace, del.Name) require.NoError(t, err) assert.False(t, hasFinalizer(t, r.Client, &temporaliov1alpha1.Connection{}, types.NamespacedName{Name: "conn", Namespace: "default"})) @@ -242,7 +252,7 @@ func TestRemoveConnectionFinalizerIfUnused(t *testing.T) { other := makeWDWithKind("other", "default", "conn", "Connection") r, _ := newTestReconciler([]client.Object{conn, del, other}) - err := r.removeConnectionFinalizerIfUnused(ctx, logr.Discard(), del) + err := r.releaseConnectionFinalizerIfUnused(ctx, logr.Discard(), del.Spec.WorkerOptions.ConnectionRef, del.Namespace, del.Name) require.NoError(t, err) assert.True(t, hasFinalizer(t, r.Client, &temporaliov1alpha1.Connection{}, types.NamespacedName{Name: "conn", Namespace: "default"})) @@ -255,7 +265,7 @@ func TestRemoveConnectionFinalizerIfUnused(t *testing.T) { otherNS := makeWDWithKind("other", "ns-b", "conn", "Connection") r, _ := newTestReconciler([]client.Object{conn, del, otherNS}) - err := r.removeConnectionFinalizerIfUnused(ctx, logr.Discard(), del) + err := r.releaseConnectionFinalizerIfUnused(ctx, logr.Discard(), del.Spec.WorkerOptions.ConnectionRef, del.Namespace, del.Name) require.NoError(t, err) assert.False(t, hasFinalizer(t, r.Client, &temporaliov1alpha1.Connection{}, types.NamespacedName{Name: "conn", Namespace: "default"})) @@ -266,7 +276,7 @@ func TestRemoveConnectionFinalizerIfUnused(t *testing.T) { del := makeWDWithKind("del", "ns-a", "conn", "ClusterConnection") r, _ := newTestReconciler([]client.Object{cc, del}) - err := r.removeConnectionFinalizerIfUnused(ctx, logr.Discard(), del) + err := r.releaseConnectionFinalizerIfUnused(ctx, logr.Discard(), del.Spec.WorkerOptions.ConnectionRef, del.Namespace, del.Name) require.NoError(t, err) assert.False(t, hasFinalizer(t, r.Client, &temporaliov1alpha1.ClusterConnection{}, types.NamespacedName{Name: "conn"})) @@ -281,7 +291,7 @@ func TestRemoveConnectionFinalizerIfUnused(t *testing.T) { otherNS := makeWDWithKind("other", "ns-b", "conn", "ClusterConnection") r, _ := newTestReconciler([]client.Object{cc, del, otherNS}) - err := r.removeConnectionFinalizerIfUnused(ctx, logr.Discard(), del) + err := r.releaseConnectionFinalizerIfUnused(ctx, logr.Discard(), del.Spec.WorkerOptions.ConnectionRef, del.Namespace, del.Name) require.NoError(t, err) assert.True(t, hasFinalizer(t, r.Client, &temporaliov1alpha1.ClusterConnection{}, types.NamespacedName{Name: "conn"}), @@ -298,7 +308,7 @@ func TestRemoveConnectionFinalizerIfUnused(t *testing.T) { nsUser := makeWDWithKind("ns-user", "default", "foo", "Connection") r, _ := newTestReconciler([]client.Object{conn, cc, del, nsUser}) - err := r.removeConnectionFinalizerIfUnused(ctx, logr.Discard(), del) + err := r.releaseConnectionFinalizerIfUnused(ctx, logr.Discard(), del.Spec.WorkerOptions.ConnectionRef, del.Namespace, del.Name) require.NoError(t, err) // ClusterConnection "foo" released (no other cluster referrer) assert.False(t, hasFinalizer(t, r.Client, &temporaliov1alpha1.ClusterConnection{}, @@ -315,7 +325,7 @@ func TestRemoveConnectionFinalizerIfUnused(t *testing.T) { other := makeWDWithKind("samename", "ns-b", "conn", "ClusterConnection") r, _ := newTestReconciler([]client.Object{cc, del, other}) - err := r.removeConnectionFinalizerIfUnused(ctx, logr.Discard(), del) + err := r.releaseConnectionFinalizerIfUnused(ctx, logr.Discard(), del.Spec.WorkerOptions.ConnectionRef, del.Namespace, del.Name) require.NoError(t, err) // the other "samename" in ns-b is a real referrer, not "self" — keep finalizer assert.True(t, hasFinalizer(t, r.Client, &temporaliov1alpha1.ClusterConnection{}, @@ -328,7 +338,7 @@ func TestRemoveConnectionFinalizerIfUnused(t *testing.T) { del := makeWDWithKind("del", "default", "missing", "Connection") r, _ := newTestReconciler([]client.Object{del}) - err := r.removeConnectionFinalizerIfUnused(ctx, logr.Discard(), del) + err := r.releaseConnectionFinalizerIfUnused(ctx, logr.Discard(), del.Spec.WorkerOptions.ConnectionRef, del.Namespace, del.Name) require.NoError(t, err, "missing connection must be treated as already released") }) } @@ -437,7 +447,7 @@ func TestReleaseConnectionFinalizerIfUnused_ReleasesUnused(t *testing.T) { self := makeWDWithKind("w", "default", "new-conn", "Connection") r, _ := newTestReconciler([]client.Object{oldConn, self}) - oldRef := temporaliov1alpha1.ConnectionReference{Name: "old-conn", Kind: "Connection"} + oldRef := temporaliov1alpha1.ConnectionReference{Name: "old-conn"} require.NoError(t, r.releaseConnectionFinalizerIfUnused(ctx, logr.Discard(), oldRef, "default", "w")) assert.False(t, hasFinalizer(t, r.Client, &temporaliov1alpha1.Connection{}, @@ -456,7 +466,13 @@ func TestReleaseConnectionFinalizerIfUnused_KeepsSharedStillUsed(t *testing.T) { r, _ := newTestReconciler([]client.Object{shared, wdB}) // Simulate the WD "w" in ns-a migrating away from "shared". - sharedRef := temporaliov1alpha1.ConnectionReference{Name: "shared", Kind: "ClusterConnection"} + sharedRef := temporaliov1alpha1.ConnectionReference{ + ObjectRef: &corev1.TypedObjectReference{ + APIGroup: ptr("temporal.io"), + Kind: "ClusterConnection", + Name: "shared", + }, + } require.NoError(t, r.releaseConnectionFinalizerIfUnused(ctx, logr.Discard(), sharedRef, "ns-a", "w")) assert.True(t, hasFinalizer(t, r.Client, &temporaliov1alpha1.ClusterConnection{}, @@ -464,12 +480,23 @@ func TestReleaseConnectionFinalizerIfUnused_KeepsSharedStillUsed(t *testing.T) { "shared ClusterConnection finalizer must be KEPT while a WD in another namespace references it") } -// A connectionRef whose Kind was defaulted from -// "" to "Connection" must NOT be seen as a change, or every pre-existing WD would -// try to release its own connection on the first reconcile after upgrade. +// TestSameConnectionRef verifies that the shorthand (Name) and full (ObjectRef) +// forms of the same connection compare equal — so re-expressing a ref from +// shorthand to ObjectRef isn't mistaken for a migration. func TestSameConnectionRef(t *testing.T) { - ref := func(name, kind string) temporaliov1alpha1.ConnectionReference { - return temporaliov1alpha1.ConnectionReference{Name: name, Kind: kind} + // shorthand: namespaced Connection by name + short := func(name string) temporaliov1alpha1.ConnectionReference { + return temporaliov1alpha1.ConnectionReference{Name: name} + } + // full form: ObjectRef with explicit kind + full := func(name, kind string) temporaliov1alpha1.ConnectionReference { + return temporaliov1alpha1.ConnectionReference{ + ObjectRef: &corev1.TypedObjectReference{ + APIGroup: ptr("temporal.io"), + Kind: kind, + Name: name, + }, + } } tests := []struct { @@ -478,27 +505,27 @@ func TestSameConnectionRef(t *testing.T) { want bool }{ { - name: "empty kind equals Connection (normalization)", - a: ref("c", ""), - b: ref("c", "Connection"), + name: "shorthand equals ObjectRef Connection form (normalization)", + a: short("foo"), + b: full("foo", "Connection"), want: true, }, { name: "Connection differs from ClusterConnection", - a: ref("c", "Connection"), - b: ref("c", "ClusterConnection"), + a: full("foo", "Connection"), + b: full("foo", "ClusterConnection"), want: false, }, { name: "different names differ", - a: ref("a", "Connection"), - b: ref("b", "Connection"), + a: short("a"), + b: short("b"), want: false, }, { name: "same cluster ref equals itself", - a: ref("c", "ClusterConnection"), - b: ref("c", "ClusterConnection"), + a: full("foo", "ClusterConnection"), + b: full("foo", "ClusterConnection"), want: true, }, } diff --git a/internal/controller/worker_controller.go b/internal/controller/worker_controller.go index 11e5e183..aa0565db 100644 --- a/internal/controller/worker_controller.go +++ b/internal/controller/worker_controller.go @@ -167,7 +167,7 @@ func (r *WorkerDeploymentReconciler) Reconcile(ctx context.Context, req ctrl.Req } // Remove our finalizer from the Connection if no other WDs reference it. - if err := r.removeConnectionFinalizerIfUnused(ctx, l, &workerDeploy); err != nil { + if err := r.releaseConnectionFinalizerIfUnused(ctx, l, workerDeploy.Spec.WorkerOptions.ConnectionRef, workerDeploy.Namespace, workerDeploy.Name); err != nil { return ctrl.Result{}, err } @@ -222,7 +222,7 @@ func (r *WorkerDeploymentReconciler) Reconcile(ctx context.Context, req ctrl.Req // Note: ConnectionRef.Name is validated by webhook due to +kubebuilder:validation:Required // Fetch the connection parameters - connSpec, connObj, err := r.resolveConnection(ctx, &workerDeploy) + connSpec, connObj, err := r.getConnectionByRef(ctx, workerDeploy.Spec.WorkerOptions.ConnectionRef, workerDeploy.Namespace) if err != nil { l.Error(err, "unable to fetch Connection") r.recordWarningAndSetBlocked(ctx, &workerDeploy, @@ -240,14 +240,16 @@ func (r *WorkerDeploymentReconciler) Reconcile(ctx context.Context, req ctrl.Req return ctrl.Result{}, err } - // If connectionRef changed since we last finalized (name or kind), release the + // If connectionRef changed since we last reconciled (name or kind), release the // finalizer from the previously-referenced connection if no other WD uses it. // The new connection is already protected by ensureConnectionFinalizer above, // so the WD is never left unprotected. - current := workerDeploy.Spec.WorkerOptions.ConnectionRef - if observed := workerDeploy.Status.ObservedConnectionRef; observed != nil && !sameConnectionRef(*observed, current) { - if err := r.releaseConnectionFinalizerIfUnused(ctx, l, *observed, workerDeploy.Namespace, workerDeploy.Name); err != nil { - return ctrl.Result{}, err + if workerDeploy.Generation != workerDeploy.Status.ObservedGeneration { + current := workerDeploy.Spec.WorkerOptions.ConnectionRef + if observed := workerDeploy.Status.ObservedConnectionRef; observed != nil && !sameConnectionRef(*observed, current) { + if err := r.releaseConnectionFinalizerIfUnused(ctx, l, *observed, workerDeploy.Namespace, workerDeploy.Name); err != nil { + return ctrl.Result{}, err + } } } @@ -360,6 +362,7 @@ func (r *WorkerDeploymentReconciler) Reconcile(ctx context.Context, req ctrl.Req // Preserve conditions that were set during this reconciliation status.Conditions = workerDeploy.Status.Conditions status.ObservedConnectionRef = workerDeploy.Spec.WorkerOptions.ConnectionRef.DeepCopy() + status.ObservedGeneration = workerDeploy.Generation workerDeploy.Status = *status // TODO(jlegrone): Set defaults via webhook rather than manually @@ -575,7 +578,7 @@ func (r *WorkerDeploymentReconciler) handleDeletion( // Resolve Connection. // The Connection is guaranteed to exist because we hold a finalizer on it // that prevents deletion while any WD references it. - connSpec, _, err := r.resolveConnection(ctx, workerDeploy) + connSpec, _, err := r.getConnectionByRef(ctx, workerDeploy.Spec.WorkerOptions.ConnectionRef, workerDeploy.Namespace) if err != nil { return fmt.Errorf("unable to fetch Connection: %w", err) } @@ -818,17 +821,22 @@ func (r *WorkerDeploymentReconciler) recordWarningAndSetBlocked( _ = r.Status().Update(ctx, workerDeploy) } -// connectionRefIsCluster reports whether a connectionRef targets a -// cluster-scoped ClusterConnection. +// connectionRefIsCluster reports whether ref targets a cluster-scoped ClusterConnection. func connectionRefIsCluster(ref temporaliov1alpha1.ConnectionReference) bool { - return ref.Kind == "ClusterConnection" + return ref.ObjectRef != nil && ref.ObjectRef.Kind == "ClusterConnection" } -// sameConnectionRef reports whether two connectionRefs point at the same -// "" and "Connection" are treated as equal so a connectionRef whose Kind was -// defaulted from "" to "Connection" must not be seen as a change. +// connectionRefName returns the connection resource name from either form. +func connectionRefName(ref temporaliov1alpha1.ConnectionReference) string { + if ref.ObjectRef != nil { + return ref.ObjectRef.Name + } + return ref.Name +} + +// sameConnectionRef reports whether two connectionRefs resolve to the same connection. func sameConnectionRef(a, b temporaliov1alpha1.ConnectionReference) bool { - return a.Name == b.Name && connectionRefIsCluster(a) == connectionRefIsCluster(b) + return connectionRefName(a) == connectionRefName(b) && connectionRefIsCluster(a) == connectionRefIsCluster(b) } // getConnectionByRef fetches the connection resource identified by ref and @@ -841,16 +849,17 @@ func (r *WorkerDeploymentReconciler) getConnectionByRef( ref temporaliov1alpha1.ConnectionReference, namespace string, ) (temporaliov1alpha1.ConnectionSpec, client.Object, error) { + name := connectionRefName(ref) if connectionRefIsCluster(ref) { var cc temporaliov1alpha1.ClusterConnection - if err := r.Get(ctx, types.NamespacedName{Name: ref.Name}, &cc); err != nil { + if err := r.Get(ctx, types.NamespacedName{Name: name}, &cc); err != nil { return temporaliov1alpha1.ConnectionSpec{}, nil, err } return cc.Spec, &cc, nil } var conn temporaliov1alpha1.Connection if err := r.Get(ctx, types.NamespacedName{ - Name: ref.Name, + Name: name, Namespace: namespace, }, &conn); err != nil { return temporaliov1alpha1.ConnectionSpec{}, nil, err @@ -858,15 +867,6 @@ func (r *WorkerDeploymentReconciler) getConnectionByRef( return conn.Spec, &conn, nil } -// resolveConnection fetches the connection resource referenced by the -// WorkerDeployment's current connectionRef. -func (r *WorkerDeploymentReconciler) resolveConnection( - ctx context.Context, - workerDeploy *temporaliov1alpha1.WorkerDeployment, -) (temporaliov1alpha1.ConnectionSpec, client.Object, error) { - return r.getConnectionByRef(ctx, workerDeploy.Spec.WorkerOptions.ConnectionRef, workerDeploy.Namespace) -} - // ensureConnectionFinalizer adds our finalizer to the Connection so it // cannot be deleted while this WD still needs it for cleanup. func (r *WorkerDeploymentReconciler) ensureConnectionFinalizer( @@ -923,51 +923,36 @@ func (r *WorkerDeploymentReconciler) releaseConnectionFinalizerIfUnused( otherRef := wd.Spec.WorkerOptions.ConnectionRef // Same target only if BOTH name and (normalized) kind match, so a // namespaced Connection "foo" and a ClusterConnection "foo" are distinct. - if otherRef.Name == ref.Name && connectionRefIsCluster(otherRef) == isCluster { + if connectionRefName(otherRef) == connectionRefName(ref) && connectionRefIsCluster(otherRef) == isCluster { l.Info("Connection still referenced by another WorkerDeployment, keeping finalizer", - "connection", ref.Name, "kind", ref.Kind, + "connection", connectionRefName(ref), "clusterScoped", isCluster, "referencedBy", wd.Name, "referencedByNamespace", wd.Namespace) return nil } } - // Fetch by the passed ref, NOT resolveConnection(deletingWD): during a - // connectionRef change the WD's current ref points at the NEW connection, so - // resolveConnection would strip the finalizer off the wrong (new) object. + // Fetch by the passed ref, not the WD's current connectionRef: during a + // connectionRef change the WD's current ref points at the new connection, so + // fetching by current ref would strip the finalizer off the wrong object. _, connObj, err := r.getConnectionByRef(ctx, ref, selfNamespace) if err != nil { if apierrors.IsNotFound(err) { return nil } - return fmt.Errorf("unable to fetch connection %q: %w", ref.Name, err) + return fmt.Errorf("unable to fetch connection %q: %w", connectionRefName(ref), err) } if controllerutil.ContainsFinalizer(connObj, finalizerName) { - l.Info("Removing finalizer from connection", "connection", ref.Name, "kind", ref.Kind) + l.Info("Removing finalizer from connection", "connection", connectionRefName(ref), "clusterScoped", isCluster) controllerutil.RemoveFinalizer(connObj, finalizerName) if err := r.Update(ctx, connObj); err != nil { - return fmt.Errorf("unable to remove finalizer from connection %q: %w", ref.Name, err) + return fmt.Errorf("unable to remove finalizer from connection %q: %w", connectionRefName(ref), err) } } return nil } -// removeConnectionFinalizerIfUnused releases the finalizer from the WD's current -// connection when no other WD references it. Used by the deletion path. -func (r *WorkerDeploymentReconciler) removeConnectionFinalizerIfUnused( - ctx context.Context, - l logr.Logger, - deletingWD *temporaliov1alpha1.WorkerDeployment, -) error { - return r.releaseConnectionFinalizerIfUnused( - ctx, l, - deletingWD.Spec.WorkerOptions.ConnectionRef, - deletingWD.Namespace, - deletingWD.Name, - ) -} - // SetupWithManager sets up the controller with the Manager. func (r *WorkerDeploymentReconciler) SetupWithManager(mgr ctrl.Manager) error { if err := mgr.GetFieldIndexer().IndexField(context.Background(), &appsv1.Deployment{}, deployOwnerKey, func(rawObj client.Object) []string { @@ -1055,7 +1040,7 @@ func (r *WorkerDeploymentReconciler) findTWDsUsingConnection(ctx context.Context for _, twd := range twds.Items { ref := twd.Spec.WorkerOptions.ConnectionRef // Only namespaced Connection refs are driven by this (Connection) watch. - if !connectionRefIsCluster(ref) && ref.Name == tc.GetName() { + if !connectionRefIsCluster(ref) && connectionRefName(ref) == tc.GetName() { requests = append(requests, reconcile.Request{ NamespacedName: types.NamespacedName{ Name: twd.Name, @@ -1079,7 +1064,7 @@ func (r *WorkerDeploymentReconciler) findTWDsUsingClusterConnection( } for _, twd := range twds.Items { ref := twd.Spec.WorkerOptions.ConnectionRef - if connectionRefIsCluster(ref) && ref.Name == cc.GetName() { + if connectionRefIsCluster(ref) && connectionRefName(ref) == cc.GetName() { requests = append(requests, reconcile.Request{ NamespacedName: types.NamespacedName{ Name: twd.Name, From f41c869e19bdbf1e3b3b83bdfe749794a3034607 Mon Sep 17 00:00:00 2001 From: Niyomukiza Mechack Date: Mon, 31 Aug 2026 10:23:13 -0700 Subject: [PATCH 5/5] fix: address review feedback (Jay, Eniko) --- hack/sync-rbac-rules.py | 2 +- internal/controller/worker_controller.go | 4 ++++ 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/hack/sync-rbac-rules.py b/hack/sync-rbac-rules.py index dfcea9af..3e52ef4a 100644 --- a/hack/sync-rbac-rules.py +++ b/hack/sync-rbac-rules.py @@ -70,7 +70,7 @@ def filter_namespaced(rules_text): if current: blocks.append("".join(current)) - filtered = [] + filtered = [] for block in blocks: out_lines = [] in_resources = False diff --git a/internal/controller/worker_controller.go b/internal/controller/worker_controller.go index aa0565db..e4086062 100644 --- a/internal/controller/worker_controller.go +++ b/internal/controller/worker_controller.go @@ -1053,6 +1053,10 @@ func (r *WorkerDeploymentReconciler) findTWDsUsingConnection(ctx context.Context return requests } +// Note: findTWDsUsingClusterConnection performs a cluster-wide list and O(n) scan of WorkerDeployments +// for each ClusterConnection event. This is expected to be acceptable while +// the number of Workerdeployments isn't large, but may increase the controller load +// as the number of workerdeployments grow. func (r *WorkerDeploymentReconciler) findTWDsUsingClusterConnection( ctx context.Context, cc client.Object,