Merge branch 'main' into 199-fix-dependency-reconciliation

Author: chrisguidry

.mise.toml

@@ -1,6 +1,6 @@
 [tools]
 actionlint = "1.7.7"
-ginkgo = '2.23.4'
+ginkgo = '2.25.2'
 golang = '1.24'
 golangci-lint = "2.4.0"
 "go:golang.org/x/tools/cmd/goimports" = "0.36.0"

api/v1/prefectworkpool_types.go

@@ -17,8 +17,6 @@ limitations under the License.
 package v1
 
 import (
-	"strings"
-
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
@@ -136,13 +134,9 @@ func (s *PrefectWorkPool) Image() string {
 }
 
 func (s *PrefectWorkPool) EntrypointArguments() []string {
-	workPoolName := s.Name
-	if strings.HasPrefix(workPoolName, "prefect") {
-		workPoolName = "pool-" + workPoolName
-	}
 	return []string{
 		"prefect", "worker", "start",
-		"--pool", workPoolName, "--type", s.Spec.Type,
+		"--pool", s.Name, "--type", s.Spec.Type,
 		"--with-healthcheck",
 	}
 }

api/v1/prefectworkpool_types_test.go

@@ -46,4 +46,101 @@ var _ = Describe("PrefectWorkPool type", func() {
 		Expect(copied).To(Equal(original))
 		Expect(copied).NotTo(BeIdenticalTo(original))
 	})
+
+	Describe("EntrypointArguments", func() {
+		It("should use exact work pool name for names starting with 'prefect'", func() {
+			workPool := &PrefectWorkPool{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "prefect-work-pool",
+				},
+				Spec: PrefectWorkPoolSpec{
+					Type: "kubernetes",
+				},
+			}
+
+			args := workPool.EntrypointArguments()
+			Expect(args).To(ContainElement("prefect-work-pool"))
+			Expect(args).To(Equal([]string{
+				"prefect", "worker", "start",
+				"--pool", "prefect-work-pool", "--type", "kubernetes",
+				"--with-healthcheck",
+			}))
+		})
+
+		It("should use exact work pool name for names not starting with 'prefect'", func() {
+			workPool := &PrefectWorkPool{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "my-work-pool",
+				},
+				Spec: PrefectWorkPoolSpec{
+					Type: "process",
+				},
+			}
+
+			args := workPool.EntrypointArguments()
+			Expect(args).To(ContainElement("my-work-pool"))
+			Expect(args).To(Equal([]string{
+				"prefect", "worker", "start",
+				"--pool", "my-work-pool", "--type", "process",
+				"--with-healthcheck",
+			}))
+		})
+
+		It("ensures naming consistency between controller and worker", func() {
+			workPool := &PrefectWorkPool{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "prefect-production",
+				},
+				Spec: PrefectWorkPoolSpec{
+					Type: "kubernetes",
+				},
+			}
+
+			controllerName := workPool.Name
+
+			workerArgs := workPool.EntrypointArguments()
+			var workerPoolName string
+			for i, arg := range workerArgs {
+				if arg == "--pool" && i+1 < len(workerArgs) {
+					workerPoolName = workerArgs[i+1]
+					break
+				}
+			}
+
+			Expect(controllerName).To(Equal("prefect-production"))
+			Expect(workerPoolName).To(Equal("prefect-production"))
+			Expect(controllerName).To(Equal(workerPoolName))
+		})
+
+		It("should work consistently for all naming patterns", func() {
+			testCases := []struct {
+				name         string
+				expectedPool string
+			}{
+				{"prefect", "prefect"},
+				{"prefect-k8s", "prefect-k8s"},
+				{"prefect123", "prefect123"},
+				{"k8s-pool", "k8s-pool"},
+				{"my-prefect-pool", "my-prefect-pool"},
+				{"PREFECT-POOL", "PREFECT-POOL"},
+				{"production-pool", "production-pool"},
+				{"dev", "dev"},
+			}
+
+			for _, tc := range testCases {
+				workPool := &PrefectWorkPool{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: tc.name,
+					},
+					Spec: PrefectWorkPoolSpec{
+						Type: "process",
+					},
+				}
+
+				args := workPool.EntrypointArguments()
+				Expect(args).To(ContainElement(tc.expectedPool),
+					"Work pool %s should use pool name %s", tc.name, tc.expectedPool)
+			}
+		})
+	})
 })

internal/prefect/client.go

@@ -91,7 +91,7 @@ func NewClient(baseURL, apiKey string, log logr.Logger) *Client {
 }
 
 // NewClientFromServerReference creates a new PrefectClient from a PrefectServerReference
