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

// 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.log.Warn("woodpecker enablement requested but no token is mounted", "request", r.ID)
			return nil
		}
		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
	}
}

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
}