prow/tide/tide.go
func (c *Controller) Sync() error {
	start := time.Now()
	defer func() {
		duration := time.Since(start)
		c.logger.WithField("duration", duration.String()).Info("Synced")
		tideMetrics.syncDuration.Set(duration.Seconds())
	}()
	defer c.changedFiles.prune()

	c.logger.Debug("Building tide pool.")
	prs := make(map[string]PullRequest)
	for _, query := range c.config().Tide.Queries {
		q := query.Query()
		results, err := search(c.ghc.Query, c.logger, q, time.Time{}, time.Now())
		if err != nil && len(results) == 0 {
			return fmt.Errorf("query %q, err: %v", q, err)
		}
		if err != nil {
			c.logger.WithError(err).WithField("query", q).Warning("found partial results")
		}
		for _, pr := range results {
			prs[prKey(&pr)] = pr
		}
	}
	c.logger.WithField(
		"duration", time.Since(start).String(),
	).Debugf("Found %d (unfiltered) pool PRs.", len(prs))

	var pjs []prowapi.ProwJob
	var blocks blockers.Blockers
	var err error
	if len(prs) > 0 {
		start := time.Now()
		pjList, err := c.prowJobClient.List(metav1.ListOptions{LabelSelector: labels.Everything().String()})
		if err != nil {
			c.logger.WithField("duration", time.Since(start).String()).Debug("Failed to list ProwJobs from the cluster.")
			return err
		}
		c.logger.WithField("duration", time.Since(start).String()).Debug("Listed ProwJobs from the cluster.")
		pjs = pjList.Items

		if label := c.config().Tide.BlockerLabel; label != "" {
			c.logger.Debugf("Searching for blocking issues (label %q).", label)
			orgExcepts, repos := c.config().Tide.Queries.OrgExceptionsAndRepos()
			orgs := make([]string, 0, len(orgExcepts))
			for org := range orgExcepts {
				orgs = append(orgs, org)
			}
			orgRepoQuery := orgRepoQueryString(orgs, repos.UnsortedList(), orgExcepts)
			blocks, err = blockers.FindAll(c.ghc, c.logger, label, orgRepoQuery)
			if err != nil {
				return err
			}
		}
	}
	// Partition PRs into subpools and filter out non-pool PRs.
	rawPools, err := c.dividePool(prs, pjs)
	if err != nil {
		return err
	}
	filteredPools := c.filterSubpools(c.config().Tide.MaxGoroutines, rawPools)

	// Notify statusController about the new pool.
	c.sc.Lock()
	c.sc.poolPRs = poolPRMap(filteredPools)
	select {
	case c.sc.newPoolPending <- true:
	default:
	}
	c.sc.Unlock()

	// Sync subpools in parallel.
	poolChan := make(chan Pool, len(filteredPools))
	subpoolsInParallel(
		c.config().Tide.MaxGoroutines,
		filteredPools,
		func(sp *subpool) {
			pool, err := c.syncSubpool(*sp, blocks.GetApplicable(sp.org, sp.repo, sp.branch))
			if err != nil {
				sp.log.WithError(err).Errorf("Error syncing subpool.")
			}
			poolChan <- pool
		},
	)

	close(poolChan)
	pools := make([]Pool, 0, len(poolChan))
	for pool := range poolChan {
		pools = append(pools, pool)
	}
	sortPools(pools)
	c.m.Lock()
	c.pools = pools
	c.m.Unlock()

	c.History.Flush()
	return nil
}
func (c *Controller) Sync() error {
	start := time.Now()
	defer func() {
		duration := time.Since(start)
		c.logger.WithField("duration", duration.String()).Info("Synced")
		tideMetrics.syncDuration.Set(duration.Seconds())
	}()
	defer c.changedFiles.prune()

	c.logger.Debug("Building tide pool.")
	prs := make(map[string]PullRequest)
	for _, query := range c.config().Tide.Queries {
		q := query.Query()
		results, err := search(c.ghc.Query, c.logger, q, time.Time{}, time.Now())
		if err != nil && len(results) == 0 {
			return fmt.Errorf("query %q, err: %v", q, err)
		}
		if err != nil {
			c.logger.WithError(err).WithField("query", q).Warning("found partial results")
		}
		for _, pr := range results {
			prs[prKey(&pr)] = pr
		}
	}
	c.logger.WithField(
		"duration", time.Since(start).String(),
	).Debugf("Found %d (unfiltered) pool PRs.", len(prs))

	var pjs []prowapi.ProwJob
	var blocks blockers.Blockers
	var err error
	if len(prs) > 0 {
		start := time.Now()
		pjList, err := c.prowJobClient.List(metav1.ListOptions{LabelSelector: labels.Everything().String()})
		if err != nil {
			c.logger.WithField("duration", time.Since(start).String()).Debug("Failed to list ProwJobs from the cluster.")
			return err
		}
		c.logger.WithField("duration", time.Since(start).String()).Debug("Listed ProwJobs from the cluster.")
		pjs = pjList.Items

		if label := c.config().Tide.BlockerLabel; label != "" {
			c.logger.Debugf("Searching for blocking issues (label %q).", label)
			orgExcepts, repos := c.config().Tide.Queries.OrgExceptionsAndRepos()
			orgs := make([]string, 0, len(orgExcepts))
			for org := range orgExcepts {
				orgs = append(orgs, org)
			}
			orgRepoQuery := orgRepoQueryString(orgs, repos.UnsortedList(), orgExcepts)
			blocks, err = blockers.FindAll(c.ghc, c.logger, label, orgRepoQuery)
			if err != nil {
				return err
			}
		}
	}
	// Partition PRs into subpools and filter out non-pool PRs.
	rawPools, err := c.dividePool(prs, pjs)
	if err != nil {
		return err
	}
	filteredPools := c.filterSubpools(c.config().Tide.MaxGoroutines, rawPools)

	// Notify statusController about the new pool.
	c.sc.Lock()
	c.sc.poolPRs = poolPRMap(filteredPools)
	select {
	case c.sc.newPoolPending <- true:
	default:
	}
	c.sc.Unlock()

	// Sync subpools in parallel.
	poolChan := make(chan Pool, len(filteredPools))
	subpoolsInParallel(
		c.config().Tide.MaxGoroutines,
		filteredPools,
		func(sp *subpool) {
			pool, err := c.syncSubpool(*sp, blocks.GetApplicable(sp.org, sp.repo, sp.branch))
			if err != nil {
				sp.log.WithError(err).Errorf("Error syncing subpool.")
			}
			poolChan <- pool
		},
	)

	close(poolChan)
	pools := make([]Pool, 0, len(poolChan))
	for pool := range poolChan {
		pools = append(pools, pool)
	}
	sortPools(pools)
	c.m.Lock()
	c.pools = pools
	c.m.Unlock()

	c.History.Flush()
	return nil
}
// Sync runs one sync iteration.