【kubernetes/k8s源码分析】kubectl-controller-manager之job源码分析

逃离我推掉我的手 2022-04-12 13:27 465阅读 0赞

# job介绍 #

[Job][]: 批量一次性任务，并保证处理的一个或者多个Pod成功结束

*  非并行Job:
 *  固定完成次数的并行Job:
 *  带有工作队列的并行Job:

# SPEC参数 #

*  .spec.completions:  指定job成功运行Pods的次数。默认值: 1
 *  .spec.parallelism:    指定job并发运行Pods的数量。默认值: 1
 *  .spec.activeDeadlineSeconds: 指定job可运行的时间期限。
 *  .spec.backoffLimit: 指定job失败后进行重试的次数。默认是6次，每次失败后重试会有延迟时间，该时间是指数级增长，最长时间是6min。

# **0. 入口NewControllerInitializers函数** #

注册job，controllers\["job"\] = startJobController

// NewControllerInitializers is a public map of named controller groups (you can start more than one in an init func)
    // paired to their InitFunc.  This allows for structured downstream composition and subdivision.
    func NewControllerInitializers(loopMode ControllerLoopMode) map[string]InitFunc {
    	controllers := map[string]InitFunc{}
    
    	controllers["daemonset"] = startDaemonSetController
    	controllers["job"] = startJobController
    
    	return controllers
    }

# 1. startJobController函数 #

*      调用NewJobController函数构造对象
 *      调用其Run方法执行主要逻辑

func startJobController(ctx ControllerContext) (http.Handler, bool, error) {
    	if !ctx.AvailableResources[schema.GroupVersionResource{Group: "batch", Version: "v1", Resource: "jobs"}] {
    		return nil, false, nil
    	}
    	go job.NewJobController(
    		ctx.InformerFactory.Core().V1().Pods(),
    		ctx.InformerFactory.Batch().V1().Jobs(),
    		ctx.ClientBuilder.ClientOrDie("job-controller"),
    	).Run(int(ctx.ComponentConfig.JobController.ConcurrentJobSyncs), ctx.Stop)
    	return nil, true, nil
    }

# 2. NewJobController函数 #

2.1 初始化JobController结构体

*     注册jobInformer 的Add、Update、Delete 函数
 *     注册 podInformer 的Add、Update、Delete 函数
 *     jm.syncHandler = jm.syncJob核心函数

jm := &JobController{
    		kubeClient: kubeClient,
    		podControl: controller.RealPodControl{
    			KubeClient: kubeClient,
    			Recorder:   eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "job-controller"}),
    		},
    		expectations: controller.NewControllerExpectations(),
    		queue:        workqueue.NewNamedRateLimitingQueue(workqueue.NewItemExponentialFailureRateLimiter(DefaultJobBackOff, MaxJobBackOff), "job"),
    		recorder:     eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "job-controller"}),
    	}
    
    	jobInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
    		AddFunc: func(obj interface{}) {
    			jm.enqueueController(obj, true)
    		},
    		UpdateFunc: jm.updateJob,
    		DeleteFunc: func(obj interface{}) {
    			jm.enqueueController(obj, true)
    		},
    	})
    	jm.jobLister = jobInformer.Lister()
    	jm.jobStoreSynced = jobInformer.Informer().HasSynced
    
    	podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
    		AddFunc:    jm.addPod,
    		UpdateFunc: jm.updatePod,
    		DeleteFunc: jm.deletePod,
    	})
    	jm.podStore = podInformer.Lister()
    	jm.podStoreSynced = podInformer.Informer().HasSynced
    
    	jm.updateHandler = jm.updateJobStatus
    	jm.syncHandler = jm.syncJob

# 3. Run函数 #

启动n个goroutine间隔1s执行worker函数

// Run the main goroutine responsible for watching and syncing jobs.
    func (jm *JobController) Run(workers int, stopCh <-chan struct{}) {
    	defer utilruntime.HandleCrash()
    	defer jm.queue.ShutDown()
    
    	glog.Infof("Starting job controller")
    	defer glog.Infof("Shutting down job controller")
    
    	if !controller.WaitForCacheSync("job", stopCh, jm.podStoreSynced, jm.jobStoreSynced) {
    		return
    	}
    
    	for i := 0; i < workers; i++ {
    		go wait.Until(jm.worker, time.Second, stopCh)
    	}
    
    	<-stopCh
    }

###   3.1 worker函数 ###

