Admission refactor.

Signed-off-by: Klaus Ma <klaus1982.cn@gmail.com>

Author: k82cn

cmd/admission/app/options/options.go

@@ -46,6 +46,7 @@ type Config struct {
 	ValidateWebhookConfigName string
 	ValidateWebhookName       string
 	PrintVersion              bool
+	PrintYAML                 bool
 	AdmissionServiceName      string
 	AdmissionServiceNamespace string
 	SchedulerName             string
@@ -76,6 +77,7 @@ func (c *Config) AddFlags(fs *pflag.FlagSet) {
 	fs.StringVar(&c.ValidateWebhookName, "validate-webhook-name", "",
 		"Name of the webhook entry in the webhook config. [Deprecated]: it will be generated when not specified")
 	fs.BoolVar(&c.PrintVersion, "version", false, "Show version and quit")
+	fs.BoolVar(&c.PrintYAML, "yaml", false, "Show the yaml of all webhook")
 	fs.StringVar(&c.AdmissionServiceNamespace, "webhook-namespace", "default", "The namespace of this webhook")
 	fs.StringVar(&c.AdmissionServiceName, "webhook-service-name", "admission-service", "The name of this admission service")
 	fs.StringVar(&c.SchedulerName, "scheduler-name", defaultSchedulerName, "Volcano will handle pods whose .spec.SchedulerName is same as scheduler-name")
@@ -235,7 +237,7 @@ func registerMutateWebhook(client admissionregistrationv1beta1.MutatingWebhookCo
 			return err
 		}
 		if err == nil && existing != nil {
-			klog.Infof("Updating MutatingWebhookConfiguration %v", hook)
+			klog.V(4).Infof("Updating MutatingWebhookConfiguration %v", hook)
 			existing.Webhooks = hook.Webhooks
 			if _, err := client.Update(existing); err != nil {
 				return err
@@ -259,12 +261,12 @@ func registerValidateWebhook(client admissionregistrationv1beta1.ValidatingWebho
 		}
 		if err == nil && existing != nil {
 			existing.Webhooks = hook.Webhooks
-			klog.Infof("Updating ValidatingWebhookConfiguration %v", hook)
+			klog.V(4).Infof("Updating ValidatingWebhookConfiguration %v", hook)
 			if _, err := client.Update(existing); err != nil {
 				return err
 			}
 		} else {
-			klog.Infof("Creating ValidatingWebhookConfiguration %v", hook)
+			klog.V(4).Infof("Creating ValidatingWebhookConfiguration %v", hook)
 			if _, err := client.Create(&hook); err != nil {
 				return err
 			}

cmd/admission/app/server.go

@@ -18,17 +18,99 @@ package app
 
 import (
 	"crypto/tls"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"os"
+	"os/signal"
+	"strconv"
+	"syscall"
 
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/clientcmd"
 	"k8s.io/klog"
-
+	
 	"volcano.sh/volcano/cmd/admission/app/options"
+	"volcano.sh/volcano/pkg/admission/router"
 	"volcano.sh/volcano/pkg/client/clientset/versioned"
+	"volcano.sh/volcano/pkg/version"
 )
 
+// Run start the service of admission controller.
+func Run(config *options.Config) error {
+	if config.PrintVersion {
+		version.PrintVersionAndExit()
+		return nil
+	}
+
+	if config.PrintYAML {
+		router.ForEachAdmission(func(service *router.AdmissionService) {
+			fmt.Print("---\n")
+			fmt.Printf("%s\n", service.Path)
+		})
+		return nil
+	}
+
+	restConfig, err := clientcmd.BuildConfigFromFlags(config.Master, config.Kubeconfig)
+	if err != nil {
+		return fmt.Errorf("unable to build k8s config: %v", err)
+	}
+
+	vClient := getVolcanoClient(restConfig)
+	router.ForEachAdmission(func(service *router.AdmissionService) {
+		if service.Config != nil {
+			service.Config.VolcanoClient = vClient
+			service.Config.SchedulerName = config.SchedulerName
+		}
+
+		klog.V(3).Infof("Registered '%s' as webhook.", service.Path)
+		http.HandleFunc(service.Path, service.Handler)
+	})
+
+	caBundle, err := ioutil.ReadFile(config.CaCertFile)
+	if err != nil {
+		return fmt.Errorf("unable to read cacert file: %v", err)
+	}
+
+	err = options.RegisterWebhooks(config, getClient(restConfig), caBundle)
+	if err != nil {
+		return fmt.Errorf("unable to register webhook configs: %v", err)
+	}
+
+	klog.V(3).Info("Registered all webhooks to the api-server.")
+
+	webhookServeError := make(chan struct{})
+	stopChannel := make(chan os.Signal)
+	signal.Notify(stopChannel, syscall.SIGTERM, syscall.SIGINT)
+
+	server := &http.Server{
+		Addr:      ":" + strconv.Itoa(config.Port),
+		TLSConfig: configTLS(config, restConfig),
+	}
+	go func() {
+		err = server.ListenAndServeTLS("", "")
+		if err != nil && err != http.ErrServerClosed {
+			klog.Fatalf("ListenAndServeTLS for admission webhook failed: %v", err)
+			close(webhookServeError)
+		}
+
+		klog.Info("Volcano Webhook manager started.")
+	}()
+
+	select {
+	case <-stopChannel:
+		if err := server.Close(); err != nil {
+			return fmt.Errorf("close admission server failed: %v", err)
+		}
+		return nil
+	case <-webhookServeError:
+		return fmt.Errorf("unknown webhook server error")
+	}
+}
+
 // GetClient Get a clientset with restConfig.
-func GetClient(restConfig *rest.Config) *kubernetes.Clientset {
+func getClient(restConfig *rest.Config) *kubernetes.Clientset {
 	clientset, err := kubernetes.NewForConfig(restConfig)
 	if err != nil {
 		klog.Fatal(err)
@@ -37,18 +119,18 @@ func GetClient(restConfig *rest.Config) *kubernetes.Clientset {
 }
 
 // GetVolcanoClient get a clientset for volcano
-func GetVolcanoClient(restConfig *rest.Config) *versioned.Clientset {
+func getVolcanoClient(restConfig *rest.Config) *versioned.Clientset {
 	clientset, err := versioned.NewForConfig(restConfig)
 	if err != nil {
 		klog.Fatal(err)
 	}
 	return clientset
 }
 
-// ConfigTLS is a helper function that generate tls certificates from directly defined tls config or kubeconfig
+// configTLS is a helper function that generate tls certificates from directly defined tls config or kubeconfig
 // These are passed in as command line for cluster certification. If tls config is passed in, we use the directly
 // defined tls config, else use that defined in kubeconfig
-func ConfigTLS(config *options.Config, restConfig *rest.Config) *tls.Config {
+func configTLS(config *options.Config, restConfig *rest.Config) *tls.Config {
 	if len(config.CertFile) != 0 && len(config.KeyFile) != 0 {
 		sCert, err := tls.LoadX509KeyPair(config.CertFile, config.KeyFile)
 		if err != nil {

cmd/admission/main.go

@@ -16,35 +16,24 @@ limitations under the License.
 package main
 
 import (
-	"io/ioutil"
-	"net/http"
+	"fmt"
 	"os"
-	"os/signal"
 	"runtime"
-	"strconv"
-	"syscall"
 	"time"
 
 	"github.com/spf13/pflag"
 
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/apiserver/pkg/util/flag"
-	"k8s.io/client-go/tools/clientcmd"
 	"k8s.io/klog"
 
 	"volcano.sh/volcano/cmd/admission/app"
 	"volcano.sh/volcano/cmd/admission/app/options"
-	"volcano.sh/volcano/pkg/admission"
-	"volcano.sh/volcano/pkg/version"
-)
-
-func serveJobs(w http.ResponseWriter, r *http.Request) {
-	admission.Serve(w, r, admission.AdmitJobs)
-}
 
-func serveMutateJobs(w http.ResponseWriter, r *http.Request) {
-	admission.Serve(w, r, admission.MutateJobs)
-}
+	_ "volcano.sh/volcano/pkg/admission/jobs/mutate"
+	_ "volcano.sh/volcano/pkg/admission/jobs/validate"
+	_ "volcano.sh/volcano/pkg/admission/pods"
+)
 
 var logFlushFreq = pflag.Duration("log-flush-frequency", 5*time.Second, "Maximum number of seconds between log flushes")
 
@@ -57,73 +46,15 @@ func main() {
 
 	flag.InitFlags()
 
-	if config.PrintVersion {
-		version.PrintVersionAndExit()
-	}
-
 	go wait.Until(klog.Flush, *logFlushFreq, wait.NeverStop)
 	defer klog.Flush()
 
-	http.HandleFunc(admission.AdmitJobPath, serveJobs)
-	http.HandleFunc(admission.MutateJobPath, serveMutateJobs)
-
 	if err := config.CheckPortOrDie(); err != nil {
 		klog.Fatalf("Configured port is invalid: %v", err)
 	}
-	addr := ":" + strconv.Itoa(config.Port)
-
-	restConfig, err := clientcmd.BuildConfigFromFlags(config.Master, config.Kubeconfig)
-	if err != nil {
-		klog.Fatalf("Unable to build k8s config: %v", err)
-	}
-
-	admission.VolcanoClientSet = app.GetVolcanoClient(restConfig)
-
-	servePods(config)
 
-	caBundle, err := ioutil.ReadFile(config.CaCertFile)
-	if err != nil {
-		klog.Fatalf("Unable to read cacert file: %v", err)
+	if err := app.Run(config); err != nil {
+		fmt.Fprintf(os.Stderr, "%v\n", err)
+		os.Exit(1)
 	}
-
-	err = options.RegisterWebhooks(config, app.GetClient(restConfig), caBundle)
-	if err != nil {
-		klog.Fatalf("Unable to register webhook configs: %v", err)
-	}
-
-	stopChannel := make(chan os.Signal)
-	signal.Notify(stopChannel, syscall.SIGTERM, syscall.SIGINT)
-
-	server := &http.Server{
-		Addr:      addr,
-		TLSConfig: app.ConfigTLS(config, restConfig),
-	}
-	webhookServeError := make(chan struct{})
-	go func() {
-		err = server.ListenAndServeTLS("", "")
-		if err != nil && err != http.ErrServerClosed {
-			klog.Fatalf("ListenAndServeTLS for admission webhook failed: %v", err)
-			close(webhookServeError)
-		}
-	}()
-
-	select {
-	case <-stopChannel:
-		if err := server.Close(); err != nil {
-			klog.Fatalf("Close admission server failed: %v", err)
-		}
-		return
-	case <-webhookServeError:
-		return
-	}
-}
-
-func servePods(config *options.Config) {
-	admController := &admission.Controller{
-		VcClients:     admission.VolcanoClientSet,
-		SchedulerName: config.SchedulerName,
-	}
-	http.HandleFunc(admission.AdmitPodPath, admController.ServerPods)
-
-	return
 }

hack/.golint_failures

@@ -1,3 +1,6 @@
+volcano.sh/volcano/pkg/admission/jobs/mutate
+volcano.sh/volcano/pkg/admission/router
+volcano.sh/volcano/pkg/admission/schema
 volcano.sh/volcano/pkg/apis/scheduling
 volcano.sh/volcano/pkg/apis/scheduling/v1alpha1
 volcano.sh/volcano/pkg/apis/scheduling/v1alpha2

pkg/admission/mutate_job.go pkg/admission/jobs/mutate/mutate_job.gopkg/admission/mutate_job.go renamed to pkg/admission/jobs/mutate/mutate_job.go

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package admission
+package mutate
 
 import (
 	"encoding/json"
@@ -25,6 +25,9 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/klog"
 
+	"volcano.sh/volcano/pkg/admission/router"
+	"volcano.sh/volcano/pkg/admission/schema"
+	"volcano.sh/volcano/pkg/admission/util"
 	"volcano.sh/volcano/pkg/apis/batch/v1alpha1"
 )
 
@@ -33,6 +36,15 @@ const (
 	DefaultQueue = "default"
 )
 
+func init() {
+	router.RegisterAdmission(service)
+}
+
+var service = &router.AdmissionService{
+	Path: "/mutating-jobs",
+	Func: MutateJobs,
+}
+
 type patchOperation struct {
 	Op    string      `json:"op"`
 	Path  string      `json:"path"`
@@ -43,9 +55,9 @@ type patchOperation struct {
 func MutateJobs(ar v1beta1.AdmissionReview) *v1beta1.AdmissionResponse {
 	klog.V(3).Infof("mutating jobs")
 
-	job, err := DecodeJob(ar.Request.Object, ar.Request.Resource)
+	job, err := schema.DecodeJob(ar.Request.Object, ar.Request.Resource)
 	if err != nil {
-		return ToAdmissionResponse(err)
+		return util.ToAdmissionResponse(err)
 	}
 
 	reviewResponse := v1beta1.AdmissionResponse{}
@@ -58,7 +70,7 @@ func MutateJobs(ar v1beta1.AdmissionReview) *v1beta1.AdmissionResponse {
 		break
 	default:
 		err = fmt.Errorf("expect operation to be 'CREATE' ")
-		return ToAdmissionResponse(err)
+		return util.ToAdmissionResponse(err)
 	}
 
 	if err != nil {
@@ -73,7 +85,7 @@ func MutateJobs(ar v1beta1.AdmissionReview) *v1beta1.AdmissionResponse {
 	return &reviewResponse
 }
 
-func createPatch(job v1alpha1.Job) ([]byte, error) {
+func createPatch(job *v1alpha1.Job) ([]byte, error) {
 	var patch []patchOperation
 	pathQueue := patchDefaultQueue(job)
 	if pathQueue != nil {
@@ -86,7 +98,7 @@ func createPatch(job v1alpha1.Job) ([]byte, error) {
 	return json.Marshal(patch)
 }
 
-func patchDefaultQueue(job v1alpha1.Job) *patchOperation {
+func patchDefaultQueue(job *v1alpha1.Job) *patchOperation {
 	//Add default queue if not specified.
 	if job.Spec.Queue == "" {
 		return &patchOperation{Op: "add", Path: "/spec/queue", Value: DefaultQueue}

pkg/admission/mutate_job_test.go …admission/jobs/mutate/mutate_job_test.gopkg/admission/mutate_job_test.go renamed to pkg/admission/jobs/mutate/mutate_job_test.go pkg/admission/mutate_job_test.go renamed to pkg/admission/jobs/mutate/mutate_job_test.go

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package admission
+package mutate
 
 import (
 	"testing"