add admitPod

Author: wangyuqing (C)

cmd/admission/app/configure/configure.go cmd/admission/app/options/options.go

@@ -14,11 +14,12 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package configure
+package options
 
 import (
 	"flag"
 	"fmt"
+
 	"github.com/golang/glog"
 
 	"k8s.io/api/admissionregistration/v1beta1"
@@ -29,6 +30,10 @@ import (
 	admissionregistrationv1beta1client "k8s.io/client-go/kubernetes/typed/admissionregistration/v1beta1"
 )
 
+const (
+	defaultSchedulerName = "volcano"
+)
+
 // Config admission-controller server config.
 type Config struct {
 	Master                    string
@@ -44,6 +49,7 @@ type Config struct {
 	PrintVersion              bool
 	AdmissionServiceName      string
 	AdmissionServiceNamespace string
+	SchedulerName             string
 }
 
 // NewConfig create new config
@@ -73,6 +79,7 @@ func (c *Config) AddFlags() {
 	flag.BoolVar(&c.PrintVersion, "version", false, "Show version and quit")
 	flag.StringVar(&c.AdmissionServiceNamespace, "webhook-namespace", "default", "The namespace of this webhook")
 	flag.StringVar(&c.AdmissionServiceName, "webhook-service-name", "admission-service", "The name of this admission service")
+	flag.StringVar(&c.SchedulerName, "scheduler-name", defaultSchedulerName, "Volcano will handle pods whose .spec.SchedulerName is same as scheduler-name")
 }
 
 const (
@@ -84,6 +91,10 @@ const (
 	ValidateHookName = "validatejob.volcano.sh"
 	// MutateHookName Default name for webhooks in MutatingWebhookConfiguration
 	MutateHookName = "mutatejob.volcano.sh"
+	// ValidatePodConfigName ValidatingWebhookPodConfiguration name format
+	ValidatePodConfigName = "%s-validate-pod"
+	// ValidatePodHookName Default name for webhooks in ValidatingWebhookPodConfiguration
+	ValidatePodHookName = "validatepod.volcano.sh"
 )
 
 // CheckPortOrDie check valid port range
@@ -177,6 +188,42 @@ func RegisterWebhooks(c *Config, clienset *kubernetes.Clientset, cabundle []byte
 		return err
 	}
 
+	// Prepare validate pods
+	path = "/pods"
+	PodValidateHooks := v1beta1.ValidatingWebhookConfiguration{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: useGeneratedNameIfRequired("",
+				fmt.Sprintf(ValidatePodConfigName, c.AdmissionServiceName)),
+		},
+		Webhooks: []v1beta1.Webhook{{
+			Name: useGeneratedNameIfRequired("", ValidatePodHookName),
+			Rules: []v1beta1.RuleWithOperations{
+				{
+					Operations: []v1beta1.OperationType{v1beta1.Create},
+					Rule: v1beta1.Rule{
+						APIGroups:   []string{""},
+						APIVersions: []string{"v1"},
+						Resources:   []string{"pods"},
+					},
+				},
+			},
+			ClientConfig: v1beta1.WebhookClientConfig{
+				Service: &v1beta1.ServiceReference{
+					Name:      c.AdmissionServiceName,
+					Namespace: c.AdmissionServiceNamespace,
+					Path:      &path,
+				},
+				CABundle: cabundle,
+			},
+			FailurePolicy: &ignorePolicy,
+		}},
+	}
+
+	if err := registerValidateWebhook(clienset.AdmissionregistrationV1beta1().ValidatingWebhookConfigurations(),
+		[]v1beta1.ValidatingWebhookConfiguration{PodValidateHooks}); err != nil {
+		return err
+	}
+
 	return nil
 
 }

cmd/admission/app/server.go

@@ -18,28 +18,13 @@ package app
 
 import (
 	"crypto/tls"
-	"encoding/json"
-	"io/ioutil"
-	"net/http"
 
 	"github.com/golang/glog"
-	"volcano.sh/volcano/pkg/client/clientset/versioned"
 
-	"k8s.io/api/admission/v1beta1"
-	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/kubernetes"
 	restclient "k8s.io/client-go/rest"
-
-	appConf "volcano.sh/volcano/cmd/admission/app/configure"
-	admissioncontroller "volcano.sh/volcano/pkg/admission"
-)
-
-const (
-	//CONTENTTYPE http content-type
-	CONTENTTYPE = "Content-Type"
-
-	//APPLICATIONJSON json content
-	APPLICATIONJSON = "application/json"
+	appConf "volcano.sh/volcano/cmd/admission/app/options"
+	"volcano.sh/volcano/pkg/client/clientset/versioned"
 )
 
 // GetClient Get a clientset with restConfig.
@@ -51,8 +36,8 @@ func GetClient(restConfig *restclient.Config) *kubernetes.Clientset {
 	return clientset
 }
 
-//GetKubeBatchClient get a clientset for kubebatch
-func GetKubeBatchClient(restConfig *restclient.Config) *versioned.Clientset {
+// GetVolcanoClient get a clientset for volcano
+func GetVolcanoClient(restConfig *restclient.Config) *versioned.Clientset {
 	clientset, err := versioned.NewForConfig(restConfig)
 	if err != nil {
 		glog.Fatal(err)
@@ -89,52 +74,3 @@ func ConfigTLS(config *appConf.Config, restConfig *restclient.Config) *tls.Confi
 	glog.Fatal("tls: failed to find any tls config data")
 	return &tls.Config{}
 }
-
-//Serve the http request
-func Serve(w http.ResponseWriter, r *http.Request, admit admissioncontroller.AdmitFunc) {
-	var body []byte
-	if r.Body != nil {
-		if data, err := ioutil.ReadAll(r.Body); err == nil {
-			body = data
-		}
-	}
-
-	// verify the content type is accurate
-	contentType := r.Header.Get(CONTENTTYPE)
-	if contentType != APPLICATIONJSON {
-		glog.Errorf("contentType=%s, expect application/json", contentType)
-		return
-	}
-
-	var reviewResponse *v1beta1.AdmissionResponse
-	ar := v1beta1.AdmissionReview{}
-	deserializer := admissioncontroller.Codecs.UniversalDeserializer()
-	if _, _, err := deserializer.Decode(body, nil, &ar); err != nil {
-		reviewResponse = admissioncontroller.ToAdmissionResponse(err)
-	} else {
-		reviewResponse = admit(ar)
-	}
-	glog.V(3).Infof("sending response: %v", reviewResponse)
-
-	response := createResponse(reviewResponse, &ar)
-	resp, err := json.Marshal(response)
-	if err != nil {
-		glog.Error(err)
-	}
-	if _, err := w.Write(resp); err != nil {
-		glog.Error(err)
-	}
-}
-
-func createResponse(reviewResponse *v1beta1.AdmissionResponse, ar *v1beta1.AdmissionReview) v1beta1.AdmissionReview {
-	response := v1beta1.AdmissionReview{}
-	if reviewResponse != nil {
-		response.Response = reviewResponse
-		response.Response.UID = ar.Request.UID
-	}
-	// reset the Object and OldObject, they are not needed in a response.
-	ar.Request.Object = runtime.RawExtension{}
-	ar.Request.OldObject = runtime.RawExtension{}
-
-	return response
-}

cmd/admission/main.go

@@ -17,28 +17,29 @@ package main
 
 import (
 	"flag"
-	"github.com/golang/glog"
 	"io/ioutil"
 	"net/http"
 	"os"
 	"os/signal"
 	"strconv"
 	"syscall"
 
+	"github.com/golang/glog"
+
 	"k8s.io/client-go/tools/clientcmd"
 
 	"volcano.sh/volcano/cmd/admission/app"
-	appConf "volcano.sh/volcano/cmd/admission/app/configure"
+	appConf "volcano.sh/volcano/cmd/admission/app/options"
 	admissioncontroller "volcano.sh/volcano/pkg/admission"
 	"volcano.sh/volcano/pkg/version"
 )
 
 func serveJobs(w http.ResponseWriter, r *http.Request) {
-	app.Serve(w, r, admissioncontroller.AdmitJobs)
+	admissioncontroller.Serve(w, r, admissioncontroller.AdmitJobs)
 }
 
 func serveMutateJobs(w http.ResponseWriter, r *http.Request) {
-	app.Serve(w, r, admissioncontroller.MutateJobs)
+	admissioncontroller.Serve(w, r, admissioncontroller.MutateJobs)
 }
 
 func main() {
@@ -63,7 +64,9 @@ func main() {
 		glog.Fatalf("Unable to build k8s config: %v\n", err)
 	}
 
-	admissioncontroller.KubeBatchClientSet = app.GetKubeBatchClient(restConfig)
+	admissioncontroller.VolcanoClientSet = app.GetVolcanoClient(restConfig)
+
+	servePods(config)
 
 	caBundle, err := ioutil.ReadFile(config.CaCertFile)
 	if err != nil {
@@ -101,3 +104,13 @@ func main() {
 		return
 	}
 }
+
+func servePods(config *appConf.Config) {
+	admController := &admissioncontroller.Controller{
+		VcClients:     admissioncontroller.VolcanoClientSet,
+		SchedulerName: config.SchedulerName,
+	}
+	http.HandleFunc(admissioncontroller.AdmitPodPath, admController.ServerPods)
+
+	return
+}

hack/e2e-admission-config.yaml

@@ -32,6 +32,9 @@ rules:
   - apiGroups: [""]
     resources: ["services"]
     verbs: ["get"]
+  - apiGroups: ["scheduling.incubator.k8s.io", "scheduling.sigs.dev"]
+    resources: ["podgroups"]
+    verbs: ["get", "list", "watch"]
 
 ---
 kind: ClusterRoleBinding

installer/helm/chart/volcano/templates/admission.yaml

@@ -187,6 +187,9 @@ rules:
   - apiGroups: [""]
     resources: ["services"]
     verbs: ["get"]
+  - apiGroups: ["scheduling.incubator.k8s.io", "scheduling.sigs.dev"]
+    resources: ["podgroups"]
+    verbs: ["get", "list", "watch"]
 
 ---
 kind: ClusterRoleBinding

installer/volcano-development.yaml

@@ -31,18 +31,31 @@ import (
 	"k8s.io/apimachinery/pkg/util/validation/field"
 
 	"volcano.sh/volcano/pkg/apis/batch/v1alpha1"
+	vcver "volcano.sh/volcano/pkg/client/clientset/versioned"
 )
 
 const (
-	//AdmitJobPath is the pattern for the jobs admission
+	// AdmitJobPath is the pattern for the jobs admission
 	AdmitJobPath = "/jobs"
-	//MutateJobPath is the pattern for the mutating jobs
+	// MutateJobPath is the pattern for the mutating jobs
 	MutateJobPath = "/mutating-jobs"
+	// AdmitPodPath is the pattern for the pods admission
+	AdmitPodPath = "/pods"
+	// CONTENTTYPE http content-type
+	CONTENTTYPE = "Content-Type"
+	// APPLICATIONJSON json content
+	APPLICATIONJSON = "application/json"
 )
 
 //The AdmitFunc returns response
 type AdmitFunc func(v1beta1.AdmissionReview) *v1beta1.AdmissionResponse
 
+// Controller the Admission Controller type
+type Controller struct {
+	VcClients     vcver.Interface
+	SchedulerName string
+}
+
 var scheme = runtime.NewScheme()
 
 //Codecs is for retrieving serializers for the supported wire formats

pkg/admission/admission_controller.go

@@ -21,7 +21,6 @@ import (
 	"strings"
 
 	"github.com/golang/glog"
-	"volcano.sh/volcano/pkg/client/clientset/versioned"
 
 	"k8s.io/api/admission/v1beta1"
 	"k8s.io/api/core/v1"
@@ -33,11 +32,12 @@ import (
 	k8scorevalid "k8s.io/kubernetes/pkg/apis/core/validation"
 
 	"volcano.sh/volcano/pkg/apis/batch/v1alpha1"
+	"volcano.sh/volcano/pkg/client/clientset/versioned"
 	"volcano.sh/volcano/pkg/controllers/job/plugins"
 )
 
-// KubeBatchClientSet is volcano clientset
-var KubeBatchClientSet versioned.Interface
+// VolcanoClientSet is volcano clientset
+var VolcanoClientSet versioned.Interface
 
 // AdmitJobs is to admit jobs and return response
 func AdmitJobs(ar v1beta1.AdmissionReview) *v1beta1.AdmissionResponse {
@@ -151,9 +151,9 @@ func validateJob(job v1alpha1.Job, reviewResponse *v1beta1.AdmissionResponse) st
 	}
 
 	// Check whether Queue already present or not
-	if _, err := KubeBatchClientSet.SchedulingV1alpha2().Queues().Get(job.Spec.Queue, metav1.GetOptions{}); err != nil {
+	if _, err := VolcanoClientSet.SchedulingV1alpha2().Queues().Get(job.Spec.Queue, metav1.GetOptions{}); err != nil {
 		// TODO: deprecate v1alpha1
-		if _, err := KubeBatchClientSet.SchedulingV1alpha1().Queues().Get(job.Spec.Queue, metav1.GetOptions{}); err != nil {
+		if _, err := VolcanoClientSet.SchedulingV1alpha1().Queues().Get(job.Spec.Queue, metav1.GetOptions{}); err != nil {
 			msg = msg + fmt.Sprintf(" unable to find job queue: %v", err)
 		}
 	}

pkg/admission/admit_job.go

@@ -957,10 +957,10 @@ func TestValidateExecution(t *testing.T) {
 				},
 			}
 			// create fake volcano clientset
-			KubeBatchClientSet = fakeclient.NewSimpleClientset()
+			VolcanoClientSet = fakeclient.NewSimpleClientset()
 
 			//create default queue
-			_, err := KubeBatchClientSet.SchedulingV1alpha2().Queues().Create(&defaultqueue)
+			_, err := VolcanoClientSet.SchedulingV1alpha2().Queues().Create(&defaultqueue)
 			if err != nil {
 				t.Error("Queue Creation Failed")
 			}