队列取数据调用核心函数syncHandler处理，也就是syncJob函数

// worker runs a worker thread that just dequeues items, processes them, and marks them done.
    // It enforces that the syncHandler is never invoked concurrently with the same key.
    func (jm *JobController) worker() {
    	for jm.processNextWorkItem() {
    	}
    }
    
    func (jm *JobController) processNextWorkItem() bool {
    	key, quit := jm.queue.Get()
    	if quit {
    		return false
    	}
    	defer jm.queue.Done(key)
    
    	forget, err := jm.syncHandler(key.(string))
    	if err == nil {
    		if forget {
    			jm.queue.Forget(key)
    		}
    		return true
    	}
    
    	utilruntime.HandleError(fmt.Errorf("Error syncing job: %v", err))
    	jm.queue.AddRateLimited(key)
    
    	return true
    }

# 4. syncJob函数 #

###   4.1 获得job信息，如果执行完则直接返回 ###

*  c.Type == batch.JobComplete || c.Type == batch.JobFailed
 *  c.Status == v1.ConditionTrue

ns, name, err := cache.SplitMetaNamespaceKey(key)
    	if err != nil {
    		return false, err
    	}
    	if len(ns) == 0 || len(name) == 0 {
    		return false, fmt.Errorf("invalid job key %q: either namespace or name is missing", key)
    	}
    	sharedJob, err := jm.jobLister.Jobs(ns).Get(name)
    	if err != nil {
    		if errors.IsNotFound(err) {
    			glog.V(4).Infof("Job has been deleted: %v", key)
    			jm.expectations.DeleteExpectations(key)
    			return true, nil
    		}
    		return false, err
    	}
    	job := *sharedJob
    
    	// if job was finished previously, we don't want to redo the termination
    	if IsJobFinished(&job) {
    		return true, nil
    	}

###   4.2 getPodsForJob函数 ###

获取该job下所有的pod，主要靠label selector获得

// getPodsForJob returns the set of pods that this Job should manage.
    // It also reconciles ControllerRef by adopting/orphaning.
    // Note that the returned Pods are pointers into the cache.
    func (jm *JobController) getPodsForJob(j *batch.Job) ([]*v1.Pod, error) {
    	selector, err := metav1.LabelSelectorAsSelector(j.Spec.Selector)
    	if err != nil {
    		return nil, fmt.Errorf("couldn't convert Job selector: %v", err)
    	}
    	// List all pods to include those that don't match the selector anymore
    	// but have a ControllerRef pointing to this controller.
    	pods, err := jm.podStore.Pods(j.Namespace).List(labels.Everything())
    	if err != nil {
    		return nil, err
    	}
    	// If any adoptions are attempted, we should first recheck for deletion
    	// with an uncached quorum read sometime after listing Pods (see #42639).
    	canAdoptFunc := controller.RecheckDeletionTimestamp(func() (metav1.Object, error) {
    		fresh, err := jm.kubeClient.BatchV1().Jobs(j.Namespace).Get(j.Name, metav1.GetOptions{})
    		if err != nil {
    			return nil, err
    		}
    		if fresh.UID != j.UID {
    			return nil, fmt.Errorf("original Job %v/%v is gone: got uid %v, wanted %v", j.Namespace, j.Name, fresh.UID, j.UID)
    		}
    		return fresh, nil
    	})
    	cm := controller.NewPodControllerRefManager(jm.podControl, j, selector, controllerKind, canAdoptFunc)
    	return cm.ClaimPods(pods)
    }

###   4.3 获得active pod（不是success，不是failed，不是删除的） ###

首次执行，设置start time，如果设置了deadline则加入队列中

activePods := controller.FilterActivePods(pods)
    	active := int32(len(activePods))
    	succeeded, failed := getStatus(pods)
    	conditions := len(job.Status.Conditions)
    	// job first start
    	if job.Status.StartTime == nil {
    		now := metav1.Now()
    		job.Status.StartTime = &now
    		// enqueue a sync to check if job past ActiveDeadlineSeconds
    		if job.Spec.ActiveDeadlineSeconds != nil {
    			glog.V(4).Infof("Job %s have ActiveDeadlineSeconds will sync after %d seconds",
    				key, *job.Spec.ActiveDeadlineSeconds)
    			jm.queue.AddAfter(key, time.Duration(*job.Spec.ActiveDeadlineSeconds)*time.Second)
    		}
    	}

###   4.4 失败标注失败原因 ###

