add pod webhook and pod reconciler for new exclusive placement strategy

Dockerfile

@@ -16,6 +16,7 @@ COPY main.go main.go
 COPY api/ api/
 COPY pkg/controllers/ pkg/controllers/
 COPY pkg/util/ pkg/util/
+COPY pkg/webhooks pkg/webhooks
 
 # Build
 # the GOARCH has not a default value to allow the binary be built according to the host where the command

api/jobset/v1alpha2/jobset_webhook.go

@@ -29,7 +29,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/webhook/admission"
 
 	"sigs.k8s.io/jobset/pkg/util/collections"
-	"sigs.k8s.io/jobset/pkg/util/names"
+	shared "sigs.k8s.io/jobset/pkg/util/shared"
 
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -109,7 +109,7 @@ func (js *JobSet) ValidateCreate() (admission.Warnings, error) {
 		}
 		// Check that the generated job names for this replicated job will be DNS 1035 compliant.
 		// Use the largest job index as it will have the longest name.
-		testJobName := names.GenJobName(js.Name, rjob.Name, int(rjob.Replicas-1))
+		testJobName := shared.GenJobName(js.Name, rjob.Name, int(rjob.Replicas-1))
 		for _, errMessage := range validation.IsDNS1035Label(testJobName) {
 			allErrs = append(allErrs, fmt.Errorf(errMessage))
 		}

config/components/manager/manager.yaml

@@ -92,11 +92,8 @@ spec:
         # TODO(user): Configure the resources accordingly based on the project requirements.
         # More info: https://kubernetes.io/docs/concepts/configuration/manage-resources-containers/
         resources:
-          limits:
-            cpu: 500m
-            memory: 128Mi
           requests:
-            cpu: 10m
-            memory: 64Mi
+            cpu: 2
+            memory: 512Mi
       serviceAccountName: controller-manager
       terminationGracePeriodSeconds: 10

config/components/rbac/role.yaml

@@ -60,6 +60,26 @@ rules:
   - get
   - patch
   - update
+- apiGroups:
+  - ""
+  resources:
+  - nodes
+  verbs:
+  - get
+  - list
+  - watch
+- apiGroups:
+  - ""
+  resources:
+  - pods
+  verbs:
+  - create
+  - delete
+  - get
+  - list
+  - patch
+  - update
+  - watch
 - apiGroups:
   - ""
   resources:

config/components/webhook/manifests.yaml

@@ -24,6 +24,25 @@ webhooks:
     resources:
     - jobsets
   sideEffects: None
+- admissionReviewVersions:
+  - v1
+  clientConfig:
+    service:
+      name: webhook-service
+      namespace: system
+      path: /mutate--v1-pod
+  failurePolicy: Fail
+  name: mpod.kb.io
+  rules:
+  - apiGroups:
+    - ""
+    apiVersions:
+    - v1
+    operations:
+    - CREATE
+    resources:
+    - pods
+  sideEffects: None
 ---
 apiVersion: admissionregistration.k8s.io/v1
 kind: ValidatingWebhookConfiguration
@@ -50,3 +69,22 @@ webhooks:
     resources:
     - jobsets
   sideEffects: None
+- admissionReviewVersions:
+  - v1
+  clientConfig:
+    service:
+      name: webhook-service
+      namespace: system
+      path: /validate--v1-pod
+  failurePolicy: Fail
+  name: vpod.kb.io
+  rules:
+  - apiGroups:
+    - ""
+    apiVersions:
+    - v1
+    operations:
+    - CREATE
+    resources:
+    - pods
+  sideEffects: None

examples/simple/exclusive-placement.yaml

@@ -0,0 +1,26 @@
+apiVersion: jobset.x-k8s.io/v1alpha2
+kind: JobSet
+metadata:
+  name: exclusive-placement
+  annotations:
+    alpha.jobset.sigs.k8s.io/exclusive-topology: cloud.google.com/gke-nodepool # 1:1 job replica to node pool assignment
+spec:
+  failurePolicy:
+    maxRestarts: 3
+  replicatedJobs:
+  - name: workers
+    replicas: 3
+    template:
+      spec:
+        parallelism: 3
+        completions: 3
+        backoffLimit: 10
+        template:
+          spec:
+            containers:
+            - name: sleep
+              image: busybox
+              command: 
+                - sleep
+              args:
+                - 1000s

main.go

@@ -36,6 +36,7 @@ import (
 	jobset "sigs.k8s.io/jobset/api/jobset/v1alpha2"
 	"sigs.k8s.io/jobset/pkg/controllers"
 	"sigs.k8s.io/jobset/pkg/util/cert"
+	"sigs.k8s.io/jobset/pkg/webhooks"
 	//+kubebuilder:scaffold:imports
 )
 
@@ -68,7 +69,11 @@ func main() {
 
 	ctrl.SetLogger(zap.New(zap.UseFlagOptions(&opts)))
 
-	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
+	kubeConfig := ctrl.GetConfigOrDie()
+	kubeConfig.QPS = 500
+	kubeConfig.Burst = 500
+
+	mgr, err := ctrl.NewManager(kubeConfig, ctrl.Options{
 		Scheme: scheme,
 		Metrics: server.Options{
 			BindAddress: metricsAddr,
@@ -104,8 +109,12 @@ func main() {
 	}
 
 	ctx := ctrl.SetupSignalHandler()
-	if err := controllers.SetupIndexes(ctx, mgr.GetFieldIndexer()); err != nil {
-		setupLog.Error(err, "unable to setup indexes")
+	if err := controllers.SetupJobSetIndexes(ctx, mgr.GetFieldIndexer()); err != nil {
+		setupLog.Error(err, "unable to setup jobset reconciler indexes")
+		os.Exit(1)
+	}
+	if err := controllers.SetupPodReconcilerIndexes(ctx, mgr.GetFieldIndexer()); err != nil {
+		setupLog.Error(err, "unable to setup pod reconciler indexes")
 		os.Exit(1)
 	}
 
@@ -130,13 +139,30 @@ func setupControllers(mgr ctrl.Manager, certsReady chan struct{}) {
 	<-certsReady
 	setupLog.Info("certs ready")
 
+	// Set up JobSet controller.
 	jobSetController := controllers.NewJobSetReconciler(mgr.GetClient(), mgr.GetScheme(), mgr.GetEventRecorderFor("jobset"))
 	if err := jobSetController.SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "JobSet")
 		os.Exit(1)
 	}
+
+	// Set up pod reconciler.
+	podController := controllers.NewPodReconciler(mgr.GetClient(), mgr.GetScheme(), mgr.GetEventRecorderFor("pod"))
+	if err := podController.SetupWithManager(mgr); err != nil {
+		setupLog.Error(err, "unable to create controller", "controller", "Pod")
+		os.Exit(1)
+	}
+
+	// Set up JobSet validating/defaulting webhook.
 	if err := (&jobset.JobSet{}).SetupWebhookWithManager(mgr); err != nil {
-		setupLog.Error(err, "unable to create webhook", "webhook", "JobSet")
+		setupLog.Error(err, "unable to create validation/defaulting webhook", "webhook", "JobSet")
+		os.Exit(1)
+	}
+
+	// Set up pod mutating and admission webhook.
+	podWebhook := webhooks.NewPodWebhook(mgr)
+	if err := podWebhook.SetupWebhookWithManager(mgr); err != nil {
+		setupLog.Error(err, "unable to create mutating webhook", "webhook", "Pod")
 		os.Exit(1)
 	}
 	//+kubebuilder:scaffold:builder