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_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 f0aaf615..3d1ba04a 100644 --- a/api/v1alpha1/workerdeployment_types.go +++ b/api/v1alpha1/workerdeployment_types.go @@ -10,19 +10,39 @@ 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 contains the name of a Connection resource -// in the same namespace as the WorkerDeployment. +// 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"` + 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 + ObjectRef *corev1.TypedObjectReference `json:"objectRef,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 @@ -229,6 +249,18 @@ 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"` + + // 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 056a0b95..5446601e 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 @@ -117,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. @@ -784,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. @@ -829,6 +893,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) + (*in).DeepCopyInto(*out) + } } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new WorkerDeploymentStatus. @@ -844,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..3e52ef4a 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): @@ -67,18 +72,27 @@ def filter_namespaced(rules_text): 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_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..62a34f1a 100644 --- a/helm/temporal-worker-controller-crds/templates/temporal.io_workerdeployments.yaml +++ b/helm/temporal-worker-controller-crds/templates/temporal.io_workerdeployments.yaml @@ -3983,9 +3983,30 @@ spec: 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 @@ -4136,6 +4157,38 @@ spec: type: string managerIdentity: type: string + observedConnectionRef: + properties: + name: + pattern: ^[a-z0-9]([-a-z0-9]*[a-z0-9])?$ + type: string + 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 756bec18..7a6fa6e8 100644 --- a/helm/temporal-worker-controller/templates/rbac.yaml +++ b/helm/temporal-worker-controller/templates/rbac.yaml @@ -228,6 +228,7 @@ rules: - apiGroups: - temporal.io resources: + - clusterconnections - connections - temporalconnections - workerresourcetemplates @@ -240,6 +241,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..b2ad7e61 --- /dev/null +++ b/internal/controller/clusterconnection_test.go @@ -0,0 +1,538 @@ +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" + 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" + "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 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) + if kind == "ClusterConnection" { + wd.Spec.WorkerOptions.ConnectionRef = temporaliov1alpha1.ConnectionReference{ + ObjectRef: &corev1.TypedObjectReference{ + APIGroup: ptr("temporal.io"), + Kind: "ClusterConnection", + Name: connName, + }, + } + } + 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.getConnectionByRef(ctx, wd.Spec.WorkerOptions.ConnectionRef, wd.Namespace) + 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.getConnectionByRef(ctx, wd.Spec.WorkerOptions.ConnectionRef, wd.Namespace) + 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.getConnectionByRef(ctx, wd.Spec.WorkerOptions.ConnectionRef, wd.Namespace) + 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.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.getConnectionByRef(ctx, wdCC.Spec.WorkerOptions.ConnectionRef, wdCC.Namespace) + 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.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) + }) + + t.Run("NotFound_namespaced", func(t *testing.T) { + wd := makeWDWithKind("wd", "default", "missing", "Connection") + r, _ := newTestReconciler([]client.Object{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) + }) + + t.Run("NotFound_cluster", func(t *testing.T) { + wd := makeWDWithKind("wd", "default", "missing", "ClusterConnection") + r, _ := newTestReconciler([]client.Object{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) + }) +} + +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.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"})) + }) + + 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.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"})) + }) + + 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.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"})) + }) + + 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.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"})) + }) + + // 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.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"}), + "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.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{}, + 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.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{}, + 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.releaseConnectionFinalizerIfUnused(ctx, logr.Discard(), del.Spec.WorkerOptions.ConnectionRef, del.Namespace, del.Name) + 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) + }) +} + +// 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"} + 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{ + 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{}, + types.NamespacedName{Name: "shared"}), + "shared ClusterConnection finalizer must be KEPT while a WD in another namespace references it") +} + +// 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) { + // 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 { + name string + a, b temporaliov1alpha1.ConnectionReference + want bool + }{ + { + name: "shorthand equals ObjectRef Connection form (normalization)", + a: short("foo"), + b: full("foo", "Connection"), + want: true, + }, + { + name: "Connection differs from ClusterConnection", + a: full("foo", "Connection"), + b: full("foo", "ClusterConnection"), + want: false, + }, + { + name: "different names differ", + a: short("a"), + b: short("b"), + want: false, + }, + { + name: "same cluster ref equals itself", + a: full("foo", "ClusterConnection"), + b: full("foo", "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 099074e2..e4086062 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 @@ -165,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 } @@ -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.getConnectionByRef(ctx, workerDeploy.Spec.WorkerOptions.ConnectionRef, workerDeploy.Namespace) + if err != nil { l.Error(err, "unable to fetch Connection") r.recordWarningAndSetBlocked(ctx, &workerDeploy, temporaliov1alpha1.ReasonConnectionNotFound, @@ -233,13 +232,27 @@ 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 } + // 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. + 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 + } + } + } + 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 @@ -248,7 +261,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 } @@ -348,6 +361,8 @@ 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 @@ -563,13 +578,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.getConnectionByRef(ctx, workerDeploy.Spec.WorkerOptions.ConnectionRef, workerDeploy.Namespace) + 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,68 +821,132 @@ func (r *WorkerDeploymentReconciler) recordWarningAndSetBlocked( _ = r.Status().Update(ctx, workerDeploy) } +// connectionRefIsCluster reports whether ref targets a cluster-scoped ClusterConnection. +func connectionRefIsCluster(ref temporaliov1alpha1.ConnectionReference) bool { + return ref.ObjectRef != nil && ref.ObjectRef.Kind == "ClusterConnection" +} + +// 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 connectionRefName(a) == connectionRefName(b) && 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, + 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: 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: name, + Namespace: 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 } -// 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 { - connectionName := deletingWD.Spec.WorkerOptions.ConnectionRef.Name + 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 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(selfNamespace)) + } - // 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 == selfNamespace && wd.Name == selfName { 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 connectionRefName(otherRef) == connectionRefName(ref) && connectionRefIsCluster(otherRef) == isCluster { + l.Info("Connection still referenced by another WorkerDeployment, keeping finalizer", + "connection", connectionRefName(ref), "clusterScoped", isCluster, + "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 { + // 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 // already gone + return nil } - return fmt.Errorf("unable to fetch Connection %q: %w", connectionName, err) + return fmt.Errorf("unable to fetch connection %q: %w", connectionRefName(ref), 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", 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", connectionRefName(ref), err) } } @@ -915,6 +992,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 +1038,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) && connectionRefName(ref) == tc.GetName() { requests = append(requests, reconcile.Request{ NamespacedName: types.NamespacedName{ Name: twd.Name, @@ -974,6 +1053,33 @@ 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, +) []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) && connectionRefName(ref) == 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) {