-
Notifications
You must be signed in to change notification settings - Fork 674
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Implement webhook validations for the PyTorchJob #2035
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -19,6 +19,7 @@ package main | |
import ( | ||
"errors" | ||
"flag" | ||
"net/http" | ||
"os" | ||
"strings" | ||
|
||
|
@@ -40,9 +41,11 @@ import ( | |
volcanoclient "volcano.sh/apis/pkg/client/clientset/versioned" | ||
|
||
kubeflowv1 "github.com/kubeflow/training-operator/pkg/apis/kubeflow.org/v1" | ||
"github.com/kubeflow/training-operator/pkg/cert" | ||
"github.com/kubeflow/training-operator/pkg/config" | ||
controllerv1 "github.com/kubeflow/training-operator/pkg/controller.v1" | ||
"github.com/kubeflow/training-operator/pkg/controller.v1/common" | ||
"github.com/kubeflow/training-operator/pkg/webhooks" | ||
//+kubebuilder:scaffold:imports | ||
) | ||
|
||
|
@@ -72,8 +75,11 @@ func main() { | |
var enabledSchemes controllerv1.EnabledSchemes | ||
var gangSchedulerName string | ||
var namespace string | ||
var webhookServerPort int | ||
var controllerThreads int | ||
var webhookServerPort int | ||
var webhookServiceName string | ||
var webhookSecretName string | ||
|
||
flag.StringVar(&metricsAddr, "metrics-bind-address", ":8080", "The address the metric endpoint binds to.") | ||
flag.StringVar(&probeAddr, "health-probe-bind-address", ":8081", "The address the probe endpoint binds to.") | ||
flag.BoolVar(&enableLeaderElection, "leader-elect", false, | ||
|
@@ -86,7 +92,6 @@ func main() { | |
" Note: If you set another scheduler name, the training-operator assumes it's the scheduler-plugins.") | ||
flag.StringVar(&namespace, "namespace", os.Getenv(EnvKubeflowNamespace), "The namespace to monitor kubeflow jobs. If unset, it monitors all namespaces cluster-wide."+ | ||
"If set, it only monitors kubeflow jobs in the given namespace.") | ||
flag.IntVar(&webhookServerPort, "webhook-server-port", 9443, "Endpoint port for the webhook server.") | ||
flag.IntVar(&controllerThreads, "controller-threads", 1, "Number of worker threads used by the controller.") | ||
|
||
// PyTorch related flags | ||
|
@@ -101,6 +106,11 @@ func main() { | |
flag.StringVar(&config.Config.MPIKubectlDeliveryImage, "mpi-kubectl-delivery-image", | ||
config.MPIKubectlDeliveryImageDefault, "The image for mpi launcher init container") | ||
|
||
// Cert generation flags | ||
flag.IntVar(&webhookServerPort, "webhook-server-port", 9443, "Endpoint port for the webhook server.") | ||
flag.StringVar(&webhookServiceName, "webhook-service-name", "training-operator", "Name of the Service used as part of the DNSName") | ||
flag.StringVar(&webhookSecretName, "webhook-secret-name", "training-operator-webhook-cert", "Name of the Secret to store CA and server certs") | ||
|
||
opts := zap.Options{ | ||
Development: true, | ||
StacktraceLevel: zapcore.DPanicLevel, | ||
|
@@ -124,9 +134,9 @@ func main() { | |
Metrics: metricsserver.Options{ | ||
BindAddress: metricsAddr, | ||
}, | ||
WebhookServer: &webhook.DefaultServer{Options: webhook.Options{ | ||
WebhookServer: webhook.NewServer(webhook.Options{ | ||
Port: webhookServerPort, | ||
}}, | ||
}), | ||
HealthProbeBindAddress: probeAddr, | ||
LeaderElection: enableLeaderElection, | ||
LeaderElectionID: leaderElectionID, | ||
|
@@ -137,30 +147,36 @@ func main() { | |
os.Exit(1) | ||
} | ||
|
||
// Set up controllers using goroutines to start the manager quickly. | ||
go setupControllers(mgr, enabledSchemes, gangSchedulerName, controllerThreads) | ||
|
||
//+kubebuilder:scaffold:builder | ||
|
||
if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil { | ||
setupLog.Error(err, "unable to set up health check") | ||
os.Exit(1) | ||
certsReady := make(chan struct{}) | ||
defer close(certsReady) | ||
certGenerationConfig := cert.Config{ | ||
WebhookSecretName: webhookSecretName, | ||
WebhookServiceName: webhookServiceName, | ||
} | ||
if err := mgr.AddReadyzCheck("readyz", healthz.Ping); err != nil { | ||
setupLog.Error(err, "unable to set up ready check") | ||
if err = cert.ManageCerts(mgr, certGenerationConfig, certsReady); err != nil { | ||
setupLog.Error(err, "Unable to set up cert rotation") | ||
os.Exit(1) | ||
} | ||
|
||
setupProbeEndpoints(mgr, certsReady) | ||
// Set up controllers using goroutines to start the manager quickly. | ||
go setupControllers(mgr, enabledSchemes, gangSchedulerName, controllerThreads, certsReady) | ||
|
||
//+kubebuilder:scaffold:builder | ||
|
||
setupLog.Info("starting manager") | ||
if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil { | ||
setupLog.Error(err, "problem running manager") | ||
os.Exit(1) | ||
} | ||
} | ||
|
||
func setupControllers(mgr ctrl.Manager, enabledSchemes controllerv1.EnabledSchemes, gangSchedulerName string, controllerThreads int) { | ||
setupLog.Info("registering controllers...") | ||
func setupControllers(mgr ctrl.Manager, enabledSchemes controllerv1.EnabledSchemes, gangSchedulerName string, controllerThreads int, certsReady <-chan struct{}) { | ||
setupLog.Info("Waiting for certificate generation to complete") | ||
<-certsReady | ||
setupLog.Info("Certs ready") | ||
|
||
setupLog.Info("registering controllers...") | ||
// Prepare GangSchedulingSetupFunc | ||
gangSchedulingSetupFunc := common.GenNonGangSchedulerSetupFunc() | ||
if strings.EqualFold(gangSchedulerName, string(common.GangSchedulerVolcano)) { | ||
|
@@ -182,15 +198,52 @@ func setupControllers(mgr ctrl.Manager, enabledSchemes controllerv1.EnabledSchem | |
} | ||
errMsg := "failed to set up controllers" | ||
for _, s := range enabledSchemes { | ||
setupFunc, supported := controllerv1.SupportedSchemeReconciler[s] | ||
if !supported { | ||
setupReconcilerFunc, supportedReconciler := controllerv1.SupportedSchemeReconciler[s] | ||
if !supportedReconciler { | ||
setupLog.Error(errors.New(errMsg), "scheme is not supported", "scheme", s) | ||
os.Exit(1) | ||
} | ||
if err := setupFunc(mgr, gangSchedulingSetupFunc, controllerThreads); err != nil { | ||
if err := setupReconcilerFunc(mgr, gangSchedulingSetupFunc, controllerThreads); err != nil { | ||
setupLog.Error(errors.New(errMsg), "unable to create controller", "scheme", s) | ||
os.Exit(1) | ||
} | ||
setupWebhookFunc, supportedWebhook := webhooks.SupportedSchemeWebhook[s] | ||
if !supportedWebhook { | ||
setupLog.Error(errors.New(errMsg), "scheme is not supported", "scheme", s) | ||
os.Exit(1) | ||
} | ||
if err := setupWebhookFunc(mgr); err != nil { | ||
setupLog.Error(errors.New(errMsg), "unable to start webhook server", "scheme", s) | ||
os.Exit(1) | ||
} | ||
} | ||
} | ||
|
||
func setupProbeEndpoints(mgr ctrl.Manager, certsReady <-chan struct{}) { | ||
defer setupLog.Info("Probe endpoints are configured on healthz and readyz") | ||
|
||
if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil { | ||
setupLog.Error(err, "unable to set up health check") | ||
os.Exit(1) | ||
} | ||
|
||
// Wait for the webhook server to be listening before advertising the | ||
// training-operator replica as ready. This allows users to wait with sending the first | ||
// requests, requiring webhooks, until the training-operator deployment is available, so | ||
// that the early requests are not rejected during the traininig-operator's startup. | ||
// We wrap the call to GetWebhookServer in a closure to delay calling | ||
// the function, otherwise a not fully-initialized webhook server (without | ||
// ready certs) fails the start of the manager. | ||
if err := mgr.AddReadyzCheck("readyz", func(req *http.Request) error { | ||
select { | ||
case <-certsReady: | ||
return mgr.GetWebhookServer().StartedChecker()(req) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Can you explain about this closure? how is There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. @johnugeorge If we don't have this extended readiness probe, the controller-runtime readiness server responds "Ready" even if the training operator is still preparing due to certificate generation. Also, you can find the details here: https://github.com/kubernetes-sigs/kueue/pull/1676/files#r1472752662 |
||
default: | ||
return errors.New("certificates are not ready") | ||
} | ||
}); err != nil { | ||
setupLog.Error(err, "unable to set up ready check") | ||
os.Exit(1) | ||
} | ||
} | ||
|
||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,15 @@ | ||
apiVersion: kustomize.config.k8s.io/v1beta1 | ||
kind: Kustomization | ||
resources: | ||
- manifests.yaml | ||
commonLabels: | ||
control-plane: kubeflow-training-operator | ||
andreyvelich marked this conversation as resolved.
Show resolved
Hide resolved
|
||
patches: | ||
- path: patch.yaml | ||
target: | ||
group: admissionregistration.k8s.io | ||
version: v1 | ||
kind: ValidatingWebhookConfiguration | ||
|
||
configurations: | ||
- kustomizeconfig.yaml |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,10 @@ | ||
# the following config is for teaching kustomize where to look at when substituting vars. | ||
# It requires kustomize v2.1.0 or newer to work properly. | ||
namespace: | ||
- kind: ValidatingWebhookConfiguration | ||
group: admissionregistration.k8s.io | ||
path: webhooks/clientConfig/service/namespace | ||
create: true | ||
|
||
varReference: | ||
- path: metadata/annotations |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,26 @@ | ||
--- | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Can we name this file as Or kubebuilder sets this name by default ? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
Yea, this name is default value 😓 |
||
apiVersion: admissionregistration.k8s.io/v1 | ||
kind: ValidatingWebhookConfiguration | ||
metadata: | ||
name: validating-webhook-configuration | ||
andreyvelich marked this conversation as resolved.
Show resolved
Hide resolved
|
||
webhooks: | ||
- admissionReviewVersions: | ||
- v1 | ||
clientConfig: | ||
service: | ||
name: webhook-service | ||
namespace: system | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Make it There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. We have no way to modify this namespace since this manifest is automatically generated by the controller-gen. But, this namespace is replaced by https://github.com/kubeflow/training-operator/blob/a618d1c3f77f769da7d4f535e25db88d64aadb88/manifests/base/webhook/kustomizeconfig.yaml. |
||
path: /validate-kubeflow-org-v1-pytorchjob | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Do we need the full path? Maybe just: There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. We need to follow this form based on the https://book.kubebuilder.io/reference/markers/webhook. |
||
failurePolicy: Fail | ||
andreyvelich marked this conversation as resolved.
Show resolved
Hide resolved
|
||
name: validator.pytorchjob.training-operator.kubeflow.org | ||
rules: | ||
- apiGroups: | ||
- kubeflow.org | ||
apiVersions: | ||
- v1 | ||
operations: | ||
- CREATE | ||
- UPDATE | ||
resources: | ||
- pytorchjobs | ||
sideEffects: None |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,6 @@ | ||
- op: replace | ||
path: /webhooks/0/clientConfig/service/name | ||
value: training-operator | ||
- op: replace | ||
path: /metadata/name | ||
value: validator.training-operator.kubeflow.org |
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.
Do you need to check if certs are ready in
setupControllers
func ?Maybe we can check it before we run the func and not pass
certsReady
to thesetupControllers
function:training-operator/cmd/training-operator.v1/main.go
Line 162 in a618d1c
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.
We need this checking since the
setupController
is started with the go routine.By doing this, we can improve the start-up time and avoid readiness probe errors (#1841) since the manager can start unless not waiting for the completion of setting up controllers.