Merge pull request #50 from volcano-sh/feature/support-plugins

Support Plugins

Author: Klaus Ma

Gopkg.lock

@@ -88,10 +88,15 @@ required = [
   name = "k8s.io/code-generator"
   version = "kubernetes-1.13.2"
 
+[[constraint]]
+  name = "golang.org/x/crypto"  
+  branch = "release-branch.go1.12"
+
 [prune]
   go-tests = true
   unused-packages = true
 
   [[prune.project]]
     name = "k8s.io/code-generator"
     unused-packages = false
+

Gopkg.toml

@@ -20,6 +20,7 @@ spec:
           storage: 1Gi
   tasks:
   - replicas: 6
+    name: "default-nginx"
     template:
       metadata:
         name: web

example/job.yaml

@@ -1,9 +1,11 @@
 actions: "reclaim, allocate, backfill, preempt"
 tiers:
-- plugins:
-  - name: priority
-  - name: gang
-- plugins:
-  - name: drf
-  - name: predicates
-  - name: proportion
+  - plugins:
+      - name: priority
+      - name: gang
+      - name: conformance
+  - plugins:
+      - name: drf
+      - name: predicates
+      - name: proportion
+      - name: nodeorder

example/kube-batch-conf.yaml

@@ -10,6 +10,11 @@ volcano.sh/volcano/pkg/cli/job
 volcano.sh/volcano/pkg/controllers/job
 volcano.sh/volcano/pkg/controllers/job/apis
 volcano.sh/volcano/pkg/controllers/job/cache
+volcano.sh/volcano/pkg/controllers/job/helpers
+volcano.sh/volcano/pkg/controllers/job/plugins
+volcano.sh/volcano/pkg/controllers/job/plugins/env
+volcano.sh/volcano/pkg/controllers/job/plugins/interface
+volcano.sh/volcano/pkg/controllers/job/plugins/ssh
 volcano.sh/volcano/pkg/controllers/job/state
 volcano.sh/volcano/pkg/scheduler/actions/allocate
 volcano.sh/volcano/pkg/scheduler/actions/backfill

hack/.golint_failures

@@ -6,7 +6,6 @@ metadata:
     "helm.sh/hook": pre-install,pre-upgrade,post-delete
 webhooks:
   - clientConfig:
-      caBundle: ""
       service:
         name: {{ .Release.Name }}-admission-service
         namespace: {{ .Release.Namespace }}
@@ -33,7 +32,6 @@ metadata:
     "helm.sh/hook": pre-install,pre-upgrade,post-delete
 webhooks:
   - clientConfig:
-      caBundle: ""
       service:
         name: {{ .Release.Name }}-admission-service
         namespace: {{ .Release.Namespace }}

installer/chart/volcano/templates/admission-config.yaml

@@ -57,6 +57,10 @@ spec:
         admission: "true"
     spec:
       serviceAccount: {{ .Release.Name }}-admission
+      {{ if .Values.basic.image_pull_secret }}
+      imagePullSecrets:
+        - name: {{ .Values.basic.image_pull_secret }}
+      {{ end }}
       containers:
         - args:
             - --tls-cert-file=/admission.local.config/certificates/tls.crt

installer/chart/volcano/templates/admission.yaml

@@ -36,7 +36,7 @@ rules:
     resources: ["persistentvolumeclaims"]
     verbs: ["get", "list", "watch", "create"]
   - apiGroups: [""]
-    resources: ["services"]
+    resources: ["services", "configmaps"]
     verbs: ["get", "list", "watch", "create", "delete"]
   - apiGroups: ["scheduling.incubator.k8s.io"]
     resources: ["podgroups"]
@@ -74,6 +74,10 @@ spec:
         vk-controllers: test
     spec:
       serviceAccount: {{ .Release.Name }}-controllers
+      {{ if .Values.basic.image_pull_secret }}
+      imagePullSecrets:
+          - name: {{ .Values.basic.image_pull_secret }}
+      {{ end }}
       containers:
           - name: {{ .Release.Name }}-controllers
             image: {{.Values.basic.controller_image_name}}:{{.Values.basic.image_tag_version}}

installer/chart/volcano/templates/controllers.yaml

@@ -3,4 +3,5 @@ basic:
   controller_image_name: "volcano-controllers"
   scheduler_image_name: "volcano-scheduler"
   admission_image_name: "volcano-admission"
-  admission_secret_name: "volcano-admission-secret"
+  admission_secret_name: "volcano-admission-secret"
+  image_pull_secret: ""

installer/chart/volcano/values.yaml

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"reflect"
 	"strings"
+	"volcano.sh/volcano/pkg/controllers/job/plugins"
 
 	"github.com/golang/glog"
 
@@ -43,7 +44,7 @@ func AdmitJobs(ar v1beta1.AdmissionReview) *v1beta1.AdmissionResponse {
 
 	switch ar.Request.Operation {
 	case v1beta1.Create:
-		msg = validateJob(job, &reviewResponse)
+		msg = validateJobSpec(job.Spec, &reviewResponse)
 		break
 	case v1beta1.Update:
 		oldJob, err := DecodeJob(ar.Request.OldObject, ar.Request.Resource)
@@ -63,18 +64,18 @@ func AdmitJobs(ar v1beta1.AdmissionReview) *v1beta1.AdmissionResponse {
 	return &reviewResponse
 }
 
-func validateJob(job v1alpha1.Job, reviewResponse *v1beta1.AdmissionResponse) string {
+func validateJobSpec(jobSpec v1alpha1.JobSpec, reviewResponse *v1beta1.AdmissionResponse) string {
 
 	var msg string
 	taskNames := map[string]string{}
 	var totalReplicas int32
 
-	if len(job.Spec.Tasks) == 0 {
+	if len(jobSpec.Tasks) == 0 {
 		reviewResponse.Allowed = false
 		return fmt.Sprintf("No task specified in job spec")
 	}
 
-	for _, task := range job.Spec.Tasks {
+	for _, task := range jobSpec.Tasks {
 		if task.Replicas == 0 {
 			msg = msg + fmt.Sprintf(" 'replicas' is set '0' in task: %s;", task.Name)
 		}
@@ -96,15 +97,24 @@ func validateJob(job v1alpha1.Job, reviewResponse *v1beta1.AdmissionResponse) st
 		}
 	}
 
-	if totalReplicas < job.Spec.MinAvailable {
+	if totalReplicas < jobSpec.MinAvailable {
 		msg = msg + " 'minAvailable' should not be greater than total replicas in tasks;"
 	}
 
 	//duplicate job event policies
-	if duplicateInfo, ok := CheckPolicyDuplicate(job.Spec.Policies); ok {
+	if duplicateInfo, ok := CheckPolicyDuplicate(jobSpec.Policies); ok {
 		msg = msg + fmt.Sprintf(" duplicated job event policies: %s;", duplicateInfo)
 	}
 
+	//invalid job plugins
+	if len(jobSpec.Plugins) != 0 {
+		for name := range jobSpec.Plugins {
+			if _, found := plugins.GetPluginBuilder(name); !found {
+				msg = msg + fmt.Sprintf(" unable to find job plugin: %s", name)
+			}
+		}
+	}
+
 	if msg != "" {
 		reviewResponse.Allowed = false
 	}

pkg/admission/admit_job.go

@@ -63,6 +63,11 @@ type JobSpec struct {
 
 	//Specifies the queue that will be used in the scheduler, "default" queue is used this leaves empty.
 	Queue string `json:"queue,omitempty" protobuf:"bytes,7,opt,name=queue"`
+
+	// Specifies the plugin of job
+	// Key is plugin name, value is the arguments of the plugin
+	// +optional
+	Plugins map[string][]string `json:"plugins,omitempty" protobuf:"bytes,7,opt,name=plugins"`
 }
 
 // VolumeSpec defines the specification of Volume, e.g. PVC
@@ -79,6 +84,7 @@ type JobEvent string
 
 const (
 	CommandIssued JobEvent = "CommandIssued"
+	PluginError   JobEvent = "PluginError"
 )
 
 // Event represent the phase of Job, e.g. pod-failed.
@@ -232,6 +238,8 @@ type JobStatus struct {
 	Terminating int32 `json:"terminating,omitempty" protobuf:"bytes,7,opt,name=terminating"`
 	//Current version of job
 	Version int32 `json:"version,omitempty" protobuf:"bytes,8,opt,name=version"`
+	// The resources that controlled by this job, e.g. Service, ConfigMap
+	ControlledResources map[string]string
 }
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object