jobHaveNewFailure := failed > job.Status.Failed
    	// new failures happen when status does not reflect the failures and active
    	// is different than parallelism, otherwise the previous controller loop
    	// failed updating status so even if we pick up failure it is not a new one
    	exceedsBackoffLimit := jobHaveNewFailure && (active != *job.Spec.Parallelism) &&
    		(int32(previousRetry)+1 > *job.Spec.BackoffLimit)
    
    	if exceedsBackoffLimit || pastBackoffLimitOnFailure(&job, pods) {
    		// check if the number of pod restart exceeds backoff (for restart OnFailure only)
    		// OR if the number of failed jobs increased since the last syncJob
    		jobFailed = true
    		failureReason = "BackoffLimitExceeded"
    		failureMessage = "Job has reached the specified backoff limit"
    	} else if pastActiveDeadline(&job) {
    		jobFailed = true
    		failureReason = "DeadlineExceeded"
    		failureMessage = "Job was active longer than specified deadline"
    	}

###    4.5 job失败的情况 ###

*    失败了调用deleteJobPods并发删除active pod
 *   并相应的更新job状态

if jobFailed {
    		errCh := make(chan error, active)
    		jm.deleteJobPods(&job, activePods, errCh)
    		select {
    		case manageJobErr = <-errCh:
    			if manageJobErr != nil {
    				break
    			}
    		default:
    		}
    
    		// update status values accordingly
    		failed += active
    		active = 0
    		job.Status.Conditions = append(job.Status.Conditions, newCondition(batch.JobFailed, failureReason, failureMessage))
    		jm.recorder.Event(&job, v1.EventTypeWarning, failureReason, failureMessage)
    	}

###   4.6 job未失败的情况 ###

*  job是否需要sync，且job 未删除调用mangeJob(第5章节讲解)
 *  Completions 代表job成功创建指定数量的pods --> 成功

else {
    		if jobNeedsSync && job.DeletionTimestamp == nil {
    			active, manageJobErr = jm.manageJob(activePods, succeeded, &job)
    		}
    		completions := succeeded
    		complete := false
    		if job.Spec.Completions == nil {
    			// This type of job is complete when any pod exits with success.
    			// Each pod is capable of
    			// determining whether or not the entire Job is done.  Subsequent pods are
    			// not expected to fail, but if they do, the failure is ignored.  Once any
    			// pod succeeds, the controller waits for remaining pods to finish, and
    			// then the job is complete.
    			if succeeded > 0 && active == 0 {
    				complete = true
    			}
    		} else {
    			// Job specifies a number of completions.  This type of job signals
    			// success by having that number of successes.  Since we do not
    			// start more pods than there are remaining completions, there should
    			// not be any remaining active pods once this count is reached.
    			if completions >= *job.Spec.Completions {
    				complete = true
    				if active > 0 {
    					jm.recorder.Event(&job, v1.EventTypeWarning, "TooManyActivePods", "Too many active pods running after completion count reached")
    				}
    				if completions > *job.Spec.Completions {
    					jm.recorder.Event(&job, v1.EventTypeWarning, "TooManySucceededPods", "Too many succeeded pods running after completion count reached")
    				}
    			}
    		}
    		if complete {
    			job.Status.Conditions = append(job.Status.Conditions, newCondition(batch.JobComplete, "", ""))
    			now := metav1.Now()
    			job.Status.CompletionTime = &now
    		}
    	}

# 5. manageJob函数 #

###   5.1 active > parallelism情况 ###

排序 not-ready < ready, unscheduled < scheduled, and pending < running

按照排序后的优先，删除并发数量与期待的数量多出来的pod

var errCh chan error
    	if active > parallelism {
    		diff := active - parallelism
    		errCh = make(chan error, diff)
    		jm.expectations.ExpectDeletions(jobKey, int(diff))
    		glog.V(4).Infof("Too many pods running job %q, need %d, deleting %d", jobKey, parallelism, diff)
    		// Sort the pods in the order such that not-ready < ready, unscheduled
    		// < scheduled, and pending < running. This ensures that we delete pods
    		// in the earlier stages whenever possible.
    		sort.Sort(controller.ActivePods(activePods))
    
    		active -= diff
    		wait := sync.WaitGroup{}
    		wait.Add(int(diff))
    		for i := int32(0); i < diff; i++ {
    			go func(ix int32) {
    				defer wait.Done()
    				if err := jm.podControl.DeletePod(job.Namespace, activePods[ix].Name, job); err != nil {
    					defer utilruntime.HandleError(err)
    					// Decrement the expected number of deletes because the informer won't observe this deletion
    					glog.V(2).Infof("Failed to delete %v, decrementing expectations for job %q/%q", activePods[ix].Name, job.Namespace, job.Name)
    					jm.expectations.DeletionObserved(jobKey)
    					activeLock.Lock()
    					active++
    					activeLock.Unlock()
    					errCh <- err
    				}
    			}(i)
    		}
    		wait.Wait()
    
    	}

