add support for job template patches and configmap reference

Author: tinkerborg

api/v1/prefectworkpool_types.go

@@ -21,7 +21,6 @@ import (
 
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/intstr"
 )
 
@@ -55,7 +54,7 @@ type PrefectWorkPoolSpec struct {
 	DeploymentLabels map[string]string `json:"deploymentLabels,omitempty"`
 
 	// Base job template for flow runs on this Work Pool
-	BaseJobTemplate *runtime.RawExtension `json:"baseJobTemplate,omitempty"`
+	BaseJobTemplate *RawValueSource `json:"baseJobTemplate,omitempty"`
 }
 
 // PrefectWorkPoolStatus defines the observed state of PrefectWorkPool
@@ -84,6 +83,10 @@ type PrefectWorkPoolStatus struct {
 	// +optional
 	ObservedGeneration int64 `json:"observedGeneration,omitempty"`
 
+	// BaseJobTemplateVersion tracks the version of BaseJobTemplate ConfigMap, if any is defined
+	// +optional
+	BaseJobTemplateVersion string `json:"baseJobTemplateVersion,omitempty"`
+
 	// Conditions store the status conditions of the PrefectWorkPool instances
 	Conditions []metav1.Condition `json:"conditions,omitempty" patchStrategy:"merge" patchMergeKey:"type" protobuf:"bytes,1,rep,name=conditions"`
 }

api/v1/raw_value_source.go

@@ -0,0 +1,31 @@
+package v1
+
+import (
+	"fmt"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+)
+
+type RawValueSource struct {
+	Value     *runtime.RawExtension        `json:"value,omitempty"`
+	ConfigMap *corev1.ConfigMapKeySelector `json:"configMap,omitempty"`
+	Patches   []JsonPatch                  `json:"patches,omitempty"`
+}
+
+type JsonPatch struct {
+	Operation string `json:"op"`
+	Path      string `json:"path"`
+
+	// +kubebuilder:pruning:PreserveUnknownFields
+	// +kubebuilder:validation:Schemaless
+	Value *runtime.RawExtension `json:"value,omitempty"`
+}
+
+// TODO - no admission webhook yet
+func (spec *RawValueSource) Validate() error {
+	if spec.Value != nil && spec.ConfigMap != nil {
+		return fmt.Errorf("value and configMap are mutually exclusive")
+	}
+	return nil
+}

api/v1/zz_generated.deepcopy.go

@@ -63,9 +63,51 @@ spec:
             description: PrefectWorkPoolSpec defines the desired state of PrefectWorkPool
             properties:
               baseJobTemplate:
-                description: Base job template for flow runs on this Work Pool
+                description: |-
+                  Base job template for flow runs on this Work Pool
+                  BaseJobTemplate *runtime.RawExtension `json:"baseJobTemplate,omitempty"`
+                properties:
+                  configMap:
+                    description: Selects a key from a ConfigMap.
+                    properties:
+                      key:
+                        description: The key to select.
+                        type: string
+                      name:
+                        default: ""
+                        description: |-
+                          Name of the referent.
+                          This field is effectively required, but due to backwards compatibility is
+                          allowed to be empty. Instances of this type with an empty value here are
+                          almost certainly wrong.
+                          More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names
+                        type: string
+                      optional:
+                        description: Specify whether the ConfigMap or its key must
+                          be defined
+                        type: boolean
+                    required:
+                    - key
+                    type: object
+                    x-kubernetes-map-type: atomic
+                  patches:
+                    items:
+                      properties:
+                        op:
+                          type: string
+                        path:
+                          type: string
+                        value:
+                          x-kubernetes-preserve-unknown-fields: true
+                      required:
+                      - op
+                      - path
+                      type: object
+                    type: array
+                  value:
+                    type: object
+                    x-kubernetes-preserve-unknown-fields: true
                 type: object
-                x-kubernetes-preserve-unknown-fields: true
               deploymentLabels:
                 additionalProperties:
                   type: string
@@ -1995,6 +2037,10 @@ spec:
           status:
             description: PrefectWorkPoolStatus defines the observed state of PrefectWorkPool
             properties:
+              baseJobTemplateVersion:
+                description: BaseJobTemplateVersion tracks the version of BaseJobTemplate
+                  ConfigMap, if any is defined
+                type: string
               conditions:
                 description: Conditions store the status conditions of the PrefectWorkPool
                   instances

deploy/charts/prefect-operator/crds/prefect.io_prefectworkpools.yaml

