diff --git a/Makefile b/Makefile index e5d6f5044..abe53f472 100644 --- a/Makefile +++ b/Makefile @@ -101,7 +101,7 @@ help: ## Display this help. manifests: controller-gen ## Generate WebhookConfiguration, ClusterRole and CustomResourceDefinition objects. $(CONTROLLER_GEN) crd webhook paths=$(GENPATH) output:crd:artifacts:config=config/crd/bases $(CONTROLLER_GEN) rbac:roleName=nodeconfigurator-role paths="./internal/rebooter/..." output:artifacts:config=config/rbac/nodeconfigurator/ - $(CONTROLLER_GEN) rbac:roleName=manager-role paths="./internal/controller/clustercontroller/...; ./internal/controller/topologyconfcontroller/...; ./internal/controller/nodeconfigurator/...; ./internal/controller/nodesetcontroller/..." output:artifacts:config=config/rbac/clustercontroller/ + $(CONTROLLER_GEN) rbac:roleName=manager-role paths="./internal/controller/clustercontroller/...; ./internal/controller/topologyconfcontroller/...; ./internal/controller/nodeconfigurator/...; ./internal/controller/nodesetcontroller/...; ./internal/controller/resourcepatchpolicy/..." output:artifacts:config=config/rbac/clustercontroller/ $(CONTROLLER_GEN) rbac:roleName=soperator-checks-role paths="./internal/controller/soperatorchecks/..." output:artifacts:config=config/rbac/soperatorchecks/ .PHONY: generate generate: controller-gen ## Generate code containing DeepCopy, DeepCopyInto, and DeepCopyObject method implementations. diff --git a/api/v1alpha1/resourcepatchpolicy_types.go b/api/v1alpha1/resourcepatchpolicy_types.go new file mode 100644 index 000000000..1dfb6be5e --- /dev/null +++ b/api/v1alpha1/resourcepatchpolicy_types.go @@ -0,0 +1,193 @@ +/* +Copyright 2024 Nebius B.V. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package v1alpha1 + +import ( + apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +const ( + KindResourcePatchPolicy = "ResourcePatchPolicy" + + // ConditionTypeResourcePatchPolicyAccepted indicates whether the policy + // passed static validation and was accepted by the operator. + ConditionTypeResourcePatchPolicyAccepted = "Accepted" +) + +// PatchType is the patch mechanism used by a ResourcePatchPolicy. +// +kubebuilder:validation:Enum=JSONPatch;JSONMergePatch +type PatchType string + +const ( + // JSONPatchType applies RFC 6902 JSON Patch operations. + JSONPatchType PatchType = "JSONPatch" + // JSONMergePatchType applies an RFC 7386 JSON Merge Patch object. + JSONMergePatchType PatchType = "JSONMergePatch" +) + +// ResourcePatchPolicySpec defines the desired state of ResourcePatchPolicy. +type ResourcePatchPolicySpec struct { + // TargetRef identifies the SlurmCluster, NodeSet or NodeConfigurator this + // policy applies to. + TargetRef PolicyTargetReference `json:"targetRef"` + + // Priority determines the order in which policies are applied to the same + // resource. Lower values are applied first. + // + // +kubebuilder:validation:Optional + // +kubebuilder:default=0 + Priority *int32 `json:"priority,omitempty"` + + // Type is the patch mechanism: "JSONPatch" (RFC 6902) or "JSONMergePatch" + // (RFC 7386). + Type PatchType `json:"type"` + + // Patches is the list of patches to apply to generated resources. + // + // +kubebuilder:validation:MinItems=1 + Patches []ResourcePatch `json:"patches"` +} + +// PolicyTargetReference points at the operator-managed parent object a policy +// attaches to. +type PolicyTargetReference struct { + // Group is the API group of the target resource. + // + // +kubebuilder:default="slurm.nebius.ai" + Group string `json:"group"` + + // Kind is the kind of the target resource. + // + // +kubebuilder:validation:Enum=SlurmCluster;NodeSet;NodeConfigurator + Kind string `json:"kind"` + + // Name is the name of the target resource. + // + // +kubebuilder:validation:MinLength=1 + Name string `json:"name"` + + // Namespace is the namespace of the target resource. Defaults to the + // namespace of the ResourcePatchPolicy. + // + // +kubebuilder:validation:Optional + Namespace *string `json:"namespace,omitempty"` +} + +// ResourcePatch targets a specific generated Kubernetes resource. +type ResourcePatch struct { + // ResourceRef selects which generated Kubernetes resource to patch. + ResourceRef ResourceSelector `json:"resourceRef"` + + // JSONPatch contains RFC 6902 operations. Used when type is "JSONPatch". + // + // +kubebuilder:validation:Optional + JSONPatch []JSONPatchOperation `json:"jsonPatch,omitempty"` + + // JSONMergePatch contains an RFC 7386 merge patch object. Used when type is + // "JSONMergePatch". + // + // +kubebuilder:validation:Optional + // +kubebuilder:pruning:PreserveUnknownFields + JSONMergePatch *apiextensionsv1.JSON `json:"jsonMergePatch,omitempty"` +} + +// ResourceSelector selects generated Kubernetes resources by kind and name. +type ResourceSelector struct { + // Kind of the generated Kubernetes resource (e.g. StatefulSet, Service, + // ConfigMap, DaemonSet). + // + // +kubebuilder:validation:MinLength=1 + Kind string `json:"kind"` + + // Name is the exact name of the generated resource to patch. When empty, + // all resources of the given kind are matched. + // + // +kubebuilder:validation:Optional + Name *string `json:"name,omitempty"` + + // APIVersion of the resource (e.g. "apps/v1"). When set, it must match the + // generated resource's apiVersion. Defaults to matching any version. + // + // +kubebuilder:validation:Optional + APIVersion *string `json:"apiVersion,omitempty"` +} + +// JSONPatchOperation is a single RFC 6902 operation. +type JSONPatchOperation struct { + // Op is the operation: "add", "remove", "replace", "move", "copy", "test". + // + // +kubebuilder:validation:Enum=add;remove;replace;move;copy;test + Op string `json:"op"` + + // Path is the JSON Pointer (RFC 6901) to the target field. + // + // +kubebuilder:validation:MinLength=1 + Path string `json:"path"` + + // Value is the value to apply. Required for add, replace and test. + // + // +kubebuilder:validation:Optional + // +kubebuilder:pruning:PreserveUnknownFields + Value *apiextensionsv1.JSON `json:"value,omitempty"` + + // From is the source path for move and copy operations. + // + // +kubebuilder:validation:Optional + From *string `json:"from,omitempty"` +} + +// ResourcePatchPolicyStatus defines the observed state of ResourcePatchPolicy. +type ResourcePatchPolicyStatus struct { + // Conditions describe the current state of the policy. + // + // +kubebuilder:validation:Optional + // +listType=map + // +listMapKey=type + Conditions []metav1.Condition `json:"conditions,omitempty"` +} + +// ResourcePatchPolicy enables modifications to Kubernetes resources generated +// by soperator for SlurmCluster, NodeSet and NodeConfigurator objects. +// +// +kubebuilder:object:root=true +// +kubebuilder:subresource:status +// +kubebuilder:printcolumn:name="Target Kind",type="string",JSONPath=".spec.targetRef.kind",description="Kind of the target resource" +// +kubebuilder:printcolumn:name="Target Name",type="string",JSONPath=".spec.targetRef.name",description="Name of the target resource" +// +kubebuilder:printcolumn:name="Type",type="string",JSONPath=".spec.type",description="Patch type" +// +kubebuilder:printcolumn:name="Accepted",type="string",JSONPath=".status.conditions[?(@.type==\"Accepted\")].status",description="Whether the policy was accepted" +// +kubebuilder:printcolumn:name="Age",type="date",JSONPath=".metadata.creationTimestamp" +type ResourcePatchPolicy struct { + metav1.TypeMeta `json:",inline"` + metav1.ObjectMeta `json:"metadata,omitempty"` + + Spec ResourcePatchPolicySpec `json:"spec"` + Status ResourcePatchPolicyStatus `json:"status,omitempty"` +} + +// +kubebuilder:object:root=true + +// ResourcePatchPolicyList contains a list of ResourcePatchPolicy. +type ResourcePatchPolicyList struct { + metav1.TypeMeta `json:",inline"` + metav1.ListMeta `json:"metadata,omitempty"` + Items []ResourcePatchPolicy `json:"items"` +} + +func init() { + SchemeBuilder.Register(&ResourcePatchPolicy{}, &ResourcePatchPolicyList{}) +} diff --git a/api/v1alpha1/zz_generated.deepcopy.go b/api/v1alpha1/zz_generated.deepcopy.go index 75a58d7ff..e421c3775 100644 --- a/api/v1alpha1/zz_generated.deepcopy.go +++ b/api/v1alpha1/zz_generated.deepcopy.go @@ -6,6 +6,7 @@ package v1alpha1 import ( "k8s.io/api/core/v1" + apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/util/intstr" @@ -572,6 +573,31 @@ func (in *Image) DeepCopy() *Image { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *JSONPatchOperation) DeepCopyInto(out *JSONPatchOperation) { + *out = *in + if in.Value != nil { + in, out := &in.Value, &out.Value + *out = new(apiextensionsv1.JSON) + (*in).DeepCopyInto(*out) + } + if in.From != nil { + in, out := &in.From, &out.From + *out = new(string) + **out = **in + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new JSONPatchOperation. +func (in *JSONPatchOperation) DeepCopy() *JSONPatchOperation { + if in == nil { + return nil + } + out := new(JSONPatchOperation) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *JailedConfig) DeepCopyInto(out *JailedConfig) { *out = *in @@ -1180,6 +1206,26 @@ func (in *PodConfig) DeepCopy() *PodConfig { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *PolicyTargetReference) DeepCopyInto(out *PolicyTargetReference) { + *out = *in + if in.Namespace != nil { + in, out := &in.Namespace, &out.Namespace + *out = new(string) + **out = **in + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new PolicyTargetReference. +func (in *PolicyTargetReference) DeepCopy() *PolicyTargetReference { + if in == nil { + return nil + } + out := new(PolicyTargetReference) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *Reactions) DeepCopyInto(out *Reactions) { *out = *in @@ -1247,6 +1293,168 @@ func (in *ReservationSpec) DeepCopy() *ReservationSpec { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *ResourcePatch) DeepCopyInto(out *ResourcePatch) { + *out = *in + in.ResourceRef.DeepCopyInto(&out.ResourceRef) + if in.JSONPatch != nil { + in, out := &in.JSONPatch, &out.JSONPatch + *out = make([]JSONPatchOperation, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } + if in.JSONMergePatch != nil { + in, out := &in.JSONMergePatch, &out.JSONMergePatch + *out = new(apiextensionsv1.JSON) + (*in).DeepCopyInto(*out) + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ResourcePatch. +func (in *ResourcePatch) DeepCopy() *ResourcePatch { + if in == nil { + return nil + } + out := new(ResourcePatch) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *ResourcePatchPolicy) DeepCopyInto(out *ResourcePatchPolicy) { + *out = *in + out.TypeMeta = in.TypeMeta + in.ObjectMeta.DeepCopyInto(&out.ObjectMeta) + in.Spec.DeepCopyInto(&out.Spec) + in.Status.DeepCopyInto(&out.Status) +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ResourcePatchPolicy. +func (in *ResourcePatchPolicy) DeepCopy() *ResourcePatchPolicy { + if in == nil { + return nil + } + out := new(ResourcePatchPolicy) + in.DeepCopyInto(out) + return out +} + +// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object. +func (in *ResourcePatchPolicy) 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 *ResourcePatchPolicyList) DeepCopyInto(out *ResourcePatchPolicyList) { + *out = *in + out.TypeMeta = in.TypeMeta + in.ListMeta.DeepCopyInto(&out.ListMeta) + if in.Items != nil { + in, out := &in.Items, &out.Items + *out = make([]ResourcePatchPolicy, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ResourcePatchPolicyList. +func (in *ResourcePatchPolicyList) DeepCopy() *ResourcePatchPolicyList { + if in == nil { + return nil + } + out := new(ResourcePatchPolicyList) + in.DeepCopyInto(out) + return out +} + +// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object. +func (in *ResourcePatchPolicyList) 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 *ResourcePatchPolicySpec) DeepCopyInto(out *ResourcePatchPolicySpec) { + *out = *in + in.TargetRef.DeepCopyInto(&out.TargetRef) + if in.Priority != nil { + in, out := &in.Priority, &out.Priority + *out = new(int32) + **out = **in + } + if in.Patches != nil { + in, out := &in.Patches, &out.Patches + *out = make([]ResourcePatch, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ResourcePatchPolicySpec. +func (in *ResourcePatchPolicySpec) DeepCopy() *ResourcePatchPolicySpec { + if in == nil { + return nil + } + out := new(ResourcePatchPolicySpec) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *ResourcePatchPolicyStatus) DeepCopyInto(out *ResourcePatchPolicyStatus) { + *out = *in + if in.Conditions != nil { + in, out := &in.Conditions, &out.Conditions + *out = make([]metav1.Condition, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ResourcePatchPolicyStatus. +func (in *ResourcePatchPolicyStatus) DeepCopy() *ResourcePatchPolicyStatus { + if in == nil { + return nil + } + out := new(ResourcePatchPolicyStatus) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *ResourceSelector) DeepCopyInto(out *ResourceSelector) { + *out = *in + if in.Name != nil { + in, out := &in.Name, &out.Name + *out = new(string) + **out = **in + } + if in.APIVersion != nil { + in, out := &in.APIVersion, &out.APIVersion + *out = new(string) + **out = **in + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ResourceSelector. +func (in *ResourceSelector) DeepCopy() *ResourceSelector { + if in == nil { + return nil + } + out := new(ResourceSelector) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *SlurmJobSpec) DeepCopyInto(out *SlurmJobSpec) { *out = *in diff --git a/cmd/main.go b/cmd/main.go index f89802273..7af448100 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -53,6 +53,7 @@ import ( "nebius.ai/slurm-operator/internal/controller/clustercontroller" "nebius.ai/slurm-operator/internal/controller/nodeconfigurator" "nebius.ai/slurm-operator/internal/controller/nodesetcontroller" + "nebius.ai/slurm-operator/internal/controller/resourcepatchpolicy" "nebius.ai/slurm-operator/internal/controller/topologyconfcontroller" "nebius.ai/slurm-operator/internal/controllersenabled" webhookv1 "nebius.ai/slurm-operator/internal/webhook/v1" @@ -180,7 +181,7 @@ func main() { controllersSpec = controllersFlag controllersSource = "flag" } - availableControllers := []string{"cluster", "nodeconfigurator", "nodeset", "topology"} + availableControllers := []string{"cluster", "nodeconfigurator", "nodeset", "topology", "resourcepatchpolicy"} controllersSet, err := controllersenabled.New( controllersSpec, availableControllers, @@ -188,6 +189,10 @@ func main() { if err != nil { cli.Fail(setupLog, err, "unable to parse SLURM_OPERATOR_CONTROLLERS") } + // The experimental ResourcePatchPolicy feature is toggled like a controller. + // It both registers the policy status controller and enables in-memory + // patching inside the cluster, nodeset and nodeconfigurator reconcilers. + enableResourcePatchPolicy := controllersSet.Enabled("resourcepatchpolicy") if controllersSpec != "" { for _, name := range availableControllers { if !controllersSet.Enabled(name) { @@ -261,6 +266,7 @@ func main() { mgr.GetClient(), mgr.GetScheme(), mgr.GetEventRecorderFor(consts.SlurmCluster+"-controller"), + enableResourcePatchPolicy, ).SetupWithManager(mgr, maxConcurrency, cacheSyncTimeout); err != nil { cli.Fail(setupLog, err, "unable to create controller", "controller", slurmClusterName) } @@ -279,8 +285,9 @@ func main() { // region Reconciler/NodeConfigurator if controllersSet.Enabled("nodeconfigurator") { if err = (&nodeconfigurator.NodeConfiguratorReconciler{ - Client: mgr.GetClient(), - Scheme: mgr.GetScheme(), + Client: mgr.GetClient(), + Scheme: mgr.GetScheme(), + EnableResourcePatchPolicy: enableResourcePatchPolicy, }).SetupWithManager(mgr, maxConcurrency, cacheSyncTimeout); err != nil { cli.Fail(setupLog, err, "unable to create controller", "controller", "NodeConfigurator") } @@ -296,6 +303,7 @@ func main() { mgr.GetClient(), mgr.GetScheme(), mgr.GetEventRecorderFor(nodeSetNameLower+"-controller"), + enableResourcePatchPolicy, ). SetupWithManager(mgr, nodeSetNameLower, maxConcurrency, cacheSyncTimeout); err != nil { cli.Fail(setupLog, err, "unable to create controller", "controller", nodeSetName) @@ -339,6 +347,17 @@ func main() { } // endregion Reconciler/Topology + // region Reconciler/ResourcePatchPolicy + if enableResourcePatchPolicy { + if err = (&resourcepatchpolicy.ResourcePatchPolicyReconciler{ + Client: mgr.GetClient(), + Scheme: mgr.GetScheme(), + }).SetupWithManager(mgr, maxConcurrency, cacheSyncTimeout); err != nil { + cli.Fail(setupLog, err, "unable to create controller", "controller", resourcepatchpolicy.ControllerName) + } + } + // endregion Reconciler/ResourcePatchPolicy + //+kubebuilder:scaffold:builder if err = mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil { diff --git a/config/crd/bases/kustomization.yaml b/config/crd/bases/kustomization.yaml index 534b210b7..03fcc88c1 100644 --- a/config/crd/bases/kustomization.yaml +++ b/config/crd/bases/kustomization.yaml @@ -5,3 +5,4 @@ resources: - slurm.nebius.ai_nodesets.yaml - slurm.nebius.ai_slurmclusters.yaml - slurm.nebius.ai_jailedconfigs.yaml +- slurm.nebius.ai_resourcepatchpolicies.yaml diff --git a/config/crd/bases/slurm.nebius.ai_resourcepatchpolicies.yaml b/config/crd/bases/slurm.nebius.ai_resourcepatchpolicies.yaml new file mode 100644 index 000000000..1307a528b --- /dev/null +++ b/config/crd/bases/slurm.nebius.ai_resourcepatchpolicies.yaml @@ -0,0 +1,259 @@ +--- +apiVersion: apiextensions.k8s.io/v1 +kind: CustomResourceDefinition +metadata: + annotations: + controller-gen.kubebuilder.io/version: v0.19.0 + name: resourcepatchpolicies.slurm.nebius.ai +spec: + group: slurm.nebius.ai + names: + kind: ResourcePatchPolicy + listKind: ResourcePatchPolicyList + plural: resourcepatchpolicies + singular: resourcepatchpolicy + scope: Namespaced + versions: + - additionalPrinterColumns: + - description: Kind of the target resource + jsonPath: .spec.targetRef.kind + name: Target Kind + type: string + - description: Name of the target resource + jsonPath: .spec.targetRef.name + name: Target Name + type: string + - description: Patch type + jsonPath: .spec.type + name: Type + type: string + - description: Whether the policy was accepted + jsonPath: .status.conditions[?(@.type=="Accepted")].status + name: Accepted + type: string + - jsonPath: .metadata.creationTimestamp + name: Age + type: date + name: v1alpha1 + schema: + openAPIV3Schema: + description: |- + ResourcePatchPolicy enables modifications to Kubernetes resources generated + by soperator for SlurmCluster, NodeSet and NodeConfigurator objects. + properties: + apiVersion: + description: |- + APIVersion defines the versioned schema of this representation of an object. + Servers should convert recognized schemas to the latest internal value, and + may reject unrecognized values. + More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources + type: string + kind: + description: |- + Kind is a string value representing the REST resource this object represents. + Servers may infer this from the endpoint the client submits requests to. + Cannot be updated. + In CamelCase. + More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds + type: string + metadata: + type: object + spec: + description: ResourcePatchPolicySpec defines the desired state of ResourcePatchPolicy. + properties: + patches: + description: Patches is the list of patches to apply to generated + resources. + items: + description: ResourcePatch targets a specific generated Kubernetes + resource. + properties: + jsonMergePatch: + description: |- + JSONMergePatch contains an RFC 7386 merge patch object. Used when type is + "JSONMergePatch". + x-kubernetes-preserve-unknown-fields: true + jsonPatch: + description: JSONPatch contains RFC 6902 operations. Used when + type is "JSONPatch". + items: + description: JSONPatchOperation is a single RFC 6902 operation. + properties: + from: + description: From is the source path for move and copy + operations. + type: string + op: + description: 'Op is the operation: "add", "remove", "replace", + "move", "copy", "test".' + enum: + - add + - remove + - replace + - move + - copy + - test + type: string + path: + description: Path is the JSON Pointer (RFC 6901) to the + target field. + minLength: 1 + type: string + value: + description: Value is the value to apply. Required for + add, replace and test. + x-kubernetes-preserve-unknown-fields: true + required: + - op + - path + type: object + type: array + resourceRef: + description: ResourceRef selects which generated Kubernetes + resource to patch. + properties: + apiVersion: + description: |- + APIVersion of the resource (e.g. "apps/v1"). When set, it must match the + generated resource's apiVersion. Defaults to matching any version. + type: string + kind: + description: |- + Kind of the generated Kubernetes resource (e.g. StatefulSet, Service, + ConfigMap, DaemonSet). + minLength: 1 + type: string + name: + description: |- + Name is the exact name of the generated resource to patch. When empty, + all resources of the given kind are matched. + type: string + required: + - kind + type: object + required: + - resourceRef + type: object + minItems: 1 + type: array + priority: + default: 0 + description: |- + Priority determines the order in which policies are applied to the same + resource. Lower values are applied first. + format: int32 + type: integer + targetRef: + description: |- + TargetRef identifies the SlurmCluster, NodeSet or NodeConfigurator this + policy applies to. + properties: + group: + default: slurm.nebius.ai + description: Group is the API group of the target resource. + type: string + kind: + description: Kind is the kind of the target resource. + enum: + - SlurmCluster + - NodeSet + - NodeConfigurator + type: string + name: + description: Name is the name of the target resource. + minLength: 1 + type: string + namespace: + description: |- + Namespace is the namespace of the target resource. Defaults to the + namespace of the ResourcePatchPolicy. + type: string + required: + - group + - kind + - name + type: object + type: + description: |- + Type is the patch mechanism: "JSONPatch" (RFC 6902) or "JSONMergePatch" + (RFC 7386). + enum: + - JSONPatch + - JSONMergePatch + type: string + required: + - patches + - targetRef + - type + type: object + status: + description: ResourcePatchPolicyStatus defines the observed state of ResourcePatchPolicy. + properties: + conditions: + description: Conditions describe the current state of the policy. + items: + description: Condition contains details for one aspect of the current + state of this API Resource. + properties: + lastTransitionTime: + description: |- + lastTransitionTime is the last time the condition transitioned from one status to another. + This should be when the underlying condition changed. If that is not known, then using the time when the API field changed is acceptable. + format: date-time + type: string + message: + description: |- + message is a human readable message indicating details about the transition. + This may be an empty string. + maxLength: 32768 + type: string + observedGeneration: + description: |- + observedGeneration represents the .metadata.generation that the condition was set based upon. + For instance, if .metadata.generation is currently 12, but the .status.conditions[x].observedGeneration is 9, the condition is out of date + with respect to the current state of the instance. + format: int64 + minimum: 0 + type: integer + reason: + description: |- + reason contains a programmatic identifier indicating the reason for the condition's last transition. + Producers of specific condition types may define expected values and meanings for this field, + and whether the values are considered a guaranteed API. + The value should be a CamelCase string. + This field may not be empty. + maxLength: 1024 + minLength: 1 + pattern: ^[A-Za-z]([A-Za-z0-9_,:]*[A-Za-z0-9_])?$ + type: string + status: + description: status of the condition, one of True, False, Unknown. + enum: + - "True" + - "False" + - Unknown + type: string + type: + description: type of condition in CamelCase or in foo.example.com/CamelCase. + maxLength: 316 + pattern: ^([a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*/)?(([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9])$ + type: string + required: + - lastTransitionTime + - message + - reason + - status + - type + type: object + type: array + x-kubernetes-list-map-keys: + - type + x-kubernetes-list-type: map + type: object + required: + - spec + type: object + served: true + storage: true + subresources: + status: {} diff --git a/config/crd/kustomization.yaml b/config/crd/kustomization.yaml index 9a504cc6d..edd28a5b5 100644 --- a/config/crd/kustomization.yaml +++ b/config/crd/kustomization.yaml @@ -7,6 +7,7 @@ resources: - bases/slurm.nebius.ai_nodesets.yaml - bases/slurm.nebius.ai_slurmclusters.yaml - bases/slurm.nebius.ai_jailedconfigs.yaml +- bases/slurm.nebius.ai_resourcepatchpolicies.yaml #+kubebuilder:scaffold:crdkustomizeresource diff --git a/config/rbac/clustercontroller/role.yaml b/config/rbac/clustercontroller/role.yaml index 4295785bd..2e375ba1e 100644 --- a/config/rbac/clustercontroller/role.yaml +++ b/config/rbac/clustercontroller/role.yaml @@ -142,6 +142,7 @@ rules: - nodeconfigurators - nodesetpowerstates - nodesets + - resourcepatchpolicies - slurmclusters verbs: - create @@ -157,6 +158,7 @@ rules: - jailedconfigs/finalizers - nodeconfigurators/finalizers - nodesets/finalizers + - resourcepatchpolicies/finalizers - slurmclusters/finalizers verbs: - update @@ -167,6 +169,7 @@ rules: - nodeconfigurators/status - nodesetpowerstates/status - nodesets/status + - resourcepatchpolicies/status - slurmclusters/status verbs: - get diff --git a/config/samples/kustomization.yaml b/config/samples/kustomization.yaml index cc76b920b..5fcc74e02 100644 --- a/config/samples/kustomization.yaml +++ b/config/samples/kustomization.yaml @@ -4,4 +4,5 @@ resources: - slurm_v1alpha1_nodeconfigurator.yaml - slurm_v1alpha1_nodeset.yaml - slurm_v1alpha1_jailedconfig.yaml +- slurm_v1alpha1_resourcepatchpolicy.yaml #+kubebuilder:scaffold:manifestskustomizesamples diff --git a/config/samples/slurm_v1alpha1_resourcepatchpolicy.yaml b/config/samples/slurm_v1alpha1_resourcepatchpolicy.yaml new file mode 100644 index 000000000..a76aa8969 --- /dev/null +++ b/config/samples/slurm_v1alpha1_resourcepatchpolicy.yaml @@ -0,0 +1,53 @@ +apiVersion: slurm.nebius.ai/v1alpha1 +kind: ResourcePatchPolicy +metadata: + name: gpu-cluster-customizations + namespace: default +spec: + targetRef: + group: slurm.nebius.ai + kind: SlurmCluster + name: my-cluster + type: JSONPatch + priority: 0 + patches: + # Add a service-mesh sidecar annotation to worker pods. + - resourceRef: + kind: StatefulSet + name: "my-cluster-worker" + jsonPatch: + - op: add + path: "/spec/template/metadata/annotations/sidecar.istio.io~1inject" + value: "true" + # Set an NLB annotation on the login service. + - resourceRef: + kind: Service + name: "my-cluster-login" + jsonPatch: + - op: add + path: "/metadata/annotations/service.beta.kubernetes.io~1aws-load-balancer-type" + value: "nlb" +--- +apiVersion: slurm.nebius.ai/v1alpha1 +kind: ResourcePatchPolicy +metadata: + name: slurm-config-extras + namespace: default +spec: + targetRef: + group: slurm.nebius.ai + kind: SlurmCluster + name: my-cluster + type: JSONMergePatch + priority: 10 + patches: + - resourceRef: + kind: ConfigMap + name: "my-cluster-slurm-config" + jsonMergePatch: + metadata: + annotations: + custom-key: "custom-value" + data: + extra.conf: | + SomeExtraConfig=value diff --git a/docs/resource-patch-policy.md b/docs/resource-patch-policy.md new file mode 100644 index 000000000..22bb64636 --- /dev/null +++ b/docs/resource-patch-policy.md @@ -0,0 +1,108 @@ +# ResourcePatchPolicy + +ResourcePatchPolicy is an experimental, opt-in escape hatch that lets you patch +the Kubernetes resources soperator generates for `SlurmCluster`, `NodeSet` and +`NodeConfigurator` objects — without forking the operator or fighting the +reconciliation loop. + +Patches are applied to the in-memory desired object before it is submitted to +the API server, so the operator re-applies them on every reconciliation. There +is no race with external controllers. + +This feature is permanently `v1alpha1`: the operator does not guarantee the +naming scheme or structure of generated resources across releases, so a policy +that works today may need updating after an upgrade. + +## Enabling the feature + +The feature is toggled like a controller through `controllersEnabled` and is +off by default. Enable it in the operator's Helm values: + +```yaml +controllerManager: + manager: + controllersEnabled: + resourcepatchpolicy: true +``` + +This is wired into the `SLURM_OPERATOR_CONTROLLERS` mechanism, so it can also be +toggled directly via that environment variable (or the `--controllers` flag), +e.g. `SLURM_OPERATOR_CONTROLLERS="*,resourcepatchpolicy"` / +`--controllers="cluster,nodeset,nodeconfigurator,topology,resourcepatchpolicy"`. + +Enabling it both registers the policy status controller and turns on in-memory +patching inside the cluster, nodeset and nodeconfigurator reconcilers. + +## Anatomy of a policy + +```yaml +apiVersion: slurm.nebius.ai/v1alpha1 +kind: ResourcePatchPolicy +metadata: + name: node-configurator-host-network + namespace: soperator-system +spec: + targetRef: + group: slurm.nebius.ai + kind: NodeConfigurator # or SlurmCluster / NodeSet + name: soperator-node-configurator + priority: 10 # lower numbers applied first; default 0 + type: JSONPatch # or JSONMergePatch + patches: + # The DaemonSet is named "-ds". + - resourceRef: + kind: DaemonSet + name: "soperator-node-configurator-ds" # exact resource name + jsonPatch: + - op: add + path: "/spec/template/spec/hostNetwork" + value: true +``` + +- `targetRef` binds the policy to exactly one parent object. Its namespace + defaults to the policy's namespace. +- `type` selects RFC 6902 JSON Patch (`JSONPatch`) or RFC 7386 JSON Merge Patch + (`JSONMergePatch`). +- `resourceRef.kind` plus `resourceRef.name` (exact resource name) selects + which generated resources a patch applies to. An empty `name` matches every + resource of that kind. `apiVersion` may be set to disambiguate. +- In JSON Pointers, `/` inside a key is escaped as `~1` and `~` as `~0` + (e.g. `sidecar.istio.io/inject` → `sidecar.istio.io~1inject`). + +## Ordering and conflicts + +Policies that target the same resource are applied in ascending `priority` +order; ties are broken alphabetically by `namespace/name`. A later patch that +writes the same field wins. + +If a single patch entry fails (bad path, malformed payload, protected-field +violation) it is skipped and the remaining entries still apply. A `Warning` +event is recorded on the parent object. + +## Protected fields + +The following mutations are rejected to keep reconciliation working: + +| Field | Reason | +| -------------------------- | ------------------------------- | +| `metadata.name` | identity cannot change | +| `metadata.namespace` | identity cannot change | +| `metadata.ownerReferences` | breaks garbage collection | +| `spec.selector` | immutable on workload resources | + +## Status + +The ResourcePatchPolicy controller validates each policy statically and reports +an `Accepted` condition: + +```yaml +status: + conditions: + - type: Accepted + status: "True" + reason: Accepted + message: Policy passed static validation +``` + +An invalid policy (e.g. a `JSONPatch` entry with no operations) gets +`status: "False"` and `reason: Invalid` with a descriptive message. diff --git a/go.mod b/go.mod index a6328478d..53a46dd51 100644 --- a/go.mod +++ b/go.mod @@ -96,7 +96,7 @@ require ( github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect github.com/emicklei/go-restful/v3 v3.12.2 // indirect - github.com/evanphx/json-patch/v5 v5.9.11 // indirect + github.com/evanphx/json-patch/v5 v5.9.11 github.com/go-logr/zapr v1.3.0 // indirect github.com/go-openapi/jsonpointer v0.21.1 // indirect github.com/go-openapi/jsonreference v0.21.0 // indirect @@ -130,7 +130,7 @@ require ( google.golang.org/protobuf v1.36.11 // indirect gopkg.in/inf.v0 v0.9.1 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect - k8s.io/apiextensions-apiserver v0.34.2 // indirect + k8s.io/apiextensions-apiserver v0.34.2 k8s.io/klog/v2 v2.130.1 k8s.io/kube-openapi v0.0.0-20250710124328-f3f2b991d03b // indirect sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730 // indirect diff --git a/helm/soperator-crds/templates/slurmcluster-crd.yaml b/helm/soperator-crds/templates/slurmcluster-crd.yaml index 992669587..1906187ea 100644 --- a/helm/soperator-crds/templates/slurmcluster-crd.yaml +++ b/helm/soperator-crds/templates/slurmcluster-crd.yaml @@ -27276,6 +27276,265 @@ spec: --- apiVersion: apiextensions.k8s.io/v1 kind: CustomResourceDefinition +metadata: + annotations: + controller-gen.kubebuilder.io/version: v0.19.0 + name: resourcepatchpolicies.slurm.nebius.ai +spec: + group: slurm.nebius.ai + names: + kind: ResourcePatchPolicy + listKind: ResourcePatchPolicyList + plural: resourcepatchpolicies + singular: resourcepatchpolicy + scope: Namespaced + versions: + - additionalPrinterColumns: + - description: Kind of the target resource + jsonPath: .spec.targetRef.kind + name: Target Kind + type: string + - description: Name of the target resource + jsonPath: .spec.targetRef.name + name: Target Name + type: string + - description: Patch type + jsonPath: .spec.type + name: Type + type: string + - description: Whether the policy was accepted + jsonPath: .status.conditions[?(@.type=="Accepted")].status + name: Accepted + type: string + - jsonPath: .metadata.creationTimestamp + name: Age + type: date + name: v1alpha1 + schema: + openAPIV3Schema: + description: |- + ResourcePatchPolicy enables modifications to Kubernetes resources generated + by soperator for SlurmCluster, NodeSet and NodeConfigurator objects. + properties: + apiVersion: + description: |- + APIVersion defines the versioned schema of this representation of an object. + Servers should convert recognized schemas to the latest internal value, and + may reject unrecognized values. + More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources + type: string + kind: + description: |- + Kind is a string value representing the REST resource this object represents. + Servers may infer this from the endpoint the client submits requests to. + Cannot be updated. + In CamelCase. + More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds + type: string + metadata: + type: object + spec: + description: ResourcePatchPolicySpec defines the desired state of ResourcePatchPolicy. + properties: + patches: + description: Patches is the list of patches to apply to generated + resources. + items: + description: ResourcePatch targets a specific generated Kubernetes + resource. + properties: + jsonMergePatch: + description: |- + JSONMergePatch contains an RFC 7386 merge patch object. Used when type is + "JSONMergePatch". + x-kubernetes-preserve-unknown-fields: true + jsonPatch: + description: JSONPatch contains RFC 6902 operations. Used when + type is "JSONPatch". + items: + description: JSONPatchOperation is a single RFC 6902 operation. + properties: + from: + description: From is the source path for move and copy + operations. + type: string + op: + description: 'Op is the operation: "add", "remove", "replace", + "move", "copy", "test".' + enum: + - add + - remove + - replace + - move + - copy + - test + type: string + path: + description: Path is the JSON Pointer (RFC 6901) to the + target field. + minLength: 1 + type: string + value: + description: Value is the value to apply. Required for + add, replace and test. + x-kubernetes-preserve-unknown-fields: true + required: + - op + - path + type: object + type: array + resourceRef: + description: ResourceRef selects which generated Kubernetes + resource to patch. + properties: + apiVersion: + description: |- + APIVersion of the resource (e.g. "apps/v1"). When set, it must match the + generated resource's apiVersion. Defaults to matching any version. + type: string + kind: + description: |- + Kind of the generated Kubernetes resource (e.g. StatefulSet, Service, + ConfigMap, DaemonSet). + minLength: 1 + type: string + name: + description: |- + Name is the exact name of the generated resource to patch. When empty, + all resources of the given kind are matched. + type: string + required: + - kind + type: object + required: + - resourceRef + type: object + minItems: 1 + type: array + priority: + default: 0 + description: |- + Priority determines the order in which policies are applied to the same + resource. Lower values are applied first. + format: int32 + type: integer + targetRef: + description: |- + TargetRef identifies the SlurmCluster, NodeSet or NodeConfigurator this + policy applies to. + properties: + group: + default: slurm.nebius.ai + description: Group is the API group of the target resource. + type: string + kind: + description: Kind is the kind of the target resource. + enum: + - SlurmCluster + - NodeSet + - NodeConfigurator + type: string + name: + description: Name is the name of the target resource. + minLength: 1 + type: string + namespace: + description: |- + Namespace is the namespace of the target resource. Defaults to the + namespace of the ResourcePatchPolicy. + type: string + required: + - group + - kind + - name + type: object + type: + description: |- + Type is the patch mechanism: "JSONPatch" (RFC 6902) or "JSONMergePatch" + (RFC 7386). + enum: + - JSONPatch + - JSONMergePatch + type: string + required: + - patches + - targetRef + - type + type: object + status: + description: ResourcePatchPolicyStatus defines the observed state of ResourcePatchPolicy. + properties: + conditions: + description: Conditions describe the current state of the policy. + items: + description: Condition contains details for one aspect of the current + state of this API Resource. + properties: + lastTransitionTime: + description: |- + lastTransitionTime is the last time the condition transitioned from one status to another. + This should be when the underlying condition changed. If that is not known, then using the time when the API field changed is acceptable. + format: date-time + type: string + message: + description: |- + message is a human readable message indicating details about the transition. + This may be an empty string. + maxLength: 32768 + type: string + observedGeneration: + description: |- + observedGeneration represents the .metadata.generation that the condition was set based upon. + For instance, if .metadata.generation is currently 12, but the .status.conditions[x].observedGeneration is 9, the condition is out of date + with respect to the current state of the instance. + format: int64 + minimum: 0 + type: integer + reason: + description: |- + reason contains a programmatic identifier indicating the reason for the condition's last transition. + Producers of specific condition types may define expected values and meanings for this field, + and whether the values are considered a guaranteed API. + The value should be a CamelCase string. + This field may not be empty. + maxLength: 1024 + minLength: 1 + pattern: ^[A-Za-z]([A-Za-z0-9_,:]*[A-Za-z0-9_])?$ + type: string + status: + description: status of the condition, one of True, False, Unknown. + enum: + - "True" + - "False" + - Unknown + type: string + type: + description: type of condition in CamelCase or in foo.example.com/CamelCase. + maxLength: 316 + pattern: ^([a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*/)?(([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9])$ + type: string + required: + - lastTransitionTime + - message + - reason + - status + - type + type: object + type: array + x-kubernetes-list-map-keys: + - type + x-kubernetes-list-type: map + type: object + required: + - spec + type: object + served: true + storage: true + subresources: + status: {} +--- +apiVersion: apiextensions.k8s.io/v1 +kind: CustomResourceDefinition metadata: annotations: controller-gen.kubebuilder.io/version: v0.19.0 diff --git a/helm/soperator/crds/slurmcluster-crd.yaml b/helm/soperator/crds/slurmcluster-crd.yaml index 992669587..1906187ea 100644 --- a/helm/soperator/crds/slurmcluster-crd.yaml +++ b/helm/soperator/crds/slurmcluster-crd.yaml @@ -27276,6 +27276,265 @@ spec: --- apiVersion: apiextensions.k8s.io/v1 kind: CustomResourceDefinition +metadata: + annotations: + controller-gen.kubebuilder.io/version: v0.19.0 + name: resourcepatchpolicies.slurm.nebius.ai +spec: + group: slurm.nebius.ai + names: + kind: ResourcePatchPolicy + listKind: ResourcePatchPolicyList + plural: resourcepatchpolicies + singular: resourcepatchpolicy + scope: Namespaced + versions: + - additionalPrinterColumns: + - description: Kind of the target resource + jsonPath: .spec.targetRef.kind + name: Target Kind + type: string + - description: Name of the target resource + jsonPath: .spec.targetRef.name + name: Target Name + type: string + - description: Patch type + jsonPath: .spec.type + name: Type + type: string + - description: Whether the policy was accepted + jsonPath: .status.conditions[?(@.type=="Accepted")].status + name: Accepted + type: string + - jsonPath: .metadata.creationTimestamp + name: Age + type: date + name: v1alpha1 + schema: + openAPIV3Schema: + description: |- + ResourcePatchPolicy enables modifications to Kubernetes resources generated + by soperator for SlurmCluster, NodeSet and NodeConfigurator objects. + properties: + apiVersion: + description: |- + APIVersion defines the versioned schema of this representation of an object. + Servers should convert recognized schemas to the latest internal value, and + may reject unrecognized values. + More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources + type: string + kind: + description: |- + Kind is a string value representing the REST resource this object represents. + Servers may infer this from the endpoint the client submits requests to. + Cannot be updated. + In CamelCase. + More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds + type: string + metadata: + type: object + spec: + description: ResourcePatchPolicySpec defines the desired state of ResourcePatchPolicy. + properties: + patches: + description: Patches is the list of patches to apply to generated + resources. + items: + description: ResourcePatch targets a specific generated Kubernetes + resource. + properties: + jsonMergePatch: + description: |- + JSONMergePatch contains an RFC 7386 merge patch object. Used when type is + "JSONMergePatch". + x-kubernetes-preserve-unknown-fields: true + jsonPatch: + description: JSONPatch contains RFC 6902 operations. Used when + type is "JSONPatch". + items: + description: JSONPatchOperation is a single RFC 6902 operation. + properties: + from: + description: From is the source path for move and copy + operations. + type: string + op: + description: 'Op is the operation: "add", "remove", "replace", + "move", "copy", "test".' + enum: + - add + - remove + - replace + - move + - copy + - test + type: string + path: + description: Path is the JSON Pointer (RFC 6901) to the + target field. + minLength: 1 + type: string + value: + description: Value is the value to apply. Required for + add, replace and test. + x-kubernetes-preserve-unknown-fields: true + required: + - op + - path + type: object + type: array + resourceRef: + description: ResourceRef selects which generated Kubernetes + resource to patch. + properties: + apiVersion: + description: |- + APIVersion of the resource (e.g. "apps/v1"). When set, it must match the + generated resource's apiVersion. Defaults to matching any version. + type: string + kind: + description: |- + Kind of the generated Kubernetes resource (e.g. StatefulSet, Service, + ConfigMap, DaemonSet). + minLength: 1 + type: string + name: + description: |- + Name is the exact name of the generated resource to patch. When empty, + all resources of the given kind are matched. + type: string + required: + - kind + type: object + required: + - resourceRef + type: object + minItems: 1 + type: array + priority: + default: 0 + description: |- + Priority determines the order in which policies are applied to the same + resource. Lower values are applied first. + format: int32 + type: integer + targetRef: + description: |- + TargetRef identifies the SlurmCluster, NodeSet or NodeConfigurator this + policy applies to. + properties: + group: + default: slurm.nebius.ai + description: Group is the API group of the target resource. + type: string + kind: + description: Kind is the kind of the target resource. + enum: + - SlurmCluster + - NodeSet + - NodeConfigurator + type: string + name: + description: Name is the name of the target resource. + minLength: 1 + type: string + namespace: + description: |- + Namespace is the namespace of the target resource. Defaults to the + namespace of the ResourcePatchPolicy. + type: string + required: + - group + - kind + - name + type: object + type: + description: |- + Type is the patch mechanism: "JSONPatch" (RFC 6902) or "JSONMergePatch" + (RFC 7386). + enum: + - JSONPatch + - JSONMergePatch + type: string + required: + - patches + - targetRef + - type + type: object + status: + description: ResourcePatchPolicyStatus defines the observed state of ResourcePatchPolicy. + properties: + conditions: + description: Conditions describe the current state of the policy. + items: + description: Condition contains details for one aspect of the current + state of this API Resource. + properties: + lastTransitionTime: + description: |- + lastTransitionTime is the last time the condition transitioned from one status to another. + This should be when the underlying condition changed. If that is not known, then using the time when the API field changed is acceptable. + format: date-time + type: string + message: + description: |- + message is a human readable message indicating details about the transition. + This may be an empty string. + maxLength: 32768 + type: string + observedGeneration: + description: |- + observedGeneration represents the .metadata.generation that the condition was set based upon. + For instance, if .metadata.generation is currently 12, but the .status.conditions[x].observedGeneration is 9, the condition is out of date + with respect to the current state of the instance. + format: int64 + minimum: 0 + type: integer + reason: + description: |- + reason contains a programmatic identifier indicating the reason for the condition's last transition. + Producers of specific condition types may define expected values and meanings for this field, + and whether the values are considered a guaranteed API. + The value should be a CamelCase string. + This field may not be empty. + maxLength: 1024 + minLength: 1 + pattern: ^[A-Za-z]([A-Za-z0-9_,:]*[A-Za-z0-9_])?$ + type: string + status: + description: status of the condition, one of True, False, Unknown. + enum: + - "True" + - "False" + - Unknown + type: string + type: + description: type of condition in CamelCase or in foo.example.com/CamelCase. + maxLength: 316 + pattern: ^([a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*/)?(([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9])$ + type: string + required: + - lastTransitionTime + - message + - reason + - status + - type + type: object + type: array + x-kubernetes-list-map-keys: + - type + x-kubernetes-list-type: map + type: object + required: + - spec + type: object + served: true + storage: true + subresources: + status: {} +--- +apiVersion: apiextensions.k8s.io/v1 +kind: CustomResourceDefinition metadata: annotations: controller-gen.kubebuilder.io/version: v0.19.0 diff --git a/helm/soperator/templates/_helpers.tpl b/helm/soperator/templates/_helpers.tpl index 0112151df..87ff0e76c 100644 --- a/helm/soperator/templates/_helpers.tpl +++ b/helm/soperator/templates/_helpers.tpl @@ -53,7 +53,7 @@ Create the name of the service account to use {{- end }} {{- define "soperator.controllersAvailable" -}} -cluster,nodeconfigurator,nodeset,topology +cluster,nodeconfigurator,nodeset,topology,resourcepatchpolicy {{- end }} {{- define "soperator.controllersSpec" -}} @@ -66,19 +66,22 @@ cluster,nodeconfigurator,nodeset,topology {{- fail (printf "unknown controller %q in controllerManager.manager.controllersEnabled, available controllers: %q" $name $available) -}} {{- end -}} {{- end -}} -{{- /* generate comma spearated list */}} +{{- /* Generate a positive allowlist: enabled controllers are listed by name, + disabled ones (explicit false) are simply omitted. Controllers absent + from the values map default to enabled. */}} {{- $spec := list -}} {{- range $available -}} {{- if hasKey $controllers . -}} {{- if (get $controllers .) -}} {{- $spec = append $spec . -}} -{{- else -}} -{{- $spec = append $spec (printf "-%s" .) -}} {{- end -}} {{- else -}} {{- $spec = append $spec . -}} {{- end -}} {{- end -}} +{{- if not $spec -}} +{{- fail "controllerManager.manager.controllersEnabled disables every controller; enable at least one" -}} +{{- end -}} {{- join "," $spec -}} {{- end -}} {{- end }} diff --git a/helm/soperator/templates/manager-rbac.yaml b/helm/soperator/templates/manager-rbac.yaml index 53489e222..e7ab7041c 100644 --- a/helm/soperator/templates/manager-rbac.yaml +++ b/helm/soperator/templates/manager-rbac.yaml @@ -143,6 +143,7 @@ rules: - nodeconfigurators - nodesetpowerstates - nodesets + - resourcepatchpolicies - slurmclusters verbs: - create @@ -158,6 +159,7 @@ rules: - jailedconfigs/finalizers - nodeconfigurators/finalizers - nodesets/finalizers + - resourcepatchpolicies/finalizers - slurmclusters/finalizers verbs: - update @@ -168,6 +170,7 @@ rules: - nodeconfigurators/status - nodesetpowerstates/status - nodesets/status + - resourcepatchpolicies/status - slurmclusters/status verbs: - get diff --git a/helm/soperator/values.yaml b/helm/soperator/values.yaml index 09a79ed8e..bc37f36b0 100644 --- a/helm/soperator/values.yaml +++ b/helm/soperator/values.yaml @@ -29,6 +29,10 @@ controllerManager: nodeconfigurator: true nodeset: true topology: true + # Experimental ResourcePatchPolicy feature: registers the policy status + # controller and enables in-memory patching of generated resources in the + # cluster, nodeset and nodeconfigurator reconcilers. Opt-in. + resourcepatchpolicy: false containerSecurityContext: allowPrivilegeEscalation: false capabilities: diff --git a/internal/controller/clustercontroller/reconcile.go b/internal/controller/clustercontroller/reconcile.go index 6bb24dfc7..3a5fba4d6 100644 --- a/internal/controller/clustercontroller/reconcile.go +++ b/internal/controller/clustercontroller/reconcile.go @@ -40,6 +40,7 @@ import ( "nebius.ai/slurm-operator/internal/controller/state" "nebius.ai/slurm-operator/internal/controllerconfig" "nebius.ai/slurm-operator/internal/logfield" + "nebius.ai/slurm-operator/internal/resourcepatch" "nebius.ai/slurm-operator/internal/utils" "nebius.ai/slurm-operator/internal/utils/resourcegetter" "nebius.ai/slurm-operator/internal/values" @@ -99,8 +100,9 @@ type SlurmClusterReconciler struct { AppArmorProfile *reconciler.AppArmorProfileReconciler } -func NewSlurmClusterReconciler(client client.Client, scheme *runtime.Scheme, recorder record.EventRecorder) *SlurmClusterReconciler { +func NewSlurmClusterReconciler(client client.Client, scheme *runtime.Scheme, recorder record.EventRecorder, enableResourcePatchPolicy bool) *SlurmClusterReconciler { r := reconciler.NewReconciler(client, scheme, recorder) + r.EnableResourcePatchPolicy = enableResourcePatchPolicy return &SlurmClusterReconciler{ Reconciler: r, ConfigMap: reconciler.NewConfigMapReconciler(r), @@ -618,6 +620,13 @@ func (r *SlurmClusterReconciler) SetupWithManager(mgr ctrl.Manager, maxConcurren builder.WithPredicates(predicate.ResourceVersionChangedPredicate{}), ) + if r.EnableResourcePatchPolicy { + controllerBuilder.Watches( + &slurmv1alpha1.ResourcePatchPolicy{}, + handler.EnqueueRequestsFromMapFunc(resourcepatch.MapPolicyToTarget(slurmv1.KindSlurmCluster)), + ) + } + resourceChecks := r.createResourceChecks(saPredicate) for _, resourceCheck := range resourceChecks { diff --git a/internal/controller/nodeconfigurator/nodeconfigurator_controller.go b/internal/controller/nodeconfigurator/nodeconfigurator_controller.go index 6aa79b291..33dda0b0d 100644 --- a/internal/controller/nodeconfigurator/nodeconfigurator_controller.go +++ b/internal/controller/nodeconfigurator/nodeconfigurator_controller.go @@ -25,6 +25,7 @@ import ( "k8s.io/apimachinery/pkg/types" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/handler" "sigs.k8s.io/controller-runtime/pkg/log" slurmv1alpha1 "nebius.ai/slurm-operator/api/v1alpha1" @@ -32,12 +33,17 @@ import ( "nebius.ai/slurm-operator/internal/controllerconfig" "nebius.ai/slurm-operator/internal/logfield" render "nebius.ai/slurm-operator/internal/render/nodeconfigurator" + "nebius.ai/slurm-operator/internal/resourcepatch" ) // NodeConfiguratorReconciler reconciles a NodeConfigurator object type NodeConfiguratorReconciler struct { client.Client Scheme *runtime.Scheme + + // EnableResourcePatchPolicy gates the experimental ResourcePatchPolicy + // feature for the DaemonSet generated by this controller. + EnableResourcePatchPolicy bool } // +kubebuilder:rbac:groups=slurm.nebius.ai,resources=nodeconfigurators,verbs=get;list;watch;create;update;patch;delete @@ -72,6 +78,8 @@ func (r *NodeConfiguratorReconciler) Reconcile(ctx context.Context, req ctrl.Req existing := &appsv1.DaemonSet{} desired := render.RenderDaemonSet(nodeConfigurator, req.Namespace) + r.applyResourcePatchPolicies(ctx, nodeConfigurator, desired) + err := r.Get(ctx, types.NamespacedName{ Namespace: req.Namespace, Name: desired.GetName(), @@ -153,12 +161,66 @@ func (r *NodeConfiguratorReconciler) EnsureResourceDeployed( return nil } +// applyResourcePatchPolicies mutates desired in place by applying every +// ResourcePatchPolicy that targets the given NodeConfigurator. It is a no-op +// unless the feature is enabled. +func (r *NodeConfiguratorReconciler) applyResourcePatchPolicies( + ctx context.Context, + owner *slurmv1alpha1.NodeConfigurator, + desired client.Object, +) { + if !r.EnableResourcePatchPolicy { + return + } + logger := log.FromContext(ctx) + + var list slurmv1alpha1.ResourcePatchPolicyList + if err := r.List(ctx, &list, client.InNamespace(owner.GetNamespace())); err != nil { + logger.Error(err, "Failed to list ResourcePatchPolicy objects") + return + } + matching := resourcepatch.FilterPoliciesForTarget( + list.Items, + slurmv1alpha1.GroupVersion.Group, + slurmv1alpha1.KindNodeConfigurator, + owner.GetName(), + owner.GetNamespace(), + ) + if len(matching) == 0 { + return + } + + results, err := resourcepatch.Apply(r.Scheme, desired, matching) + if err != nil { + logger.Error(err, "Failed to apply ResourcePatchPolicy") + return + } + for _, res := range results { + if !res.Applied { + logger.Info("Skipped ResourcePatchPolicy patch", + "policy", res.PolicyName, + "resourceKind", res.Resource.Kind, + "resourceName", res.Resource.Name, + "reason", res.Message, + ) + } + } +} + // SetupWithManager sets up the controller with the Manager. func (r *NodeConfiguratorReconciler) SetupWithManager(mgr ctrl.Manager, maxConcurrency int, cacheSyncTimeout time.Duration) error { - return ctrl.NewControllerManagedBy(mgr). + b := ctrl.NewControllerManagedBy(mgr). For(&slurmv1alpha1.NodeConfigurator{}). Owns(&appsv1.DaemonSet{}). Named("nodeconfigurator"). - WithOptions(controllerconfig.ControllerOptions(maxConcurrency, cacheSyncTimeout)). - Complete(r) + WithOptions(controllerconfig.ControllerOptions(maxConcurrency, cacheSyncTimeout)) + + if r.EnableResourcePatchPolicy { + b = b.Watches( + &slurmv1alpha1.ResourcePatchPolicy{}, + handler.EnqueueRequestsFromMapFunc(resourcepatch.MapPolicyToTarget(slurmv1alpha1.KindNodeConfigurator)), + ) + } + + return b.Complete(r) } diff --git a/internal/controller/nodeconfigurator/resourcepatch_test.go b/internal/controller/nodeconfigurator/resourcepatch_test.go new file mode 100644 index 000000000..0c40149d0 --- /dev/null +++ b/internal/controller/nodeconfigurator/resourcepatch_test.go @@ -0,0 +1,94 @@ +/* +Copyright 2024 Nebius B.V. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package nodeconfigurator + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + clientgoscheme "k8s.io/client-go/kubernetes/scheme" + "k8s.io/utils/ptr" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + slurmv1alpha1 "nebius.ai/slurm-operator/api/v1alpha1" + render "nebius.ai/slurm-operator/internal/render/nodeconfigurator" +) + +func ncScheme(t *testing.T) *runtime.Scheme { + t.Helper() + scheme := runtime.NewScheme() + require.NoError(t, clientgoscheme.AddToScheme(scheme)) + require.NoError(t, slurmv1alpha1.AddToScheme(scheme)) + return scheme +} + +func nodeConfiguratorPolicy() *slurmv1alpha1.ResourcePatchPolicy { + return &slurmv1alpha1.ResourcePatchPolicy{ + ObjectMeta: metav1.ObjectMeta{Name: "nc", Namespace: "test-namespace"}, + Spec: slurmv1alpha1.ResourcePatchPolicySpec{ + TargetRef: slurmv1alpha1.PolicyTargetReference{ + Group: slurmv1alpha1.GroupVersion.Group, + Kind: slurmv1alpha1.KindNodeConfigurator, + Name: "node-configurator", + }, + Type: slurmv1alpha1.JSONPatchType, + Patches: []slurmv1alpha1.ResourcePatch{{ + ResourceRef: slurmv1alpha1.ResourceSelector{Kind: "DaemonSet", Name: ptr.To("node-configurator-ds")}, + JSONPatch: []slurmv1alpha1.JSONPatchOperation{ + {Op: "add", Path: "/metadata/labels/team", Value: &apiextensionsv1.JSON{Raw: []byte(`"hpc"`)}}, + }, + }}, + }, + } +} + +func TestApplyResourcePatchPolicies_PatchesDaemonSet(t *testing.T) { + scheme := ncScheme(t) + c := fake.NewClientBuilder().WithScheme(scheme).WithObjects(nodeConfiguratorPolicy()).Build() + r := &NodeConfiguratorReconciler{Client: c, Scheme: scheme, EnableResourcePatchPolicy: true} + + nc := &slurmv1alpha1.NodeConfigurator{ + ObjectMeta: metav1.ObjectMeta{Name: "node-configurator", Namespace: "test-namespace"}, + Spec: slurmv1alpha1.NodeConfiguratorSpec{Rebooter: slurmv1alpha1.Rebooter{Enabled: true}}, + } + ds := render.RenderDaemonSet(nc, "test-namespace") + + r.applyResourcePatchPolicies(context.Background(), nc, ds) + + assert.Equal(t, "hpc", ds.Labels["team"]) +} + +func TestApplyResourcePatchPolicies_DisabledNoOp(t *testing.T) { + scheme := ncScheme(t) + c := fake.NewClientBuilder().WithScheme(scheme).WithObjects(nodeConfiguratorPolicy()).Build() + r := &NodeConfiguratorReconciler{Client: c, Scheme: scheme, EnableResourcePatchPolicy: false} + + nc := &slurmv1alpha1.NodeConfigurator{ + ObjectMeta: metav1.ObjectMeta{Name: "node-configurator", Namespace: "test-namespace"}, + Spec: slurmv1alpha1.NodeConfiguratorSpec{Rebooter: slurmv1alpha1.Rebooter{Enabled: true}}, + } + ds := render.RenderDaemonSet(nc, "test-namespace") + + r.applyResourcePatchPolicies(context.Background(), nc, ds) + _, ok := ds.Labels["team"] + assert.False(t, ok) +} diff --git a/internal/controller/nodesetcontroller/controller.go b/internal/controller/nodesetcontroller/controller.go index 0c6b706f5..1c83fc0a2 100644 --- a/internal/controller/nodesetcontroller/controller.go +++ b/internal/controller/nodesetcontroller/controller.go @@ -42,6 +42,7 @@ import ( "nebius.ai/slurm-operator/internal/controller/reconciler" "nebius.ai/slurm-operator/internal/controllerconfig" "nebius.ai/slurm-operator/internal/logfield" + "nebius.ai/slurm-operator/internal/resourcepatch" ) // +kubebuilder:rbac:groups=slurm.nebius.ai,resources=nodesets,verbs=get;list;watch;create;update;patch;delete @@ -71,8 +72,9 @@ type NodeSetReconciler struct { NodeSetPowerState *reconciler.NodeSetPowerStateReconciler } -func NewNodeSetReconciler(client client.Client, scheme *runtime.Scheme, recorder record.EventRecorder) *NodeSetReconciler { +func NewNodeSetReconciler(client client.Client, scheme *runtime.Scheme, recorder record.EventRecorder, enableResourcePatchPolicy bool) *NodeSetReconciler { r := reconciler.NewReconciler(client, scheme, recorder) + r.EnableResourcePatchPolicy = enableResourcePatchPolicy return &NodeSetReconciler{ Reconciler: r, AdvancedStatefulSet: reconciler.NewAdvancedStatefulSetReconciler(r), @@ -117,6 +119,13 @@ func (r *NodeSetReconciler) SetupWithManager(mgr ctrl.Manager, name string, maxC builder.WithPredicates(predicate.ResourceVersionChangedPredicate{}), ) + if r.EnableResourcePatchPolicy { + controllerBuilder.Watches( + &slurmv1alpha1.ResourcePatchPolicy{}, + handler.EnqueueRequestsFromMapFunc(resourcepatch.MapPolicyToTarget(slurmv1alpha1.KindNodeSet)), + ) + } + resourceChecks := r.createResourceChecks(controllercommon.CreateServiceAccountPredicate()) for _, resourceCheck := range resourceChecks { if resourceCheck.Check { diff --git a/internal/controller/reconciler/reconciler.go b/internal/controller/reconciler/reconciler.go index 5fd747779..2efbbc7da 100644 --- a/internal/controller/reconciler/reconciler.go +++ b/internal/controller/reconciler/reconciler.go @@ -33,6 +33,10 @@ type ( Scheme *runtime.Scheme Recorder record.EventRecorder + + // EnableResourcePatchPolicy gates the experimental ResourcePatchPolicy + // feature. When false, ApplyResourcePatchPolicies is a no-op. + EnableResourcePatchPolicy bool } patchFunc func(existing, desired client.Object) (client.Patch, error) patcher interface { @@ -220,6 +224,10 @@ func (r Reconciler) reconcile( } } + // Apply user-defined ResourcePatchPolicy patches to the in-memory + // desired object before it is submitted to the API server. + r.ApplyResourcePatchPolicies(ctx, owner, desired) + err := r.EnsureDeployed(ctx, owner, existing, desired, deps...) if err != nil { logger.Error(err, "Failed to deploy") diff --git a/internal/controller/reconciler/resourcepatch.go b/internal/controller/reconciler/resourcepatch.go new file mode 100644 index 000000000..8f30c48cf --- /dev/null +++ b/internal/controller/reconciler/resourcepatch.go @@ -0,0 +1,103 @@ +/* +Copyright 2024 Nebius B.V. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package reconciler + +import ( + "context" + + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/apiutil" + "sigs.k8s.io/controller-runtime/pkg/log" + + slurmv1alpha1 "nebius.ai/slurm-operator/api/v1alpha1" + "nebius.ai/slurm-operator/internal/logfield" + "nebius.ai/slurm-operator/internal/resourcepatch" +) + +// ApplyResourcePatchPolicies mutates desired in place by applying every +// ResourcePatchPolicy whose targetRef points at owner. It is a no-op unless the +// feature is enabled on the Reconciler. +// +// Patches are applied to the in-memory desired object before it is submitted to +// the API server, so the operator's reconciliation loop naturally re-applies +// them on every pass. Failed patch entries are logged and skipped; they never +// abort reconciliation of the resource. +func (r Reconciler) ApplyResourcePatchPolicies( + ctx context.Context, + owner client.Object, + desired client.Object, +) { + if !r.EnableResourcePatchPolicy { + return + } + + logger := log.FromContext(ctx) + + ownerGVK, err := apiutil.GVKForObject(owner, r.Scheme) + if err != nil { + logger.Error(err, "Failed to resolve owner GVK for ResourcePatchPolicy") + return + } + + var list slurmv1alpha1.ResourcePatchPolicyList + if err := r.List(ctx, &list, client.InNamespace(owner.GetNamespace())); err != nil { + logger.Error(err, "Failed to list ResourcePatchPolicy objects") + return + } + if len(list.Items) == 0 { + return + } + + matching := resourcepatch.FilterPoliciesForTarget( + list.Items, + ownerGVK.Group, + ownerGVK.Kind, + owner.GetName(), + owner.GetNamespace(), + ) + if len(matching) == 0 { + return + } + + results, err := resourcepatch.Apply(r.Scheme, desired, matching) + if err != nil { + logger.Error(err, "Failed to apply ResourcePatchPolicy", logfield.ResourceKV(desired)...) + return + } + + for _, res := range results { + if res.Applied { + logger.V(1).Info("Applied ResourcePatchPolicy", + "policy", res.PolicyName, + "resourceKind", res.Resource.Kind, + "resourceName", res.Resource.Name, + ) + continue + } + logger.Info("Skipped ResourcePatchPolicy patch", + "policy", res.PolicyName, + "resourceKind", res.Resource.Kind, + "resourceName", res.Resource.Name, + "reason", res.Message, + ) + if r.Recorder != nil { + r.Recorder.Eventf(owner, "Warning", "ResourcePatchSkipped", + "Patch from policy %q on %s/%s skipped: %s", + res.PolicyName, res.Resource.Kind, res.Resource.Name, res.Message) + } + } +} diff --git a/internal/controller/reconciler/resourcepatch_test.go b/internal/controller/reconciler/resourcepatch_test.go new file mode 100644 index 000000000..baf3bf6fe --- /dev/null +++ b/internal/controller/reconciler/resourcepatch_test.go @@ -0,0 +1,130 @@ +/* +Copyright 2024 Nebius B.V. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package reconciler + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + appsv1 "k8s.io/api/apps/v1" + apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + clientgoscheme "k8s.io/client-go/kubernetes/scheme" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + slurmv1 "nebius.ai/slurm-operator/api/v1" + slurmv1alpha1 "nebius.ai/slurm-operator/api/v1alpha1" +) + +func patchScheme(t *testing.T) *runtime.Scheme { + t.Helper() + scheme := runtime.NewScheme() + require.NoError(t, clientgoscheme.AddToScheme(scheme)) + require.NoError(t, slurmv1.AddToScheme(scheme)) + require.NoError(t, slurmv1alpha1.AddToScheme(scheme)) + return scheme +} + +func clusterPatchPolicy() *slurmv1alpha1.ResourcePatchPolicy { + return &slurmv1alpha1.ResourcePatchPolicy{ + ObjectMeta: metav1.ObjectMeta{Name: "p", Namespace: "default"}, + Spec: slurmv1alpha1.ResourcePatchPolicySpec{ + TargetRef: slurmv1alpha1.PolicyTargetReference{ + Group: "slurm.nebius.ai", Kind: slurmv1.KindSlurmCluster, Name: "test-cluster", + }, + Type: slurmv1alpha1.JSONPatchType, + Patches: []slurmv1alpha1.ResourcePatch{{ + ResourceRef: slurmv1alpha1.ResourceSelector{Kind: "StatefulSet", Name: ptrString("test-cluster-worker")}, + JSONPatch: []slurmv1alpha1.JSONPatchOperation{ + {Op: "add", Path: "/metadata/annotations/patched", Value: &apiextensionsv1.JSON{Raw: []byte(`"yes"`)}}, + }, + }}, + }, + } +} + +func ptrString(s string) *string { return &s } + +func workerStatefulSet() *appsv1.StatefulSet { + return &appsv1.StatefulSet{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-cluster-worker", + Namespace: "default", + Annotations: map[string]string{"existing": "v"}, + }, + } +} + +func TestApplyResourcePatchPolicies_PatchesDesired(t *testing.T) { + scheme := patchScheme(t) + cluster := &slurmv1.SlurmCluster{ObjectMeta: metav1.ObjectMeta{Name: "test-cluster", Namespace: "default"}} + + c := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(clusterPatchPolicy()). + Build() + + r := Reconciler{Client: c, Scheme: scheme, EnableResourcePatchPolicy: true} + sts := workerStatefulSet() + + r.ApplyResourcePatchPolicies(context.Background(), cluster, sts) + + assert.Equal(t, "yes", sts.Annotations["patched"]) + assert.Equal(t, "v", sts.Annotations["existing"]) +} + +func TestApplyResourcePatchPolicies_DisabledIsNoOp(t *testing.T) { + scheme := patchScheme(t) + cluster := &slurmv1.SlurmCluster{ObjectMeta: metav1.ObjectMeta{Name: "test-cluster", Namespace: "default"}} + + c := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(clusterPatchPolicy()). + Build() + + r := Reconciler{Client: c, Scheme: scheme, EnableResourcePatchPolicy: false} + sts := workerStatefulSet() + + r.ApplyResourcePatchPolicies(context.Background(), cluster, sts) + + _, ok := sts.Annotations["patched"] + assert.False(t, ok, "feature disabled must not patch") +} + +func TestApplyResourcePatchPolicies_NonMatchingTargetIgnored(t *testing.T) { + scheme := patchScheme(t) + // Policy targets a different cluster name. + policy := clusterPatchPolicy() + policy.Spec.TargetRef.Name = "other-cluster" + + cluster := &slurmv1.SlurmCluster{ObjectMeta: metav1.ObjectMeta{Name: "test-cluster", Namespace: "default"}} + c := fake.NewClientBuilder().WithScheme(scheme).WithObjects(policy).Build() + + r := Reconciler{Client: c, Scheme: scheme, EnableResourcePatchPolicy: true} + sts := workerStatefulSet() + + r.ApplyResourcePatchPolicies(context.Background(), cluster, sts) + + _, ok := sts.Annotations["patched"] + assert.False(t, ok, "policy for a different target must not patch") +} + +var _ client.Object = (*slurmv1alpha1.ResourcePatchPolicy)(nil) diff --git a/internal/controller/resourcepatchpolicy/controller.go b/internal/controller/resourcepatchpolicy/controller.go new file mode 100644 index 000000000..f4910058a --- /dev/null +++ b/internal/controller/resourcepatchpolicy/controller.go @@ -0,0 +1,106 @@ +/* +Copyright 2024 Nebius B.V. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +// Package resourcepatchpolicy contains the controller that validates +// ResourcePatchPolicy objects and reports their acceptance status. The actual +// application of patches happens in the parent controllers (cluster, nodeset, +// nodeconfigurator) during their reconciliation. +package resourcepatchpolicy + +import ( + "context" + "time" + + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/log" + + slurmv1alpha1 "nebius.ai/slurm-operator/api/v1alpha1" + "nebius.ai/slurm-operator/internal/controllerconfig" + "nebius.ai/slurm-operator/internal/resourcepatch" +) + +const ControllerName = "resourcepatchpolicy" + +// ResourcePatchPolicyReconciler validates ResourcePatchPolicy objects and +// records their acceptance condition. +type ResourcePatchPolicyReconciler struct { + client.Client + Scheme *runtime.Scheme +} + +// +kubebuilder:rbac:groups=slurm.nebius.ai,resources=resourcepatchpolicies,verbs=get;list;watch;create;update;patch;delete +// +kubebuilder:rbac:groups=slurm.nebius.ai,resources=resourcepatchpolicies/status,verbs=get;update;patch +// +kubebuilder:rbac:groups=slurm.nebius.ai,resources=resourcepatchpolicies/finalizers,verbs=update + +func (r *ResourcePatchPolicyReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) { + logger := log.FromContext(ctx) + + policy := &slurmv1alpha1.ResourcePatchPolicy{} + if err := r.Get(ctx, req.NamespacedName, policy); err != nil { + return ctrl.Result{}, client.IgnoreNotFound(err) + } + if !policy.DeletionTimestamp.IsZero() { + return ctrl.Result{}, nil + } + + condition := metav1.Condition{ + Type: slurmv1alpha1.ConditionTypeResourcePatchPolicyAccepted, + ObservedGeneration: policy.Generation, + LastTransitionTime: metav1.Now(), + } + + if err := resourcepatch.ValidatePolicy(policy); err != nil { + condition.Status = metav1.ConditionFalse + condition.Reason = "Invalid" + condition.Message = err.Error() + logger.Info("ResourcePatchPolicy rejected", "reason", err.Error()) + } else { + condition.Status = metav1.ConditionTrue + condition.Reason = "Accepted" + condition.Message = "Policy passed static validation" + } + + if !meta.SetStatusCondition(&policy.Status.Conditions, condition) { + // No change in the condition; nothing to persist. + return ctrl.Result{}, nil + } + + if err := r.Status().Update(ctx, policy); err != nil { + // The policy may have been deleted between the cached Get and this + // update; that is not an error worth retrying. + if apierrors.IsNotFound(err) { + return ctrl.Result{}, nil + } + logger.Error(err, "Failed to update ResourcePatchPolicy status") + return ctrl.Result{}, err + } + + return ctrl.Result{}, nil +} + +// SetupWithManager sets up the controller with the Manager. +func (r *ResourcePatchPolicyReconciler) SetupWithManager(mgr ctrl.Manager, maxConcurrency int, cacheSyncTimeout time.Duration) error { + return ctrl.NewControllerManagedBy(mgr). + For(&slurmv1alpha1.ResourcePatchPolicy{}). + Named(ControllerName). + WithOptions(controllerconfig.ControllerOptions(maxConcurrency, cacheSyncTimeout)). + Complete(r) +} diff --git a/internal/controller/resourcepatchpolicy/controller_test.go b/internal/controller/resourcepatchpolicy/controller_test.go new file mode 100644 index 000000000..52e18f6ca --- /dev/null +++ b/internal/controller/resourcepatchpolicy/controller_test.go @@ -0,0 +1,108 @@ +/* +Copyright 2024 Nebius B.V. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package resourcepatchpolicy + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" + "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/client/fake" + + slurmv1alpha1 "nebius.ai/slurm-operator/api/v1alpha1" +) + +func newReconciler(t *testing.T, objs ...client.Object) *ResourcePatchPolicyReconciler { + t.Helper() + scheme := runtime.NewScheme() + require.NoError(t, slurmv1alpha1.AddToScheme(scheme)) + c := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(objs...). + WithStatusSubresource(&slurmv1alpha1.ResourcePatchPolicy{}). + Build() + return &ResourcePatchPolicyReconciler{Client: c, Scheme: scheme} +} + +func basePolicy() *slurmv1alpha1.ResourcePatchPolicy { + return &slurmv1alpha1.ResourcePatchPolicy{ + ObjectMeta: metav1.ObjectMeta{Name: "p", Namespace: "default"}, + Spec: slurmv1alpha1.ResourcePatchPolicySpec{ + TargetRef: slurmv1alpha1.PolicyTargetReference{Group: "slurm.nebius.ai", Kind: "SlurmCluster", Name: "c"}, + Type: slurmv1alpha1.JSONPatchType, + Patches: []slurmv1alpha1.ResourcePatch{{ + ResourceRef: slurmv1alpha1.ResourceSelector{Kind: "StatefulSet"}, + JSONPatch: []slurmv1alpha1.JSONPatchOperation{ + {Op: "add", Path: "/metadata/annotations/a", Value: &apiextensionsv1.JSON{Raw: []byte(`"b"`)}}, + }, + }}, + }, + } +} + +func TestReconcile_AcceptsValidPolicy(t *testing.T) { + policy := basePolicy() + r := newReconciler(t, policy) + + _, err := r.Reconcile(context.Background(), ctrl.Request{ + NamespacedName: types.NamespacedName{Namespace: "default", Name: "p"}, + }) + require.NoError(t, err) + + var got slurmv1alpha1.ResourcePatchPolicy + require.NoError(t, r.Get(context.Background(), types.NamespacedName{Namespace: "default", Name: "p"}, &got)) + cond := meta.FindStatusCondition(got.Status.Conditions, slurmv1alpha1.ConditionTypeResourcePatchPolicyAccepted) + require.NotNil(t, cond) + assert.Equal(t, metav1.ConditionTrue, cond.Status) + assert.Equal(t, "Accepted", cond.Reason) +} + +func TestReconcile_RejectsInvalidPolicy(t *testing.T) { + policy := basePolicy() + // Make it invalid: JSONPatch type but no operations. + policy.Spec.Patches[0].JSONPatch = nil + r := newReconciler(t, policy) + + _, err := r.Reconcile(context.Background(), ctrl.Request{ + NamespacedName: types.NamespacedName{Namespace: "default", Name: "p"}, + }) + require.NoError(t, err) + + var got slurmv1alpha1.ResourcePatchPolicy + require.NoError(t, r.Get(context.Background(), types.NamespacedName{Namespace: "default", Name: "p"}, &got)) + cond := meta.FindStatusCondition(got.Status.Conditions, slurmv1alpha1.ConditionTypeResourcePatchPolicyAccepted) + require.NotNil(t, cond) + assert.Equal(t, metav1.ConditionFalse, cond.Status) + assert.Equal(t, "Invalid", cond.Reason) + assert.NotEmpty(t, cond.Message) +} + +func TestReconcile_NotFoundIsNoError(t *testing.T) { + r := newReconciler(t) + _, err := r.Reconcile(context.Background(), ctrl.Request{ + NamespacedName: types.NamespacedName{Namespace: "default", Name: "missing"}, + }) + assert.NoError(t, err) +} diff --git a/internal/render/controller/patch_test.go b/internal/render/controller/patch_test.go new file mode 100644 index 000000000..667804245 --- /dev/null +++ b/internal/render/controller/patch_test.go @@ -0,0 +1,184 @@ +package controller + +import ( + "encoding/json" + "testing" + + kruisev1b1 "github.com/openkruise/kruise-api/apps/v1beta1" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" + "k8s.io/apimachinery/pkg/api/resource" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/util/intstr" + clientgoscheme "k8s.io/client-go/kubernetes/scheme" + "k8s.io/utils/ptr" + + slurmv1 "nebius.ai/slurm-operator/api/v1" + slurmv1alpha1 "nebius.ai/slurm-operator/api/v1alpha1" + "nebius.ai/slurm-operator/internal/consts" + "nebius.ai/slurm-operator/internal/resourcepatch" + "nebius.ai/slurm-operator/internal/values" +) + +func patchTestScheme(t *testing.T) *runtime.Scheme { + t.Helper() + scheme := runtime.NewScheme() + require.NoError(t, clientgoscheme.AddToScheme(scheme)) + require.NoError(t, slurmv1alpha1.AddToScheme(scheme)) + require.NoError(t, kruisev1b1.AddToScheme(scheme)) + return scheme +} + +func jsonValue(t *testing.T, v any) *apiextensionsv1.JSON { + t.Helper() + raw, err := json.Marshal(v) + require.NoError(t, err) + return &apiextensionsv1.JSON{Raw: raw} +} + +// renderControllerStatefulSet renders a representative controller StatefulSet +// (an OpenKruise Advanced StatefulSet). +func renderControllerStatefulSet(t *testing.T) *kruisev1b1.StatefulSet { + t.Helper() + controller := &values.SlurmController{ + K8sNodeFilterName: "test-filter", + StatefulSet: values.StatefulSet{ + Name: "test-cluster-controller", + Replicas: 1, + MaxUnavailable: intstr.FromInt32(1), + }, + Service: values.Service{Name: "test-controller-svc"}, + ContainerSlurmctld: values.Container{ + NodeContainer: slurmv1.NodeContainer{ + Image: "test-image:latest", + ImagePullPolicy: corev1.PullAlways, + Port: 6817, + Resources: corev1.ResourceList{ + corev1.ResourceCPU: resource.MustParse("500m"), + corev1.ResourceMemory: resource.MustParse("1Gi"), + }, + AppArmorProfile: consts.AppArmorProfileUnconfined, + }, + Name: "slurmctld", + }, + ContainerMunge: values.Container{ + NodeContainer: slurmv1.NodeContainer{ + Image: "munge-image:latest", + ImagePullPolicy: corev1.PullAlways, + Resources: corev1.ResourceList{ + corev1.ResourceCPU: resource.MustParse("100m"), + corev1.ResourceMemory: resource.MustParse("256Mi"), + }, + AppArmorProfile: consts.AppArmorProfileUnconfined, + }, + }, + VolumeSpool: slurmv1.NodeVolume{VolumeSourceName: ptr.To("test-volume")}, + VolumeJail: slurmv1.NodeVolume{VolumeSourceName: ptr.To("test-volume")}, + PriorityClass: "test-priority", + } + + nodeFilters := []slurmv1.K8sNodeFilter{{ + Name: "test-filter", + Affinity: &corev1.Affinity{ + NodeAffinity: &corev1.NodeAffinity{ + RequiredDuringSchedulingIgnoredDuringExecution: &corev1.NodeSelector{ + NodeSelectorTerms: []corev1.NodeSelectorTerm{{ + MatchExpressions: []corev1.NodeSelectorRequirement{{ + Key: "node-type", + Operator: corev1.NodeSelectorOpIn, + Values: []string{"controller"}, + }}, + }}, + }, + }, + }, + }} + volumeSources := []slurmv1.VolumeSource{{ + Name: "test-volume", + VolumeSource: corev1.VolumeSource{EmptyDir: &corev1.EmptyDirVolumeSource{}}, + }} + + sts, err := RenderStatefulSet("test-namespace", "test-cluster", nodeFilters, volumeSources, controller, true) + require.NoError(t, err) + return &sts +} + +func TestPatchControllerStatefulSet_MeshAnnotationAndResources(t *testing.T) { + scheme := patchTestScheme(t) + sts := renderControllerStatefulSet(t) + require.NotEmpty(t, sts.Spec.Template.Spec.Containers) + + policy := slurmv1alpha1.ResourcePatchPolicy{ + ObjectMeta: metav1.ObjectMeta{Name: "ctl-patch", Namespace: "test-namespace"}, + Spec: slurmv1alpha1.ResourcePatchPolicySpec{ + TargetRef: slurmv1alpha1.PolicyTargetReference{ + Group: "slurm.nebius.ai", Kind: "SlurmCluster", Name: "test-cluster", + }, + Type: slurmv1alpha1.JSONPatchType, + Patches: []slurmv1alpha1.ResourcePatch{{ + ResourceRef: slurmv1alpha1.ResourceSelector{ + Kind: "StatefulSet", Name: ptr.To(sts.Name), + }, + JSONPatch: []slurmv1alpha1.JSONPatchOperation{ + { + Op: "add", + Path: "/spec/template/metadata/annotations", + Value: jsonValue(t, map[string]string{"sidecar.istio.io/inject": "true"}), + }, + { + Op: "add", + Path: "/spec/template/spec/containers/0/resources/limits", + Value: jsonValue(t, corev1.ResourceList{ + corev1.ResourceCPU: resource.MustParse("2"), + }), + }, + }, + }}, + }, + } + + results, err := resourcepatch.Apply(scheme, sts, []slurmv1alpha1.ResourcePatchPolicy{policy}) + require.NoError(t, err) + require.Len(t, results, 1) + assert.True(t, results[0].Applied, "message: %s", results[0].Message) + + assert.Equal(t, "true", sts.Spec.Template.Annotations["sidecar.istio.io/inject"]) + cpu := sts.Spec.Template.Spec.Containers[0].Resources.Limits[corev1.ResourceCPU] + assert.Equal(t, "2", cpu.String()) + + // Identity and selector must be untouched. + assert.Equal(t, "test-cluster-controller", sts.Name) + require.NotNil(t, sts.Spec.Selector) +} + +func TestPatchControllerStatefulSet_RejectsOwnerReferenceInjection(t *testing.T) { + scheme := patchTestScheme(t) + sts := renderControllerStatefulSet(t) + + policy := slurmv1alpha1.ResourcePatchPolicy{ + ObjectMeta: metav1.ObjectMeta{Name: "ctl-bad", Namespace: "test-namespace"}, + Spec: slurmv1alpha1.ResourcePatchPolicySpec{ + TargetRef: slurmv1alpha1.PolicyTargetReference{ + Group: "slurm.nebius.ai", Kind: "SlurmCluster", Name: "test-cluster", + }, + Type: slurmv1alpha1.JSONPatchType, + Patches: []slurmv1alpha1.ResourcePatch{{ + ResourceRef: slurmv1alpha1.ResourceSelector{Kind: "StatefulSet", Name: ptr.To(sts.Name)}, + JSONPatch: []slurmv1alpha1.JSONPatchOperation{ + {Op: "add", Path: "/metadata/ownerReferences", Value: jsonValue(t, []metav1.OwnerReference{ + {APIVersion: "v1", Kind: "Pod", Name: "evil", UID: "1"}, + })}, + }, + }}, + }, + } + + results, err := resourcepatch.Apply(scheme, sts, []slurmv1alpha1.ResourcePatchPolicy{policy}) + require.NoError(t, err) + require.Len(t, results, 1) + assert.False(t, results[0].Applied) + assert.Empty(t, sts.OwnerReferences) +} diff --git a/internal/render/nodeconfigurator/patch_test.go b/internal/render/nodeconfigurator/patch_test.go new file mode 100644 index 000000000..112a6ce71 --- /dev/null +++ b/internal/render/nodeconfigurator/patch_test.go @@ -0,0 +1,155 @@ +package nodeconfigurator + +import ( + "encoding/json" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" + "k8s.io/apimachinery/pkg/api/resource" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + clientgoscheme "k8s.io/client-go/kubernetes/scheme" + "k8s.io/utils/ptr" + + slurmv1alpha1 "nebius.ai/slurm-operator/api/v1alpha1" + "nebius.ai/slurm-operator/internal/resourcepatch" +) + +func patchTestScheme(t *testing.T) *runtime.Scheme { + t.Helper() + scheme := runtime.NewScheme() + require.NoError(t, clientgoscheme.AddToScheme(scheme)) + require.NoError(t, slurmv1alpha1.AddToScheme(scheme)) + return scheme +} + +func jsonValue(t *testing.T, v any) *apiextensionsv1.JSON { + t.Helper() + raw, err := json.Marshal(v) + require.NoError(t, err) + return &apiextensionsv1.JSON{Raw: raw} +} + +func newNodeConfigurator() *slurmv1alpha1.NodeConfigurator { + return &slurmv1alpha1.NodeConfigurator{ + ObjectMeta: metav1.ObjectMeta{Name: "node-configurator"}, + Spec: slurmv1alpha1.NodeConfiguratorSpec{ + Rebooter: slurmv1alpha1.Rebooter{Enabled: true}, + }, + } +} + +// TestPatchNodeConfiguratorDaemonSet applies a ResourcePatchPolicy to the real +// DaemonSet generated by RenderDaemonSet. +func TestPatchNodeConfiguratorDaemonSet_AddAnnotationAndResources(t *testing.T) { + scheme := patchTestScheme(t) + ds := RenderDaemonSet(newNodeConfigurator(), "test-namespace") + require.NotNil(t, ds) + require.NotEmpty(t, ds.Spec.Template.Spec.Containers) + + policy := slurmv1alpha1.ResourcePatchPolicy{ + ObjectMeta: metav1.ObjectMeta{Name: "nc-patch", Namespace: "test-namespace"}, + Spec: slurmv1alpha1.ResourcePatchPolicySpec{ + TargetRef: slurmv1alpha1.PolicyTargetReference{ + Group: "slurm.nebius.ai", Kind: "NodeConfigurator", Name: "node-configurator", + }, + Type: slurmv1alpha1.JSONPatchType, + Patches: []slurmv1alpha1.ResourcePatch{ + { + ResourceRef: slurmv1alpha1.ResourceSelector{ + Kind: "DaemonSet", Name: ptr.To(ds.Name), + }, + JSONPatch: []slurmv1alpha1.JSONPatchOperation{ + { + Op: "add", + Path: "/spec/template/metadata/annotations", + Value: jsonValue(t, map[string]string{"ad.datadoghq.com/inject": "true"}), + }, + { + Op: "add", + Path: "/spec/template/spec/containers/0/resources", + Value: jsonValue(t, corev1.ResourceRequirements{ + Limits: corev1.ResourceList{corev1.ResourceMemory: resource.MustParse("256Mi")}, + }), + }, + }, + }, + }, + }, + } + + results, err := resourcepatch.Apply(scheme, ds, []slurmv1alpha1.ResourcePatchPolicy{policy}) + require.NoError(t, err) + require.Len(t, results, 1) + assert.True(t, results[0].Applied, "message: %s", results[0].Message) + + assert.Equal(t, "true", ds.Spec.Template.Annotations["ad.datadoghq.com/inject"]) + mem := ds.Spec.Template.Spec.Containers[0].Resources.Limits[corev1.ResourceMemory] + assert.Equal(t, "256Mi", mem.String()) + + // The operator-managed update strategy must remain intact. + require.NotNil(t, ds.Spec.UpdateStrategy.RollingUpdate) +} + +func TestPatchNodeConfiguratorDaemonSet_AddToleration(t *testing.T) { + scheme := patchTestScheme(t) + ds := RenderDaemonSet(newNodeConfigurator(), "test-namespace") + before := len(ds.Spec.Template.Spec.Tolerations) + + tol := corev1.Toleration{Key: "dedicated", Operator: corev1.TolerationOpExists, Effect: corev1.TaintEffectNoSchedule} + policy := slurmv1alpha1.ResourcePatchPolicy{ + ObjectMeta: metav1.ObjectMeta{Name: "nc-tol", Namespace: "test-namespace"}, + Spec: slurmv1alpha1.ResourcePatchPolicySpec{ + TargetRef: slurmv1alpha1.PolicyTargetReference{ + Group: "slurm.nebius.ai", Kind: "NodeConfigurator", Name: "node-configurator", + }, + Type: slurmv1alpha1.JSONPatchType, + Patches: []slurmv1alpha1.ResourcePatch{{ + ResourceRef: slurmv1alpha1.ResourceSelector{Kind: "DaemonSet", Name: ptr.To(ds.Name)}, + JSONPatch: []slurmv1alpha1.JSONPatchOperation{ + {Op: "add", Path: "/spec/template/spec/tolerations", Value: jsonValue(t, []corev1.Toleration{})}, + {Op: "add", Path: "/spec/template/spec/tolerations/-", Value: jsonValue(t, tol)}, + }, + }}, + }, + } + + results, err := resourcepatch.Apply(scheme, ds, []slurmv1alpha1.ResourcePatchPolicy{policy}) + require.NoError(t, err) + require.Len(t, results, 1) + assert.True(t, results[0].Applied, "message: %s", results[0].Message) + require.Len(t, ds.Spec.Template.Spec.Tolerations, before+1) + assert.Equal(t, "dedicated", ds.Spec.Template.Spec.Tolerations[before].Key) +} + +func TestPatchNodeConfiguratorDaemonSet_RejectsSelectorChange(t *testing.T) { + scheme := patchTestScheme(t) + ds := RenderDaemonSet(newNodeConfigurator(), "test-namespace") + require.NotNil(t, ds.Spec.Selector) + originalSelector := ds.Spec.Selector.DeepCopy() + + policy := slurmv1alpha1.ResourcePatchPolicy{ + ObjectMeta: metav1.ObjectMeta{Name: "nc-bad", Namespace: "test-namespace"}, + Spec: slurmv1alpha1.ResourcePatchPolicySpec{ + TargetRef: slurmv1alpha1.PolicyTargetReference{ + Group: "slurm.nebius.ai", Kind: "NodeConfigurator", Name: "node-configurator", + }, + Type: slurmv1alpha1.JSONPatchType, + Patches: []slurmv1alpha1.ResourcePatch{{ + ResourceRef: slurmv1alpha1.ResourceSelector{Kind: "DaemonSet", Name: ptr.To(ds.Name)}, + JSONPatch: []slurmv1alpha1.JSONPatchOperation{ + {Op: "add", Path: "/spec/selector/matchLabels/hijack", Value: jsonValue(t, "yes")}, + }, + }}, + }, + } + + results, err := resourcepatch.Apply(scheme, ds, []slurmv1alpha1.ResourcePatchPolicy{policy}) + require.NoError(t, err) + require.Len(t, results, 1) + assert.False(t, results[0].Applied, "selector change must be rejected") + assert.Equal(t, originalSelector, ds.Spec.Selector, "selector must remain unchanged") +} diff --git a/internal/render/worker/patch_test.go b/internal/render/worker/patch_test.go new file mode 100644 index 000000000..8cb4164c0 --- /dev/null +++ b/internal/render/worker/patch_test.go @@ -0,0 +1,184 @@ +package worker_test + +import ( + "encoding/json" + "testing" + + kruisev1b1 "github.com/openkruise/kruise-api/apps/v1beta1" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" + "k8s.io/apimachinery/pkg/api/resource" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + clientgoscheme "k8s.io/client-go/kubernetes/scheme" + "k8s.io/utils/ptr" + "sigs.k8s.io/controller-runtime/pkg/client" + + slurmv1 "nebius.ai/slurm-operator/api/v1" + slurmv1alpha1 "nebius.ai/slurm-operator/api/v1alpha1" + "nebius.ai/slurm-operator/internal/consts" + "nebius.ai/slurm-operator/internal/render/worker" + "nebius.ai/slurm-operator/internal/resourcepatch" + "nebius.ai/slurm-operator/internal/values" +) + +func patchTestScheme(t *testing.T) *runtime.Scheme { + t.Helper() + scheme := runtime.NewScheme() + require.NoError(t, clientgoscheme.AddToScheme(scheme)) + require.NoError(t, slurmv1alpha1.AddToScheme(scheme)) + require.NoError(t, kruisev1b1.AddToScheme(scheme)) + return scheme +} + +func jsonValue(t *testing.T, v any) *apiextensionsv1.JSON { + t.Helper() + raw, err := json.Marshal(v) + require.NoError(t, err) + return &apiextensionsv1.JSON{Raw: raw} +} + +func renderWorkerStatefulSet(t *testing.T) *kruisev1b1.StatefulSet { + t.Helper() + nodeSet := &values.SlurmNodeSet{ + Name: "test-nodeset", + ParentalCluster: client.ObjectKey{ + Namespace: "test-namespace", + Name: "test-cluster", + }, + ContainerSlurmd: values.Container{ + NodeContainer: slurmv1.NodeContainer{ + Image: "test-image", + ImagePullPolicy: corev1.PullIfNotPresent, + Resources: corev1.ResourceList{ + corev1.ResourceMemory: resource.MustParse("1Gi"), + corev1.ResourceCPU: resource.MustParse("100m"), + corev1.ResourceEphemeralStorage: resource.MustParse("1Gi"), + }, + }, + }, + ContainerMunge: values.Container{ + NodeContainer: slurmv1.NodeContainer{Image: "munge-image"}, + }, + VolumeSpool: corev1.VolumeSource{HostPath: &corev1.HostPathVolumeSource{Path: "/tmp/spool"}}, + VolumeJail: corev1.VolumeSource{HostPath: &corev1.HostPathVolumeSource{Path: "/tmp/jail"}}, + StatefulSet: values.StatefulSet{Replicas: 1}, + ServiceUmbrella: values.Service{Name: "test-umbrella"}, + SupervisorDConfigMapName: "supervisord-config", + SSHDConfigMapName: "sshd-config", + GPU: &slurmv1alpha1.GPUSpec{Enabled: false}, + } + + sts, err := worker.RenderNodeSetStatefulSet( + "test-cluster", + nodeSet, + &slurmv1.Secrets{}, + consts.CGroupV2, + false, + false, + "", + ) + require.NoError(t, err) + return &sts +} + +func TestPatchWorkerStatefulSet_GpuLimitAndAnnotation(t *testing.T) { + scheme := patchTestScheme(t) + sts := renderWorkerStatefulSet(t) + require.NotEmpty(t, sts.Spec.Template.Spec.Containers) + + policy := slurmv1alpha1.ResourcePatchPolicy{ + ObjectMeta: metav1.ObjectMeta{Name: "worker-patch", Namespace: "test-namespace"}, + Spec: slurmv1alpha1.ResourcePatchPolicySpec{ + TargetRef: slurmv1alpha1.PolicyTargetReference{ + Group: "slurm.nebius.ai", Kind: "NodeSet", Name: "test-nodeset", + }, + Type: slurmv1alpha1.JSONPatchType, + Patches: []slurmv1alpha1.ResourcePatch{{ + ResourceRef: slurmv1alpha1.ResourceSelector{ + Kind: "StatefulSet", Name: ptr.To(sts.Name), + }, + JSONPatch: []slurmv1alpha1.JSONPatchOperation{ + { + Op: "add", + Path: "/spec/template/metadata/annotations", + Value: jsonValue(t, map[string]string{"ad.datadoghq.com/worker.checks": "{}"}), + }, + { + Op: "add", + Path: "/spec/template/spec/containers/0/resources/limits/nvidia.com~1gpu", + Value: jsonValue(t, "8"), + }, + }, + }}, + }, + } + + results, err := resourcepatch.Apply(scheme, sts, []slurmv1alpha1.ResourcePatchPolicy{policy}) + require.NoError(t, err) + require.Len(t, results, 1) + assert.True(t, results[0].Applied, "message: %s", results[0].Message) + + assert.Equal(t, "{}", sts.Spec.Template.Annotations["ad.datadoghq.com/worker.checks"]) + gpu := sts.Spec.Template.Spec.Containers[0].Resources.Limits["nvidia.com/gpu"] + assert.Equal(t, "8", gpu.String()) +} + +func TestPatchWorkerStatefulSet_AddNodeSelector(t *testing.T) { + scheme := patchTestScheme(t) + sts := renderWorkerStatefulSet(t) + + policy := slurmv1alpha1.ResourcePatchPolicy{ + ObjectMeta: metav1.ObjectMeta{Name: "worker-ns", Namespace: "test-namespace"}, + Spec: slurmv1alpha1.ResourcePatchPolicySpec{ + TargetRef: slurmv1alpha1.PolicyTargetReference{ + Group: "slurm.nebius.ai", Kind: "NodeSet", Name: "test-nodeset", + }, + Type: slurmv1alpha1.JSONPatchType, + Patches: []slurmv1alpha1.ResourcePatch{{ + ResourceRef: slurmv1alpha1.ResourceSelector{Kind: "StatefulSet", Name: ptr.To(sts.Name)}, + JSONPatch: []slurmv1alpha1.JSONPatchOperation{ + {Op: "add", Path: "/spec/template/spec/nodeSelector", Value: jsonValue(t, map[string]string{ + "node-pool": "gpu", + })}, + }, + }}, + }, + } + + results, err := resourcepatch.Apply(scheme, sts, []slurmv1alpha1.ResourcePatchPolicy{policy}) + require.NoError(t, err) + require.Len(t, results, 1) + assert.True(t, results[0].Applied, "message: %s", results[0].Message) + assert.Equal(t, "gpu", sts.Spec.Template.Spec.NodeSelector["node-pool"]) +} + +func TestPatchWorkerStatefulSet_RejectsNameChange(t *testing.T) { + scheme := patchTestScheme(t) + sts := renderWorkerStatefulSet(t) + original := sts.Name + + policy := slurmv1alpha1.ResourcePatchPolicy{ + ObjectMeta: metav1.ObjectMeta{Name: "worker-bad", Namespace: "test-namespace"}, + Spec: slurmv1alpha1.ResourcePatchPolicySpec{ + TargetRef: slurmv1alpha1.PolicyTargetReference{ + Group: "slurm.nebius.ai", Kind: "NodeSet", Name: "test-nodeset", + }, + Type: slurmv1alpha1.JSONPatchType, + Patches: []slurmv1alpha1.ResourcePatch{{ + ResourceRef: slurmv1alpha1.ResourceSelector{Kind: "StatefulSet", Name: ptr.To(sts.Name)}, + JSONPatch: []slurmv1alpha1.JSONPatchOperation{ + {Op: "replace", Path: "/metadata/name", Value: jsonValue(t, "hijacked")}, + }, + }}, + }, + } + + results, err := resourcepatch.Apply(scheme, sts, []slurmv1alpha1.ResourcePatchPolicy{policy}) + require.NoError(t, err) + require.Len(t, results, 1) + assert.False(t, results[0].Applied) + assert.Equal(t, original, sts.Name) +} diff --git a/internal/resourcepatch/engine.go b/internal/resourcepatch/engine.go new file mode 100644 index 000000000..65d6196f2 --- /dev/null +++ b/internal/resourcepatch/engine.go @@ -0,0 +1,267 @@ +/* +Copyright 2024 Nebius B.V. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +// Package resourcepatch implements the in-memory patching engine used by the +// ResourcePatchPolicy feature. It applies RFC 6902 (JSON Patch) and RFC 7386 +// (JSON Merge Patch) operations to Kubernetes objects generated by the operator +// before they are submitted to the API server. +package resourcepatch + +import ( + "encoding/json" + "fmt" + "reflect" + "sort" + + jsonpatch "github.com/evanphx/json-patch/v5" + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/apiutil" + + slurmv1alpha1 "nebius.ai/slurm-operator/api/v1alpha1" +) + +// Result describes the outcome of applying a single policy patch entry to one +// object. +type Result struct { + // PolicyName / PolicyNamespace identify the source policy. + PolicyName string + PolicyNamespace string + + // Resource identifies the object the patch targeted. + Resource corev1.ObjectReference + + // Applied reports whether the patch was applied to the object. + Applied bool + + // Message provides human-readable details, typically on failure. + Message string +} + +// Apply applies all matching patches from policies to obj in place. +// +// policies must already be filtered to those whose targetRef points at obj's +// parent (see FilterPoliciesForTarget). Policies are applied in ascending +// priority order, with namespace/name used as a deterministic tiebreaker. +// +// A patch that fails (bad path, protected-field violation, malformed payload) +// leaves obj untouched for that entry and is reported as a non-applied Result; +// the remaining patches still run. The returned error is non-nil only for +// problems that prevent matching altogether (e.g. unknown object type). +func Apply( + scheme *runtime.Scheme, + obj client.Object, + policies []slurmv1alpha1.ResourcePatchPolicy, +) ([]Result, error) { + gvk, err := apiutil.GVKForObject(obj, scheme) + if err != nil { + return nil, fmt.Errorf("resolving GVK for object: %w", err) + } + kind := gvk.Kind + apiVersion := gvk.GroupVersion().String() + name := obj.GetName() + + sorted := sortPolicies(policies) + + var results []Result + for i := range sorted { + policy := &sorted[i] + + // Never mutate a resource using a policy that would be rejected by + // static validation; this keeps the apply path consistent with the + // Accepted status condition set by the policy controller. + if err := ValidatePolicy(policy); err != nil { + results = append(results, Result{ + PolicyName: policy.Name, + PolicyNamespace: policy.Namespace, + Resource: corev1.ObjectReference{ + APIVersion: apiVersion, + Kind: kind, + Name: name, + Namespace: obj.GetNamespace(), + }, + Message: fmt.Sprintf("policy failed validation: %v", err), + }) + continue + } + + for j := range policy.Spec.Patches { + patch := &policy.Spec.Patches[j] + if !matchSelector(patch.ResourceRef, kind, apiVersion, name) { + continue + } + res := applyOne(obj, policy, patch, kind, apiVersion) + results = append(results, res) + } + } + return results, nil +} + +// FilterPoliciesForTarget returns the subset of policies whose targetRef points +// at the parent identified by group/kind/name/namespace. A policy with an empty +// targetRef namespace defaults to the policy's own namespace. +func FilterPoliciesForTarget( + policies []slurmv1alpha1.ResourcePatchPolicy, + group, kind, name, namespace string, +) []slurmv1alpha1.ResourcePatchPolicy { + var out []slurmv1alpha1.ResourcePatchPolicy + for i := range policies { + ref := policies[i].Spec.TargetRef + targetNS := policies[i].Namespace + if ref.Namespace != nil && *ref.Namespace != "" { + targetNS = *ref.Namespace + } + if ref.Group == group && ref.Kind == kind && ref.Name == name && targetNS == namespace { + out = append(out, policies[i]) + } + } + return out +} + +func applyOne( + obj client.Object, + policy *slurmv1alpha1.ResourcePatchPolicy, + patch *slurmv1alpha1.ResourcePatch, + kind, apiVersion string, +) Result { + res := Result{ + PolicyName: policy.Name, + PolicyNamespace: policy.Namespace, + Resource: corev1.ObjectReference{ + APIVersion: apiVersion, + Kind: kind, + Name: obj.GetName(), + Namespace: obj.GetNamespace(), + }, + } + + original, err := json.Marshal(obj) + if err != nil { + res.Message = fmt.Sprintf("marshalling object: %v", err) + return res + } + + patched, err := runPatch(policy.Spec.Type, patch, original) + if err != nil { + res.Message = err.Error() + return res + } + + if violation := protectedFieldViolation(original, patched); violation != "" { + res.Message = violation + return res + } + + if err := resetAndUnmarshal(obj, patched); err != nil { + res.Message = fmt.Sprintf("decoding patched object: %v", err) + return res + } + + res.Applied = true + return res +} + +func runPatch( + patchType slurmv1alpha1.PatchType, + patch *slurmv1alpha1.ResourcePatch, + doc []byte, +) ([]byte, error) { + switch patchType { + case slurmv1alpha1.JSONPatchType: + if len(patch.JSONPatch) == 0 { + return nil, fmt.Errorf("policy type is JSONPatch but jsonPatch is empty") + } + raw, err := json.Marshal(patch.JSONPatch) + if err != nil { + return nil, fmt.Errorf("marshalling jsonPatch: %w", err) + } + decoded, err := jsonpatch.DecodePatch(raw) + if err != nil { + return nil, fmt.Errorf("decoding jsonPatch: %w", err) + } + patched, err := decoded.Apply(doc) + if err != nil { + return nil, fmt.Errorf("applying jsonPatch: %w", err) + } + return patched, nil + + case slurmv1alpha1.JSONMergePatchType: + if patch.JSONMergePatch == nil || len(patch.JSONMergePatch.Raw) == 0 { + return nil, fmt.Errorf("policy type is JSONMergePatch but jsonMergePatch is empty") + } + patched, err := jsonpatch.MergePatch(doc, patch.JSONMergePatch.Raw) + if err != nil { + return nil, fmt.Errorf("applying jsonMergePatch: %w", err) + } + return patched, nil + + default: + return nil, fmt.Errorf("unsupported patch type %q", patchType) + } +} + +// matchSelector reports whether ref matches a resource of the given kind, +// apiVersion and name. Name matching is exact; an empty Name matches every +// resource of the kind. +func matchSelector(ref slurmv1alpha1.ResourceSelector, kind, apiVersion, name string) bool { + if ref.Kind != kind { + return false + } + if ref.APIVersion != nil && *ref.APIVersion != "" && *ref.APIVersion != apiVersion { + return false + } + if ref.Name == nil || *ref.Name == "" { + return true + } + return *ref.Name == name +} + +func sortPolicies(policies []slurmv1alpha1.ResourcePatchPolicy) []slurmv1alpha1.ResourcePatchPolicy { + sorted := make([]slurmv1alpha1.ResourcePatchPolicy, len(policies)) + copy(sorted, policies) + sort.SliceStable(sorted, func(i, j int) bool { + pi, pj := priorityOf(&sorted[i]), priorityOf(&sorted[j]) + if pi != pj { + return pi < pj + } + if sorted[i].Namespace != sorted[j].Namespace { + return sorted[i].Namespace < sorted[j].Namespace + } + return sorted[i].Name < sorted[j].Name + }) + return sorted +} + +func priorityOf(policy *slurmv1alpha1.ResourcePatchPolicy) int32 { + if policy.Spec.Priority == nil { + return 0 + } + return *policy.Spec.Priority +} + +// resetAndUnmarshal replaces the contents of obj with the decoded JSON document. +// obj is zeroed first so that fields removed by the patch are cleared rather +// than retained from the pre-patch state. +func resetAndUnmarshal(obj client.Object, data []byte) error { + v := reflect.ValueOf(obj) + if v.Kind() != reflect.Ptr || v.IsNil() { + return fmt.Errorf("object must be a non-nil pointer, got %T", obj) + } + elem := v.Elem() + elem.Set(reflect.Zero(elem.Type())) + return json.Unmarshal(data, obj) +} diff --git a/internal/resourcepatch/engine_test.go b/internal/resourcepatch/engine_test.go new file mode 100644 index 000000000..89e9d509e --- /dev/null +++ b/internal/resourcepatch/engine_test.go @@ -0,0 +1,618 @@ +/* +Copyright 2024 Nebius B.V. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package resourcepatch_test + +import ( + "encoding/json" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" + "k8s.io/apimachinery/pkg/api/resource" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/util/intstr" + clientgoscheme "k8s.io/client-go/kubernetes/scheme" + "k8s.io/utils/ptr" + "sigs.k8s.io/controller-runtime/pkg/client" + + slurmv1alpha1 "nebius.ai/slurm-operator/api/v1alpha1" + "nebius.ai/slurm-operator/internal/resourcepatch" +) + +func testScheme(t *testing.T) *runtime.Scheme { + t.Helper() + scheme := runtime.NewScheme() + require.NoError(t, clientgoscheme.AddToScheme(scheme)) + require.NoError(t, slurmv1alpha1.AddToScheme(scheme)) + return scheme +} + +// jsonValue marshals v into an *apiextensionsv1.JSON for use as a JSON Patch +// operation value. +func jsonValue(t *testing.T, v any) *apiextensionsv1.JSON { + t.Helper() + raw, err := json.Marshal(v) + require.NoError(t, err) + return &apiextensionsv1.JSON{Raw: raw} +} + +func mustQuantity(s string) resource.Quantity { + return resource.MustParse(s) +} + +// policy builds a JSONPatch policy targeting a SlurmCluster. +func policy(name string, priority int32, patches ...slurmv1alpha1.ResourcePatch) slurmv1alpha1.ResourcePatchPolicy { + return slurmv1alpha1.ResourcePatchPolicy{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "default"}, + Spec: slurmv1alpha1.ResourcePatchPolicySpec{ + TargetRef: slurmv1alpha1.PolicyTargetReference{ + Group: "slurm.nebius.ai", + Kind: "SlurmCluster", + Name: "my-cluster", + }, + Priority: ptr.To(priority), + Type: slurmv1alpha1.JSONPatchType, + Patches: patches, + }, + } +} + +func nameSel(kind, name string) slurmv1alpha1.ResourceSelector { + return slurmv1alpha1.ResourceSelector{Kind: kind, Name: ptr.To(name)} +} + +// representativeStatefulSet mirrors the structure of a controller/worker +// StatefulSet generated by the operator (one main container, labelled pod +// template, immutable selector). +func representativeStatefulSet() *appsv1.StatefulSet { + return &appsv1.StatefulSet{ + ObjectMeta: metav1.ObjectMeta{ + Name: "my-cluster-worker", + Namespace: "default", + Labels: map[string]string{"app.kubernetes.io/component": "worker"}, + Annotations: map[string]string{"existing": "true"}, + }, + Spec: appsv1.StatefulSetSpec{ + Replicas: ptr.To(int32(3)), + Selector: &metav1.LabelSelector{ + MatchLabels: map[string]string{"app.kubernetes.io/component": "worker"}, + }, + Template: corev1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{ + Labels: map[string]string{"app.kubernetes.io/component": "worker"}, + Annotations: map[string]string{"kubectl.kubernetes.io/default-container": "slurmd"}, + }, + Spec: corev1.PodSpec{ + Containers: []corev1.Container{ + { + Name: "slurmd", + Image: "slurmd:latest", + Resources: corev1.ResourceRequirements{ + Limits: corev1.ResourceList{corev1.ResourceCPU: resource.MustParse("500m")}, + }, + }, + }, + }, + }, + }, + } +} + +func TestApply_AddPodTemplateAnnotation(t *testing.T) { + scheme := testScheme(t) + sts := representativeStatefulSet() + + p := policy("mesh", 0, slurmv1alpha1.ResourcePatch{ + ResourceRef: nameSel("StatefulSet", "my-cluster-worker"), + JSONPatch: []slurmv1alpha1.JSONPatchOperation{ + { + Op: "add", + Path: "/spec/template/metadata/annotations/sidecar.istio.io~1inject", + Value: jsonValue(t, "true"), + }, + }, + }) + + results, err := resourcepatch.Apply(scheme, sts, []slurmv1alpha1.ResourcePatchPolicy{p}) + require.NoError(t, err) + require.Len(t, results, 1) + assert.True(t, results[0].Applied, "message: %s", results[0].Message) + assert.Equal(t, "true", sts.Spec.Template.Annotations["sidecar.istio.io/inject"]) +} + +func TestApply_ReplaceContainerResourceLimit(t *testing.T) { + scheme := testScheme(t) + sts := representativeStatefulSet() + + p := policy("gpu", 0, slurmv1alpha1.ResourcePatch{ + ResourceRef: nameSel("StatefulSet", "my-cluster-worker"), + JSONPatch: []slurmv1alpha1.JSONPatchOperation{ + { + Op: "add", + Path: "/spec/template/spec/containers/0/resources/limits/nvidia.com~1gpu", + Value: jsonValue(t, "8"), + }, + }, + }) + + results, err := resourcepatch.Apply(scheme, sts, []slurmv1alpha1.ResourcePatchPolicy{p}) + require.NoError(t, err) + require.Len(t, results, 1) + assert.True(t, results[0].Applied, "message: %s", results[0].Message) + + gpu := sts.Spec.Template.Spec.Containers[0].Resources.Limits["nvidia.com/gpu"] + assert.Equal(t, "8", gpu.String()) +} + +func TestApply_AddTolerationToList(t *testing.T) { + scheme := testScheme(t) + sts := representativeStatefulSet() + + toleration := corev1.Toleration{ + Key: "nvidia.com/gpu", + Operator: corev1.TolerationOpExists, + Effect: corev1.TaintEffectNoSchedule, + } + p := policy("tol", 0, + slurmv1alpha1.ResourcePatch{ + ResourceRef: nameSel("StatefulSet", "my-cluster-worker"), + JSONPatch: []slurmv1alpha1.JSONPatchOperation{ + // First ensure the tolerations array exists, then append. + {Op: "add", Path: "/spec/template/spec/tolerations", Value: jsonValue(t, []corev1.Toleration{})}, + {Op: "add", Path: "/spec/template/spec/tolerations/-", Value: jsonValue(t, toleration)}, + }, + }, + ) + + results, err := resourcepatch.Apply(scheme, sts, []slurmv1alpha1.ResourcePatchPolicy{p}) + require.NoError(t, err) + require.Len(t, results, 1) + assert.True(t, results[0].Applied, "message: %s", results[0].Message) + require.Len(t, sts.Spec.Template.Spec.Tolerations, 1) + assert.Equal(t, "nvidia.com/gpu", sts.Spec.Template.Spec.Tolerations[0].Key) +} + +func TestApply_RemoveOpClearsField(t *testing.T) { + scheme := testScheme(t) + sts := representativeStatefulSet() + require.Equal(t, "true", sts.Annotations["existing"]) + + p := policy("rm", 0, slurmv1alpha1.ResourcePatch{ + ResourceRef: nameSel("StatefulSet", "my-cluster-worker"), + JSONPatch: []slurmv1alpha1.JSONPatchOperation{ + {Op: "remove", Path: "/metadata/annotations/existing"}, + }, + }) + + results, err := resourcepatch.Apply(scheme, sts, []slurmv1alpha1.ResourcePatchPolicy{p}) + require.NoError(t, err) + require.Len(t, results, 1) + assert.True(t, results[0].Applied, "message: %s", results[0].Message) + _, ok := sts.Annotations["existing"] + assert.False(t, ok, "annotation should have been removed") +} + +func TestApply_JSONMergePatch(t *testing.T) { + scheme := testScheme(t) + cm := &corev1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{ + Name: "my-cluster-slurm-config", + Namespace: "default", + }, + Data: map[string]string{"slurm.conf": "base"}, + } + + p := slurmv1alpha1.ResourcePatchPolicy{ + ObjectMeta: metav1.ObjectMeta{Name: "merge", Namespace: "default"}, + Spec: slurmv1alpha1.ResourcePatchPolicySpec{ + TargetRef: slurmv1alpha1.PolicyTargetReference{ + Group: "slurm.nebius.ai", Kind: "SlurmCluster", Name: "my-cluster", + }, + Type: slurmv1alpha1.JSONMergePatchType, + Patches: []slurmv1alpha1.ResourcePatch{ + { + ResourceRef: nameSel("ConfigMap", "my-cluster-slurm-config"), + JSONMergePatch: jsonValue(t, map[string]any{ + "metadata": map[string]any{ + "annotations": map[string]string{"custom-key": "custom-value"}, + }, + "data": map[string]string{"extra.conf": "SomeExtraConfig=value"}, + }), + }, + }, + }, + } + + results, err := resourcepatch.Apply(scheme, cm, []slurmv1alpha1.ResourcePatchPolicy{p}) + require.NoError(t, err) + require.Len(t, results, 1) + assert.True(t, results[0].Applied, "message: %s", results[0].Message) + assert.Equal(t, "custom-value", cm.Annotations["custom-key"]) + assert.Equal(t, "SomeExtraConfig=value", cm.Data["extra.conf"]) + assert.Equal(t, "base", cm.Data["slurm.conf"], "merge patch must preserve existing keys") +} + +func TestApply_ExactNameSelector(t *testing.T) { + scheme := testScheme(t) + + tests := []struct { + name string + selector string + objName string + want bool + }{ + {"exact match", "my-cluster-worker", "my-cluster-worker", true}, + {"exact mismatch", "my-cluster-worker", "my-cluster-login", false}, + // Glob patterns are not interpreted: they only match a literal name. + {"glob is treated literally", "my-cluster-worker-*", "my-cluster-worker-0", false}, + {"glob literal exact", "my-cluster-worker-*", "my-cluster-worker-*", true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + sts := representativeStatefulSet() + sts.Name = tt.objName + p := policy("p", 0, slurmv1alpha1.ResourcePatch{ + ResourceRef: slurmv1alpha1.ResourceSelector{Kind: "StatefulSet", Name: ptr.To(tt.selector)}, + JSONPatch: []slurmv1alpha1.JSONPatchOperation{ + {Op: "add", Path: "/metadata/annotations/patched", Value: jsonValue(t, "yes")}, + }, + }) + results, err := resourcepatch.Apply(scheme, sts, []slurmv1alpha1.ResourcePatchPolicy{p}) + require.NoError(t, err) + if tt.want { + require.Len(t, results, 1) + assert.True(t, results[0].Applied) + assert.Equal(t, "yes", sts.Annotations["patched"]) + } else { + assert.Empty(t, results, "selector should not have matched") + _, ok := sts.Annotations["patched"] + assert.False(t, ok) + } + }) + } +} + +func TestApply_EmptyNameSelectorMatchesAllOfKind(t *testing.T) { + scheme := testScheme(t) + sts := representativeStatefulSet() + sts.Name = "anything" + + p := policy("p", 0, slurmv1alpha1.ResourcePatch{ + ResourceRef: slurmv1alpha1.ResourceSelector{Kind: "StatefulSet"}, // no NameSelector + JSONPatch: []slurmv1alpha1.JSONPatchOperation{ + {Op: "add", Path: "/metadata/annotations/patched", Value: jsonValue(t, "yes")}, + }, + }) + results, err := resourcepatch.Apply(scheme, sts, []slurmv1alpha1.ResourcePatchPolicy{p}) + require.NoError(t, err) + require.Len(t, results, 1) + assert.True(t, results[0].Applied) +} + +func TestApply_KindMismatchSkipped(t *testing.T) { + scheme := testScheme(t) + sts := representativeStatefulSet() + + p := policy("p", 0, slurmv1alpha1.ResourcePatch{ + ResourceRef: nameSel("Service", "my-cluster-worker"), // wrong kind for a StatefulSet + JSONPatch: []slurmv1alpha1.JSONPatchOperation{ + {Op: "add", Path: "/metadata/annotations/patched", Value: jsonValue(t, "yes")}, + }, + }) + results, err := resourcepatch.Apply(scheme, sts, []slurmv1alpha1.ResourcePatchPolicy{p}) + require.NoError(t, err) + assert.Empty(t, results) +} + +func TestApply_APIVersionSelector(t *testing.T) { + scheme := testScheme(t) + sts := representativeStatefulSet() + + mkPolicy := func(apiVersion string) slurmv1alpha1.ResourcePatchPolicy { + return policy("p", 0, slurmv1alpha1.ResourcePatch{ + ResourceRef: slurmv1alpha1.ResourceSelector{ + Kind: "StatefulSet", + Name: ptr.To("my-cluster-worker"), + APIVersion: ptr.To(apiVersion), + }, + JSONPatch: []slurmv1alpha1.JSONPatchOperation{ + {Op: "add", Path: "/metadata/annotations/patched", Value: jsonValue(t, "yes")}, + }, + }) + } + + t.Run("matching apiVersion", func(t *testing.T) { + obj := representativeStatefulSet() + results, err := resourcepatch.Apply(scheme, obj, []slurmv1alpha1.ResourcePatchPolicy{mkPolicy("apps/v1")}) + require.NoError(t, err) + require.Len(t, results, 1) + assert.True(t, results[0].Applied) + }) + + t.Run("mismatching apiVersion", func(t *testing.T) { + results, err := resourcepatch.Apply(scheme, sts, []slurmv1alpha1.ResourcePatchPolicy{mkPolicy("apps/v1beta1")}) + require.NoError(t, err) + assert.Empty(t, results) + }) +} + +func TestApply_PriorityOrdering(t *testing.T) { + scheme := testScheme(t) + sts := representativeStatefulSet() + + // Two policies write the same annotation; the higher-priority number is + // applied last and therefore wins. + low := policy("a-low", 5, slurmv1alpha1.ResourcePatch{ + ResourceRef: nameSel("StatefulSet", "my-cluster-worker"), + JSONPatch: []slurmv1alpha1.JSONPatchOperation{ + {Op: "add", Path: "/metadata/annotations/winner", Value: jsonValue(t, "low")}, + }, + }) + high := policy("z-high", 10, slurmv1alpha1.ResourcePatch{ + ResourceRef: nameSel("StatefulSet", "my-cluster-worker"), + JSONPatch: []slurmv1alpha1.JSONPatchOperation{ + {Op: "add", Path: "/metadata/annotations/winner", Value: jsonValue(t, "high")}, + }, + }) + + // Pass them out of order to prove sorting happens inside Apply. + results, err := resourcepatch.Apply(scheme, sts, []slurmv1alpha1.ResourcePatchPolicy{high, low}) + require.NoError(t, err) + require.Len(t, results, 2) + assert.Equal(t, "high", sts.Annotations["winner"]) +} + +func TestApply_EqualPriorityAlphabeticalTiebreak(t *testing.T) { + scheme := testScheme(t) + sts := representativeStatefulSet() + + first := policy("aaa", 0, slurmv1alpha1.ResourcePatch{ + ResourceRef: nameSel("StatefulSet", "my-cluster-worker"), + JSONPatch: []slurmv1alpha1.JSONPatchOperation{{Op: "add", Path: "/metadata/annotations/winner", Value: jsonValue(t, "aaa")}}, + }) + second := policy("bbb", 0, slurmv1alpha1.ResourcePatch{ + ResourceRef: nameSel("StatefulSet", "my-cluster-worker"), + JSONPatch: []slurmv1alpha1.JSONPatchOperation{{Op: "add", Path: "/metadata/annotations/winner", Value: jsonValue(t, "bbb")}}, + }) + + results, err := resourcepatch.Apply(scheme, sts, []slurmv1alpha1.ResourcePatchPolicy{second, first}) + require.NoError(t, err) + require.Len(t, results, 2) + assert.Equal(t, "bbb", sts.Annotations["winner"], "later alphabetical name applied last") +} + +func TestApply_FailedPatchDoesNotMutateAndOthersContinue(t *testing.T) { + scheme := testScheme(t) + sts := representativeStatefulSet() + + p := policy("mixed", 0, + slurmv1alpha1.ResourcePatch{ + ResourceRef: nameSel("StatefulSet", "my-cluster-worker"), + // "test" op against a non-matching value fails. + JSONPatch: []slurmv1alpha1.JSONPatchOperation{ + {Op: "test", Path: "/metadata/name", Value: jsonValue(t, "wrong-name")}, + }, + }, + slurmv1alpha1.ResourcePatch{ + ResourceRef: nameSel("StatefulSet", "my-cluster-worker"), + JSONPatch: []slurmv1alpha1.JSONPatchOperation{ + {Op: "add", Path: "/metadata/annotations/good", Value: jsonValue(t, "yes")}, + }, + }, + ) + + results, err := resourcepatch.Apply(scheme, sts, []slurmv1alpha1.ResourcePatchPolicy{p}) + require.NoError(t, err) + require.Len(t, results, 2) + assert.False(t, results[0].Applied) + assert.NotEmpty(t, results[0].Message) + assert.True(t, results[1].Applied, "message: %s", results[1].Message) + assert.Equal(t, "yes", sts.Annotations["good"]) + assert.Equal(t, "my-cluster-worker", sts.Name, "failed patch must not have mutated the object") +} + +func TestApply_InvalidPolicyDoesNotMutate(t *testing.T) { + scheme := testScheme(t) + sts := representativeStatefulSet() + + // Invalid: JSONPatch type but the entry also carries a merge payload, which + // ValidatePolicy rejects. The valid-looking JSONPatch must NOT be applied. + p := policy("invalid", 0, slurmv1alpha1.ResourcePatch{ + ResourceRef: nameSel("StatefulSet", "my-cluster-worker"), + JSONPatch: []slurmv1alpha1.JSONPatchOperation{{Op: "add", Path: "/metadata/annotations/x", Value: jsonValue(t, "y")}}, + JSONMergePatch: jsonValue(t, map[string]any{"metadata": map[string]any{}}), + }) + + results, err := resourcepatch.Apply(scheme, sts, []slurmv1alpha1.ResourcePatchPolicy{p}) + require.NoError(t, err) + require.Len(t, results, 1) + assert.False(t, results[0].Applied) + assert.Contains(t, results[0].Message, "failed validation") + _, ok := sts.Annotations["x"] + assert.False(t, ok, "invalid policy must not mutate the object") +} + +func TestApply_BadPathReportedAsFailure(t *testing.T) { + scheme := testScheme(t) + sts := representativeStatefulSet() + + p := policy("bad", 0, slurmv1alpha1.ResourcePatch{ + ResourceRef: nameSel("StatefulSet", "my-cluster-worker"), + JSONPatch: []slurmv1alpha1.JSONPatchOperation{ + // "replace" requires the target path to exist. + {Op: "replace", Path: "/spec/template/spec/containers/5/image", Value: jsonValue(t, "x")}, + }, + }) + results, err := resourcepatch.Apply(scheme, sts, []slurmv1alpha1.ResourcePatchPolicy{p}) + require.NoError(t, err) + require.Len(t, results, 1) + assert.False(t, results[0].Applied) + assert.NotEmpty(t, results[0].Message) +} + +func TestApply_ProtectedFields(t *testing.T) { + scheme := testScheme(t) + + tests := []struct { + name string + op slurmv1alpha1.JSONPatchOperation + }{ + { + name: "name", + op: slurmv1alpha1.JSONPatchOperation{Op: "replace", Path: "/metadata/name", Value: jsonValue(t, "renamed")}, + }, + { + name: "namespace", + op: slurmv1alpha1.JSONPatchOperation{Op: "replace", Path: "/metadata/namespace", Value: jsonValue(t, "other")}, + }, + { + name: "ownerReferences", + op: slurmv1alpha1.JSONPatchOperation{Op: "add", Path: "/metadata/ownerReferences", Value: jsonValue(t, []metav1.OwnerReference{ + {APIVersion: "v1", Kind: "Pod", Name: "x", UID: "123"}, + })}, + }, + { + name: "selector", + op: slurmv1alpha1.JSONPatchOperation{Op: "replace", Path: "/spec/selector/matchLabels", Value: jsonValue(t, map[string]string{ + "app.kubernetes.io/component": "hijacked", + })}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + sts := representativeStatefulSet() + p := policy("protected", 0, slurmv1alpha1.ResourcePatch{ + ResourceRef: nameSel("StatefulSet", "my-cluster-worker"), + JSONPatch: []slurmv1alpha1.JSONPatchOperation{tt.op}, + }) + results, err := resourcepatch.Apply(scheme, sts, []slurmv1alpha1.ResourcePatchPolicy{p}) + require.NoError(t, err) + require.Len(t, results, 1) + assert.False(t, results[0].Applied, "protected field %s must be rejected", tt.name) + assert.NotEmpty(t, results[0].Message) + // The object must be unchanged. + assert.Equal(t, "my-cluster-worker", sts.Name) + assert.Equal(t, "default", sts.Namespace) + assert.Empty(t, sts.OwnerReferences) + assert.Equal(t, "worker", sts.Spec.Selector.MatchLabels["app.kubernetes.io/component"]) + }) + } +} + +func TestApply_TypeMismatchEmptyPatchReported(t *testing.T) { + scheme := testScheme(t) + sts := representativeStatefulSet() + + // Policy type is JSONPatch but the entry only carries a merge patch. + p := policy("empty", 0, slurmv1alpha1.ResourcePatch{ + ResourceRef: nameSel("StatefulSet", "my-cluster-worker"), + JSONMergePatch: jsonValue(t, map[string]any{"metadata": map[string]any{}}), + }) + results, err := resourcepatch.Apply(scheme, sts, []slurmv1alpha1.ResourcePatchPolicy{p}) + require.NoError(t, err) + require.Len(t, results, 1) + assert.False(t, results[0].Applied) + assert.Contains(t, results[0].Message, "jsonPatch is empty") +} + +func TestFilterPoliciesForTarget(t *testing.T) { + mk := func(name, ns string, refKind, refName string, refNS *string) slurmv1alpha1.ResourcePatchPolicy { + return slurmv1alpha1.ResourcePatchPolicy{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: ns}, + Spec: slurmv1alpha1.ResourcePatchPolicySpec{ + TargetRef: slurmv1alpha1.PolicyTargetReference{ + Group: "slurm.nebius.ai", Kind: refKind, Name: refName, Namespace: refNS, + }, + }, + } + } + + policies := []slurmv1alpha1.ResourcePatchPolicy{ + mk("match-same-ns", "hpc", "SlurmCluster", "gpu", nil), + mk("wrong-kind", "hpc", "NodeSet", "gpu", nil), + mk("wrong-name", "hpc", "SlurmCluster", "other", nil), + mk("explicit-ns", "default", "SlurmCluster", "gpu", ptr.To("hpc")), + mk("explicit-other-ns", "default", "SlurmCluster", "gpu", ptr.To("elsewhere")), + } + + got := resourcepatch.FilterPoliciesForTarget(policies, "slurm.nebius.ai", "SlurmCluster", "gpu", "hpc") + var names []string + for _, p := range got { + names = append(names, p.Name) + } + assert.ElementsMatch(t, []string{"match-same-ns", "explicit-ns"}, names) +} + +func TestApply_NodeConfiguratorTargetGenericKind(t *testing.T) { + // The engine itself is agnostic to the target kind; matching is by the + // generated resource's kind. This proves a DaemonSet is patchable. + scheme := testScheme(t) + ds := &appsv1.DaemonSet{ + ObjectMeta: metav1.ObjectMeta{ + Name: "node-configurator", + Namespace: "default", + }, + Spec: appsv1.DaemonSetSpec{ + Selector: &metav1.LabelSelector{MatchLabels: map[string]string{"app": "nc"}}, + UpdateStrategy: appsv1.DaemonSetUpdateStrategy{ + Type: appsv1.RollingUpdateDaemonSetStrategyType, + RollingUpdate: &appsv1.RollingUpdateDaemonSet{ + MaxUnavailable: ptr.To(intstr.FromInt32(1)), + }, + }, + Template: corev1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{"app": "nc"}}, + Spec: corev1.PodSpec{ + Containers: []corev1.Container{{Name: "configurator", Image: "nc:latest"}}, + }, + }, + }, + } + + p := slurmv1alpha1.ResourcePatchPolicy{ + ObjectMeta: metav1.ObjectMeta{Name: "nc", Namespace: "default"}, + Spec: slurmv1alpha1.ResourcePatchPolicySpec{ + TargetRef: slurmv1alpha1.PolicyTargetReference{Group: "slurm.nebius.ai", Kind: "NodeConfigurator", Name: "node-configurator"}, + Type: slurmv1alpha1.JSONPatchType, + Patches: []slurmv1alpha1.ResourcePatch{{ + ResourceRef: nameSel("DaemonSet", "node-configurator"), + JSONPatch: []slurmv1alpha1.JSONPatchOperation{ + {Op: "add", Path: "/spec/template/spec/containers/0/resources", Value: jsonValue(t, corev1.ResourceRequirements{ + Limits: corev1.ResourceList{corev1.ResourceMemory: mustQuantity("512Mi")}, + })}, + }, + }}, + }, + } + + results, err := resourcepatch.Apply(scheme, ds, []slurmv1alpha1.ResourcePatchPolicy{p}) + require.NoError(t, err) + require.Len(t, results, 1) + assert.True(t, results[0].Applied, "message: %s", results[0].Message) + mem := ds.Spec.Template.Spec.Containers[0].Resources.Limits[corev1.ResourceMemory] + assert.Equal(t, "512Mi", mem.String()) +} + +// ensure client.Object is satisfied by the test objects (compile-time guard). +var _ client.Object = (*appsv1.StatefulSet)(nil) +var _ client.Object = (*appsv1.DaemonSet)(nil) diff --git a/internal/resourcepatch/protected.go b/internal/resourcepatch/protected.go new file mode 100644 index 000000000..8eeac1fcf --- /dev/null +++ b/internal/resourcepatch/protected.go @@ -0,0 +1,75 @@ +/* +Copyright 2024 Nebius B.V. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package resourcepatch + +import ( + "encoding/json" + "fmt" + "reflect" +) + +// protectedFieldViolation compares the pre-patch and post-patch JSON documents +// and returns a non-empty message if the patch mutated a field the operator +// relies on for correct reconciliation. An empty string means the patch is +// safe to apply. +// +// Rejected (these break the operator's contract with the API server): +// - metadata.name / metadata.namespace — identity cannot change +// - metadata.ownerReferences — breaks garbage collection +// - spec.selector — immutable after creation on workload resources +func protectedFieldViolation(original, patched []byte) string { + var before, after genericObject + if err := json.Unmarshal(original, &before); err != nil { + return fmt.Sprintf("decoding original object for protected-field check: %v", err) + } + if err := json.Unmarshal(patched, &after); err != nil { + return fmt.Sprintf("decoding patched object for protected-field check: %v", err) + } + + if before.Metadata.Name != after.Metadata.Name { + return fmt.Sprintf("patch must not change metadata.name (%q -> %q)", + before.Metadata.Name, after.Metadata.Name) + } + if before.Metadata.Namespace != after.Metadata.Namespace { + return fmt.Sprintf("patch must not change metadata.namespace (%q -> %q)", + before.Metadata.Namespace, after.Metadata.Namespace) + } + if !reflect.DeepEqual(before.Metadata.OwnerReferences, after.Metadata.OwnerReferences) { + return "patch must not change metadata.ownerReferences" + } + if !reflect.DeepEqual(before.Spec.Selector, after.Spec.Selector) { + return "patch must not change spec.selector" + } + return "" +} + +// genericObject captures only the fields relevant to protected-field checks, +// independent of the concrete Kubernetes type. Unmodelled fields are ignored. +// The comparable fields are decoded into interface{} so that comparison is +// insensitive to JSON key ordering produced by the patch library. +type genericObject struct { + Metadata struct { + Name string `json:"name"` + Namespace string `json:"namespace"` + OwnerReferences any `json:"ownerReferences"` + } `json:"metadata"` + Spec struct { + // Selector is present on StatefulSet, Deployment, DaemonSet, etc. and is + // immutable after creation. nil for resources without a selector. + Selector any `json:"selector"` + } `json:"spec"` +} diff --git a/internal/resourcepatch/validate.go b/internal/resourcepatch/validate.go new file mode 100644 index 000000000..714aa6a05 --- /dev/null +++ b/internal/resourcepatch/validate.go @@ -0,0 +1,98 @@ +/* +Copyright 2024 Nebius B.V. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package resourcepatch + +import ( + "encoding/json" + "fmt" + + jsonpatch "github.com/evanphx/json-patch/v5" + + slurmv1alpha1 "nebius.ai/slurm-operator/api/v1alpha1" +) + +// ValidatePolicy performs static validation of a ResourcePatchPolicy that does +// not depend on the resources being targeted: it checks that every patch entry +// carries a payload consistent with the policy type and that the payload is +// syntactically valid. It returns nil when the policy is acceptable. +func ValidatePolicy(policy *slurmv1alpha1.ResourcePatchPolicy) error { + if len(policy.Spec.Patches) == 0 { + return fmt.Errorf("policy must define at least one patch") + } + + for i := range policy.Spec.Patches { + patch := &policy.Spec.Patches[i] + if patch.ResourceRef.Kind == "" { + return fmt.Errorf("patches[%d]: resourceRef.kind is required", i) + } + + switch policy.Spec.Type { + case slurmv1alpha1.JSONPatchType: + if len(patch.JSONPatch) == 0 { + return fmt.Errorf("patches[%d]: type is JSONPatch but jsonPatch is empty", i) + } + if patch.JSONMergePatch != nil { + return fmt.Errorf("patches[%d]: jsonMergePatch must not be set when type is JSONPatch", i) + } + raw, err := json.Marshal(patch.JSONPatch) + if err != nil { + return fmt.Errorf("patches[%d]: marshalling jsonPatch: %w", i, err) + } + if _, err := jsonpatch.DecodePatch(raw); err != nil { + return fmt.Errorf("patches[%d]: invalid jsonPatch: %w", i, err) + } + for j := range patch.JSONPatch { + if err := validateOperation(&patch.JSONPatch[j]); err != nil { + return fmt.Errorf("patches[%d].jsonPatch[%d]: %w", i, j, err) + } + } + + case slurmv1alpha1.JSONMergePatchType: + if patch.JSONMergePatch == nil || len(patch.JSONMergePatch.Raw) == 0 { + return fmt.Errorf("patches[%d]: type is JSONMergePatch but jsonMergePatch is empty", i) + } + if len(patch.JSONPatch) > 0 { + return fmt.Errorf("patches[%d]: jsonPatch must not be set when type is JSONMergePatch", i) + } + if !json.Valid(patch.JSONMergePatch.Raw) { + return fmt.Errorf("patches[%d]: jsonMergePatch is not valid JSON", i) + } + + default: + return fmt.Errorf("unsupported patch type %q", policy.Spec.Type) + } + } + return nil +} + +func validateOperation(op *slurmv1alpha1.JSONPatchOperation) error { + switch op.Op { + case "add", "replace", "test": + if op.Value == nil { + return fmt.Errorf("op %q requires a value", op.Op) + } + case "remove": + // no value or from required + case "move", "copy": + if op.From == nil || *op.From == "" { + return fmt.Errorf("op %q requires a from path", op.Op) + } + default: + return fmt.Errorf("unknown op %q", op.Op) + } + return nil +} diff --git a/internal/resourcepatch/validate_test.go b/internal/resourcepatch/validate_test.go new file mode 100644 index 000000000..89e4e2a6b --- /dev/null +++ b/internal/resourcepatch/validate_test.go @@ -0,0 +1,146 @@ +/* +Copyright 2024 Nebius B.V. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package resourcepatch_test + +import ( + "testing" + + "github.com/stretchr/testify/assert" + apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/utils/ptr" + + slurmv1alpha1 "nebius.ai/slurm-operator/api/v1alpha1" + "nebius.ai/slurm-operator/internal/resourcepatch" +) + +func mkPolicy(patchType slurmv1alpha1.PatchType, patches ...slurmv1alpha1.ResourcePatch) *slurmv1alpha1.ResourcePatchPolicy { + return &slurmv1alpha1.ResourcePatchPolicy{ + ObjectMeta: metav1.ObjectMeta{Name: "p", Namespace: "default"}, + Spec: slurmv1alpha1.ResourcePatchPolicySpec{ + TargetRef: slurmv1alpha1.PolicyTargetReference{Group: "slurm.nebius.ai", Kind: "SlurmCluster", Name: "c"}, + Type: patchType, + Patches: patches, + }, + } +} + +func TestValidatePolicy(t *testing.T) { + val := func(v any) *apiextensionsv1.JSON { + return jsonValue(t, v) + } + + tests := []struct { + name string + policy *slurmv1alpha1.ResourcePatchPolicy + wantErr bool + }{ + { + name: "valid json patch", + policy: mkPolicy(slurmv1alpha1.JSONPatchType, slurmv1alpha1.ResourcePatch{ + ResourceRef: nameSel("StatefulSet", "x"), + JSONPatch: []slurmv1alpha1.JSONPatchOperation{ + {Op: "add", Path: "/metadata/annotations/a", Value: val("b")}, + }, + }), + }, + { + name: "valid merge patch", + policy: mkPolicy(slurmv1alpha1.JSONMergePatchType, slurmv1alpha1.ResourcePatch{ + ResourceRef: nameSel("ConfigMap", "x"), + JSONMergePatch: &apiextensionsv1.JSON{Raw: []byte(`{"data":{"k":"v"}}`)}, + }), + }, + { + name: "no patches", + policy: mkPolicy(slurmv1alpha1.JSONPatchType), + wantErr: true, + }, + { + name: "json patch empty", + policy: mkPolicy(slurmv1alpha1.JSONPatchType, slurmv1alpha1.ResourcePatch{ + ResourceRef: nameSel("StatefulSet", "x"), + }), + wantErr: true, + }, + { + name: "json patch with merge payload", + policy: mkPolicy(slurmv1alpha1.JSONPatchType, slurmv1alpha1.ResourcePatch{ + ResourceRef: nameSel("StatefulSet", "x"), + JSONPatch: []slurmv1alpha1.JSONPatchOperation{{Op: "add", Path: "/a", Value: val("b")}}, + JSONMergePatch: &apiextensionsv1.JSON{Raw: []byte(`{}`)}, + }), + wantErr: true, + }, + { + name: "add without value", + policy: mkPolicy(slurmv1alpha1.JSONPatchType, slurmv1alpha1.ResourcePatch{ + ResourceRef: nameSel("StatefulSet", "x"), + JSONPatch: []slurmv1alpha1.JSONPatchOperation{{Op: "add", Path: "/a"}}, + }), + wantErr: true, + }, + { + name: "move without from", + policy: mkPolicy(slurmv1alpha1.JSONPatchType, slurmv1alpha1.ResourcePatch{ + ResourceRef: nameSel("StatefulSet", "x"), + JSONPatch: []slurmv1alpha1.JSONPatchOperation{{Op: "move", Path: "/a"}}, + }), + wantErr: true, + }, + { + name: "move with from", + policy: mkPolicy(slurmv1alpha1.JSONPatchType, slurmv1alpha1.ResourcePatch{ + ResourceRef: nameSel("StatefulSet", "x"), + JSONPatch: []slurmv1alpha1.JSONPatchOperation{{Op: "move", Path: "/a", From: ptr.To("/b")}}, + }), + }, + { + name: "merge patch empty", + policy: mkPolicy(slurmv1alpha1.JSONMergePatchType, slurmv1alpha1.ResourcePatch{ + ResourceRef: nameSel("ConfigMap", "x"), + }), + wantErr: true, + }, + { + name: "merge patch invalid json", + policy: mkPolicy(slurmv1alpha1.JSONMergePatchType, slurmv1alpha1.ResourcePatch{ + ResourceRef: nameSel("ConfigMap", "x"), + JSONMergePatch: &apiextensionsv1.JSON{Raw: []byte(`{not json`)}, + }), + wantErr: true, + }, + { + name: "missing resourceRef kind", + policy: mkPolicy(slurmv1alpha1.JSONPatchType, slurmv1alpha1.ResourcePatch{ + JSONPatch: []slurmv1alpha1.JSONPatchOperation{{Op: "add", Path: "/a", Value: val("b")}}, + }), + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := resourcepatch.ValidatePolicy(tt.policy) + if tt.wantErr { + assert.Error(t, err) + } else { + assert.NoError(t, err) + } + }) + } +} diff --git a/internal/resourcepatch/watch.go b/internal/resourcepatch/watch.go new file mode 100644 index 000000000..876bbadc4 --- /dev/null +++ b/internal/resourcepatch/watch.go @@ -0,0 +1,59 @@ +/* +Copyright 2024 Nebius B.V. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package resourcepatch + +import ( + "context" + + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/handler" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + + slurmv1alpha1 "nebius.ai/slurm-operator/api/v1alpha1" +) + +// MapPolicyToTarget returns a handler.MapFunc that, given a ResourcePatchPolicy +// targeting a resource of kind targetKind, enqueues a reconcile request for the +// target resource. It is used to re-reconcile the parent (SlurmCluster, +// NodeSet, NodeConfigurator) whenever a policy that affects it changes. +func MapPolicyToTarget(targetKind string) handler.MapFunc { + return func(_ context.Context, obj client.Object) []reconcile.Request { + policy, ok := obj.(*slurmv1alpha1.ResourcePatchPolicy) + if !ok { + return nil + } + ref := policy.Spec.TargetRef + if ref.Kind != targetKind { + return nil + } + + namespace := policy.Namespace + if ref.Namespace != nil && *ref.Namespace != "" { + namespace = *ref.Namespace + } + + return []reconcile.Request{ + { + NamespacedName: types.NamespacedName{ + Namespace: namespace, + Name: ref.Name, + }, + }, + } + } +}