prow/plank/controller.go
func (c *Controller) Sync() error {
	pjs, err := c.kc.ListProwJobs(c.selector)
	if err != nil {
		return fmt.Errorf("error listing prow jobs: %v", err)
	}
	selector := fmt.Sprintf("%s=true", kube.CreatedByProw)
	if len(c.selector) > 0 {
		selector = strings.Join([]string{c.selector, selector}, ",")
	}

	pm := map[string]kube.Pod{}
	for alias, client := range c.pkcs {
		pods, err := client.ListPods(selector)
		if err != nil {
			return fmt.Errorf("error listing pods in cluster %q: %v", alias, err)
		}
		for _, pod := range pods {
			pm[pod.ObjectMeta.Name] = pod
		}
	}
	// TODO: Replace the following filtering with a field selector once CRDs support field selectors.
	// https://github.com/kubernetes/kubernetes/issues/53459
	var k8sJobs []prowapi.ProwJob
	for _, pj := range pjs {
		if pj.Spec.Agent == prowapi.KubernetesAgent {
			k8sJobs = append(k8sJobs, pj)
		}
	}
	pjs = k8sJobs

	var syncErrs []error
	if err := c.terminateDupes(pjs, pm); err != nil {
		syncErrs = append(syncErrs, err)
	}

	// Share what we have for gathering metrics.
	c.pjLock.Lock()
	c.pjs = pjs
	c.pjLock.Unlock()

	pendingCh, triggeredCh := pjutil.PartitionActive(pjs)
	errCh := make(chan error, len(pjs))
	reportCh := make(chan prowapi.ProwJob, len(pjs))

	// Reinstantiate on every resync of the controller instead of trying
	// to keep this in sync with the state of the world.
	c.pendingJobs = make(map[string]int)
	// Sync pending jobs first so we can determine what is the maximum
	// number of new jobs we can trigger when syncing the non-pendings.
	maxSyncRoutines := c.config().Plank.MaxGoroutines
	c.log.Debugf("Handling %d pending prowjobs", len(pendingCh))
	syncProwJobs(c.log, c.syncPendingJob, maxSyncRoutines, pendingCh, reportCh, errCh, pm)
	c.log.Debugf("Handling %d triggered prowjobs", len(triggeredCh))
	syncProwJobs(c.log, c.syncTriggeredJob, maxSyncRoutines, triggeredCh, reportCh, errCh, pm)

	close(errCh)
	close(reportCh)

	for err := range errCh {
		syncErrs = append(syncErrs, err)
	}

	var reportErrs []error
	if !c.skipReport {
		reportTemplate := c.config().Plank.ReportTemplate
		reportTypes := c.config().GitHubReporter.JobTypesToReport
		for report := range reportCh {
			if err := reportlib.Report(c.ghc, reportTemplate, report, reportTypes); err != nil {
				reportErrs = append(reportErrs, err)
				c.log.WithFields(pjutil.ProwJobFields(&report)).WithError(err).Warn("Failed to report ProwJob status")
			}

			// plank is not retrying on errors, so we just set the current state as reported
			if err := c.setPreviousReportState(report); err != nil {
				c.log.WithFields(pjutil.ProwJobFields(&report)).WithError(err).Error("Failed to patch PrevReportStates")
			}
		}
	}

	if len(syncErrs) == 0 && len(reportErrs) == 0 {
		return nil
	}
	return fmt.Errorf("errors syncing: %v, errors reporting: %v", syncErrs, reportErrs)
}
func (c *Controller) Sync() error {
	pjs, err := c.kc.ListProwJobs(c.selector)
	if err != nil {
		return fmt.Errorf("error listing prow jobs: %v", err)
	}
	selector := fmt.Sprintf("%s=true", kube.CreatedByProw)
	if len(c.selector) > 0 {
		selector = strings.Join([]string{c.selector, selector}, ",")
	}

	pm := map[string]kube.Pod{}
	for alias, client := range c.pkcs {
		pods, err := client.ListPods(selector)
		if err != nil {
			return fmt.Errorf("error listing pods in cluster %q: %v", alias, err)
		}
		for _, pod := range pods {
			pm[pod.ObjectMeta.Name] = pod
		}
	}
	// TODO: Replace the following filtering with a field selector once CRDs support field selectors.
	// https://github.com/kubernetes/kubernetes/issues/53459
	var k8sJobs []prowapi.ProwJob
	for _, pj := range pjs {
		if pj.Spec.Agent == prowapi.KubernetesAgent {
			k8sJobs = append(k8sJobs, pj)
		}
	}
	pjs = k8sJobs

	var syncErrs []error
	if err := c.terminateDupes(pjs, pm); err != nil {
		syncErrs = append(syncErrs, err)
	}

	// Share what we have for gathering metrics.
	c.pjLock.Lock()
	c.pjs = pjs
	c.pjLock.Unlock()

	pendingCh, triggeredCh := pjutil.PartitionActive(pjs)
	errCh := make(chan error, len(pjs))
	reportCh := make(chan prowapi.ProwJob, len(pjs))

	// Reinstantiate on every resync of the controller instead of trying
	// to keep this in sync with the state of the world.
	c.pendingJobs = make(map[string]int)
	// Sync pending jobs first so we can determine what is the maximum
	// number of new jobs we can trigger when syncing the non-pendings.
	maxSyncRoutines := c.config().Plank.MaxGoroutines
	c.log.Debugf("Handling %d pending prowjobs", len(pendingCh))
	syncProwJobs(c.log, c.syncPendingJob, maxSyncRoutines, pendingCh, reportCh, errCh, pm)
	c.log.Debugf("Handling %d triggered prowjobs", len(triggeredCh))
	syncProwJobs(c.log, c.syncTriggeredJob, maxSyncRoutines, triggeredCh, reportCh, errCh, pm)

	close(errCh)
	close(reportCh)

	for err := range errCh {
		syncErrs = append(syncErrs, err)
	}

	var reportErrs []error
	if !c.skipReport {
		reportTemplate := c.config().Plank.ReportTemplate
		reportTypes := c.config().GitHubReporter.JobTypesToReport
		for report := range reportCh {
			if err := reportlib.Report(c.ghc, reportTemplate, report, reportTypes); err != nil {
				reportErrs = append(reportErrs, err)
				c.log.WithFields(pjutil.ProwJobFields(&report)).WithError(err).Warn("Failed to report ProwJob status")
			}

			// plank is not retrying on errors, so we just set the current state as reported
			if err := c.setPreviousReportState(report); err != nil {
				c.log.WithFields(pjutil.ProwJobFields(&report)).WithError(err).Error("Failed to patch PrevReportStates")
			}
		}
	}

	if len(syncErrs) == 0 && len(reportErrs) == 0 {
		return nil
	}
	return fmt.Errorf("errors syncing: %v, errors reporting: %v", syncErrs, reportErrs)
}
func ( c * Controller ) Sync ( ) error { pjs , err := c . kc . ListProwJobs ( c . selector ) if err != nil { return fmt . Errorf ( " " , err ) } selector := fmt . Sprintf ( " " , kube . CreatedByProw ) if len ( c . selector ) > 0 { selector = strings . Join ( [ ] string { c . selector , selector } , " " ) } pm := map [ string ] kube . Pod { } for alias , client := range c . pkcs { pods , err := client . ListPods ( selector ) if err != nil { return fmt . Errorf ( " " , alias , err ) } for _ , pod := range pods { pm [ pod . ObjectMeta . Name ] = pod } } // TODO: Replace the following filtering with a field selector once CRDs support field selectors. // https://github.com/kubernetes/kubernetes/issues/53459 var k8sJobs [ ] prowapi . ProwJob for _ , pj := range pjs { if pj . Spec . Agent == prowapi . KubernetesAgent { k8sJobs = append ( k8sJobs , pj ) } } pjs = k8sJobs var syncErrs [ ] error if err := c . terminateDupes ( pjs , pm ) ; err != nil { syncErrs = append ( syncErrs , err ) } // Share what we have for gathering metrics. c . pjLock . Lock ( ) c . pjs = pjs c . pjLock . Unlock ( ) pendingCh , triggeredCh := pjutil . PartitionActive ( pjs ) errCh := make ( chan error , len ( pjs ) ) reportCh := make ( chan prowapi . ProwJob , len ( pjs ) ) // Reinstantiate on every resync of the controller instead of trying // to keep this in sync with the state of the world. c . pendingJobs = make ( map [ string ] int ) // Sync pending jobs first so we can determine what is the maximum // number of new jobs we can trigger when syncing the non-pendings. maxSyncRoutines := c . config ( ) . Plank . MaxGoroutines c . log . Debugf ( " " , len ( pendingCh ) ) syncProwJobs ( c . log , c . syncPendingJob , maxSyncRoutines , pendingCh , reportCh , errCh , pm ) c . log . Debugf ( " " , len ( triggeredCh ) ) syncProwJobs ( c . log , c . syncTriggeredJob , maxSyncRoutines , triggeredCh , reportCh , errCh , pm ) close ( errCh ) close ( reportCh ) for err := range errCh { syncErrs = append ( syncErrs , err ) } var reportErrs [ ] error if ! c . skipReport { reportTemplate := c . config ( ) . Plank . ReportTemplate reportTypes := c . config ( ) . GitHubReporter . JobTypesToReport for report := range reportCh { if err := reportlib . Report ( c . ghc , reportTemplate , report , reportTypes ) ; err != nil { reportErrs = append ( reportErrs , err ) c . log . WithFields ( pjutil . ProwJobFields ( & report ) ) . WithError ( err ) . Warn ( " " ) } // plank is not retrying on errors, so we just set the current state as reported if err := c . setPreviousReportState ( report ) ; err != nil { c . log . WithFields ( pjutil . ProwJobFields ( & report ) ) . WithError ( err ) . Error ( " " ) } } } if len ( syncErrs ) == 0 && len ( reportErrs ) == 0 { return nil } return fmt . Errorf ( " " , syncErrs , reportErrs ) }
// Sync does one sync iteration.
Sync does one sync iteration .