@@ -4,6 +4,7 @@ go 1.24.2
 
 require (
 	dario.cat/mergo v1.0.2
+	github.com/evanphx/json-patch v0.5.2
 	github.com/go-logr/logr v1.4.3
 	github.com/google/uuid v1.6.0
 	github.com/onsi/ginkgo/v2 v2.25.2

go.mod

@@ -50,6 +50,7 @@ github.com/google/pprof v0.0.0-20250403155104-27863c87afa6 h1:BHT72Gu3keYf3ZEu2J
 github.com/google/pprof v0.0.0-20250403155104-27863c87afa6/go.mod h1:boTsfXsheKC2y+lKOCMpSfarhxDeIzfZG1jqGcPl3cA=
 github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/jessevdk/go-flags v1.4.0/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
 github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY=
 github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y=
 github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=

go.sum

@@ -18,6 +18,7 @@ package controller
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	"github.com/PrefectHQ/prefect-operator/internal/prefect"
@@ -28,10 +29,13 @@ import (
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
 	prefectiov1 "github.com/PrefectHQ/prefect-operator/api/v1"
 	"github.com/PrefectHQ/prefect-operator/internal/conditions"
@@ -108,15 +112,25 @@ func (r *PrefectWorkPoolReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 		return ctrl.Result{RequeueAfter: time.Second}, nil
 	}
 
+	var baseJobTemplateConfigMap corev1.ConfigMap
+
+	if workPool.Spec.BaseJobTemplate != nil && workPool.Spec.BaseJobTemplate.ConfigMap != nil {
+		configMapRef := workPool.Spec.BaseJobTemplate.ConfigMap
+		err := r.Get(ctx, types.NamespacedName{Name: configMapRef.Name, Namespace: workPool.Namespace}, &baseJobTemplateConfigMap)
+		if err != nil {
+			return ctrl.Result{}, err
+		}
+	}
+
 	specHash, err := utils.Hash(workPool.Spec, 16)
 	if err != nil {
 		log.Error(err, "Failed to calculate spec hash", "workPool", workPool.Name)
 		return ctrl.Result{}, err
 	}
 
-	if r.needsSync(&workPool, specHash) {
+	if r.needsSync(&workPool, specHash, &baseJobTemplateConfigMap) {
 		log.Info("Starting sync with Prefect API", "deployment", workPool.Name)
-		err := r.syncWithPrefect(ctx, &workPool)
+		err := r.syncWithPrefect(ctx, &workPool, &baseJobTemplateConfigMap)
 		if err != nil {
 			return ctrl.Result{}, err
 		}
@@ -221,7 +235,7 @@ func (r *PrefectWorkPoolReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 	return ctrl.Result{}, nil
 }
 
-func (r *PrefectWorkPoolReconciler) needsSync(workPool *prefectiov1.PrefectWorkPool, currentSpecHash string) bool {
+func (r *PrefectWorkPoolReconciler) needsSync(workPool *prefectiov1.PrefectWorkPool, currentSpecHash string, baseJobTemplateConfigMap *corev1.ConfigMap) bool {
 	if workPool.Status.Id == nil || *workPool.Status.Id == "" {
 		return true
 	}
@@ -234,6 +248,10 @@ func (r *PrefectWorkPoolReconciler) needsSync(workPool *prefectiov1.PrefectWorkP
 		return true
 	}
 
+	if workPool.Status.BaseJobTemplateVersion != baseJobTemplateConfigMap.ResourceVersion {
+		return true
+	}
+
 	// Drift detection: sync if last sync was too long ago
 	if workPool.Status.LastSyncTime == nil {
 		return true
@@ -243,7 +261,7 @@ func (r *PrefectWorkPoolReconciler) needsSync(workPool *prefectiov1.PrefectWorkP
 	return timeSinceLastSync > 10*time.Minute
 }
 
-func (r *PrefectWorkPoolReconciler) syncWithPrefect(ctx context.Context, workPool *prefectiov1.PrefectWorkPool) error {
+func (r *PrefectWorkPoolReconciler) syncWithPrefect(ctx context.Context, workPool *prefectiov1.PrefectWorkPool, baseJobTemplateConfigMap *corev1.ConfigMap) error {
 	name := workPool.Name
 	log := log.FromContext(ctx)
 
@@ -259,8 +277,21 @@ func (r *PrefectWorkPoolReconciler) syncWithPrefect(ctx context.Context, workPoo
 		return err
 	}
 
+	var baseJobTemplate []byte
+
+	if baseJobTemplateConfigMap.Name != "" {
+		key := workPool.Spec.BaseJobTemplate.ConfigMap.Key
+
+		baseJobTemplateJson, exists := baseJobTemplateConfigMap.Data[key]
+		if !exists {
+			return fmt.Errorf("can't find key %s in ConfigMap %s", key, baseJobTemplateConfigMap.Name)
+		}
+
+		baseJobTemplate = []byte(baseJobTemplateJson)
+	}
+
 	if prefectWorkPool == nil {
-		workPoolSpec, err := prefect.ConvertToWorkPoolSpec(workPool)
+		workPoolSpec, err := prefect.ConvertToWorkPoolSpec(ctx, workPool, baseJobTemplate, prefectClient)
 		if err != nil {
 			log.Error(err, "Failed to convert work pool spec", "workPool", name)
 			r.setCondition(workPool, PrefectWorkPoolConditionSynced, metav1.ConditionFalse, "ConversionError", err.Error())
@@ -274,7 +305,7 @@ func (r *PrefectWorkPoolReconciler) syncWithPrefect(ctx context.Context, workPoo
 			return err
 		}
 	} else {
-		workPoolSpec, err := prefect.ConvertToWorkPoolUpdateSpec(workPool)
+		workPoolSpec, err := prefect.ConvertToWorkPoolUpdateSpec(ctx, workPool, baseJobTemplate, prefectClient)
 		if err != nil {
 			log.Error(err, "Failed to convert work pool spec", "workPool", name)
 			r.setCondition(workPool, PrefectWorkPoolConditionSynced, metav1.ConditionFalse, "ConversionError", err.Error())
@@ -302,6 +333,7 @@ func (r *PrefectWorkPoolReconciler) syncWithPrefect(ctx context.Context, workPoo
 	workPool.Status.SpecHash = specHash
 	workPool.Status.ObservedGeneration = workPool.Generation
 	workPool.Status.LastSyncTime = &now
+	workPool.Status.BaseJobTemplateVersion = baseJobTemplateConfigMap.ResourceVersion
 
 	r.setCondition(workPool, PrefectWorkPoolConditionSynced, metav1.ConditionTrue, "SyncSuccessful", "Work pool successfully synced with Prefect API")
 
@@ -384,10 +416,49 @@ func (r *PrefectWorkPoolReconciler) getPrefectClient(ctx context.Context, workPo
 	return prefectClient, nil
 }
 
+// func (r *PrefectWorkPoolReconciler) mapConfigMapToWorkPools() handler.MapFunc {
+// return func(ctx context.Context, obj handler.MapObject) []reconcile.Request {
+
+// func (r *PrefectWorkPoolReconciler) mapConfigMapToWorkPools(obj client.Object) []reconcile.Request {
+func (r *PrefectWorkPoolReconciler) mapConfigMapToWorkPools(ctx context.Context, obj client.Object) []reconcile.Request {
+	configMap, ok := obj.(*corev1.ConfigMap)
+	if !ok {
+		return nil
+	}
+
+	// List all CRs in the ConfigMap's namespace
+	workPools := &prefectiov1.PrefectWorkPoolList{}
+	if err := r.List(ctx, workPools, client.InNamespace(configMap.Namespace)); err != nil {
+		// Log error if needed
+		return nil
+	}
+
+	if len(workPools.Items) == 0 {
+		return nil
+	}
+
+	var requests []reconcile.Request
+	for _, workPool := range workPools.Items {
+		if workPool.Spec.BaseJobTemplate != nil &&
+			workPool.Spec.BaseJobTemplate.ConfigMap != nil &&
+			workPool.Spec.BaseJobTemplate.ConfigMap.Name == configMap.Name {
+			requests = append(requests, reconcile.Request{
+				NamespacedName: types.NamespacedName{
+					Name:      workPool.Name,
+					Namespace: workPool.Namespace,
+				},
+			})
+		}
+	}
+
+	return requests
+}
+
 // SetupWithManager sets up the controller with the Manager.
 func (r *PrefectWorkPoolReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&prefectiov1.PrefectWorkPool{}).
 		Owns(&appsv1.Deployment{}).
+		Watches(&corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Namespace: "prefect"}}, handler.EnqueueRequestsFromMapFunc(r.mapConfigMapToWorkPools)).
 		Complete(r)
 }