Allows work pools for Prefect servers without API keys, and includes …

…examples (#70)

We were inadvertently requiring an API key to connect to a remote Prefect
server, but that is only required for Prefect Cloud at this time.  This removes
that restriction, and adds examples of Prefect servers and Prefect Cloud
configurations for reference.

api/v1/prefectworkpool_types.go

@@ -143,11 +143,10 @@ func (s *PrefectWorkPool) Command() []string {
 	}
 }
 
-// PrefectAPIURL returns the API URL for the Prefect Server.
-// If an API Key is provided, it will return the RemoteAPIURL.
-// Otherwise, it will default to the local, in-cluster API URL.
+// PrefectAPIURL returns the API URL for the Prefect Server, either from the RemoteAPIURL or
+// from the in-cluster server
 func (s *PrefectWorkPool) PrefectAPIURL() string {
-	if s.Spec.Server.APIKey != nil && s.Spec.Server.RemoteAPIURL != nil {
+	if s.Spec.Server.RemoteAPIURL != nil {
 		remote := *s.Spec.Server.RemoteAPIURL
 		if !strings.HasSuffix(remote, "/api") {
 			remote = fmt.Sprintf("%s/api", remote)

config/samples/v1_prefectworkpool_external_prefect_cloud.yaml

@@ -0,0 +1,27 @@
+apiVersion: v1
+kind: Secret
+metadata:
+  name: the-api-key-secret
+type: Opaque
+stringData:
+  the-api-key: "pnu_XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"
+---
+apiVersion: prefect.io/v1
+kind: PrefectWorkPool
+metadata:
+  labels:
+    app.kubernetes.io/name: prefect-operator
+    app.kubernetes.io/managed-by: kustomize
+  name: external-pool
+spec:
+  type: process
+  server:
+    remoteApiUrl: https://api.prefect.cloud
+    accountId:   aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa
+    workspaceId: bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb
+    apiKey:
+      valueFrom:
+        secretKeyRef:
+          name: 'the-api-key-secret'
+          key: 'the-api-key'
+  workers: 3

config/samples/v1_prefectworkpool_external_prefect_server.yaml

@@ -0,0 +1,12 @@
+apiVersion: prefect.io/v1
+kind: PrefectWorkPool
+metadata:
+  labels:
+    app.kubernetes.io/name: prefect-operator
+    app.kubernetes.io/managed-by: kustomize
+  name: external-pool
+spec:
+  type: process
+  server:
+    remoteApiUrl: https://my-server.example.com
+  workers: 3

internal/controller/prefectworkpool_controller_test.go

@@ -608,6 +608,94 @@ var _ = Describe("PrefectWorkPool Controller", func() {
 		})
 	})
 
+	It("should set PREFECT_API_URL when provided", func() {
+		name := types.NamespacedName{
+			Namespace: namespaceName,
+			Name:      "example-work-pool-with-api-key",
+		}
+
+		prefectworkpool := &prefectiov1.PrefectWorkPool{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: name.Namespace,
+				Name:      name.Name,
+			},
+			Spec: prefectiov1.PrefectWorkPoolSpec{
+				Server: prefectiov1.PrefectServerReference{
+					Name:         "test-server",
+					Namespace:    name.Namespace,
+					RemoteAPIURL: ptr.To("https://some-server.example.com/api"),
+				},
+			},
+		}
+		Expect(k8sClient.Create(ctx, prefectworkpool)).To(Succeed())
+
+		controllerReconciler := &PrefectWorkPoolReconciler{
+			Client: k8sClient,
+			Scheme: k8sClient.Scheme(),
+		}
+
+		_, err := controllerReconciler.Reconcile(ctx, reconcile.Request{
+			NamespacedName: name,
+		})
+		Expect(err).NotTo(HaveOccurred())
+
+		deployment := &appsv1.Deployment{}
+		Eventually(func() error {
+			return k8sClient.Get(ctx, name, deployment)
+		}).Should(Succeed())
+
+		Expect(deployment.Spec.Template.Spec.Containers).To(HaveLen(1))
+		container := deployment.Spec.Template.Spec.Containers[0]
+		Expect(container.Env).To(ContainElement(corev1.EnvVar{
+			Name:  "PREFECT_API_URL",
+			Value: "https://some-server.example.com/api",
+		}))
+	})
+
+	It("should ensure PREFECT_API_URL ends with /api when provided", func() {
+		name := types.NamespacedName{
+			Namespace: namespaceName,
+			Name:      "example-work-pool-with-api-key",
+		}
+
+		prefectworkpool := &prefectiov1.PrefectWorkPool{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: name.Namespace,
+				Name:      name.Name,
+			},
+			Spec: prefectiov1.PrefectWorkPoolSpec{
+				Server: prefectiov1.PrefectServerReference{
+					Name:         "test-server",
+					Namespace:    name.Namespace,
+					RemoteAPIURL: ptr.To("https://some-server.example.com"),
+				},
+			},
+		}
+		Expect(k8sClient.Create(ctx, prefectworkpool)).To(Succeed())
+
+		controllerReconciler := &PrefectWorkPoolReconciler{
+			Client: k8sClient,
+			Scheme: k8sClient.Scheme(),
+		}
+
+		_, err := controllerReconciler.Reconcile(ctx, reconcile.Request{
+			NamespacedName: name,
+		})
+		Expect(err).NotTo(HaveOccurred())
+
+		deployment := &appsv1.Deployment{}
+		Eventually(func() error {
+			return k8sClient.Get(ctx, name, deployment)
+		}).Should(Succeed())
+
+		Expect(deployment.Spec.Template.Spec.Containers).To(HaveLen(1))
+		container := deployment.Spec.Template.Spec.Containers[0]
+		Expect(container.Env).To(ContainElement(corev1.EnvVar{
+			Name:  "PREFECT_API_URL",
+			Value: "https://some-server.example.com/api",
+		}))
+	})
+
 	It("should set PREFECT_API_KEY and a remote PREFECT_API_URL when apiKey.value is provided", func() {
 		name := types.NamespacedName{
 			Namespace: namespaceName,