###   5.2 active < parallelism情况 ###

batch创建pod使用慢开始算法

} else if active < parallelism {
    		wantActive := int32(0)
    		if job.Spec.Completions == nil {
    			// Job does not specify a number of completions.  Therefore, number active
    			// should be equal to parallelism, unless the job has seen at least
    			// once success, in which leave whatever is running, running.
    			if succeeded > 0 {
    				wantActive = active
    			} else {
    				wantActive = parallelism
    			}
    		} else {
    			// Job specifies a specific number of completions.  Therefore, number
    			// active should not ever exceed number of remaining completions.
    			wantActive = *job.Spec.Completions - succeeded
    			if wantActive > parallelism {
    				wantActive = parallelism
    			}
    		}
    		diff := wantActive - active
    		if diff < 0 {
    			utilruntime.HandleError(fmt.Errorf("More active than wanted: job %q, want %d, have %d", jobKey, wantActive, active))
    			diff = 0
    		}
    		jm.expectations.ExpectCreations(jobKey, int(diff))
    		errCh = make(chan error, diff)
    		glog.V(4).Infof("Too few pods running job %q, need %d, creating %d", jobKey, wantActive, diff)
    
    		active += diff
    		wait := sync.WaitGroup{}
    
    		// Batch the pod creates. Batch sizes start at SlowStartInitialBatchSize
    		// and double with each successful iteration in a kind of "slow start".
    		// This handles attempts to start large numbers of pods that would
    		// likely all fail with the same error. For example a project with a
    		// low quota that attempts to create a large number of pods will be
    		// prevented from spamming the API service with the pod create requests
    		// after one of its pods fails.  Conveniently, this also prevents the
    		// event spam that those failures would generate.
    		for batchSize := int32(integer.IntMin(int(diff), controller.SlowStartInitialBatchSize)); diff > 0; batchSize = integer.Int32Min(2*batchSize, diff) {
    			errorCount := len(errCh)
    			wait.Add(int(batchSize))
    			for i := int32(0); i < batchSize; i++ {
    				go func() {
    					defer wait.Done()
    					err := jm.podControl.CreatePodsWithControllerRef(job.Namespace, &job.Spec.Template, job, metav1.NewControllerRef(job, controllerKind))
    					if err != nil && errors.IsTimeout(err) {
    						// Pod is created but its initialization has timed out.
    						// If the initialization is successful eventually, the
    						// controller will observe the creation via the informer.
    						// If the initialization fails, or if the pod keeps
    						// uninitialized for a long time, the informer will not
    						// receive any update, and the controller will create a new
    						// pod when the expectation expires.
    						return
    					}
    					if err != nil {
    						defer utilruntime.HandleError(err)
    						// Decrement the expected number of creates because the informer won't observe this pod
    						glog.V(2).Infof("Failed creation, decrementing expectations for job %q/%q", job.Namespace, job.Name)
    						jm.expectations.CreationObserved(jobKey)
    						activeLock.Lock()
    						active--
    						activeLock.Unlock()
    						errCh <- err
    					}
    				}()
    			}
    			wait.Wait()
    			// any skipped pods that we never attempted to start shouldn't be expected.
    			skippedPods := diff - batchSize
    			if errorCount < len(errCh) && skippedPods > 0 {
    				glog.V(2).Infof("Slow-start failure. Skipping creation of %d pods, decrementing expectations for job %q/%q", skippedPods, job.Namespace, job.Name)
    				active -= skippedPods
    				for i := int32(0); i < skippedPods; i++ {
    					// Decrement the expected number of creates because the informer won't observe this pod
    					jm.expectations.CreationObserved(jobKey)
    				}
    				// The skipped pods will be retried later. The next controller resync will
    				// retry the slow start process.
    				break
    			}
    			diff -= batchSize
    		}
    	}

[Job]: https://kubernetes.io/docs/concepts/workloads/controllers/jobs-run-to-completion/