KEP-2170: Adding validation webhook for v2 trainjob #2307
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change | ||||
|---|---|---|---|---|---|---|
|
|
@@ -28,17 +28,20 @@ import ( | |||||
| metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" | ||||||
| apiruntime "k8s.io/apimachinery/pkg/runtime" | ||||||
| "k8s.io/apimachinery/pkg/runtime/schema" | ||||||
| "k8s.io/apimachinery/pkg/util/validation/field" | ||||||
| "k8s.io/utils/ptr" | ||||||
| ctrl "sigs.k8s.io/controller-runtime" | ||||||
| "sigs.k8s.io/controller-runtime/pkg/builder" | ||||||
| "sigs.k8s.io/controller-runtime/pkg/client" | ||||||
| ctrlutil "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" | ||||||
| "sigs.k8s.io/controller-runtime/pkg/webhook/admission" | ||||||
| jobsetv1alpha2 "sigs.k8s.io/jobset/api/jobset/v1alpha2" | ||||||
|
|
||||||
| kubeflowv2 "github.com/kubeflow/training-operator/pkg/apis/kubeflow.org/v2alpha1" | ||||||
| "github.com/kubeflow/training-operator/pkg/constants" | ||||||
| runtime "github.com/kubeflow/training-operator/pkg/runtime.v2" | ||||||
| "github.com/kubeflow/training-operator/pkg/runtime.v2/framework" | ||||||
| util_v2 "github.com/kubeflow/training-operator/pkg/util.v2" | ||||||
| ) | ||||||
|
|
||||||
| type JobSet struct { | ||||||
|
|
@@ -50,6 +53,7 @@ type JobSet struct { | |||||
|
|
||||||
| var _ framework.WatchExtensionPlugin = (*JobSet)(nil) | ||||||
| var _ framework.ComponentBuilderPlugin = (*JobSet)(nil) | ||||||
| var _ framework.CustomValidationPlugin = (*JobSet)(nil) | ||||||
|
|
||||||
| const Name = constants.JobSetKind | ||||||
|
|
||||||
|
|
@@ -140,3 +144,104 @@ func (j *JobSet) ReconcilerBuilders() []runtime.ReconcilerBuilder { | |||||
| }, | ||||||
| } | ||||||
| } | ||||||
|
|
||||||
| func (j *JobSet) Validate(runtimeJobTemplate client.Object, runtimeInfo *runtime.Info, oldObj, newObj *kubeflowv2.TrainJob) (admission.Warnings, field.ErrorList) { | ||||||
|
|
||||||
| var allErrs field.ErrorList | ||||||
| specPath := field.NewPath("spec") | ||||||
| runtimeRefPath := specPath.Child("runtimeRef") | ||||||
|
|
||||||
| jobSet, ok := runtimeJobTemplate.(*jobsetv1alpha2.JobSet) | ||||||
| if !ok { | ||||||
| return nil, nil | ||||||
| } | ||||||
|
|
||||||
| if newObj.Spec.ModelConfig != nil && newObj.Spec.ModelConfig.Input != nil { | ||||||
|
There was a problem hiding this comment. I think, for now we should check the initContainers in JobSet, as I mentioned here: https://github.com/kubeflow/training-operator/blob/master/pkg/runtime.v2/framework/plugins/jobset/builder.go#L87-L89 |
||||||
| if !util_v2.JobExists(jobSet.Spec.ReplicatedJobs, constants.JobInitializer) { | ||||||
|
There was a problem hiding this comment. @tenzen-y Can we use Go-native libs to verify that Name exists in the slice ? |
||||||
| allErrs = append(allErrs, field.Invalid(runtimeRefPath, newObj.Spec.RuntimeRef, "trainingRuntime should have initializer job when trainJob is configured with input modelConfig")) | ||||||
|
There was a problem hiding this comment. Let's use name from var here.
Suggested change
|
||||||
| } else { | ||||||
| for _, job := range jobSet.Spec.ReplicatedJobs { | ||||||
| if job.Name == constants.JobInitializer { | ||||||
| if !util_v2.ContainerExists(job.Template.Spec.Template.Spec.InitContainers, constants.ContainerModelInitializer) { | ||||||
|
|
||||||
| allErrs = append(allErrs, field.Invalid(runtimeRefPath, newObj.Spec.RuntimeRef, "trainingRuntime should have container with name - model-initializer in the initializer job")) | ||||||
|
There was a problem hiding this comment. Same suggestion for the Container and Job name. |
||||||
| } | ||||||
| } | ||||||
| } | ||||||
| } | ||||||
| } | ||||||
|
|
||||||
| if newObj.Spec.DatasetConfig != nil { | ||||||
| if !util_v2.JobExists(jobSet.Spec.ReplicatedJobs, constants.JobInitializer) { | ||||||
| allErrs = append(allErrs, field.Invalid(runtimeRefPath, newObj.Spec.RuntimeRef, "trainingRuntime should have initializer job when trainJob is configured with datasetConfig")) | ||||||
|
|
||||||
| } else { | ||||||
| for _, job := range jobSet.Spec.ReplicatedJobs { | ||||||
| if job.Name == constants.JobInitializer { | ||||||
| if !util_v2.ContainerExists(job.Template.Spec.Template.Spec.InitContainers, constants.ContainerDatasetInitializer) { | ||||||
| allErrs = append(allErrs, field.Invalid(runtimeRefPath, newObj.Spec.RuntimeRef, "trainingRuntime should have container with name - dataset-initializer in the initializer job")) | ||||||
|
|
||||||
| } | ||||||
| } | ||||||
| } | ||||||
| } | ||||||
| } | ||||||
|
|
||||||
| if newObj.Spec.ModelConfig != nil && newObj.Spec.ModelConfig.Output != nil { | ||||||
| if !util_v2.JobExists(jobSet.Spec.ReplicatedJobs, constants.JobExporter) { | ||||||
| allErrs = append(allErrs, field.Invalid(runtimeRefPath, newObj.Spec.RuntimeRef, "trainingRuntime should have exporter job when trainJob is configured with output modelConfig")) | ||||||
| } else { | ||||||
| for _, job := range jobSet.Spec.ReplicatedJobs { | ||||||
| if job.Name == constants.JobExporter { | ||||||
| if !util_v2.ContainerExists(job.Template.Spec.Template.Spec.InitContainers, constants.ContainerModelExporter) { | ||||||
| allErrs = append(allErrs, field.Invalid(runtimeRefPath, newObj.Spec.RuntimeRef, "trainingRuntime should have container with name - model-exporter in the exporter job")) | ||||||
| } | ||||||
| } | ||||||
| } | ||||||
| } | ||||||
| } | ||||||
|
Comment on lines
+187
to
+199
There was a problem hiding this comment. Let's implement it once we support Output. |
||||||
|
|
||||||
| if len(newObj.Spec.PodSpecOverrides) != 0 { | ||||||
|
There was a problem hiding this comment. Should we implement this validation when we support PodSpecOverride ? |
||||||
| podSpecOverridesPath := specPath.Child("podSpecOverrides") | ||||||
| jobsMap := map[string]bool{} | ||||||
| for _, job := range jobSet.Spec.ReplicatedJobs { | ||||||
| jobsMap[job.Name] = true | ||||||
| } | ||||||
| // validate if jobOverrides are valid | ||||||
| for idx, override := range newObj.Spec.PodSpecOverrides { | ||||||
| for _, job := range override.TargetJobs { | ||||||
| if _, found := jobsMap[job.Name]; !found { | ||||||
| allErrs = append(allErrs, field.Invalid(podSpecOverridesPath, newObj.Spec.PodSpecOverrides, fmt.Sprintf("job: %s, configured in the podOverride should be present in the referenced training runtime", job))) | ||||||
| } | ||||||
| } | ||||||
| if len(override.Containers) != 0 { | ||||||
| // validate if containerOverrides are valid | ||||||
| containerMap := map[string]bool{} | ||||||
| for _, job := range jobSet.Spec.ReplicatedJobs { | ||||||
| for _, container := range job.Template.Spec.Template.Spec.Containers { | ||||||
| containerMap[container.Name] = true | ||||||
| } | ||||||
| } | ||||||
| containerOverridePath := podSpecOverridesPath.Index(idx) | ||||||
| for _, container := range override.Containers { | ||||||
| if _, found := containerMap[container.Name]; !found { | ||||||
| allErrs = append(allErrs, field.Invalid(containerOverridePath, override.Containers, fmt.Sprintf("container: %s, configured in the containerOverride should be present in the referenced training runtime", container.Name))) | ||||||
| } | ||||||
| } | ||||||
| } | ||||||
| if len(override.InitContainers) != 0 { | ||||||
| // validate if initContainerOverrides are valid | ||||||
| initContainerMap := map[string]bool{} | ||||||
| for _, job := range jobSet.Spec.ReplicatedJobs { | ||||||
| for _, initContainer := range job.Template.Spec.Template.Spec.InitContainers { | ||||||
| initContainerMap[initContainer.Name] = true | ||||||
| } | ||||||
| } | ||||||
| initContainerOverridePath := podSpecOverridesPath.Index(idx) | ||||||
| for _, container := range override.Containers { | ||||||
| if _, found := initContainerMap[container.Name]; !found { | ||||||
| allErrs = append(allErrs, field.Invalid(initContainerOverridePath, override.InitContainers, fmt.Sprintf("initContainer: %s, configured in the initContainerOverride should be present in the referenced training runtime", container.Name))) | ||||||
| } | ||||||
| } | ||||||
| } | ||||||
| } | ||||||
| } | ||||||
| return nil, allErrs | ||||||
| } | ||||||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -18,6 +18,7 @@ package mpi | |
|
|
||
| import ( | ||
| "context" | ||
| "strconv" | ||
|
|
||
| "k8s.io/apimachinery/pkg/util/validation/field" | ||
| "sigs.k8s.io/controller-runtime/pkg/client" | ||
|
|
@@ -55,7 +56,16 @@ func (m *MPI) EnforceMLPolicy(info *runtime.Info, trainJob *kubeflowv2.TrainJob) | |
| return nil | ||
| } | ||
|
|
||
| func (m *MPI) Validate(runtimeJobTemplate client.Object, runtimeInfo *runtime.Info, oldJobObj, newJobObj *kubeflowv2.TrainJob) (admission.Warnings, field.ErrorList) { | ||
| var allErrs field.ErrorList | ||
| specPath := field.NewPath("spec") | ||
| if newJobObj.Spec.Trainer != nil { | ||
| numProcPerNodePath := specPath.Child("trainer").Child("numProcPerNode") | ||
| if runtimeInfo.RuntimePolicy.MLPolicy.MPI != nil { | ||
| if _, err := strconv.Atoi(*newJobObj.Spec.Trainer.NumProcPerNode); err != nil { | ||
| allErrs = append(allErrs, field.Invalid(numProcPerNodePath, newJobObj.Spec.Trainer.NumProcPerNode, "should have an int value")) | ||
|
There was a problem hiding this comment. Uhm, based on considering these string and integer conversion everywhere, we want to define the numProcPerNode as a typed IntOrString. There was a problem hiding this comment. I think so, is this value compatible with the k8s API conventions: https://github.com/kubernetes/community/blob/master/contributors/devel/sig-architecture/api-conventions.md ? |
||
| } | ||
| } | ||
| } | ||
| return nil, allErrs | ||
| } | ||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Please can we implement the validation for exporter in the future once we design it as part of: #2245 ?
We should discuss whether we want to use sidecar container or another ReplicatedJob for model checkpointing.
cc @saileshd1402 @akshaychitneni @tenzen-y