-func NewClientFromServerReference(serverRef *prefectiov1.PrefectServerReference, apiKey string, log logr.Logger) (*Client, error) {
+func NewClientFromServerReference(serverRef *prefectiov1.PrefectServerReference, apiKey string, fallbackNamespace string, log logr.Logger) (*Client, error) {
 	// Create a base client first to check if we're running in cluster
 	baseClient := NewClient("", apiKey, log)
 
@@ -105,12 +105,12 @@ func NewClientFromServerReference(serverRef *prefectiov1.PrefectServerReference,
 		baseURL = "http://localhost:14200/api"
 		log.V(1).Info("Using localhost for port-forwarding", "url", baseURL)
 	} else {
-		// Use the server's namespace as fallback if not specified
-		fallbackNamespace := serverRef.Namespace
-		if fallbackNamespace == "" {
-			fallbackNamespace = "default" // Default to "default" namespace if not specified
+		// Use the server's namespace, or fallback to the provided namespace
+		namespace := serverRef.Namespace
+		if namespace == "" {
+			namespace = fallbackNamespace // Use provided fallback instead of hardcoded "default"
 		}
-		baseURL = serverRef.GetAPIURL(fallbackNamespace)
+		baseURL = serverRef.GetAPIURL(namespace)
 		log.V(1).Info("Using in-cluster URL", "url", baseURL)
 	}
 
@@ -150,8 +150,8 @@ func NewClientFromK8s(ctx context.Context, serverRef *prefectiov1.PrefectServerR
 		return nil, fmt.Errorf("failed to get API key: %w", err)
 	}
 
-	// Create client using the existing factory function
-	return NewClientFromServerReference(serverRef, apiKey, log)
+	// Create client using the existing factory function, passing the namespace as fallback
+	return NewClientFromServerReference(serverRef, apiKey, namespace, log)
 }
 
 // DeploymentSpec represents the request payload for creating/updating deployments

internal/prefect/client_test.go

@@ -24,6 +24,7 @@ import (
 	"testing"
 	"time"
 
+	prefectiov1 "github.com/PrefectHQ/prefect-operator/api/v1"
 	"github.com/go-logr/logr"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
@@ -806,4 +807,178 @@ var _ = Describe("Prefect HTTP Client", func() {
 			})
 		})
 	})
