FEATURE: Add ClusterConnection cluster-scoped resource

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

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

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: <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.

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)