Function ManageJobResources

pkg/operator/resources/job/taskapi/cron.go:47–190

()

var _inProgressJobSpecMap = map[string]*spec.TaskJob{}

func ManageJobResources() error {
	inProgressJobKeys, err := job.ListAllInProgressJobKeys(userconfig.TaskAPIKind)
	if err != nil {
		return err
	}

	inProgressJobIDSet := strset.Set{}
	for _, jobKey := range inProgressJobKeys {
		inProgressJobIDSet.Add(jobKey.ID)
	}

	for jobID := range _inProgressJobSpecMap {
		if !inProgressJobIDSet.Has(jobID) {
			delete(_inProgressJobSpecMap, jobID)
		}
	}

	jobs, err := config.K8s.ListJobs(
		&kmeta.ListOptions{
			LabelSelector: klabels.SelectorFromSet(
				map[string]string{"apiKind": userconfig.TaskAPIKind.String()},
			).String(),
		},
	)
	if err != nil {
		return err
	}

	k8sJobMap := map[string]kbatch.Job{}
	k8sJobIDSet := strset.Set{}
	for _, kJob := range jobs {
		k8sJobMap[kJob.Labels["jobID"]] = kJob
		k8sJobIDSet.Add(kJob.Labels["jobID"])
	}

	for _, jobKey := range inProgressJobKeys {
		jobLogger, err := operator.GetJobLogger(jobKey)
		if err != nil {
			telemetry.Error(err)
			operatorLogger.Error(err)
			continue
		}

		k8sJob, jobFound := k8sJobMap[jobKey.ID]

		jobState, err := job.GetJobState(jobKey)
		if err != nil {
			jobLogger.Error(err)
			jobLogger.Error("terminating job and cleaning up job resources")
			err := errors.FirstError(
				job.DeleteInProgressFile(jobKey),
				deleteJobRuntimeResources(jobKey),
				recordFailure(jobKey),
			)
			if err != nil {
				telemetry.Error(err)
				operatorLogger.Error(err)
			}