repospawner/internal/server/reconcile.go

package server

import (
	"context"
	"time"

	corev1 "k8s.io/api/core/v1"

	"git.unkin.net/unkin/repospawner/internal/jobs"
	"git.unkin.net/unkin/repospawner/internal/store"
)

// reconcileInterval is how often the server folds Job state into the store. The
// UI polls on the same cadence, so a change surfaces within two ticks.
const reconcileInterval = 10 * time.Second

// maxWoodpeckerTokenWaits bounds how many reconcile passes a merged request
// waits for a Woodpecker token that vanished after the request was accepted.
// Past it the request fails with the reason, rather than sitting in
// enabling-ci forever with the trouble visible only in the server's logs.
const maxWoodpeckerTokenWaits = 30

// woodpeckerTokenUnavailable is the error a request carries when the token
// never came back.
const woodpeckerTokenUnavailable = "woodpecker token unavailable"

// Run reconciles until ctx is cancelled, starting with an immediate pass so a
// restarted server rebuilds its state before serving its first request.
func (s *Server) Run(ctx context.Context) {
	if err := s.Reconcile(ctx); err != nil {
		s.log.Error("initial reconcile failed", "err", err)
	}
	t := time.NewTicker(reconcileInterval)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-t.C:
			if err := s.Reconcile(ctx); err != nil {
				s.log.Error("reconcile failed", "err", err)
			}
		}
	}
}

// Reconcile reads every repospawner Job, advances each request's state and
// creates whatever Job comes next. It is also the startup recovery path: a
// request the store has never seen is rebuilt from its Job annotations.
func (s *Server) Reconcile(ctx context.Context) error {
	jobList, err := s.cluster.ListJobs(ctx)
	if err != nil {
		return err
	}
	podList, err := s.cluster.ListPods(ctx)
	if err != nil {
		return err
	}
	results := terminationMessages(podList)

	byRequest := map[string]map[jobs.Type]jobs.View{}
	for _, j := range jobList {
		id := j.Labels[jobs.LabelRequest]
		t := jobs.Type(j.Labels[jobs.LabelType])
		if id == "" || t == "" {
			continue
		}
		v := jobs.ViewOf(j, results[resultKey{id, t}])
		if byRequest[id] == nil {
			byRequest[id] = map[jobs.Type]jobs.View{}
		}
		byRequest[id][t] = v
	}

	for id, views := range byRequest {
		current, ok := s.store.Get(id)
		if !ok {
			current = rebuild(views)
		}
		next, action := jobs.Advance(current, views)
		s.store.Put(next)
		if err := s.act(ctx, next, action); err != nil {
			s.log.Error("create follow-up job", "request", id, "action", string(action), "err", err)
		}
	}
	return nil
}

// rebuild reconstructs a request the store lost, preferring the newest job's
// annotations because those carry the pull request coordinates.
func rebuild(views map[jobs.Type]jobs.View) store.Request {
	for _, t := range []jobs.Type{jobs.TypeWoodpecker, jobs.TypeWatch, jobs.TypePR} {
		if v, ok := views[t]; ok {
			return jobs.RequestFrom(v)
		}
	}
	return store.Request{}
}

func (s *Server) act(ctx context.Context, r store.Request, action jobs.Action) error {
	switch action {
	case jobs.ActionCreateWatch:
		s.log.Info("following terraform-git pull request", "request", r.ID, "pr", r.PRNumber)
		return s.cluster.CreateJob(ctx, jobs.Watch(s.cfg, r))
	case jobs.ActionCreateWoodpecker:
		if !s.woodpeckerAvailable() {
			s.awaitWoodpeckerToken(r)
			return nil
		}
		s.forgetWoodpeckerWait(r.ID)
		s.log.Info("enabling repository in woodpecker", "request", r.ID, "name", r.Name)
		return s.cluster.CreateJob(ctx, jobs.Woodpecker(s.cfg, r))
	case jobs.ActionNone:
		return nil
	default:
		return nil
	}
}

// awaitWoodpeckerToken counts a pass spent waiting for a token that was there
// when the request was accepted, and fails the request once the wait is over.
func (s *Server) awaitWoodpeckerToken(r store.Request) {
	s.waitMu.Lock()
	s.woodpeckerWaits[r.ID]++
	waits := s.woodpeckerWaits[r.ID]
	s.waitMu.Unlock()

	if waits < maxWoodpeckerTokenWaits {
		s.log.Warn("woodpecker enablement requested but no token is mounted",
			"request", r.ID, "name", r.Name, "waits", waits)
		return
	}
	s.log.Error("failing request: woodpecker token never became available",
		"request", r.ID, "name", r.Name, "waits", waits)
	r.State = store.StateFailed
	r.Error = woodpeckerTokenUnavailable
	s.store.Put(r)
	s.forgetWoodpeckerWait(r.ID)
}

func (s *Server) forgetWoodpeckerWait(id string) {
	s.waitMu.Lock()
	defer s.waitMu.Unlock()
	delete(s.woodpeckerWaits, id)
}

type resultKey struct {
	request string
	jobType jobs.Type
}

// terminationMessages collects each job pod's termination message, preferring a
// terminated container over one that is merely waiting to restart.
func terminationMessages(pods []corev1.Pod) map[resultKey][]byte {
	out := map[resultKey][]byte{}
	for _, p := range pods {
		id := p.Labels[jobs.LabelRequest]
		t := jobs.Type(p.Labels[jobs.LabelType])
		if id == "" || t == "" {
			continue
		}
		for _, cs := range p.Status.ContainerStatuses {
			term := cs.State.Terminated
			if term == nil && cs.LastTerminationState.Terminated != nil {
				term = cs.LastTerminationState.Terminated
			}
			if term == nil || term.Message == "" {
				continue
			}
			key := resultKey{id, t}
			// A retried pod leaves several messages; the successful one wins.
			if _, seen := out[key]; !seen || term.ExitCode == 0 {
				out[key] = []byte(term.Message)
			}
		}
	}
	return out
}