feat: master/agent phase 4 — registry polling and version history (#9) (#23)

internal/master/poller/poller.go

@@ -0,0 +1,320 @@
+// Package poller checks the registry for new digests of the tags the
+// services follow. A new digest becomes the service's available version, and
+// a service in auto mode gets an update job.
+package poller
+
+import (
+	"context"
+	"errors"
+	"log/slog"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/shcizo/package-updater/internal/master/events"
+	"github.com/shcizo/package-updater/internal/master/jobs"
+	"github.com/shcizo/package-updater/internal/master/model"
+)
+
+// lookupTimeout bounds one registry lookup.
+const lookupTimeout = 30 * time.Second
+
+// maxErrorBytes caps the registry error stored in services.last_poll_error.
+const maxErrorBytes = 500
+
+// Store is the storage the poller needs. *store.Store satisfies it.
+type Store interface {
+	ListServices(ctx context.Context) ([]model.Service, error)
+	GetService(ctx context.Context, id int64) (model.Service, error)
+	SetAvailable(ctx context.Context, id int64, a model.Available) error
+	SetLastTagDigest(ctx context.Context, id int64, digest string) error
+	SetPollError(ctx context.Context, ids []int64, msg string) error
+}
+
+// Jobs starts updates. *jobs.Service satisfies it.
+type Jobs interface {
+	Start(ctx context.Context, svc model.Service, req jobs.StartRequest) (model.Job, error)
+}
+
+// Registry looks up the digest a "repo:tag" ref points at. *registry.Client
+// satisfies it.
+type Registry interface {
+	Digest(ctx context.Context, ref string) (string, error)
+}
+
+// Publisher receives change signals. *events.Broker satisfies it.
+type Publisher interface{ Publish(t events.Topic) }
+
+// Config holds the poller settings.
+type Config struct {
+	Interval   time.Duration // time between the end of a cycle and the start of the next (POLL_INTERVAL)
+	FirstDelay time.Duration // wait before the first cycle; 1 minute in production
+	Workers    int           // registry lookups in flight at once; 4 in production
+}
+
+// Poller runs polling cycles.
+type Poller struct {
+	store Store
+	jobs  Jobs
+	reg   Registry
+	pub   Publisher
+	cfg   Config
+	log   *slog.Logger
+	now   func() time.Time
+}
+
+// New returns a Poller.
+func New(st Store, j Jobs, reg Registry, pub Publisher, cfg Config, log *slog.Logger) *Poller {
+	if cfg.Workers < 1 {
+		cfg.Workers = 1
+	}
+	return &Poller{store: st, jobs: j, reg: reg, pub: pub, cfg: cfg, log: log, now: time.Now}
+}
+
+// SkipReason says why svc is not polled: "follows no tag", "built from
+// source", "no registry digest", or "" when it is polled.
+func SkipReason(svc model.Service) string {
+	switch {
+	case jobs.FollowedRef(svc.ActualRef) == "":
+		return model.ReasonNoTag
+	case svc.Built:
+		return "built from source"
+	case svc.ActualDigest == "":
+		return "no registry digest"
+	}
+	return ""
+}
+
+// Run polls until ctx is done. The first cycle starts Config.FirstDelay after
+// the call and each following one Config.Interval after the previous cycle
+// ended, so cycles never overlap. It blocks.
+func (p *Poller) Run(ctx context.Context) {
+	timer := time.NewTimer(p.cfg.FirstDelay)
+	defer timer.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-timer.C:
+		}
+		p.Cycle(ctx)
+		timer.Reset(p.cfg.Interval)
+	}
+}
+
+// lookup is the registry's answer for one followed ref.
+type lookup struct {
+	ref    string
+	digest string
+	err    error
+}
+
+// cycleStats is what one cycle did, for the poll_cycle log line.
+type cycleStats struct {
+	refs, ok, failed, suggested, jobs, skipped int
+	wrote                                      bool
+}
+
+// Cycle runs one poll over all services. It looks up every distinct followed
+// ref once, with at most Config.Workers lookups at a time, then records the
+// results. Writing results and starting jobs is not cut short by ctx ending
+// during the lookups, so a job that was marked is always started.
+func (p *Poller) Cycle(ctx context.Context) {
+	began := p.now()
+	svcs, err := p.store.ListServices(ctx)
+	if err != nil {
+		p.log.Error("poll_list_failed", "error", err)
+		return
+	}
+	var stats cycleStats
+	byRef := map[string][]model.Service{}
+	var refs []string
+	for _, svc := range svcs {
+		if svc.RemovedAt != nil {
+			continue
+		}
+		if SkipReason(svc) != "" {
+			stats.skipped++
+			continue
+		}
+		ref := jobs.FollowedRef(svc.ActualRef)
+		if _, seen := byRef[ref]; !seen {
+			refs = append(refs, ref)
+		}
+		byRef[ref] = append(byRef[ref], svc)
+	}
+	stats.refs = len(refs)
+
+	write := context.WithoutCancel(ctx)
+	for _, l := range p.lookupAll(ctx, refs) {
+		if l.err != nil && ctx.Err() != nil {
+			continue // cancelled mid-lookup: not the registry's fault
+		}
+		p.record(write, l, byRef[l.ref], &stats)
+	}
+	if stats.wrote {
+		p.pub.Publish(events.Services)
+	}
+	p.log.Info("poll_cycle", "refs", stats.refs, "ok", stats.ok, "failed", stats.failed,
+		"suggested", stats.suggested, "jobs", stats.jobs, "skipped", stats.skipped,
+		"duration_ms", p.now().Sub(began).Milliseconds())
+}
+
+// lookupAll asks the registry for every ref, Config.Workers at a time. It
+// returns the answers in ref order and stops starting new lookups once ctx
+// is done; refs never looked up are missing from the result.
+func (p *Poller) lookupAll(ctx context.Context, refs []string) []lookup {
+	results := make([]*lookup, len(refs))
+	next := make(chan int)
+	var wg sync.WaitGroup
+	for w := 0; w < p.cfg.Workers; w++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for i := range next {
+				lctx, cancel := context.WithTimeout(ctx, lookupTimeout)
+				d, err := p.reg.Digest(lctx, refs[i])
+				cancel()
+				results[i] = &lookup{ref: refs[i], digest: d, err: err}
+			}
+		}()
+	}
+feed:
+	for i := range refs {
+		select {
+		case next <- i:
+		case <-ctx.Done():
+			break feed
+		}
+	}
+	close(next)
+	wg.Wait()
+	var out []lookup
+	for _, r := range results {
+		if r != nil {
+			out = append(out, *r)
+		}
+	}
+	return out
+}
+
+// record stores one lookup result for the services that follow l.ref.
+func (p *Poller) record(ctx context.Context, l lookup, svcs []model.Service, stats *cycleStats) {
+	if l.err != nil {
+		stats.failed++
+		msg := truncate(l.err.Error())
+		p.log.Warn("poll_failed", "ref", l.ref, "error", msg)
+		p.setPollError(ctx, svcs, msg, stats, func(s model.Service) bool { return s.LastPollError != msg })
+		return
+	}
+	stats.ok++
+	p.setPollError(ctx, svcs, "", stats, func(s model.Service) bool { return s.LastPollError != "" })
+	for _, svc := range svcs {
+		if !p.suggest(ctx, svc, l, stats) {
+			continue
+		}
+		if autoWanted(svc, l.ref, l.digest) {
+			p.startAuto(ctx, svc.ID, l.ref, l.digest, stats)
+		}
+	}
+}
+
+// suggest marks l.digest as the available version of svc unless the service
+// already runs it or already has exactly that suggestion. A service that
+// runs l.digest also gets it as last_tag_digest. It returns false only when
+// writing the suggestion failed.
+func (p *Poller) suggest(ctx context.Context, svc model.Service, l lookup, stats *cycleStats) bool {
+	if svc.ActualDigest == l.digest && svc.LastTagDigest != l.digest {
+		// Seed last_tag_digest for a service that was not updated by a tag job
+		// (it predates the column or was updated by hand), so that a later
+		// rollback is not undone.
+		if err := p.store.SetLastTagDigest(ctx, svc.ID, l.digest); err != nil {
+			p.log.Error("poll_set_last_tag_digest_failed", "service_id", svc.ID, "error", err)
+		}
+	}
+	if svc.ActualDigest == l.digest || (svc.AvailableRef == l.ref && svc.AvailableDigest == l.digest) {
+		return true
+	}
+	err := p.store.SetAvailable(ctx, svc.ID, model.Available{Ref: l.ref, Digest: l.digest, Source: model.SourcePoll, At: p.now()})
+	if err != nil {
+		p.log.Error("poll_set_available_failed", "service_id", svc.ID, "ref", l.ref, "error", err)
+		return false
+	}
+	stats.wrote = true
+	stats.suggested++
+	return true
+}
+
+// autoWanted reports whether an update of svc to ref at digest should be
+// started automatically: the service is in auto mode, not removed, does not
+// already run that digest, was not already updated automatically to it
+// (otherwise a rollback to an older version would be undone at the next
+// cycle), and its last job did not already deploy it (desired_ref holds the
+// digest until the agent reports the new inventory).
+func autoWanted(svc model.Service, ref, digest string) bool {
+	return svc.RemovedAt == nil &&
+		svc.UpdateMode == model.ModeAuto &&
+		svc.ActualDigest != digest &&
+		svc.LastTagDigest != digest &&
+		svc.DesiredRef != ref+"@"+digest
+}
+
+// startAuto starts the update job of an auto service to ref and remembers
+// digest as services.last_tag_digest, so the same digest is not started
+// again, for instance after the user rolled back. It re-reads the service
+// first, because a cycle can run for a long time and the user may have
+// switched the service to manual meanwhile. A service that already has a
+// job or cannot be updated is not an error for the cycle, and last_tag_digest
+// is left alone so the next cycle tries again.
+func (p *Poller) startAuto(ctx context.Context, id int64, ref, digest string, stats *cycleStats) {
+	svc, err := p.store.GetService(ctx, id)
+	if err != nil {
+		p.log.Warn("poll_get_service_failed", "service_id", id, "error", err)
+		return
+	}
+	if !autoWanted(svc, ref, digest) {
+		return
+	}
+	_, err = p.jobs.Start(ctx, svc, jobs.StartRequest{Trigger: model.TriggerPoll, Ref: ref})
+	var notUpdatable *model.NotUpdatableError
+	switch {
+	case err == nil:
+		stats.jobs++
+		if err := p.store.SetLastTagDigest(ctx, id, digest); err != nil {
+			p.log.Error("poll_set_last_tag_digest_failed", "service_id", id, "error", err)
+		}
+	case errors.Is(err, model.ErrJobActive), errors.As(err, &notUpdatable):
+		p.log.Debug("poll_job_not_started", "service_id", id, "ref", ref, "reason", err.Error())
+	default:
+		p.log.Warn("poll_job_failed", "service_id", id, "ref", ref, "error", err)
+	}
+}
+
+// setPollError sets msg as last_poll_error on the services for which
+// changed returns true; with none, it writes nothing.
+func (p *Poller) setPollError(ctx context.Context, svcs []model.Service, msg string, stats *cycleStats, changed func(model.Service) bool) {
+	var ids []int64
+	for _, svc := range svcs {
+		if changed(svc) {
+			ids = append(ids, svc.ID)
+		}
+	}
+	if len(ids) == 0 {
+		return
+	}
+	if err := p.store.SetPollError(ctx, ids, msg); err != nil {
+		p.log.Error("poll_set_error_failed", "error", err)
+		return
+	}
+	stats.wrote = true
+}
+
+// truncate cuts msg to maxErrorBytes without leaving a broken UTF-8
+// sequence, and drops NUL bytes, which Postgres text cannot hold.
+func truncate(msg string) string {
+	msg = strings.ReplaceAll(msg, "\x00", "")
+	if len(msg) > maxErrorBytes {
+		msg = strings.ToValidUTF8(msg[:maxErrorBytes], "")
+	}
+	return msg
+}