Method Run

runtime/controller.go:174–392

Run starts and runs the controller's event loop. It returns when ctx is cancelled or an unrecoverable error occurs. Before returning, it closes the controller, so it must only be called once. The event loop schedules/invokes resource reconciliation and periodically flushes catalog changes to persist

(ctx context.Context)

// The implementation centers around these internal functions: enqueue, processQueue (uses markPending, trySchedule, invoke), and processCompletedInvocation.
// See their docstrings for further details.
func (c *Controller) Run(ctx context.Context) error {
	// Initially enqueue all resources
	c.mu.Lock()
	for _, rs := range c.catalog.resources {
		for _, r := range rs {
			c.enqueue(r.Meta.Name)
		}
	}
	c.mu.Unlock()

	// Ticker for periodically flushing catalog changes
	flushTicker := time.NewTicker(10 * time.Second)
	defer flushTicker.Stop()

	// Ticker for periodically checking for hanging reconciles that don't respond to cancelation
	hangingTicker := time.NewTicker(time.Minute)
	defer hangingTicker.Stop()

	// Timer for scheduling resources added to c.timeline.
	// Call resetTimelineTimer whenever the timeline may have been changed (must hold mu).
	timelineTimer := time.NewTimer(time.Second)
	defer timelineTimer.Stop()
	timelineTimer.Stop() // We want it stopped initially
	nextTime := time.Time{}
	resetTimelineTimer := func() {
		_, t := c.timeline.Peek()
		if t.Equal(nextTime) {
			return
		}
		nextTime = t

		timelineTimer.Stop()
		if t.IsZero() {
			return
		}

		d := time.Until(t)
		if d <= 0 {
			// must be positive
			d = time.Nanosecond
		}

		d += time.Second // Add a second to avoid rapid cancellations due to micro differences in schedule time
		timelineTimer.Reset(d)
	}

	// Run event loop
	var stop bool
	var loopErr error
	for !stop {
		select {
		case <-c.queueUpdatedCh: // There are resources we should schedule
			c.mu.Lock()
			err := c.processQueue()
			if err != nil {
				loopErr = err
				stop = true
			} else {