Method Start

runner/pool/pool.go:1846–1899

()

}

func (r *basePoolManager) Start() error {
	r.SetPoolRunningState(false, "pool manager is starting up")
	initializeEntity := make(chan struct{}, 1)
	go func() {
		// At this point, we just mark instances in `creating` and `deleting` as `pending_delete`
		// These instances were interrupted while transitioning, so they are in a state we need
		// to recover from. The call to DeleteRunner() will potentially remove them from github
		// if using JIT, and mark them for deleting. The actual deletion will happen only after
		// tools are fetched and the loops start.
		slog.InfoContext(r.ctx, "cleaning up instances that were interrupted in flight")
		if err := r.cleanupRunnersInterruptedInFlight(); err != nil {
			slog.ErrorContext(r.ctx, "startup runner cleanup failed", "error", err)
		}

		slog.Info("running initial tool update")
		for {
			slog.DebugContext(r.ctx, "waiting for tools to be available")
			hasTools, stopped := r.waitForToolsOrCancel()
			if stopped {
				return
			}
			if hasTools {
				break
			}
		}
		if err := r.updateTools(); err != nil {
			slog.With(slog.Any("error", err)).Error("failed to update tools")
		}
		initializeEntity <- struct{}{}
	}()

	go r.runWatcher()
	go func() {
		select {
		case <-r.quit:
			return
		case <-r.ctx.Done():
			return
		case <-initializeEntity:
		}
		defer close(initializeEntity)
		go r.startLoopForFunction(r.runnerCleanup, common.PoolReapTimeoutInterval, "timeout_reaper", false)
		go r.startLoopForFunction(r.scaleDown, common.PoolScaleDownInterval, "scale_down", false)
		// always run the delete pending instances routine. This way we can still remove existing runners, even if the pool is not running.
		go r.startLoopForFunction(r.deletePendingInstances, common.PoolConsilitationInterval, "consolidate[delete_pending]", true)
		go r.startLoopForFunction(r.addPendingInstances, common.PoolConsilitationInterval, "consolidate[add_pending]", false)
		go r.startLoopForFunction(r.ensureMinIdleRunners, common.PoolConsilitationInterval, "consolidate[ensure_min_idle]", false)
		go r.startLoopForFunction(r.retryFailedInstances, common.PoolConsilitationInterval, "consolidate[retry_failed]", false)
		go r.startLoopForFunction(r.updateTools, common.PoolToolUpdateInterval, "update_tools", true)
		go r.startLoopForFunction(r.consumeQueuedJobs, common.PoolConsilitationInterval, "job_queue_consumer", false)
		go r.startLoopForFunction(r.reconcileStaleJobs, common.PoolStaleJobReconcileInterval, "stale_job_reconciler", false)
	}()
	return nil
}

func (r *basePoolManager) Stop() error {
	close(r.quit)