Merge pull request #779 from hzxuzhonghu/master

Set DNSPolicy=ClusterFirstWithHostNet when hostnetwork set

pkg/controllers/job/job_controller.go

@@ -20,7 +20,10 @@ import (
 	"fmt"
 	"hash"
 	"hash/fnv"
-	"k8s.io/api/core/v1"
+	"sync"
+	"time"
+
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/api/scheduling/v1beta1"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/informers"
@@ -34,8 +37,6 @@ import (
 	"k8s.io/client-go/tools/record"
 	"k8s.io/client-go/util/workqueue"
 	"k8s.io/klog"
-	"sync"
-	"time"
 
 	batchv1alpha1 "volcano.sh/volcano/pkg/apis/batch/v1alpha1"
 	busv1alpha1 "volcano.sh/volcano/pkg/apis/bus/v1alpha1"
@@ -145,8 +146,7 @@ func NewJobController(
 
 	cc.jobInformer = informerfactory.NewSharedInformerFactory(cc.vcClient, 0).Batch().V1alpha1().Jobs()
 	cc.jobInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
-		AddFunc: cc.addJob,
-		// TODO: enable this until we find an appropriate way.
+		AddFunc:    cc.addJob,
 		UpdateFunc: cc.updateJob,
 		DeleteFunc: cc.deleteJob,
 	})

pkg/controllers/job/job_controller_actions.go

@@ -87,7 +87,7 @@ func (cc *Controller) killJob(jobInfo *apis.JobInfo, podRetainPhase state.PhaseM
 	}
 
 	job = job.DeepCopy()
-	//Job version is bumped only when job is killed
+	// Job version is bumped only when job is killed
 	job.Status.Version = job.Status.Version + 1
 
 	job.Status = batch.JobStatus{

pkg/controllers/job/job_controller_util.go

@@ -18,14 +18,14 @@ package job
 
 import (
 	"fmt"
-	"volcano.sh/volcano/pkg/apis/bus/v1alpha1"
 
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/klog"
 
 	batch "volcano.sh/volcano/pkg/apis/batch/v1alpha1"
+	"volcano.sh/volcano/pkg/apis/bus/v1alpha1"
 	"volcano.sh/volcano/pkg/apis/helpers"
 	schedulingv2 "volcano.sh/volcano/pkg/apis/scheduling/v1beta1"
 	"volcano.sh/volcano/pkg/controllers/apis"
@@ -53,11 +53,6 @@ func createJobPod(job *batch.Job, template *v1.PodTemplateSpec, ix int) *v1.Pod
 		Spec: templateCopy.Spec,
 	}
 
-	// If no scheduler name in Pod, use scheduler name from Job.
-	if len(pod.Spec.SchedulerName) == 0 {
-		pod.Spec.SchedulerName = job.Spec.SchedulerName
-	}
-
 	volumeMap := make(map[string]string)
 	for _, volume := range job.Spec.Volumes {
 		vcName := volume.VolumeClaimName

pkg/webhooks/admission/jobs/mutate/mutate_job.go

@@ -23,7 +23,7 @@ import (
 
 	"k8s.io/api/admission/v1beta1"
 	whv1beta1 "k8s.io/api/admissionregistration/v1beta1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/klog"
 
 	"volcano.sh/volcano/pkg/apis/batch/v1alpha1"
@@ -33,8 +33,10 @@ import (
 )
 
 const (
-	//DefaultQueue constant stores the name of the queue as "default"
+	// DefaultQueue constant stores the name of the queue as "default"
 	DefaultQueue = "default"
+
+	defaultSchedulerName = "volcano"
 )
 
 func init() {
@@ -77,9 +79,6 @@ func MutateJobs(ar v1beta1.AdmissionReview) *v1beta1.AdmissionResponse {
 		return util.ToAdmissionResponse(err)
 	}
 
-	reviewResponse := v1beta1.AdmissionResponse{}
-	reviewResponse.Allowed = true
-
 	var patchBytes []byte
 	switch ar.Request.Operation {
 	case v1beta1.Create:
@@ -90,12 +89,11 @@ func MutateJobs(ar v1beta1.AdmissionReview) *v1beta1.AdmissionResponse {
 		return util.ToAdmissionResponse(err)
 	}
 
-	if err != nil {
-		reviewResponse.Result = &metav1.Status{Message: err.Error()}
-		return &reviewResponse
+	klog.V(3).Infof("AdmissionResponse: patch=%v", string(patchBytes))
+	reviewResponse := v1beta1.AdmissionResponse{
+		Allowed: true,
+		Patch:   patchBytes,
 	}
-	klog.V(3).Infof("AdmissionResponse: patch=%v\n", string(patchBytes))
-	reviewResponse.Patch = patchBytes
 	pt := v1beta1.PatchTypeJSONPatch
 	reviewResponse.PatchType = &pt
 
@@ -108,6 +106,10 @@ func createPatch(job *v1alpha1.Job) ([]byte, error) {
 	if pathQueue != nil {
 		patch = append(patch, *pathQueue)
 	}
+	pathScheduler := patchDefaultScheduler(job)
+	if pathScheduler != nil {
+		patch = append(patch, *pathScheduler)
+	}
 	pathSpec := mutateSpec(job.Spec.Tasks, "/spec/tasks")
 	if pathSpec != nil {
 		patch = append(patch, *pathSpec)
@@ -123,6 +125,14 @@ func patchDefaultQueue(job *v1alpha1.Job) *patchOperation {
 	return nil
 }
 
+func patchDefaultScheduler(job *v1alpha1.Job) *patchOperation {
+	// Add default scheduler name if not specified.
+	if job.Spec.SchedulerName == "" {
+		return &patchOperation{Op: "add", Path: "/spec/schedulerName", Value: defaultSchedulerName}
+	}
+	return nil
+}
+
 func mutateSpec(tasks []v1alpha1.TaskSpec, basePath string) *patchOperation {
 	patched := false
 	for index := range tasks {
@@ -132,6 +142,11 @@ func mutateSpec(tasks []v1alpha1.TaskSpec, basePath string) *patchOperation {
 			patched = true
 			tasks[index].Name = v1alpha1.DefaultTaskSpec + strconv.Itoa(index)
 		}
+
+		if tasks[index].Template.Spec.HostNetwork && tasks[index].Template.Spec.DNSPolicy == "" {
+			tasks[index].Template.Spec.DNSPolicy = v1.DNSClusterFirstWithHostNet
+		}
+
 	}
 	if !patched {
 		return nil