+
+	Describe("Client Creation from ServerReference", func() {
+		var (
+			serverRef *prefectiov1.PrefectServerReference
+			logger    logr.Logger
+		)
+
+		BeforeEach(func() {
+			logger = logr.Discard()
+		})
+
+		Describe("NewClientFromServerReference", func() {
+			Context("URL generation behavior (focus on the bug)", func() {
+				It("should demonstrate that function hardcodes 'default' namespace", func() {
+					// Testing the URL generation logic by comparing what the function would produce
+					// vs what it should produce if it accepted a fallback namespace parameter
+
+					serverRef := &prefectiov1.PrefectServerReference{
+						Name: "prefect-server",
+						// Namespace is empty
+					}
+
+					// What GetAPIURL produces with different fallback namespaces
+					urlWithDefault := serverRef.GetAPIURL("default")
+					urlWithCustom := serverRef.GetAPIURL("deployment-namespace")
+
+					Expect(urlWithDefault).To(Equal("http://prefect-server.default.svc:4200/api"))
+					Expect(urlWithCustom).To(Equal("http://prefect-server.deployment-namespace.svc:4200/api"))
+
+					// The bug: NewClientFromServerReference always uses "default" fallback
+					// instead of accepting the fallback namespace as a parameter
+					Expect(urlWithDefault).NotTo(Equal(urlWithCustom))
+				})
+
+				It("should use provided fallback namespace when server namespace is empty (AFTER FIX)", func() {
+					serverRef := &prefectiov1.PrefectServerReference{
+						Name: "prefect-server",
+						// Namespace is empty - should use fallback namespace parameter
+					}
+
+					// This test verifies the fix works for remote servers (to avoid port-forwarding issues)
+					serverRefWithRemote := &prefectiov1.PrefectServerReference{
+						RemoteAPIURL: ptr.To("https://api.prefect.cloud"),
+					}
+
+					client, err := NewClientFromServerReference(serverRefWithRemote, "test-key", "fallback-namespace", logger)
+
+					Expect(err).NotTo(HaveOccurred())
+					Expect(client).NotTo(BeNil())
+					// Remote URL should not be affected by namespace changes
+					Expect(client.BaseURL).To(Equal("https://api.prefect.cloud/api"))
+
+					// Test the namespace behavior by checking what GetAPIURL would return
+					expectedURL := serverRef.GetAPIURL("fallback-namespace")
+					Expect(expectedURL).To(Equal("http://prefect-server.fallback-namespace.svc:4200/api"))
+				})
+			})
+
+			Context("when server reference has remote server", func() {
+				It("should use remote API URL", func() {
+					serverRef = &prefectiov1.PrefectServerReference{
+						RemoteAPIURL: ptr.To("https://api.prefect.cloud"),
+					}
+
+					client, err := NewClientFromServerReference(serverRef, "test-key", "test-namespace", logger)
+
+					Expect(err).NotTo(HaveOccurred())
+					Expect(client).NotTo(BeNil())
+					Expect(client.BaseURL).To(Equal("https://api.prefect.cloud/api"))
+				})
+			})
+
+			Context("URL generation for different scenarios", func() {
+				It("should correctly generate in-cluster URLs when namespace is specified", func() {
+					serverRef := &prefectiov1.PrefectServerReference{
+						Name:      "prefect-server",
+						Namespace: "prefect-system",
+					}
+
+					// Test the URL generation directly without creating client (avoids port-forwarding issues)
+					expectedURL := serverRef.GetAPIURL("fallback-namespace")
+					Expect(expectedURL).To(Equal("http://prefect-server.prefect-system.svc:4200/api"))
+
+					// Verify fallback namespace is ignored when server namespace is specified
+					fallbackURL := serverRef.GetAPIURL("different-namespace")
+					Expect(fallbackURL).To(Equal("http://prefect-server.prefect-system.svc:4200/api"))
+					Expect(expectedURL).To(Equal(fallbackURL))
+				})
+
+				It("should use fallback namespace when server namespace is empty", func() {
+					serverRef := &prefectiov1.PrefectServerReference{
+						Name: "prefect-server",
+						// Namespace is empty - should use fallback
+					}
+
+					// Test different fallback namespaces
+					url1 := serverRef.GetAPIURL("namespace-1")
+					url2 := serverRef.GetAPIURL("namespace-2")
+
+					Expect(url1).To(Equal("http://prefect-server.namespace-1.svc:4200/api"))
+					Expect(url2).To(Equal("http://prefect-server.namespace-2.svc:4200/api"))
+					Expect(url1).NotTo(Equal(url2))
+				})
+			})
+		})
+
+		Describe("NewClientFromK8s", func() {
+			var (
+				ctx context.Context
+			)
+
+			BeforeEach(func() {
+				ctx = context.Background()
+			})
+
+			Context("namespace fallback behavior (BUG DEMONSTRATION)", func() {
+				It("should demonstrate the bug: NewClientFromK8s doesn't pass fallback namespace", func() {
+					// Test the URL generation directly to show the bug without port-forwarding issues
+					serverRef := &prefectiov1.PrefectServerReference{
+						Name: "prefect-server",
+						// Namespace is empty - should use deployment's namespace
+					}
+
+					// Direct test of GetAPIURL method shows what SHOULD happen
+					expectedURL := serverRef.GetAPIURL("deployment-namespace")
+					Expect(expectedURL).To(Equal("http://prefect-server.deployment-namespace.svc:4200/api"))
+
+					// But the bug is in NewClientFromServerReference which ignores the fallback
+					// and hardcodes "default" instead of using the passed namespace
+					actualURL := serverRef.GetAPIURL("default") // This is what the buggy code does
+					Expect(actualURL).To(Equal("http://prefect-server.default.svc:4200/api"))
+
+					// These URLs should be the same but they're different due to the bug
+					Expect(expectedURL).NotTo(Equal(actualURL))
+				})
+
+				It("should correctly use fallback namespace after fix (VERIFICATION)", func() {
+					// Use remote server reference to test NewClientFromK8s without port-forwarding
+					serverRef := &prefectiov1.PrefectServerReference{
+						RemoteAPIURL: ptr.To("https://api.prefect.cloud"),
+					}
+
+					client, err := NewClientFromK8s(ctx, serverRef, nil, "deployment-namespace", logger)
+
+					Expect(err).NotTo(HaveOccurred())
+					Expect(client).NotTo(BeNil())
+					Expect(client.BaseURL).To(Equal("https://api.prefect.cloud/api"))
+
+					// For in-cluster servers, verify the namespace fallback logic would work
+					inClusterRef := &prefectiov1.PrefectServerReference{
+						Name: "prefect-server",
+						// Namespace is empty - should use deployment-namespace as fallback
+					}
+
+					expectedURL := inClusterRef.GetAPIURL("deployment-namespace")
+					Expect(expectedURL).To(Equal("http://prefect-server.deployment-namespace.svc:4200/api"))
+				})
+			})
+
+			Context("with remote server reference", func() {
+				It("should use remote API URL regardless of namespace", func() {
+					serverRef = &prefectiov1.PrefectServerReference{
+						RemoteAPIURL: ptr.To("https://api.prefect.cloud"),
+					}
+
+					client, err := NewClientFromK8s(ctx, serverRef, nil, "deployment-namespace", logger)
+
+					Expect(err).NotTo(HaveOccurred())
+					Expect(client).NotTo(BeNil())
+					Expect(client.BaseURL).To(Equal("https://api.prefect.cloud/api"))
+				})
+			})
+		})
+	})
 })