From af61f17dee4009dcef16caafb8e053eb6436d530 Mon Sep 17 00:00:00 2001 From: Samuel Enocsson Date: Tue, 13 Oct 2026 08:05:21 +0200 Subject: [PATCH 01/14] feat(master): record image versions and track poll suggestions in the store (#9) --- internal/e2e/agent_master_test.go | 54 +++++- internal/master/model/model.go | 31 +++- internal/master/model/model_test.go | 18 ++ internal/master/store/jobs.go | 19 ++- .../master/store/migrations/0004_polling.sql | 19 +++ internal/master/store/services.go | 103 ++++++++++- .../master/store/versions_integration_test.go | 160 ++++++++++++++++++ .../master/store/webhook_integration_test.go | 35 ++-- 8 files changed, 403 insertions(+), 36 deletions(-) create mode 100644 internal/master/store/migrations/0004_polling.sql create mode 100644 internal/master/store/versions_integration_test.go diff --git a/internal/e2e/agent_master_test.go b/internal/e2e/agent_master_test.go index 9036eb1..2974cf8 100644 --- a/internal/e2e/agent_master_test.go +++ b/internal/e2e/agent_master_test.go @@ -44,6 +44,51 @@ type memStore struct { jobs map[int64]*model.Job nextJob int64 finishApplied int // FinishJob calls that returned true + versions map[int64]model.ImageVersion +} + +// seedVersion adds an image_versions row that GetImageVersion returns. +func (m *memStore) seedVersion(v model.ImageVersion) { + m.mu.Lock() + defer m.mu.Unlock() + if m.versions == nil { + m.versions = map[int64]model.ImageVersion{} + } + m.versions[v.ID] = v +} + +func (m *memStore) GetImageVersion(_ context.Context, id int64) (model.ImageVersion, error) { + m.mu.Lock() + defer m.mu.Unlock() + v, ok := m.versions[id] + if !ok { + return model.ImageVersion{}, model.ErrNotFound + } + return v, nil +} + +func (m *memStore) ListServices(_ context.Context) ([]model.Service, error) { + m.mu.Lock() + defer m.mu.Unlock() + var out []model.Service + for _, svc := range m.svcRows { + if svc.RemovedAt == nil { + out = append(out, svc) + } + } + sort.Slice(out, func(i, j int) bool { return out[i].ID < out[j].ID }) + return out, nil +} + +func (m *memStore) SetPollError(_ context.Context, ids []int64, msg string) error { + m.mu.Lock() + defer m.mu.Unlock() + for _, id := range ids { + svc := m.svcRows[id] + svc.LastPollError = msg + m.svcRows[id] = svc + } + return nil } // seedService adds a service row that jobs can be triggered for. @@ -88,7 +133,7 @@ func (m *memStore) SetAvailable(_ context.Context, id int64, a model.Available) if !ok { return model.ErrNotFound } - svc.AvailableRef, svc.AvailableSource, svc.AvailableAt = a.Ref, a.Source, &a.At + svc.AvailableRef, svc.AvailableDigest, svc.AvailableSource, svc.AvailableAt = a.Ref, a.Digest, a.Source, &a.At m.svcRows[id] = svc return nil } @@ -185,9 +230,10 @@ func (m *memStore) FinishJob(_ context.Context, agentID, jobID int64, f model.Jo svc := m.svcRows[j.ServiceID] svc.DesiredRef = f.Ref // Same rule as the SQL: the job was created for the suggested ref after it was reported. - requested, _, _ := strings.Cut(requestedRef, "@") - if svc.AvailableAt != nil && svc.AvailableRef == requested && !svc.AvailableAt.After(j.CreatedAt) { - svc.AvailableRef, svc.AvailableSource, svc.AvailableAt = "", "", nil + requested, digest, _ := strings.Cut(requestedRef, "@") + if svc.AvailableAt != nil && svc.AvailableRef == requested && !svc.AvailableAt.After(j.CreatedAt) && + (digest == "" || svc.AvailableDigest == digest) { + svc.AvailableRef, svc.AvailableDigest, svc.AvailableSource, svc.AvailableAt = "", "", "", nil } m.svcRows[j.ServiceID] = svc } diff --git a/internal/master/model/model.go b/internal/master/model/model.go index 3b4a8ff..975538c 100644 --- a/internal/master/model/model.go +++ b/internal/master/model/model.go @@ -8,6 +8,7 @@ import ( "strings" "time" + "github.com/shcizo/package-updater/internal/imageref" "github.com/shcizo/package-updater/internal/protocol" ) @@ -56,8 +57,10 @@ type Service struct { // AvailableRef is the newest ref a webhook or poll reported for this // service; "" if none, or if a later job already deployed it. AvailableRef string - AvailableSource string // SourceWebhook; "" when AvailableRef is empty + AvailableDigest string // digest of AvailableRef when known (poll); "" from the webhook + AvailableSource string // SourceWebhook or SourcePoll; "" when AvailableRef is empty AvailableAt *time.Time + LastPollError string // last registry error for the followed ref; "" when fine FirstSeenAt time.Time UpdatedAt time.Time RemovedAt *time.Time @@ -69,18 +72,40 @@ const ( ModeAuto = "auto" ) -// Sources of an available ref. +// Sources of an available ref and of an ImageVersion. const ( SourceWebhook = "webhook" + SourcePoll = "poll" + // SourceRunning is used by image_versions only: an agent reported the + // version as running. + SourceRunning = "running" ) // Available is a newer ref reported for a service. type Available struct { Ref string - Source string // SourceWebhook + Digest string // "" when unknown (webhook) + Source string // SourceWebhook or SourcePoll At time.Time } +// ImageVersion is one version of an image repo the master has seen. Tag and +// Digest are "" when the version has no tag or its digest is unknown; they +// are never both empty. +type ImageVersion struct { + ID int64 + Repo string + Tag string + Digest string + Source string // SourceWebhook, SourcePoll or SourceRunning: the first source that saw it + FirstSeenAt time.Time +} + +// Ref returns "repo:tag@digest", "repo:tag" or "repo@digest". +func (v ImageVersion) Ref() string { + return imageref.Ref{Repo: v.Repo, Tag: v.Tag, Digest: v.Digest}.String() +} + // DisplayName is the human-facing name of the service. func (s Service) DisplayName() string { switch s.Kind { diff --git a/internal/master/model/model_test.go b/internal/master/model/model_test.go index 48c7947..96cb8af 100644 --- a/internal/master/model/model_test.go +++ b/internal/master/model/model_test.go @@ -40,6 +40,24 @@ func TestDisplayName(t *testing.T) { } } +func TestImageVersionRef(t *testing.T) { + cases := []struct { + name string + v ImageVersion + want string + }{ + {"tag and digest", ImageVersion{Repo: "r/x", Tag: "1", Digest: "sha256:a"}, "r/x:1@sha256:a"}, + {"tag only", ImageVersion{Repo: "r/x", Tag: "1"}, "r/x:1"}, + {"digest only", ImageVersion{Repo: "r/x", Digest: "sha256:a"}, "r/x@sha256:a"}, + {"registry port", ImageVersion{Repo: "localhost:5000/x", Tag: "1"}, "localhost:5000/x:1"}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + assert.Equal(t, c.want, c.v.Ref()) + }) + } +} + func TestJob_Active(t *testing.T) { cases := []struct { status JobStatus diff --git a/internal/master/store/jobs.go b/internal/master/store/jobs.go index 31a9da9..1789060 100644 --- a/internal/master/store/jobs.go +++ b/internal/master/store/jobs.go @@ -189,9 +189,11 @@ func (s *Store) JobProgress(ctx context.Context, agentID, jobID int64, chunk str // started_at (the agent's, or the existing one if the agent sent none), // finished_at; to_ref = f.Ref when set. On success it also sets the // service's desired_ref = f.Ref and clears its available ref when that is -// the ref the job was created for (ignoring any @digest) and was reported no -// later than the job was created (that version is now running), in the same -// transaction. Returns false if no +// the ref the job was created for and was reported no later than the job was +// created (that version is now running), in the same transaction. A job +// created for "repo:tag@digest" clears only a suggestion with that same +// digest, so rolling back to an older digest keeps a newer suggestion (and a +// suggestion without a digest, such as a webhook's). Returns false if no // such active job belongs to agentID. func (s *Store) FinishJob(ctx context.Context, agentID, jobID int64, f model.JobFinish) (bool, error) { tx, err := s.pool.Begin(ctx) @@ -226,13 +228,14 @@ func (s *Store) FinishJob(ctx context.Context, agentID, jobID int64, f model.Job return false, fmt.Errorf("finish job: %w", err) } if f.Status == model.JobSucceeded { - requestedRef, _, _ = strings.Cut(requestedRef, "@") + requestedRef, requestedDigest, _ := strings.Cut(requestedRef, "@") if _, err := tx.Exec(ctx, ` UPDATE services SET desired_ref = $2, - available_ref = CASE WHEN available_ref = $4 AND available_at <= $3 THEN '' ELSE available_ref END, - available_source = CASE WHEN available_ref = $4 AND available_at <= $3 THEN '' ELSE available_source END, - available_at = CASE WHEN available_ref = $4 AND available_at <= $3 THEN NULL ELSE available_at END - WHERE id = $1`, serviceID, f.Ref, createdAt, requestedRef); err != nil { + available_ref = CASE WHEN available_ref = $4 AND available_at <= $3 AND ($5 = '' OR available_digest = $5) THEN '' ELSE available_ref END, + available_digest = CASE WHEN available_ref = $4 AND available_at <= $3 AND ($5 = '' OR available_digest = $5) THEN '' ELSE available_digest END, + available_source = CASE WHEN available_ref = $4 AND available_at <= $3 AND ($5 = '' OR available_digest = $5) THEN '' ELSE available_source END, + available_at = CASE WHEN available_ref = $4 AND available_at <= $3 AND ($5 = '' OR available_digest = $5) THEN NULL ELSE available_at END + WHERE id = $1`, serviceID, f.Ref, createdAt, requestedRef, requestedDigest); err != nil { return false, fmt.Errorf("finish job: desired ref: %w", err) } } diff --git a/internal/master/store/migrations/0004_polling.sql b/internal/master/store/migrations/0004_polling.sql new file mode 100644 index 0000000..a1da48d --- /dev/null +++ b/internal/master/store/migrations/0004_polling.sql @@ -0,0 +1,19 @@ +ALTER TABLE services DROP CONSTRAINT services_available_source_check; +ALTER TABLE services ADD CONSTRAINT services_available_source_check + CHECK (available_source IN ('', 'webhook', 'poll')); +ALTER TABLE services ADD COLUMN available_digest text NOT NULL DEFAULT ''; +ALTER TABLE services ADD COLUMN last_poll_error text NOT NULL DEFAULT ''; + +-- Every version the master has seen. '' means "no tag" / "digest unknown"; +-- never NULL, so the unique constraint deduplicates. +CREATE TABLE image_versions ( + id bigserial PRIMARY KEY, + repo text NOT NULL, + tag text NOT NULL DEFAULT '', + digest text NOT NULL DEFAULT '', + source text NOT NULL CHECK (source IN ('webhook', 'poll', 'running')), + first_seen_at timestamptz NOT NULL, + CONSTRAINT image_versions_tag_or_digest CHECK (tag <> '' OR digest <> ''), + CONSTRAINT image_versions_unique UNIQUE (repo, tag, digest) +); +CREATE INDEX image_versions_recent ON image_versions (repo, first_seen_at DESC, id DESC); diff --git a/internal/master/store/services.go b/internal/master/store/services.go index a71f6d5..0649086 100644 --- a/internal/master/store/services.go +++ b/internal/master/store/services.go @@ -70,6 +70,25 @@ func upsertService(ctx context.Context, db execer, agentID int64, clusterID stri if err != nil { return fmt.Errorf("upsert service %s: %w", svc.Key(), err) } + if svc.Digest != "" { + ref := imageref.Parse(svc.Image) + if err := insertVersion(ctx, db, ref.Repo, ref.Tag, svc.Digest, model.SourceRunning, at); err != nil { + return err + } + } + return nil +} + +// insertVersion records a version in image_versions unless the same +// (repo, tag, digest) is already there; the first sighting keeps its source +// and first_seen_at. +func insertVersion(ctx context.Context, db execer, repo, tag, digest, source string, at time.Time) error { + if _, err := db.Exec(ctx, ` + INSERT INTO image_versions (repo, tag, digest, source, first_seen_at) + VALUES ($1, $2, $3, $4, $5) ON CONFLICT DO NOTHING`, + repo, tag, digest, source, at); err != nil { + return fmt.Errorf("record version %s:%s: %w", repo, tag, err) + } return nil } @@ -149,8 +168,8 @@ func (s *Store) RemoveService(ctx context.Context, agentID int64, clusterID, age const serviceColumns = `s.id, s.kind, s.location_key, s.agent_id, COALESCE(a.name, ''), s.compose_project, s.compose_service, s.working_dir, s.config_files, s.built, s.swarm_cluster_id, s.swarm_service, s.container_name, s.image_repo, s.desired_ref, s.actual_ref, s.actual_digest, - s.status, s.self, s.update_mode, s.available_ref, s.available_source, s.available_at, - s.first_seen_at, s.updated_at, s.removed_at` + s.status, s.self, s.update_mode, s.available_ref, s.available_digest, s.available_source, s.available_at, + s.last_poll_error, s.first_seen_at, s.updated_at, s.removed_at` func scanService(row pgx.Row) (model.Service, error) { var m model.Service @@ -158,8 +177,8 @@ func scanService(row pgx.Row) (model.Service, error) { err := row.Scan(&m.ID, &kind, &m.LocationKey, &m.AgentID, &m.AgentName, &m.ComposeProject, &m.ComposeService, &m.WorkingDir, &m.ConfigFiles, &m.Built, &m.SwarmClusterID, &m.SwarmService, &m.ContainerName, &m.ImageRepo, &m.DesiredRef, &m.ActualRef, &m.ActualDigest, - &m.Status, &m.Self, &m.UpdateMode, &m.AvailableRef, &m.AvailableSource, &m.AvailableAt, - &m.FirstSeenAt, &m.UpdatedAt, &m.RemovedAt) + &m.Status, &m.Self, &m.UpdateMode, &m.AvailableRef, &m.AvailableDigest, &m.AvailableSource, &m.AvailableAt, + &m.LastPollError, &m.FirstSeenAt, &m.UpdatedAt, &m.RemovedAt) m.Kind = protocol.Kind(kind) return m, err } @@ -198,16 +217,84 @@ func (s *Store) queryServices(ctx context.Context, what, extraWhere string, args } // SetAvailable records a newer ref for the service, replacing any earlier -// suggestion. +// suggestion, and records the version in image_versions in the same +// transaction (the tag is parsed from a.Ref). func (s *Store) SetAvailable(ctx context.Context, id int64, a model.Available) error { - if _, err := s.pool.Exec(ctx, ` - UPDATE services SET available_ref = $2, available_source = $3, available_at = $4 - WHERE id = $1`, id, a.Ref, a.Source, a.At); err != nil { + tx, err := s.pool.Begin(ctx) + if err != nil { + return fmt.Errorf("set available: begin: %w", err) + } + defer func() { _ = tx.Rollback(ctx) }() + if _, err := tx.Exec(ctx, ` + UPDATE services SET available_ref = $2, available_digest = $3, available_source = $4, available_at = $5 + WHERE id = $1`, id, a.Ref, a.Digest, a.Source, a.At); err != nil { return fmt.Errorf("set available: %w", err) } + ref := imageref.Parse(a.Ref) + if err := insertVersion(ctx, tx, ref.Repo, ref.Tag, a.Digest, a.Source, a.At); err != nil { + return err + } + if err := tx.Commit(ctx); err != nil { + return fmt.Errorf("set available: commit: %w", err) + } return nil } +// SetPollError sets last_poll_error on the given services; msg "" clears it. +func (s *Store) SetPollError(ctx context.Context, ids []int64, msg string) error { + if len(ids) == 0 { + return nil + } + if _, err := s.pool.Exec(ctx, + `UPDATE services SET last_poll_error = $2 WHERE id = ANY($1)`, ids, msg); err != nil { + return fmt.Errorf("set poll error: %w", err) + } + return nil +} + +// RecentVersions returns up to limit versions of repo, most recently first +// seen first. A version without a digest is left out when a version with the +// same tag and a digest exists. +func (s *Store) RecentVersions(ctx context.Context, repo string, limit int) ([]model.ImageVersion, error) { + rows, err := s.pool.Query(ctx, ` + SELECT id, repo, tag, digest, source, first_seen_at FROM image_versions v + WHERE v.repo = $1 + AND NOT (v.digest = '' AND EXISTS ( + SELECT 1 FROM image_versions d + WHERE d.repo = v.repo AND d.tag = v.tag AND d.digest <> '')) + ORDER BY v.first_seen_at DESC, v.id DESC + LIMIT $2`, repo, limit) + if err != nil { + return nil, fmt.Errorf("recent versions: %w", err) + } + defer rows.Close() + var out []model.ImageVersion + for rows.Next() { + var v model.ImageVersion + if err := rows.Scan(&v.ID, &v.Repo, &v.Tag, &v.Digest, &v.Source, &v.FirstSeenAt); err != nil { + return nil, fmt.Errorf("scan version: %w", err) + } + out = append(out, v) + } + return out, rows.Err() +} + +// GetImageVersion returns one version. Returns model.ErrNotFound if the ID +// does not exist. +func (s *Store) GetImageVersion(ctx context.Context, id int64) (model.ImageVersion, error) { + var v model.ImageVersion + err := s.pool.QueryRow(ctx, ` + SELECT id, repo, tag, digest, source, first_seen_at FROM image_versions WHERE id = $1`, id, + ).Scan(&v.ID, &v.Repo, &v.Tag, &v.Digest, &v.Source, &v.FirstSeenAt) + if errors.Is(err, pgx.ErrNoRows) { + return model.ImageVersion{}, model.ErrNotFound + } + if err != nil { + return model.ImageVersion{}, fmt.Errorf("get image version: %w", err) + } + return v, nil +} + // SetUpdateMode sets the service's update mode (model.ModeManual or // model.ModeAuto). Returns model.ErrNotFound if the service does not exist // or is marked removed. diff --git a/internal/master/store/versions_integration_test.go b/internal/master/store/versions_integration_test.go new file mode 100644 index 0000000..5d86c98 --- /dev/null +++ b/internal/master/store/versions_integration_test.go @@ -0,0 +1,160 @@ +//go:build integration + +package store + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/shcizo/package-updater/internal/master/model" + "github.com/shcizo/package-updater/internal/protocol" +) + +const sluskRepo = "gitea.shcizo.se/shcizo/slusk" + +func versionsOf(t *testing.T, s *Store, repo string) []model.ImageVersion { + t.Helper() + got, err := s.RecentVersions(context.Background(), repo, 100) + require.NoError(t, err) + return got +} + +func TestSetAvailable_RecordsVersion(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + _, svc := jobFixture(t, s, "a") + + require.NoError(t, s.SetAvailable(ctx, svc, model.Available{Ref: sluskRepo + ":1.5.0", Digest: "sha256:n1", Source: model.SourcePoll, At: t0})) + + got, err := s.GetService(ctx, svc) + require.NoError(t, err) + assert.Equal(t, sluskRepo+":1.5.0", got.AvailableRef) + assert.Equal(t, "sha256:n1", got.AvailableDigest) + assert.Equal(t, model.SourcePoll, got.AvailableSource) + vs := versionsOf(t, s, sluskRepo) + var found model.ImageVersion + for _, v := range vs { + if v.Digest == "sha256:n1" { + found = v + } + } + require.NotZero(t, found.ID) + assert.Equal(t, "1.5.0", found.Tag) + assert.Equal(t, model.SourcePoll, found.Source) + + // Seeing the same version again keeps the first sighting. + require.NoError(t, s.SetAvailable(ctx, svc, model.Available{Ref: sluskRepo + ":1.5.0", Digest: "sha256:n1", Source: model.SourceWebhook, At: t0.Add(time.Hour)})) + again, err := s.GetImageVersion(ctx, found.ID) + require.NoError(t, err) + assert.Equal(t, model.SourcePoll, again.Source) + assert.True(t, t0.Equal(again.FirstSeenAt)) + assert.Len(t, versionsOf(t, s, sluskRepo), len(vs)) +} + +func TestSnapshotRecordsRunningVersions(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + id := newAgent(t, s, "node-2") + tagless := protocol.Service{Kind: protocol.KindSwarm, SwarmService: "g_g", Image: "gitea/gitea@sha256:87a6", Digest: "sha256:87a6", Status: "running"} + bare := protocol.Service{Kind: protocol.KindContainer, ContainerName: "pg", Image: "postgres", Digest: "sha256:pg1", Status: "running"} + nodigest := protocol.Service{Kind: protocol.KindContainer, ContainerName: "nd", Image: "nodigest:1", Status: "running"} + hello := protocol.Hello{Hostname: "node-2", SwarmManager: true, SwarmClusterID: "c1"} + svcs := []protocol.Service{slusk, tagless, bare, nodigest} + + require.NoError(t, s.ApplySnapshot(ctx, id, hello, svcs, t0)) + require.NoError(t, s.ApplySnapshot(ctx, id, hello, svcs, t0.Add(time.Hour))) + + got := versionsOf(t, s, sluskRepo) + require.Len(t, got, 1) + assert.Equal(t, "1.4.2", got[0].Tag) + assert.Equal(t, "sha256:d1", got[0].Digest) + assert.Equal(t, model.SourceRunning, got[0].Source) + assert.True(t, t0.Equal(got[0].FirstSeenAt), "re-snapshot does not bump first_seen_at") + tl := versionsOf(t, s, "gitea/gitea") + require.Len(t, tl, 1) + assert.Equal(t, "", tl[0].Tag) + assert.Equal(t, "sha256:87a6", tl[0].Digest) + b := versionsOf(t, s, "postgres") + require.Len(t, b, 1) + assert.Equal(t, "", b[0].Tag, "a bare image is recorded without a tag, not as latest") + assert.Empty(t, versionsOf(t, s, "nodigest")) +} + +func TestRecentVersions(t *testing.T) { + ctx := context.Background() + t.Run("order, limit and repo filter", func(t *testing.T) { + s := openTestStore(t) + _, svc := jobFixture(t, s, "a") + require.NoError(t, s.SetAvailable(ctx, svc, model.Available{Ref: "other/repo:x", Digest: "sha256:ox", Source: model.SourcePoll, At: t0})) + // 12 versions of the service's repo, one per minute. + for i := 0; i < 12; i++ { + tag := "t" + string(rune('a'+i)) + require.NoError(t, s.SetAvailable(ctx, svc, model.Available{Ref: sluskRepo + ":" + tag, Digest: "sha256:" + tag, Source: model.SourcePoll, At: t0.Add(time.Duration(i+1) * time.Minute)})) + } + + got, err := s.RecentVersions(ctx, sluskRepo, 10) + + require.NoError(t, err) + require.Len(t, got, 10) + assert.Equal(t, "tl", got[0].Tag, "newest first") + assert.Equal(t, "tc", got[9].Tag) + for _, v := range got { + assert.Equal(t, sluskRepo, v.Repo) + } + }) + + t.Run("digest-less row is hidden once its tag has a digest", func(t *testing.T) { + s := openTestStore(t) + _, svc := jobFixture(t, s, "a") + add := func(tag, digest string, at time.Time) { + require.NoError(t, s.SetAvailable(ctx, svc, model.Available{Ref: sluskRepo + ":" + tag, Digest: digest, Source: model.SourceWebhook, At: at})) + } + add("main", "", t0.Add(time.Minute)) + add("solo", "", t0.Add(2*time.Minute)) + require.Len(t, versionsOf(t, s, sluskRepo), 3, "running 1.4.2 plus two digest-less rows") + + add("main", "sha256:m1", t0.Add(3*time.Minute)) + + var refs []string + for _, v := range versionsOf(t, s, sluskRepo) { + refs = append(refs, v.Ref()) + } + assert.Equal(t, []string{sluskRepo + ":main@sha256:m1", sluskRepo + ":solo", sluskRepo + ":1.4.2@sha256:d1"}, refs) + }) +} + +func TestGetImageVersion_NotFound(t *testing.T) { + s := openTestStore(t) + + _, err := s.GetImageVersion(context.Background(), 99999) + + assert.ErrorIs(t, err, model.ErrNotFound) +} + +func TestSetPollError(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + agent, svc := jobFixture(t, s, "a") + require.NoError(t, s.SetAvailable(ctx, svc, model.Available{Ref: sluskRepo + ":1.5.0", Digest: "sha256:n1", Source: model.SourcePoll, At: t0})) + + require.NoError(t, s.SetPollError(ctx, []int64{svc}, "boom")) + require.NoError(t, s.SetPollError(ctx, nil, "ignored")) + got, err := s.GetService(ctx, svc) + require.NoError(t, err) + assert.Equal(t, "boom", got.LastPollError) + + require.NoError(t, s.UpsertService(ctx, agent, "", slusk, t0.Add(time.Hour))) + got, err = s.GetService(ctx, svc) + require.NoError(t, err) + assert.Equal(t, "boom", got.LastPollError, "inventory writes do not touch the poll error") + assert.Equal(t, "sha256:n1", got.AvailableDigest, "nor the suggestion") + + require.NoError(t, s.SetPollError(ctx, []int64{svc}, "")) + got, err = s.GetService(ctx, svc) + require.NoError(t, err) + assert.Empty(t, got.LastPollError) +} diff --git a/internal/master/store/webhook_integration_test.go b/internal/master/store/webhook_integration_test.go index e047cec..3139647 100644 --- a/internal/master/store/webhook_integration_test.go +++ b/internal/master/store/webhook_integration_test.go @@ -91,27 +91,34 @@ func TestUpsertKeepsUpdateMode(t *testing.T) { func TestFinishJob_ClearsAvailableOnlyWhenThatVersionWasDeployed(t *testing.T) { succeed := model.JobFinish{Status: model.JobSucceeded, Ref: "r:2@sha256:d", StartedAt: t0, FinishedAt: t0} tests := []struct { - name string - availableRef string - availableAt time.Time // job is created at t0 for r:2 - finish model.JobFinish - wantCleared bool + name string + availableRef string + availableDigest string + availableAt time.Time // the job is created at t0 + toRef string // the job's to_ref as created + finish model.JobFinish + wantCleared bool }{ - {"same ref, suggestion before the job", "r:2", t0.Add(-time.Minute), succeed, true}, - {"same ref, suggestion at the same instant", "r:2", t0, succeed, true}, - {"same ref, suggestion after the job was created", "r:2", t0.Add(time.Minute), succeed, false}, - {"different ref keeps the suggestion", "r:3", t0.Add(-time.Minute), succeed, false}, - {"failed job keeps the suggestion", "r:2", t0.Add(-time.Minute), + {"same ref, suggestion before the job", "r:2", "", t0.Add(-time.Minute), "r:2", succeed, true}, + {"same ref, suggestion at the same instant", "r:2", "", t0, "r:2", succeed, true}, + {"same ref, suggestion after the job was created", "r:2", "", t0.Add(time.Minute), "r:2", succeed, false}, + {"different ref keeps the suggestion", "r:3", "", t0.Add(-time.Minute), "r:2", succeed, false}, + {"failed job keeps the suggestion", "r:2", "", t0.Add(-time.Minute), "r:2", model.JobFinish{Status: model.JobFailed, Error: "boom", StartedAt: t0, FinishedAt: t0}, false}, + {"tag job clears a poll suggestion with a digest", "r:2", "sha256:new", t0.Add(-time.Minute), "r:2", succeed, true}, + {"pick of the suggested digest clears it", "r:2", "sha256:new", t0.Add(-time.Minute), "r:2@sha256:new", succeed, true}, + {"rollback to an older digest keeps the newer suggestion", "r:2", "sha256:new", t0.Add(-time.Minute), "r:2@sha256:old", succeed, false}, + {"digest pick keeps a suggestion without digest", "r:2", "", t0.Add(-time.Minute), "r:2@sha256:old", succeed, false}, + {"tagless pick keeps the suggestion", "r:2", "sha256:new", t0.Add(-time.Minute), "r@sha256:old", succeed, false}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { s := openTestStore(t) ctx := context.Background() agent, svc := jobFixture(t, s, "a") - require.NoError(t, s.SetAvailable(ctx, svc, model.Available{Ref: tt.availableRef, Source: model.SourceWebhook, At: tt.availableAt})) + require.NoError(t, s.SetAvailable(ctx, svc, model.Available{Ref: tt.availableRef, Digest: tt.availableDigest, Source: model.SourcePoll, At: tt.availableAt})) nj := newJob(svc, agent) - nj.ToRef = "r:2" + nj.ToRef = tt.toRef j, err := s.CreateJob(ctx, nj) require.NoError(t, err) @@ -123,11 +130,13 @@ func TestFinishJob_ClearsAvailableOnlyWhenThatVersionWasDeployed(t *testing.T) { require.NoError(t, err) if tt.wantCleared { assert.Empty(t, got.AvailableRef) + assert.Empty(t, got.AvailableDigest) assert.Empty(t, got.AvailableSource) assert.Nil(t, got.AvailableAt) } else { assert.Equal(t, tt.availableRef, got.AvailableRef) - assert.Equal(t, model.SourceWebhook, got.AvailableSource) + assert.Equal(t, tt.availableDigest, got.AvailableDigest) + assert.Equal(t, model.SourcePoll, got.AvailableSource) assert.NotNil(t, got.AvailableAt) } }) -- 2.54.0 From d0726ca992fc8976f20e81ba7bdd86145f1bc0a6 Mon Sep 17 00:00:00 2001 From: Samuel Enocsson Date: Tue, 13 Oct 2026 08:07:45 +0200 Subject: [PATCH 02/14] feat(master,agent): deploy digest-pinned refs and route them to protocol 3 agents (#9) --- internal/agent/updater/compose.go | 4 +- internal/agent/updater/compose_test.go | 18 +++ internal/agent/updater/resolve.go | 21 +++- internal/agent/updater/resolve_test.go | 15 ++- internal/agent/updater/swarm.go | 4 +- internal/agent/updater/swarm_test.go | 16 +++ internal/e2e/agent_master_test.go | 4 +- internal/e2e/jobs_test.go | 4 +- internal/e2e/webhook_test.go | 3 +- internal/master/hub/jobs_test.go | 4 +- internal/master/hub/route.go | 13 +- internal/master/hub/route_test.go | 39 +++--- internal/master/jobs/jobs.go | 41 ++++++- internal/master/jobs/jobs_test.go | 157 ++++++++++++++++++++++++- internal/protocol/protocol.go | 13 +- 15 files changed, 303 insertions(+), 53 deletions(-) diff --git a/internal/agent/updater/compose.go b/internal/agent/updater/compose.go index 1e20aac..74eddff 100644 --- a/internal/agent/updater/compose.go +++ b/internal/agent/updater/compose.go @@ -91,7 +91,7 @@ func NewComposeExecutor(cli ComposeDocker, resolve *Resolver, selfID string, run return &ComposeExecutor{cli: cli, resolve: resolve, selfID: selfID, run: run, build: NewBuildChecker(cli, selfID, runOut)} } -// Execute pins job.Ref to its current digest, writes it to the stack's +// Execute pins job.Ref to its current digest (a ref that has one is used as is), writes it to the stack's // override file and recreates the service. Working directory and config // files come from the service's container labels on this machine, never from // the job. It returns the pinned ref. @@ -150,7 +150,7 @@ func (e *ComposeExecutor) Execute(ctx context.Context, job protocol.JobStart, ou if err != nil { return "", err } - fmt.Fprintf(out, "resolved %s\n", pinned) + fmt.Fprintln(out, resolveNote(job.Ref, pinned)) if _, err := WriteOverride(dir, service, pinned); err != nil { return "", err diff --git a/internal/agent/updater/compose_test.go b/internal/agent/updater/compose_test.go index ab9b637..0b9f9cb 100644 --- a/internal/agent/updater/compose_test.go +++ b/internal/agent/updater/compose_test.go @@ -108,6 +108,24 @@ func TestComposeExecutor(t *testing.T) { assert.Equal(t, pinned, serviceImage(doc, "web")) assert.Contains(t, out.String(), "resolved "+pinned) }) + t.Run("a pinned ref is written verbatim and not resolved", func(t *testing.T) { + t.Cleanup(cleanOverride) + for _, ref := range []string{"gitea.shcizo.se/shcizo/slusk:1.3@" + d, "gitea.shcizo.se/shcizo/slusk@" + d} { + rec := &recorder{} + dist := &fakeDist{digest: "sha256:other"} + var out strings.Builder + pinnedJob := job + pinnedJob.Ref = ref + got, err := setup(fakeCompose{rw, []container.Summary{ct}}, dist, rec).Execute(context.Background(), pinnedJob, &out) + require.NoError(t, err) + assert.Equal(t, ref, got) + assert.Empty(t, dist.refs, "no registry lookup") + _, doc := readOverride(t, stack) + assert.Equal(t, ref, serviceImage(doc, "web")) + assert.Contains(t, out.String(), "pinned "+ref) + assert.NotContains(t, out.String(), "resolved") + } + }) t.Run("built from source is refused before anything is written", func(t *testing.T) { t.Cleanup(cleanOverride) rec := &recorder{} diff --git a/internal/agent/updater/resolve.go b/internal/agent/updater/resolve.go index a3ddab2..677b04e 100644 --- a/internal/agent/updater/resolve.go +++ b/internal/agent/updater/resolve.go @@ -16,7 +16,8 @@ type DistributionClient interface { } // Resolver turns "repo:tag" into "repo:tag@digest" by asking the local -// daemon, which contacts the registry. +// daemon, which contacts the registry. A ref that already has a digest is +// passed through. type Resolver struct { cli DistributionClient auth *RegistryAuth @@ -27,13 +28,15 @@ func NewResolver(cli DistributionClient, auth *RegistryAuth) *Resolver { return &Resolver{cli: cli, auth: auth} } -// Resolve returns ref pinned to the digest its tag currently points at. It -// fails for a ref without a tag or with a digest already. +// Resolve returns ref pinned to the digest its tag currently points at. A ref +// that already has a digest ("repo:tag@digest" or "repo@digest") is returned +// unchanged without asking the daemon. It fails for a ref with neither tag nor +// digest. func (r *Resolver) Resolve(ctx context.Context, ref string) (string, error) { p := imageref.Parse(ref) switch { case p.Digest != "": - return "", fmt.Errorf("image %q already has a digest", ref) + return ref, nil case p.Tag == "": return "", fmt.Errorf("image %q has no tag to resolve", ref) } @@ -46,3 +49,13 @@ func (r *Resolver) Resolve(ctx context.Context, ref string) (string, error) { } return ref + "@" + info.Descriptor.Digest.String(), nil } + +// resolveNote is the job-output line that says which ref is deployed: +// "resolved" when pinned came from resolving a tag, "pinned" when ref already +// carried its digest. +func resolveNote(ref, pinned string) string { + if pinned == ref { + return "pinned " + pinned + } + return "resolved " + pinned +} diff --git a/internal/agent/updater/resolve_test.go b/internal/agent/updater/resolve_test.go index a6eab28..e84bb59 100644 --- a/internal/agent/updater/resolve_test.go +++ b/internal/agent/updater/resolve_test.go @@ -21,14 +21,21 @@ func TestResolver(t *testing.T) { assert.Equal(t, []string{"gitea.shcizo.se/shcizo/slusk:1.4"}, dist.refs) assert.Equal(t, []string{"enc-gitea"}, dist.auths) }) - t.Run("digest or missing tag is refused without asking the daemon", func(t *testing.T) { - for _, ref := range []string{"repo@sha256:x", "repo:1@sha256:x", "repo"} { + t.Run("a ref with a digest passes through without asking the daemon", func(t *testing.T) { + for _, ref := range []string{"repo@sha256:x", "repo:1@sha256:x", "localhost:5000/repo:1@sha256:x"} { dist := &fakeDist{digest: d} - _, err := NewResolver(dist, nil).Resolve(context.Background(), ref) - assert.Error(t, err, ref) + got, err := NewResolver(dist, nil).Resolve(context.Background(), ref) + require.NoError(t, err, ref) + assert.Equal(t, ref, got) assert.Empty(t, dist.refs) } }) + t.Run("a bare ref is refused without asking the daemon", func(t *testing.T) { + dist := &fakeDist{digest: d} + _, err := NewResolver(dist, nil).Resolve(context.Background(), "repo") + assert.ErrorContains(t, err, "has no tag to resolve") + assert.Empty(t, dist.refs) + }) t.Run("daemon error is wrapped", func(t *testing.T) { cause := errors.New("unauthorized") _, err := NewResolver(&fakeDist{err: cause}, nil).Resolve(context.Background(), "repo:1") diff --git a/internal/agent/updater/swarm.go b/internal/agent/updater/swarm.go index 63e2ffd..d644c2d 100644 --- a/internal/agent/updater/swarm.go +++ b/internal/agent/updater/swarm.go @@ -35,7 +35,7 @@ func NewSwarmExecutor(cli SwarmClient, resolve *Resolver, auth *RegistryAuth) *S } // Execute checks that the local daemon is a manager of the job's cluster, -// pins job.Ref to its current digest and updates the service to it. The +// pins job.Ref to its current digest (a ref that has one is used as is) and updates the service to it. The // update is addressed by the ID and version of the service that was // inspected, so it changes exactly that object. Success means the daemon // accepted the new spec, not that the tasks have converged. It returns the @@ -71,7 +71,7 @@ func (e *SwarmExecutor) Execute(ctx context.Context, job protocol.JobStart, out if err != nil { return "", err } - fmt.Fprintf(out, "resolved %s\n", pinned) + fmt.Fprintln(out, resolveNote(job.Ref, pinned)) spec := svc.Spec spec.TaskTemplate.ContainerSpec.Image = pinned diff --git a/internal/agent/updater/swarm_test.go b/internal/agent/updater/swarm_test.go index d44ddfa..429248f 100644 --- a/internal/agent/updater/swarm_test.go +++ b/internal/agent/updater/swarm_test.go @@ -77,6 +77,22 @@ func swarmSvc(name string) swarm.Service { } } +func TestSwarmUsesPinnedRef(t *testing.T) { + const ref = "gitea/gitea:1.26@sha256:abababababababababababababababababababababababababababababababab" + job := protocol.JobStart{Kind: protocol.KindSwarm, SwarmClusterID: "c1", SwarmService: "gitea_gitea", Ref: ref} + fake := &fakeSwarm{info: managerOf("c1"), svc: swarmSvc("gitea_gitea")} + dist := &fakeDist{digest: "sha256:other"} + var out bytes.Buffer + + got, err := NewSwarmExecutor(fake, NewResolver(dist, nil), nil).Execute(context.Background(), job, &out) + + require.NoError(t, err) + assert.Equal(t, ref, got) + assert.Equal(t, ref, fake.updSpec.TaskTemplate.ContainerSpec.Image) + assert.Empty(t, dist.refs, "no registry lookup") + assert.Contains(t, out.String(), "pinned "+ref) +} + func TestSwarmExecutor(t *testing.T) { job := protocol.JobStart{Kind: protocol.KindSwarm, SwarmClusterID: "c1", SwarmService: "gitea_gitea", Ref: "gitea/gitea:1.27"} workerInfo := managerOf("c1") diff --git a/internal/e2e/agent_master_test.go b/internal/e2e/agent_master_test.go index 2974cf8..bb737ef 100644 --- a/internal/e2e/agent_master_test.go +++ b/internal/e2e/agent_master_test.go @@ -328,7 +328,9 @@ type currentHub struct{ p atomic.Pointer[hub.Hub] } func (c *currentHub) ServeHTTP(w http.ResponseWriter, r *http.Request) { c.p.Load().ServeHTTP(w, r) } func (c *currentHub) Disconnect(id int64) { c.p.Load().Disconnect(id) } -func (c *currentHub) Route(svc model.Service) (model.Route, error) { return c.p.Load().Route(svc) } +func (c *currentHub) Route(svc model.Service, minProtocol int) (model.Route, error) { + return c.p.Load().Route(svc, minProtocol) +} func (c *currentHub) StartJob(r model.Route, id int64, s protocol.JobStart) error { return c.p.Load().StartJob(r, id, s) } diff --git a/internal/e2e/jobs_test.go b/internal/e2e/jobs_test.go index 912ea93..a4d1669 100644 --- a/internal/e2e/jobs_test.go +++ b/internal/e2e/jobs_test.go @@ -119,7 +119,7 @@ func (e *jobEnv) waitRoutable(t *testing.T) { t.Helper() svc, err := e.store.GetService(context.Background(), 1) require.NoError(t, err) - require.Eventually(t, func() bool { _, err := e.m.hub.Route(svc); return err == nil }, waitFor, pollEvery) + require.Eventually(t, func() bool { _, err := e.m.hub.Route(svc, protocol.MinJobVersion); return err == nil }, waitFor, pollEvery) } func (e *jobEnv) job(t *testing.T, id int64) model.Job { @@ -240,7 +240,7 @@ func TestPhaseOneAgentGetsNoJobs(t *testing.T) { require.NoError(t, wsjson.Write(ctx, c, snap)) require.Eventually(t, func() bool { - _, err := m.hub.Route(svcRow) + _, err := m.hub.Route(svcRow, protocol.MinJobVersion) return errors.Is(err, model.ErrAgentTooOld) }, waitFor, pollEvery) diff --git a/internal/e2e/webhook_test.go b/internal/e2e/webhook_test.go index 5e38c82..ed9ef10 100644 --- a/internal/e2e/webhook_test.go +++ b/internal/e2e/webhook_test.go @@ -16,6 +16,7 @@ import ( "github.com/shcizo/package-updater/internal/master/model" "github.com/shcizo/package-updater/internal/master/webhook" + "github.com/shcizo/package-updater/internal/protocol" ) // webhookServer mounts the webhook handler on a test server, wired to the @@ -111,7 +112,7 @@ func TestWebhookFailedAutoJobIs500(t *testing.T) { e.agent.stopConn() // no agent connection: the job fails at once with "agent offline" require.Eventually(t, func() bool { svc, _ := e.store.GetService(context.Background(), 1) - _, err := e.m.hub.Route(svc) + _, err := e.m.hub.Route(svc, protocol.MinJobVersion) return err != nil }, waitFor, pollEvery) diff --git a/internal/master/hub/jobs_test.go b/internal/master/hub/jobs_test.go index ce7fe00..61fad3f 100644 --- a/internal/master/hub/jobs_test.go +++ b/internal/master/hub/jobs_test.go @@ -43,7 +43,7 @@ func waitRoute(t *testing.T, h *Hub) model.Route { var r model.Route require.Eventually(t, func() bool { var err error - r, err = h.Route(svc) + r, err = h.Route(svc, protocol.MinJobVersion) return err == nil }, 2*time.Second, 5*time.Millisecond) return r @@ -61,7 +61,7 @@ func TestHub_ReconcilesBeforeRoutable(t *testing.T) { send(t, c, protocol.TypeHello, protocol.Hello{Hostname: "node-2", ProtocolVersion: 2, KnownJobs: []string{"7", "9", "x"}}) svc := model.Service{Kind: protocol.KindCompose, AgentID: ptr(node2.ID)} time.Sleep(50 * time.Millisecond) // hello is now blocked inside FailLostJobs - _, err = h.Route(svc) + _, err = h.Route(svc, protocol.MinJobVersion) assert.ErrorIs(t, err, model.ErrAgentOffline) close(store.lostGate) diff --git a/internal/master/hub/route.go b/internal/master/hub/route.go index 6013ee7..19fd603 100644 --- a/internal/master/hub/route.go +++ b/internal/master/hub/route.go @@ -14,9 +14,10 @@ type candidate struct { hello protocol.Hello } -// Route picks the connection a job for svc goes to. It returns -// model.ErrAgentOffline or model.ErrAgentTooOld when there is none. -func (h *Hub) Route(svc model.Service) (model.Route, error) { +// Route picks the connection a job for svc goes to, among agents that speak +// at least protocol version minProtocol. It returns model.ErrAgentOffline or +// model.ErrAgentTooOld when there is none. +func (h *Hub) Route(svc model.Service, minProtocol int) (model.Route, error) { h.mu.Lock() conns := make([]candidate, 0, len(h.conns)) for id, c := range h.conns { @@ -25,7 +26,7 @@ func (h *Hub) Route(svc model.Service) (model.Route, error) { } } h.mu.Unlock() - return route(svc, conns) + return route(svc, conns, minProtocol) } // route picks the connection a job for svc goes to. Compose services and @@ -33,7 +34,7 @@ func (h *Hub) Route(svc model.Service) (model.Route, error) { // run on the reporting agent if it qualifies, else on any other connected // manager of the same cluster (lowest agent ID first). Too-old wins over // offline only when a connection would otherwise have qualified. -func route(svc model.Service, conns []candidate) (model.Route, error) { +func route(svc model.Service, conns []candidate, minProtocol int) (model.Route, error) { sort.Slice(conns, func(a, b int) bool { return conns[a].agentID < conns[b].agentID }) tooOld := false try := func(c candidate) bool { @@ -42,7 +43,7 @@ func route(svc model.Service, conns []candidate) (model.Route, error) { return false } } - if c.hello.ProtocolVersion < protocol.MinJobVersion { + if c.hello.ProtocolVersion < minProtocol { tooOld = true return false } diff --git a/internal/master/hub/route_test.go b/internal/master/hub/route_test.go index 1ad86d8..9393cc8 100644 --- a/internal/master/hub/route_test.go +++ b/internal/master/hub/route_test.go @@ -12,37 +12,46 @@ import ( func TestRoute(t *testing.T) { composeSvc := model.Service{Kind: protocol.KindCompose, AgentID: ptr(int64(2))} swarmSvc := model.Service{Kind: protocol.KindSwarm, AgentID: ptr(int64(1)), SwarmClusterID: "c1"} - v2 := protocol.Hello{ProtocolVersion: protocol.Version} - mgrV2 := protocol.Hello{ProtocolVersion: protocol.Version, SwarmManager: true, SwarmClusterID: "c1"} + v2 := protocol.Hello{ProtocolVersion: 2} + v3 := protocol.Hello{ProtocolVersion: 3} + mgrV2 := protocol.Hello{ProtocolVersion: 2, SwarmManager: true, SwarmClusterID: "c1"} + mgrV3 := protocol.Hello{ProtocolVersion: 3, SwarmManager: true, SwarmClusterID: "c1"} + jobs, pinned := protocol.MinJobVersion, protocol.MinPinnedRefVersion tests := []struct { name string svc model.Service conns []candidate + min int want model.Route err error }{ - {"compose on own agent", composeSvc, []candidate{{2, 20, v2}}, model.Route{AgentID: 2, ConnID: 20}, nil}, - {"compose, own agent too old", composeSvc, []candidate{{2, 20, protocol.Hello{}}}, model.Route{}, model.ErrAgentTooOld}, - {"compose, own agent missing", composeSvc, nil, model.Route{}, model.ErrAgentOffline}, - {"compose, only another agent online", composeSvc, []candidate{{5, 50, v2}}, model.Route{}, model.ErrAgentOffline}, - {"swarm on reporting agent", swarmSvc, []candidate{{1, 10, mgrV2}, {3, 30, mgrV2}}, model.Route{AgentID: 1, ConnID: 10}, nil}, - {"swarm, reporting agent offline, other manager", swarmSvc, []candidate{{3, 30, mgrV2}}, model.Route{AgentID: 3, ConnID: 30}, nil}, - {"swarm, lowest agent id first", swarmSvc, []candidate{{4, 40, mgrV2}, {3, 30, mgrV2}}, model.Route{AgentID: 3, ConnID: 30}, nil}, + {"compose on own agent", composeSvc, []candidate{{2, 20, v2}}, jobs, model.Route{AgentID: 2, ConnID: 20}, nil}, + {"compose, own agent too old", composeSvc, []candidate{{2, 20, protocol.Hello{}}}, jobs, model.Route{}, model.ErrAgentTooOld}, + {"compose, own agent missing", composeSvc, nil, jobs, model.Route{}, model.ErrAgentOffline}, + {"compose, only another agent online", composeSvc, []candidate{{5, 50, v2}}, jobs, model.Route{}, model.ErrAgentOffline}, + {"compose, v2 agent and a pinned ref", composeSvc, []candidate{{2, 20, v2}}, pinned, model.Route{}, model.ErrAgentTooOld}, + {"compose, v3 agent and a pinned ref", composeSvc, []candidate{{2, 20, v3}}, pinned, model.Route{AgentID: 2, ConnID: 20}, nil}, + {"swarm on reporting agent", swarmSvc, []candidate{{1, 10, mgrV2}, {3, 30, mgrV2}}, jobs, model.Route{AgentID: 1, ConnID: 10}, nil}, + {"swarm, reporting agent offline, other manager", swarmSvc, []candidate{{3, 30, mgrV2}}, jobs, model.Route{AgentID: 3, ConnID: 30}, nil}, + {"swarm, lowest agent id first", swarmSvc, []candidate{{4, 40, mgrV2}, {3, 30, mgrV2}}, jobs, model.Route{AgentID: 3, ConnID: 30}, nil}, {"swarm, other manager in another cluster", swarmSvc, - []candidate{{3, 30, protocol.Hello{ProtocolVersion: 2, SwarmManager: true, SwarmClusterID: "c2"}}}, model.Route{}, model.ErrAgentOffline}, + []candidate{{3, 30, protocol.Hello{ProtocolVersion: 2, SwarmManager: true, SwarmClusterID: "c2"}}}, jobs, model.Route{}, model.ErrAgentOffline}, {"swarm, only a v0 manager of the cluster", swarmSvc, - []candidate{{1, 10, protocol.Hello{SwarmManager: true, SwarmClusterID: "c1"}}}, model.Route{}, model.ErrAgentTooOld}, + []candidate{{1, 10, protocol.Hello{SwarmManager: true, SwarmClusterID: "c1"}}}, jobs, model.Route{}, model.ErrAgentTooOld}, {"swarm, v0 manager but a v2 manager exists", swarmSvc, - []candidate{{1, 10, protocol.Hello{SwarmManager: true, SwarmClusterID: "c1"}}, {3, 30, mgrV2}}, model.Route{AgentID: 3, ConnID: 30}, nil}, + []candidate{{1, 10, protocol.Hello{SwarmManager: true, SwarmClusterID: "c1"}}, {3, 30, mgrV2}}, jobs, model.Route{AgentID: 3, ConnID: 30}, nil}, {"swarm, worker of the cluster", swarmSvc, - []candidate{{1, 10, protocol.Hello{ProtocolVersion: 2, SwarmClusterID: "c1"}}}, model.Route{}, model.ErrAgentOffline}, + []candidate{{1, 10, protocol.Hello{ProtocolVersion: 2, SwarmClusterID: "c1"}}}, jobs, model.Route{}, model.ErrAgentOffline}, {"swarm, manager with empty cluster id", model.Service{Kind: protocol.KindSwarm}, - []candidate{{1, 10, protocol.Hello{ProtocolVersion: 2, SwarmManager: true}}}, model.Route{}, model.ErrAgentOffline}, + []candidate{{1, 10, protocol.Hello{ProtocolVersion: 2, SwarmManager: true}}}, jobs, model.Route{}, model.ErrAgentOffline}, + {"swarm, pinned ref falls back to a v3 manager when the reporting one is v2", swarmSvc, + []candidate{{1, 10, mgrV2}, {3, 30, mgrV3}}, pinned, model.Route{AgentID: 3, ConnID: 30}, nil}, + {"swarm, pinned ref and only v2 managers", swarmSvc, []candidate{{1, 10, mgrV2}}, pinned, model.Route{}, model.ErrAgentTooOld}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - got, err := route(tt.svc, tt.conns) + got, err := route(tt.svc, tt.conns, tt.min) assert.ErrorIs(t, err, tt.err) assert.Equal(t, tt.want, got) }) diff --git a/internal/master/jobs/jobs.go b/internal/master/jobs/jobs.go index 6b2fe1b..f479dd3 100644 --- a/internal/master/jobs/jobs.go +++ b/internal/master/jobs/jobs.go @@ -20,13 +20,15 @@ type Store interface { CreateJob(ctx context.Context, j model.NewJob) (model.Job, error) FailJob(ctx context.Context, id int64, msg string, at time.Time) (bool, error) GetJob(ctx context.Context, id int64) (model.Job, error) + GetImageVersion(ctx context.Context, id int64) (model.ImageVersion, error) } // Hub routes jobs to agent connections. *hub.Hub satisfies it. type Hub interface { - // Route picks the connection for svc: model.ErrAgentOffline or + // Route picks the connection for svc among agents speaking at least + // protocol version minProtocol: model.ErrAgentOffline or // model.ErrAgentTooOld when there is none. - Route(svc model.Service) (model.Route, error) + Route(svc model.Service, minProtocol int) (model.Route, error) // StartJob sends job.start over exactly the routed connection. // model.ErrAgentOffline if that connection is gone or was replaced. StartJob(r model.Route, jobID int64, start protocol.JobStart) error @@ -116,8 +118,9 @@ func (s *Service) updatable(svc model.Service, explicitRef bool) error { type StartRequest struct { Trigger model.JobTrigger RequestedBy string // OIDC subject for manual jobs; "" otherwise - // Ref is the ref to deploy ("repo:tag"). Empty means the service's - // followed tag, derived from its actual ref. + // Ref is the ref to deploy ("repo:tag", "repo:tag@digest" or + // "repo@digest"). Empty means the service's followed tag, derived from + // its actual ref. Ref string } @@ -131,8 +134,30 @@ func (s *Service) Trigger(ctx context.Context, serviceID int64, requestedBy stri return s.Start(ctx, svc, StartRequest{Trigger: model.TriggerManual, RequestedBy: requestedBy}) } +// TriggerVersion starts a manual update of the service to exactly the image +// version versionID, which must be a version of the service's own repo; the +// ref is built from the stored version, never taken from the caller. It +// returns model.ErrNotFound for an unknown service, an unknown version or a +// version of another repo, and otherwise what Start returns. A version +// without a tag leaves the service following no tag. +func (s *Service) TriggerVersion(ctx context.Context, serviceID, versionID int64, requestedBy string) (model.Job, error) { + svc, err := s.store.GetService(ctx, serviceID) + if err != nil { + return model.Job{}, err + } + v, err := s.store.GetImageVersion(ctx, versionID) + if err != nil { + return model.Job{}, err + } + if v.Repo != svc.ImageRepo { + return model.Job{}, model.ErrNotFound + } + return s.Start(ctx, svc, StartRequest{Trigger: model.TriggerManual, RequestedBy: requestedBy, Ref: v.Ref()}) +} + // Start starts an update of svc. It returns the job, which may already be -// failed ("agent offline", "agent too old for updates"); a +// failed ("agent offline", "agent too old for updates", the latter also when +// the ref has a digest or no tag and the agent speaks protocol version 2); a // *model.NotUpdatableError when the service cannot be updated (no job // created); the active job and model.ErrJobActive when one is queued or // running. @@ -145,7 +170,11 @@ func (s *Service) Start(ctx context.Context, svc model.Service, req StartRequest ref = FollowedRef(svc.ActualRef) // updatable checked the tag } - r, routeErr := s.hub.Route(svc) + minProtocol := protocol.MinJobVersion + if p := imageref.Parse(ref); p.Digest != "" || p.Tag == "" { + minProtocol = protocol.MinPinnedRefVersion + } + r, routeErr := s.hub.Route(svc, minProtocol) nj := model.NewJob{ ServiceID: svc.ID, FromRef: svc.ActualRef, ToRef: ref, Trigger: req.Trigger, RequestedBy: req.RequestedBy, CreatedAt: s.now(), diff --git a/internal/master/jobs/jobs_test.go b/internal/master/jobs/jobs_test.go index 3918fb2..7ad6044 100644 --- a/internal/master/jobs/jobs_test.go +++ b/internal/master/jobs/jobs_test.go @@ -21,6 +21,7 @@ type fakeStore struct { jobs map[int64]model.Job created []model.NewJob active *model.Job // returned with ErrJobActive when set + versions map[int64]model.ImageVersion revokedAgents map[int64]bool // CreateJob returns ErrAgentOffline for these } @@ -54,6 +55,14 @@ func (f *fakeStore) FailJob(_ context.Context, id int64, msg string, at time.Tim return true, nil } +func (f *fakeStore) GetImageVersion(_ context.Context, id int64) (model.ImageVersion, error) { + v, ok := f.versions[id] + if !ok { + return model.ImageVersion{}, model.ErrNotFound + } + return v, nil +} + func (f *fakeStore) GetJob(_ context.Context, id int64) (model.Job, error) { return f.jobs[id], nil } type startCall struct { @@ -63,13 +72,21 @@ type startCall struct { } type fakeHub struct { - route model.Route - routeErr error - startErr error - starts []startCall + route model.Route + routeErr error + startErr error + starts []startCall + agentProtocol int // when set, Route fails with ErrAgentTooOld for a higher minProtocol + mins []int // minProtocol of every Route call } -func (h *fakeHub) Route(model.Service) (model.Route, error) { return h.route, h.routeErr } +func (h *fakeHub) Route(_ model.Service, minProtocol int) (model.Route, error) { + h.mins = append(h.mins, minProtocol) + if h.agentProtocol != 0 && minProtocol > h.agentProtocol { + return model.Route{}, model.ErrAgentTooOld + } + return h.route, h.routeErr +} func (h *fakeHub) StartJob(r model.Route, id int64, s protocol.JobStart) error { h.starts = append(h.starts, startCall{r, id, s}) @@ -341,3 +358,133 @@ func TestFollowedRef(t *testing.T) { assert.Equal(t, "localhost:5000/foo:latest", FollowedRef("localhost:5000/foo")) assert.Empty(t, FollowedRef("redis@sha256:abc")) } + +func TestStart_RoutesWithTheMinimumProtocolOfTheRef(t *testing.T) { + tests := []struct { + name string + ref string // "" means the followed tag + want int + }{ + {"followed tag", "", protocol.MinJobVersion}, + {"repo:tag", "x/y:2", protocol.MinJobVersion}, + {"repo:tag@digest", "x/y:2@sha256:a", protocol.MinPinnedRefVersion}, + {"repo@digest", "x/y@sha256:a", protocol.MinPinnedRefVersion}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + st := &fakeStore{services: map[int64]model.Service{1: composeSvc}, jobs: map[int64]model.Job{}} + h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}} + s, _ := newService(st, h) + + _, err := s.Start(context.Background(), composeSvc, StartRequest{Trigger: model.TriggerManual, Ref: tt.ref}) + + require.NoError(t, err) + assert.Equal(t, []int{tt.want}, h.mins) + }) + } +} + +func TestTriggerVersion_Ref(t *testing.T) { + const repo = "gitea.shcizo.se/shcizo/slusk" + noTag := model.Service{ID: 4, Kind: protocol.KindCompose, ComposeProject: "n", ComposeService: "n", + AgentID: ptr(int64(7)), ImageRepo: repo, ActualRef: repo + "@sha256:d1"} + svc := composeSvc + svc.ImageRepo = repo + tests := []struct { + name string + svc model.Service + version model.ImageVersion + wantRef string + wantMin int + }{ + {"tag and digest", svc, model.ImageVersion{ID: 10, Repo: repo, Tag: "1.3.0", Digest: "sha256:old"}, repo + ":1.3.0@sha256:old", protocol.MinPinnedRefVersion}, + {"tag only", svc, model.ImageVersion{ID: 10, Repo: repo, Tag: "main"}, repo + ":main", protocol.MinJobVersion}, + {"digest only", svc, model.ImageVersion{ID: 10, Repo: repo, Digest: "sha256:old"}, repo + "@sha256:old", protocol.MinPinnedRefVersion}, + {"service that follows no tag", noTag, model.ImageVersion{ID: 10, Repo: repo, Tag: "1.3.0", Digest: "sha256:old"}, repo + ":1.3.0@sha256:old", protocol.MinPinnedRefVersion}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + st := &fakeStore{ + services: map[int64]model.Service{tt.svc.ID: tt.svc}, + jobs: map[int64]model.Job{}, + versions: map[int64]model.ImageVersion{10: tt.version}, + } + h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}} + s, _ := newService(st, h) + + job, err := s.TriggerVersion(context.Background(), tt.svc.ID, 10, "sub-1") + + require.NoError(t, err) + assert.Equal(t, model.JobQueued, job.Status) + require.Len(t, st.created, 1) + assert.Equal(t, model.TriggerManual, st.created[0].Trigger) + assert.Equal(t, "sub-1", st.created[0].RequestedBy) + assert.Equal(t, tt.wantRef, st.created[0].ToRef) + require.Len(t, h.starts, 1) + assert.Equal(t, tt.wantRef, h.starts[0].start.Ref) + assert.Equal(t, []int{tt.wantMin}, h.mins) + }) + } +} + +func TestTriggerVersion_NotFound(t *testing.T) { + const repo = "gitea.shcizo.se/shcizo/slusk" + svc := composeSvc + svc.ImageRepo = repo + tests := []struct { + name string + serviceID int64 + versionID int64 + }{ + {"unknown service", 9, 10}, + {"unknown version", 1, 99}, + {"version of another repo", 1, 11}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + st := &fakeStore{ + services: map[int64]model.Service{1: svc}, + jobs: map[int64]model.Job{}, + versions: map[int64]model.ImageVersion{ + 10: {ID: 10, Repo: repo, Tag: "1"}, + 11: {ID: 11, Repo: "evil/other", Tag: "1"}, + }, + } + h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}} + s, pub := newService(st, h) + + _, err := s.TriggerVersion(context.Background(), tt.serviceID, tt.versionID, "sub-1") + + assert.ErrorIs(t, err, model.ErrNotFound) + assert.Empty(t, st.created, "no job") + assert.Empty(t, h.starts) + assert.Empty(t, pub.topics) + }) + } +} + +func TestTriggerVersion_RequiresPinnedProtocol(t *testing.T) { + const repo = "gitea.shcizo.se/shcizo/slusk" + svc := composeSvc + svc.ImageRepo = repo + st := &fakeStore{ + services: map[int64]model.Service{1: svc}, + jobs: map[int64]model.Job{}, + versions: map[int64]model.ImageVersion{ + 10: {ID: 10, Repo: repo, Tag: "1.3.0", Digest: "sha256:old"}, + 11: {ID: 11, Repo: repo, Tag: "main"}, + }, + } + h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}, agentProtocol: 2} + s, _ := newService(st, h) + + pinned, err := s.TriggerVersion(context.Background(), 1, 10, "sub-1") + require.NoError(t, err) + assert.Equal(t, model.JobFailed, pinned.Status) + assert.Equal(t, "agent too old for updates", pinned.Error) + assert.Empty(t, h.starts, "nothing sent to the old agent") + + plain, err := s.TriggerVersion(context.Background(), 1, 11, "sub-1") + require.NoError(t, err) + assert.Equal(t, model.JobQueued, plain.Status) +} diff --git a/internal/protocol/protocol.go b/internal/protocol/protocol.go index af25a2c..2d67989 100644 --- a/internal/protocol/protocol.go +++ b/internal/protocol/protocol.go @@ -16,11 +16,16 @@ import ( // Version is the protocol version this build speaks. Agents send it in // Hello. Phase-1 agents do not send the field, which reads as 0. -const Version = 2 +const Version = 3 // MinJobVersion is the lowest agent protocol version the master sends jobs to. const MinJobVersion = 2 +// MinPinnedRefVersion is the lowest agent protocol version the master sends +// jobs whose Ref has a digest or no tag to. Version-2 agents refuse a Ref +// with a digest. +const MinPinnedRefVersion = 3 + // MaxJobOutput caps the job output an agent keeps and the master stores, in // bytes. Only the tail is kept. With JSON escaping (worst case 6 bytes per // byte) a JobResult carrying this much output still fits the 1 MiB read limit. @@ -123,8 +128,10 @@ type Remove struct { Key string `json:"key"` } -// JobStart asks the agent to deploy Ref to one service. Ref is "repo:tag"; -// the agent resolves the tag's current digest and deploys "repo:tag@digest". +// JobStart asks the agent to deploy Ref to one service. Ref is "repo:tag", +// "repo:tag@digest" or "repo@digest". For "repo:tag" the agent resolves the +// tag's current digest and deploys "repo:tag@digest"; a Ref with a digest is +// deployed as is. // The agent reads the Compose working directory and config files from its // own Docker daemon, not from this message. type JobStart struct { -- 2.54.0 From c6ea2d371089c5ff723f86de8213952eaf0e7e36 Mon Sep 17 00:00:00 2001 From: Samuel Enocsson Date: Tue, 13 Oct 2026 08:11:37 +0200 Subject: [PATCH 03/14] feat(master): registry digest lookup and polling loop (#9) --- go.mod | 9 +- go.sum | 24 +- internal/master/poller/poller.go | 270 ++++++++++++ internal/master/poller/poller_test.go | 487 ++++++++++++++++++++++ internal/master/registry/registry.go | 55 +++ internal/master/registry/registry_test.go | 251 +++++++++++ 6 files changed, 1088 insertions(+), 8 deletions(-) create mode 100644 internal/master/poller/poller.go create mode 100644 internal/master/poller/poller_test.go create mode 100644 internal/master/registry/registry.go create mode 100644 internal/master/registry/registry_test.go diff --git a/go.mod b/go.mod index 74e3aa7..1c2dfb7 100644 --- a/go.mod +++ b/go.mod @@ -7,6 +7,7 @@ require ( github.com/coreos/go-oidc/v3 v3.21.0 github.com/distribution/reference v0.6.0 github.com/docker/docker v28.5.2+incompatible + github.com/google/go-containerregistry v0.22.1 github.com/jackc/pgx/v5 v5.10.0 github.com/opencontainers/go-digest v1.0.0 github.com/stretchr/testify v1.11.1 @@ -21,6 +22,8 @@ require ( github.com/containerd/errdefs/pkg v0.3.0 // indirect github.com/containerd/log v0.1.0 // indirect github.com/davecgh/go-spew v1.1.1 // indirect + github.com/docker/cli v29.7.2+incompatible // indirect + github.com/docker/docker-credential-helpers v0.9.3 // indirect github.com/docker/go-connections v0.7.0 // indirect github.com/docker/go-units v0.5.0 // indirect github.com/felixge/httpsnoop v1.0.4 // indirect @@ -30,6 +33,7 @@ require ( github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect github.com/jackc/puddle/v2 v2.2.2 // indirect + github.com/klauspost/compress v1.19.2 // indirect github.com/moby/docker-image-spec v1.3.1 // indirect github.com/moby/sys/atomicwriter v0.1.0 // indirect github.com/moby/term v0.5.2 // indirect @@ -37,14 +41,15 @@ require ( github.com/opencontainers/image-spec v1.1.1 // indirect github.com/pkg/errors v0.9.1 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect + github.com/sirupsen/logrus v1.9.4 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.68.0 // indirect go.opentelemetry.io/otel v1.43.0 // indirect go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.43.0 // indirect go.opentelemetry.io/otel/metric v1.43.0 // indirect go.opentelemetry.io/otel/trace v1.43.0 // indirect - golang.org/x/sync v0.20.0 // indirect - golang.org/x/sys v0.42.0 // indirect + golang.org/x/sync v0.22.0 // indirect + golang.org/x/sys v0.47.0 // indirect golang.org/x/text v0.35.0 // indirect golang.org/x/time v0.15.0 // indirect gotest.tools/v3 v3.5.2 // indirect diff --git a/go.sum b/go.sum index 7d4ff22..1ba542e 100644 --- a/go.sum +++ b/go.sum @@ -21,8 +21,12 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/distribution/reference v0.6.0 h1:0IXCQ5g4/QMHHkarYzh5l+u8T3t73zM5QvfrDyIgxBk= github.com/distribution/reference v0.6.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E= +github.com/docker/cli v29.7.2+incompatible h1:dlkwallR8XqfeVnA2ELEhdwvb4lsSwuB4IgsG8Q9cLY= +github.com/docker/cli v29.7.2+incompatible/go.mod h1:JLrzqnKDaYBop7H2jaqPtU4hHvMKP+vjCwu2uszcLI8= github.com/docker/docker v28.5.2+incompatible h1:DBX0Y0zAjZbSrm1uzOkdr1onVghKaftjlSWt4AFexzM= github.com/docker/docker v28.5.2+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk= +github.com/docker/docker-credential-helpers v0.9.3 h1:gAm/VtF9wgqJMoxzT3Gj5p4AqIjCBS4wrsOh9yRqcz8= +github.com/docker/docker-credential-helpers v0.9.3/go.mod h1:x+4Gbw9aGmChi3qTLZj8Dfn0TD20M/fuWy0E5+WDeCo= github.com/docker/go-connections v0.7.0 h1:6SsRfJddP22WMrCkj19x9WKjEDTB+ahsdiGYf0mN39c= github.com/docker/go-connections v0.7.0/go.mod h1:no1qkHdjq7kLMGUXYAduOhYPSJxxvgWBh7ogVvptn3Q= github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4= @@ -38,6 +42,8 @@ github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/go-containerregistry v0.22.1 h1:RZuuSYhTvlDvtsK+NkutoCZ//C0X2ebLK8X8l3ULs84= +github.com/google/go-containerregistry v0.22.1/go.mod h1:bJR35SK8XgisYmhg/FMQ/5RK0S/XrOAqLBV5/LR2XE0= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0 h1:HWRh5R2+9EifMyIHV7ZV+MIZqgz+PMpZ14Jynv3O2Zs= @@ -50,6 +56,8 @@ github.com/jackc/pgx/v5 v5.10.0 h1:VhSvgU2jSli8o3AqIEOTJr7rZwAEUVo4E4XhR94Zfr0= github.com/jackc/pgx/v5 v5.10.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= +github.com/klauspost/compress v1.19.2 h1:hMRETovs/pu/dVWN7zIT1PGG8t509MwT6bO7XSi26R8= +github.com/klauspost/compress v1.19.2/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= @@ -74,8 +82,8 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= -github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ= -github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ= +github.com/sirupsen/logrus v1.9.4 h1:TsZE7l11zFCLZnZ+teH4Umoq5BhEIfIzfRDZ1Uzql2w= +github.com/sirupsen/logrus v1.9.4/go.mod h1:ftWc9WdOfJ0a92nsE2jF5u5ZwH8Bv2zdeOC42RjbV2g= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= @@ -101,18 +109,22 @@ go.opentelemetry.io/otel/trace v1.43.0 h1:BkNrHpup+4k4w+ZZ86CZoHHEkohws8AY+WTX09 go.opentelemetry.io/otel/trace v1.43.0/go.mod h1:/QJhyVBUUswCphDVxq+8mld+AvhXZLhe+8WVFxiFff0= go.opentelemetry.io/proto/otlp v1.10.0 h1:IQRWgT5srOCYfiWnpqUYz9CVmbO8bFmKcwYxpuCSL2g= go.opentelemetry.io/proto/otlp v1.10.0/go.mod h1:/CV4QoCR/S9yaPj8utp3lvQPoqMtxXdzn7ozvvozVqk= +golang.org/x/mod v0.39.0 h1:UF5zwQdCRRUpHfyPwr7d4UrGiVeldIsogtzWVnczL74= +golang.org/x/mod v0.39.0/go.mod h1:bvIbwjQ0HUFFf5AKukeeYQG4ZBUG9yxQbR9aEweIwYY= golang.org/x/net v0.52.0 h1:He/TN1l0e4mmR3QqHMT2Xab3Aj3L9qjbhRm78/6jrW0= golang.org/x/net v0.52.0/go.mod h1:R1MAz7uMZxVMualyPXb+VaqGSa3LIaUqk0eEt3w36Sw= golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs= golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q= -golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4= -golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= -golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo= -golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek= +golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= +golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= golang.org/x/text v0.35.0 h1:JOVx6vVDFokkpaq1AEptVzLTpDe9KGpj5tR4/X+ybL8= golang.org/x/text v0.35.0/go.mod h1:khi/HExzZJ2pGnjenulevKNX1W67CUy0AsXcNubPGCA= golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= +golang.org/x/tools v0.49.0 h1:3NI7VXzL9+1WZD52Dx2ttoPwD5DWrFGpl9mFZDlmisI= +golang.org/x/tools v0.49.0/go.mod h1:SJNXV9DBKT0UbdttsQjbfJlAE/q+y36++zo3uL3N0Oo= google.golang.org/genproto/googleapis/api v0.0.0-20260401024825-9d38bb4040a9 h1:VPWxll4HlMw1Vs/qXtN7BvhZqsS9cdAittCNvVENElA= google.golang.org/genproto/googleapis/api v0.0.0-20260401024825-9d38bb4040a9/go.mod h1:7QBABkRtR8z+TEnmXTqIqwJLlzrZKVfAUm7tY3yGv0M= google.golang.org/genproto/googleapis/rpc v0.0.0-20260401024825-9d38bb4040a9 h1:m8qni9SQFH0tJc1X0vmnpw/0t+AImlSvp30sEupozUg= diff --git a/internal/master/poller/poller.go b/internal/master/poller/poller.go new file mode 100644 index 0000000..83fb16c --- /dev/null +++ b/internal/master/poller/poller.go @@ -0,0 +1,270 @@ +// 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) + SetAvailable(ctx context.Context, id int64, a model.Available) 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 "follows no tag" + 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 svc.ActualDigest == l.digest || (svc.AvailableRef == l.ref && svc.AvailableDigest == l.digest) { + continue + } + 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) + continue + } + stats.wrote = true + stats.suggested++ + if svc.UpdateMode != model.ModeAuto { + continue + } + p.startAuto(ctx, svc, l.ref, stats) + } +} + +// startAuto starts the update job of an auto service to ref. A service that +// already has a job or cannot be updated is not an error for the cycle. +func (p *Poller) startAuto(ctx context.Context, svc model.Service, ref string, stats *cycleStats) { + _, err := p.jobs.Start(ctx, svc, jobs.StartRequest{Trigger: model.TriggerPoll, Ref: ref}) + var notUpdatable *model.NotUpdatableError + switch { + case err == nil: + stats.jobs++ + case errors.Is(err, model.ErrJobActive), errors.As(err, ¬Updatable): + p.log.Debug("poll_job_not_started", "service_id", svc.ID, "ref", ref, "reason", err.Error()) + default: + p.log.Warn("poll_job_failed", "service_id", svc.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 +} diff --git a/internal/master/poller/poller_test.go b/internal/master/poller/poller_test.go new file mode 100644 index 0000000..23a4597 --- /dev/null +++ b/internal/master/poller/poller_test.go @@ -0,0 +1,487 @@ +package poller + +import ( + "context" + "errors" + "fmt" + "io" + "log/slog" + "sort" + "strings" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/shcizo/package-updater/internal/master/events" + "github.com/shcizo/package-updater/internal/master/jobs" + "github.com/shcizo/package-updater/internal/master/model" +) + +type availableCall struct { + id int64 + a model.Available +} + +type pollErrorCall struct { + ids []int64 + msg string +} + +type fakeStore struct { + mu sync.Mutex + services []model.Service + listErr error + available []availableCall + pollErrors []pollErrorCall + ctxErrs []error // ctx.Err() seen by every write +} + +func (f *fakeStore) ListServices(context.Context) ([]model.Service, error) { + return f.services, f.listErr +} + +func (f *fakeStore) SetAvailable(ctx context.Context, id int64, a model.Available) error { + f.mu.Lock() + defer f.mu.Unlock() + f.ctxErrs = append(f.ctxErrs, ctx.Err()) + f.available = append(f.available, availableCall{id, a}) + return nil +} + +func (f *fakeStore) SetPollError(ctx context.Context, ids []int64, msg string) error { + f.mu.Lock() + defer f.mu.Unlock() + f.ctxErrs = append(f.ctxErrs, ctx.Err()) + f.pollErrors = append(f.pollErrors, pollErrorCall{ids, msg}) + return nil +} + +type startCall struct { + svcID int64 + req jobs.StartRequest +} + +type fakeJobs struct { + mu sync.Mutex + starts []startCall + errFor map[int64]error // by service ID + ctxErrs []error +} + +func (f *fakeJobs) Start(ctx context.Context, svc model.Service, req jobs.StartRequest) (model.Job, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.ctxErrs = append(f.ctxErrs, ctx.Err()) + f.starts = append(f.starts, startCall{svc.ID, req}) + return model.Job{ID: int64(len(f.starts))}, f.errFor[svc.ID] +} + +type fakeRegistry struct { + mu sync.Mutex + digests map[string]string // ref -> digest + errs map[string]error // ref -> error + calls []string + inFlight int + maxSeen int + delay time.Duration + started chan struct{} // receives a value whenever a lookup starts, if set + release chan struct{} // lookups wait for it to be closed, if set + onResult func() // called after a lookup computed its answer +} + +func (f *fakeRegistry) Digest(ctx context.Context, ref string) (string, error) { + f.mu.Lock() + f.calls = append(f.calls, ref) + f.inFlight++ + if f.inFlight > f.maxSeen { + f.maxSeen = f.inFlight + } + f.mu.Unlock() + defer func() { + f.mu.Lock() + f.inFlight-- + f.mu.Unlock() + }() + + if f.started != nil { + f.started <- struct{}{} + } + if f.release != nil { + select { + case <-f.release: + case <-ctx.Done(): + return "", ctx.Err() + } + } + if f.delay > 0 { + time.Sleep(f.delay) + } + if f.onResult != nil { + f.onResult() + } + if err := f.errs[ref]; err != nil { + return "", err + } + return f.digests[ref], nil +} + +func (f *fakeRegistry) sortedCalls() []string { + f.mu.Lock() + defer f.mu.Unlock() + out := append([]string(nil), f.calls...) + sort.Strings(out) + return out +} + +type fakePub struct { + mu sync.Mutex + topics []events.Topic +} + +func (p *fakePub) Publish(t events.Topic) { + p.mu.Lock() + defer p.mu.Unlock() + p.topics = append(p.topics, t) +} + +const ( + repoA = "gitea.shcizo.se/shcizo/slusk" + repoB = "gitea/gitea" +) + +// svc builds a polled service whose running version has digest sha256:old. +func svc(id int64, actualRef string) model.Service { + return model.Service{ID: id, Kind: "compose", ActualRef: actualRef, ActualDigest: "sha256:old", UpdateMode: model.ModeManual} +} + +func newPoller(st *fakeStore, j *fakeJobs, reg *fakeRegistry, cfg Config) (*Poller, *fakePub) { + pub := &fakePub{} + if cfg.Workers == 0 { + cfg.Workers = 4 + } + p := New(st, j, reg, pub, cfg, slog.New(slog.NewTextHandler(io.Discard, nil))) + p.now = func() time.Time { return time.Date(2026, 10, 13, 12, 0, 0, 0, time.UTC) } + return p, pub +} + +func TestSkipReason(t *testing.T) { + tests := []struct { + name string + svc model.Service + want string + }{ + {"tag and digest", model.Service{ActualRef: "r:1", ActualDigest: "sha256:a"}, ""}, + {"bare image follows latest", model.Service{ActualRef: "r", ActualDigest: "sha256:a"}, ""}, + {"digest only follows no tag", model.Service{ActualRef: "r@sha256:a", ActualDigest: "sha256:a"}, "follows no tag"}, + {"built", model.Service{ActualRef: "r:1", ActualDigest: "sha256:a", Built: true}, "built from source"}, + {"no digest", model.Service{ActualRef: "r:1"}, "no registry digest"}, + {"self and container are polled", model.Service{ActualRef: "r:1", ActualDigest: "sha256:a", Self: true, Kind: "container"}, ""}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, SkipReason(tt.svc)) + }) + } +} + +func TestOneHeadPerDistinctRef(t *testing.T) { + st := &fakeStore{services: []model.Service{ + svc(1, repoA+":1.4"), + svc(2, repoA+":1.4@sha256:old"), + svc(3, repoB), + svc(4, repoB+":latest"), + svc(5, repoB+":1.27"), + }} + reg := &fakeRegistry{digests: map[string]string{}} + p, _ := newPoller(st, &fakeJobs{}, reg, Config{}) + + p.Cycle(context.Background()) + + assert.Equal(t, []string{repoA + ":1.4", repoB + ":1.27", repoB + ":latest"}, reg.sortedCalls()) +} + +func TestSkipped(t *testing.T) { + removed := svc(4, "r:1") + removed.RemovedAt = ptr(time.Now()) + st := &fakeStore{services: []model.Service{ + {ID: 1, ActualRef: "r@sha256:a", ActualDigest: "sha256:a"}, + {ID: 2, ActualRef: "r:1", ActualDigest: "sha256:a", Built: true}, + {ID: 3, ActualRef: "r:1"}, + removed, + }} + reg := &fakeRegistry{} + j := &fakeJobs{} + p, pub := newPoller(st, j, reg, Config{}) + + p.Cycle(context.Background()) + + assert.Empty(t, reg.calls, "no lookups") + assert.Empty(t, st.available) + assert.Empty(t, st.pollErrors) + assert.Empty(t, j.starts) + assert.Empty(t, pub.topics) +} + +func TestNewDigest(t *testing.T) { + ref := repoA + ":1.4" + tests := []struct { + name string + mode string + wantStart bool + }{ + {"auto gets one poll job to repo:tag", model.ModeAuto, true}, + {"manual is only marked", model.ModeManual, false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + s := svc(1, ref+"@sha256:old") + s.UpdateMode = tt.mode + st := &fakeStore{services: []model.Service{s}} + reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}} + j := &fakeJobs{} + p, pub := newPoller(st, j, reg, Config{}) + + p.Cycle(context.Background()) + + require.Len(t, st.available, 1) + assert.Equal(t, int64(1), st.available[0].id) + assert.Equal(t, model.Available{Ref: ref, Digest: "sha256:new", Source: model.SourcePoll, + At: time.Date(2026, 10, 13, 12, 0, 0, 0, time.UTC)}, st.available[0].a) + if tt.wantStart { + assert.Equal(t, []startCall{{1, jobs.StartRequest{Trigger: model.TriggerPoll, Ref: ref}}}, j.starts) + } else { + assert.Empty(t, j.starts) + } + assert.Equal(t, []events.Topic{events.Services}, pub.topics) + }) + } +} + +func TestSameSuggestionIsIdempotent(t *testing.T) { + ref := repoA + ":1.4" + running := svc(1, ref+"@sha256:old") + running.UpdateMode = model.ModeAuto + running.ActualDigest = "sha256:new" // already on the registry's digest + suggested := svc(2, ref) + suggested.UpdateMode = model.ModeAuto + suggested.AvailableRef, suggested.AvailableDigest = ref, "sha256:new" + staleSuggestion := svc(3, ref) + staleSuggestion.AvailableRef, staleSuggestion.AvailableDigest = ref, "sha256:older" + st := &fakeStore{services: []model.Service{running, suggested, staleSuggestion}} + reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}} + j := &fakeJobs{} + p, _ := newPoller(st, j, reg, Config{}) + + p.Cycle(context.Background()) + + require.Len(t, st.available, 1, "only the service with an older suggestion is updated") + assert.Equal(t, int64(3), st.available[0].id) + assert.Empty(t, j.starts, "no new job for an already suggested digest") +} + +func TestStartErrorsDoNotAbortCycle(t *testing.T) { + ref := repoA + ":1.4" + var svcs []model.Service + for id := int64(1); id <= 3; id++ { + s := svc(id, ref) + s.UpdateMode = model.ModeAuto + svcs = append(svcs, s) + } + st := &fakeStore{services: svcs} + reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}} + j := &fakeJobs{errFor: map[int64]error{ + 1: model.ErrJobActive, + 2: &model.NotUpdatableError{Reason: model.ReasonSelf}, + 3: errors.New("database down"), + }} + p, _ := newPoller(st, j, reg, Config{}) + + p.Cycle(context.Background()) + + assert.Len(t, st.available, 3) + assert.Len(t, j.starts, 3) +} + +func TestErrorIsPerRef(t *testing.T) { + bad, good := repoA+":1.4", repoB+":1.27" + badOld := svc(1, bad) + badSame := svc(2, bad) + badSame.LastPollError = "head: 401 unauthorized" + goodSvc := svc(3, good) + st := &fakeStore{services: []model.Service{badOld, badSame, goodSvc}} + reg := &fakeRegistry{ + errs: map[string]error{bad: errors.New("head: 401 unauthorized")}, + digests: map[string]string{good: "sha256:new"}, + } + p, pub := newPoller(st, &fakeJobs{}, reg, Config{}) + + p.Cycle(context.Background()) + + assert.Equal(t, []pollErrorCall{{[]int64{1}, "head: 401 unauthorized"}}, st.pollErrors, + "only the service whose stored error differs is written") + require.Len(t, st.available, 1, "the other ref is still processed") + assert.Equal(t, int64(3), st.available[0].id) + assert.Equal(t, []events.Topic{events.Services}, pub.topics) +} + +func TestErrorTextIsTruncated(t *testing.T) { + ref := repoA + ":1.4" + st := &fakeStore{services: []model.Service{svc(1, ref)}} + reg := &fakeRegistry{errs: map[string]error{ref: errors.New(strings.Repeat("é", 400) + "\x00")}} + p, _ := newPoller(st, &fakeJobs{}, reg, Config{}) + + p.Cycle(context.Background()) + + require.Len(t, st.pollErrors, 1) + msg := st.pollErrors[0].msg + assert.LessOrEqual(t, len(msg), maxErrorBytes) + assert.Equal(t, strings.Repeat("é", 250), msg, "cut on a character boundary, NUL removed") +} + +func TestErrorClearedOnSuccess(t *testing.T) { + ref := repoA + ":1.4" + failing := svc(1, ref) + failing.LastPollError = "head: boom" + fine := svc(2, ref) + failing.ActualDigest, fine.ActualDigest = "sha256:new", "sha256:new" + st := &fakeStore{services: []model.Service{failing, fine}} + reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}} + p, pub := newPoller(st, &fakeJobs{}, reg, Config{}) + + p.Cycle(context.Background()) + + assert.Equal(t, []pollErrorCall{{[]int64{1}, ""}}, st.pollErrors) + assert.Empty(t, st.available) + assert.Equal(t, []events.Topic{events.Services}, pub.topics) +} + +func TestNothingWrittenNothingPublished(t *testing.T) { + ref := repoA + ":1.4" + s := svc(1, ref) + s.ActualDigest = "sha256:new" + st := &fakeStore{services: []model.Service{s}} + reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}} + p, pub := newPoller(st, &fakeJobs{}, reg, Config{}) + + p.Cycle(context.Background()) + + assert.Empty(t, st.pollErrors) + assert.Empty(t, st.available) + assert.Empty(t, pub.topics) +} + +func TestListErrorEndsTheCycle(t *testing.T) { + st := &fakeStore{listErr: errors.New("db down")} + reg := &fakeRegistry{} + p, _ := newPoller(st, &fakeJobs{}, reg, Config{}) + + p.Cycle(context.Background()) + + assert.Empty(t, reg.calls) +} + +func TestConcurrencyLimit(t *testing.T) { + var svcs []model.Service + for i := 0; i < 8; i++ { + svcs = append(svcs, svc(int64(i+1), fmt.Sprintf("%s%d:1", repoA, i))) + } + st := &fakeStore{services: svcs} + reg := &fakeRegistry{delay: 20 * time.Millisecond} + p, _ := newPoller(st, &fakeJobs{}, reg, Config{Workers: 3}) + + p.Cycle(context.Background()) + + assert.Len(t, reg.calls, 8) + assert.Equal(t, 3, reg.maxSeen) +} + +func TestCycleStopsOnCancel(t *testing.T) { + var svcs []model.Service + for i := 0; i < 6; i++ { + svcs = append(svcs, svc(int64(i+1), fmt.Sprintf("%s%d:1", repoA, i))) + } + st := &fakeStore{services: svcs} + reg := &fakeRegistry{release: make(chan struct{}), started: make(chan struct{}, 6)} + p, pub := newPoller(st, &fakeJobs{}, reg, Config{Workers: 2}) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { p.Cycle(ctx); close(done) }() + <-reg.started + <-reg.started + + cancel() + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("Cycle did not return after cancel") + } + assert.Len(t, reg.calls, 2, "no new lookups after cancel") + assert.Empty(t, st.pollErrors, "cancelled lookups are not recorded as registry errors") + assert.Empty(t, st.available) + assert.Empty(t, pub.topics) +} + +func TestMarkedJobsAreStartedEvenIfCancelledDuringTheCycle(t *testing.T) { + ref := repoA + ":1.4" + s := svc(1, ref) + s.UpdateMode = model.ModeAuto + st := &fakeStore{services: []model.Service{s}} + ctx, cancel := context.WithCancel(context.Background()) + reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}, onResult: cancel} + j := &fakeJobs{} + p, _ := newPoller(st, j, reg, Config{}) + + p.Cycle(ctx) + + require.Len(t, st.available, 1) + require.Len(t, j.starts, 1) + assert.NoError(t, st.ctxErrs[0], "writes do not see the cancelled context") + assert.NoError(t, j.ctxErrs[0]) +} + +func TestRun(t *testing.T) { + ref := repoA + ":1.4" + st := &fakeStore{services: []model.Service{svc(1, ref)}} + reg := &fakeRegistry{digests: map[string]string{ref: "sha256:old"}, delay: 30 * time.Millisecond} + p, _ := newPoller(st, &fakeJobs{}, reg, Config{Interval: 5 * time.Millisecond, FirstDelay: 60 * time.Millisecond, Workers: 1}) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + began := time.Now() + go func() { p.Run(ctx); close(done) }() + + require.Eventually(t, func() bool { + reg.mu.Lock() + defer reg.mu.Unlock() + return len(reg.calls) >= 3 + }, 5*time.Second, 5*time.Millisecond) + cancel() + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("Run did not return after cancel") + } + assert.GreaterOrEqual(t, time.Since(began), 60*time.Millisecond+2*30*time.Millisecond, "first cycle waits FirstDelay") + assert.Equal(t, 1, reg.maxSeen, "cycles never overlap") +} + +func TestRunReturnsDuringFirstDelay(t *testing.T) { + p, _ := newPoller(&fakeStore{}, &fakeJobs{}, &fakeRegistry{}, Config{Interval: time.Hour, FirstDelay: time.Hour}) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { p.Run(ctx); close(done) }() + + cancel() + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("Run did not return after cancel") + } +} + +func ptr[T any](v T) *T { return &v } diff --git a/internal/master/registry/registry.go b/internal/master/registry/registry.go new file mode 100644 index 0000000..69a4dd6 --- /dev/null +++ b/internal/master/registry/registry.go @@ -0,0 +1,55 @@ +// Package registry looks up the digest a tag points at in a container +// registry, using HEAD requests only. +package registry + +import ( + "context" + "fmt" + + "github.com/google/go-containerregistry/pkg/authn" + "github.com/google/go-containerregistry/pkg/name" + "github.com/google/go-containerregistry/pkg/v1/remote" + + "github.com/shcizo/package-updater/internal/imageref" +) + +// Client looks up tag digests with HEAD requests only. Docker Hub counts a +// GET of a manifest as a pull against the rate limit and a HEAD as nothing, +// so Client never issues a manifest GET. +type Client struct { + keychain authn.Keychain + userAgent string +} + +// New returns a Client that finds credentials the way the Docker CLI does: +// in $DOCKER_CONFIG/config.json, else ~/.docker/config.json. Registries +// without an entry are accessed anonymously. userAgent is added to the +// User-Agent header of every request. +func New(userAgent string) *Client { + return &Client{keychain: authn.DefaultKeychain, userAgent: userAgent} +} + +// Digest returns the digest that "repo:tag" points at, as the registry +// reports it: for a multi-arch image that is the digest of the index, the +// same one `docker inspect` shows in RepoDigests. A digest suffix in ref is +// ignored. It sends HEAD /v2//manifests/ and nothing else to the +// manifest endpoint; it does not fall back to GET when the registry omits +// the digest header. +func (c *Client) Digest(ctx context.Context, ref string) (string, error) { + p := imageref.Parse(ref) + if p.Tag == "" { + return "", fmt.Errorf("image %q has no tag", ref) + } + r, err := name.ParseReference(p.Repo + ":" + p.Tag) + if err != nil { + return "", fmt.Errorf("parse %s: %w", ref, err) + } + desc, err := remote.Head(r, + remote.WithContext(ctx), + remote.WithAuthFromKeychain(c.keychain), + remote.WithUserAgent(c.userAgent)) + if err != nil { + return "", fmt.Errorf("head %s: %w", r, err) + } + return desc.Digest.String(), nil +} diff --git a/internal/master/registry/registry_test.go b/internal/master/registry/registry_test.go new file mode 100644 index 0000000..888de7a --- /dev/null +++ b/internal/master/registry/registry_test.go @@ -0,0 +1,251 @@ +package registry + +import ( + "context" + "encoding/base64" + "io" + "log" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync" + "testing" + + "github.com/google/go-containerregistry/pkg/authn" + "github.com/google/go-containerregistry/pkg/name" + ggcrregistry "github.com/google/go-containerregistry/pkg/registry" + "github.com/google/go-containerregistry/pkg/v1/random" + "github.com/google/go-containerregistry/pkg/v1/remote" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// request is one request the test registry received. +type request struct{ method, path string } + +// testRegistry is an in-memory registry that records every request. +type testRegistry struct { + srv *httptest.Server + host string // "127.0.0.1:port", the registry part of a ref + + mu sync.Mutex + requests []request +} + +// newTestRegistry starts a registry. When user is set every request needs +// those basic-auth credentials. +func newTestRegistry(t *testing.T, user, pass string) *testRegistry { + t.Helper() + backend := ggcrregistry.New(ggcrregistry.Logger(log.New(io.Discard, "", 0))) + tr := &testRegistry{} + tr.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + tr.mu.Lock() + tr.requests = append(tr.requests, request{r.Method, r.URL.Path}) + tr.mu.Unlock() + if user != "" { + if u, p, ok := r.BasicAuth(); !ok || u != user || p != pass { + w.Header().Set("WWW-Authenticate", `Basic realm="test"`) + w.WriteHeader(http.StatusUnauthorized) + return + } + } + backend.ServeHTTP(w, r) + })) + t.Cleanup(tr.srv.Close) + tr.host = strings.TrimPrefix(tr.srv.URL, "http://") + return tr +} + +// pushIndex stores a random multi-arch index as repo:tag and returns its +// digest. user and pass are used when the registry requires them. +func (tr *testRegistry) pushIndex(t *testing.T, repo, tag string, opts ...remote.Option) string { + t.Helper() + idx, err := random.Index(64, 1, 2) + require.NoError(t, err) + ref, err := name.ParseReference(tr.host + "/" + repo + ":" + tag) + require.NoError(t, err) + require.NoError(t, remote.WriteIndex(ref, idx, opts...)) + d, err := idx.Digest() + require.NoError(t, err) + return d.String() +} + +// manifestRequests returns the requests that touched a manifest. +func (tr *testRegistry) manifestRequests() []request { + tr.mu.Lock() + defer tr.mu.Unlock() + var out []request + for _, r := range tr.requests { + if strings.Contains(r.path, "/manifests/") { + out = append(out, r) + } + } + return out +} + +// reset forgets the requests received so far. +func (tr *testRegistry) reset() { + tr.mu.Lock() + defer tr.mu.Unlock() + tr.requests = nil +} + +// dockerConfig points DOCKER_CONFIG at a fresh directory holding config.json +// with the given content; "" leaves the directory empty. +func dockerConfig(t *testing.T, configJSON string) { + t.Helper() + dir := t.TempDir() + if configJSON != "" { + require.NoError(t, os.WriteFile(filepath.Join(dir, "config.json"), []byte(configJSON), 0o600)) + } + t.Setenv("DOCKER_CONFIG", dir) +} + +func TestDigestOfIndexIsIndexDigest(t *testing.T) { + dockerConfig(t, "") + reg := newTestRegistry(t, "", "") + want := reg.pushIndex(t, "team/app", "1.2") + + got, err := New("test").Digest(context.Background(), reg.host+"/team/app:1.2") + + require.NoError(t, err) + assert.Equal(t, want, got) +} + +func TestDigestIgnoresDigestSuffix(t *testing.T) { + dockerConfig(t, "") + reg := newTestRegistry(t, "", "") + want := reg.pushIndex(t, "team/app", "1.2") + + got, err := New("test").Digest(context.Background(), reg.host+"/team/app:1.2@sha256:"+strings.Repeat("0", 64)) + + require.NoError(t, err) + assert.Equal(t, want, got) +} + +func TestDigestUsesHeadOnly(t *testing.T) { + dockerConfig(t, "") + reg := newTestRegistry(t, "", "") + reg.pushIndex(t, "team/app", "1.2") + reg.reset() + + _, err := New("test").Digest(context.Background(), reg.host+"/team/app:1.2") + + require.NoError(t, err) + assert.Equal(t, []request{{http.MethodHead, "/v2/team/app/manifests/1.2"}}, reg.manifestRequests()) + for _, r := range reg.requests { + if !strings.Contains(r.path, "/manifests/") { + assert.Equal(t, request{http.MethodGet, "/v2/"}, r, "the only other request is the API ping") + } + } +} + +func TestDigestSendsUserAgent(t *testing.T) { + dockerConfig(t, "") + var ua string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + ua = r.Header.Get("User-Agent") + w.WriteHeader(http.StatusNotFound) + })) + t.Cleanup(srv.Close) + + _, _ = New("package-updater-test/1").Digest(context.Background(), strings.TrimPrefix(srv.URL, "http://")+"/a/b:1") + + assert.Contains(t, ua, "package-updater-test/1") +} + +func TestCredentialsFromDockerConfig(t *testing.T) { + reg := newTestRegistry(t, "bot", "s3cret") + want := reg.pushIndex(t, "team/app", "1.2", remote.WithAuth(&authn.Basic{Username: "bot", Password: "s3cret"})) + auth := base64.StdEncoding.EncodeToString([]byte("bot:s3cret")) + dockerConfig(t, `{"auths":{"`+reg.host+`":{"auth":"`+auth+`"}}}`) + + got, err := New("test").Digest(context.Background(), reg.host+"/team/app:1.2") + + require.NoError(t, err) + assert.Equal(t, want, got) +} + +func TestWrongCredentialsFail(t *testing.T) { + reg := newTestRegistry(t, "bot", "s3cret") + reg.pushIndex(t, "team/app", "1.2", remote.WithAuth(&authn.Basic{Username: "bot", Password: "s3cret"})) + auth := base64.StdEncoding.EncodeToString([]byte("bot:wrong")) + dockerConfig(t, `{"auths":{"`+reg.host+`":{"auth":"`+auth+`"}}}`) + + _, err := New("test").Digest(context.Background(), reg.host+"/team/app:1.2") + + require.Error(t, err) + assert.NotContains(t, err.Error(), "wrong", "credentials do not leak into errors") + assert.NotContains(t, err.Error(), auth) +} + +func TestAnonymousWithoutConfig(t *testing.T) { + dockerConfig(t, "") + open := newTestRegistry(t, "", "") + want := open.pushIndex(t, "team/app", "1.2") + protected := newTestRegistry(t, "bot", "s3cret") + + got, err := New("test").Digest(context.Background(), open.host+"/team/app:1.2") + require.NoError(t, err) + assert.Equal(t, want, got, "anonymous access works without a config file") + + _, err = New("test").Digest(context.Background(), protected.host+"/team/app:1.2") + assert.Error(t, err, "a registry that needs credentials fails without them") +} + +func TestUnknownTagIsAnError(t *testing.T) { + dockerConfig(t, "") + reg := newTestRegistry(t, "", "") + reg.pushIndex(t, "team/app", "1.2") + + _, err := New("test").Digest(context.Background(), reg.host+"/team/app:nope") + + assert.Error(t, err) +} + +func TestRefWithoutTagIsAnError(t *testing.T) { + _, err := New("test").Digest(context.Background(), "team/app@sha256:"+strings.Repeat("a", 64)) + + assert.ErrorContains(t, err, "has no tag") +} + +// A config.json whose credsStore names a helper that is not installed makes +// the lookup fail for every registry, including ones that would work +// anonymously. The master container has no credential helpers, so only +// inline "auths" entries are supported there. +func TestCredsStoreWithoutHelperFailsTheLookup(t *testing.T) { + reg := newTestRegistry(t, "", "") + reg.pushIndex(t, "team/app", "1.2") + dockerConfig(t, `{"credsStore":"no-such-helper-for-package-updater-tests"}`) + + _, err := New("test").Digest(context.Background(), reg.host+"/team/app:1.2") + + assert.ErrorContains(t, err, "error getting credentials") + assert.ErrorContains(t, err, "docker-credential-no-such-helper-for-package-updater-tests") +} + +// A registry that leaves out Docker-Content-Digest must not make the client +// retry with a manifest download, which Docker Hub counts as a pull. +func TestMissingDigestHeaderDoesNotFallBackToDownload(t *testing.T) { + dockerConfig(t, "") + var mu sync.Mutex + var methods []string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if !strings.Contains(r.URL.Path, "/manifests/") { + return + } + mu.Lock() + methods = append(methods, r.Method) + mu.Unlock() + w.Header().Set("Content-Type", "application/vnd.oci.image.index.v1+json") + w.Header().Set("Content-Length", "10") + })) + t.Cleanup(srv.Close) + + _, err := New("test").Digest(context.Background(), strings.TrimPrefix(srv.URL, "http://")+"/a/b:1") + + assert.Error(t, err) + assert.Equal(t, []string{http.MethodHead}, methods) +} -- 2.54.0 From 2390b6151e46bfad2ced70fa2080c20c8ab82d9d Mon Sep 17 00:00:00 2001 From: Samuel Enocsson Date: Tue, 13 Oct 2026 08:14:43 +0200 Subject: [PATCH 04/14] feat(master): POLL_INTERVAL, poller wiring and version picker in the UI (#9) --- cmd/master/main.go | 48 ++++- internal/master/config/config.go | 14 ++ internal/master/config/config_test.go | 22 +- internal/master/web/server.go | 42 +++- internal/master/web/server_test.go | 29 ++- internal/master/web/static/app.css | 1 + internal/master/web/templates/overview.html | 2 +- internal/master/web/templates/service.html | 23 ++ internal/master/web/versions_test.go | 221 ++++++++++++++++++++ internal/master/web/views.go | 47 +++++ 10 files changed, 438 insertions(+), 11 deletions(-) create mode 100644 internal/master/web/versions_test.go diff --git a/cmd/master/main.go b/cmd/master/main.go index 15cde34..83ae845 100644 --- a/cmd/master/main.go +++ b/cmd/master/main.go @@ -11,6 +11,7 @@ import ( "net/http" "os" "os/signal" + "path/filepath" "strings" "syscall" "time" @@ -20,6 +21,8 @@ import ( "github.com/shcizo/package-updater/internal/master/events" "github.com/shcizo/package-updater/internal/master/hub" "github.com/shcizo/package-updater/internal/master/jobs" + "github.com/shcizo/package-updater/internal/master/poller" + "github.com/shcizo/package-updater/internal/master/registry" "github.com/shcizo/package-updater/internal/master/store" "github.com/shcizo/package-updater/internal/master/web" "github.com/shcizo/package-updater/internal/master/webauth" @@ -37,6 +40,29 @@ func main() { } } +// pollFirstDelay is how long the master runs before its first registry +// polling cycle, so agents have reported their inventory. +const pollFirstDelay = time.Minute + +// pollWorkers is how many registry lookups run at once. +const pollWorkers = 4 + +// registryCredentialsDir returns the directory the registry client reads +// credentials from (DOCKER_CONFIG, else ~/.docker) and whether a config.json +// is in it. Only the path is logged, never the file. +func registryCredentialsDir() (dir string, found bool) { + dir = os.Getenv("DOCKER_CONFIG") + if dir == "" { + home, err := os.UserHomeDir() + if err != nil { + return "", false + } + dir = filepath.Join(home, ".docker") + } + _, err := os.Stat(filepath.Join(dir, "config.json")) + return dir, err == nil +} + func run() error { cfg, err := config.Load() if err != nil { @@ -81,6 +107,7 @@ func run() error { ui, err := web.New(st, agentHub, jobService, broker, web.Config{ AgentURL: cfg.AgentURL, MasterKey: fingerprint, AgentImage: cfg.AgentImage, + PollInterval: cfg.PollInterval, }, logger) if err != nil { return err @@ -127,22 +154,39 @@ func run() error { BaseContext: baseContext, } + credDir, credFound := registryCredentialsDir() + logger.Info("registry_credentials", "docker_config", credDir, "config_json_found", credFound) + pollCtx, stopPolling := context.WithCancel(ctx) + defer stopPolling() + registryPoller := poller.New(st, jobService, registry.New("package-updater-master/"+version), broker, poller.Config{ + Interval: cfg.PollInterval, FirstDelay: pollFirstDelay, Workers: pollWorkers, + }, logger) + pollDone := make(chan struct{}) + go func() { + defer close(pollDone) + registryPoller.Run(pollCtx) + }() + errCh := make(chan error, 2) go func() { errCh <- webSrv.ListenAndServe() }() go func() { errCh <- agentSrv.ListenAndServeTLS("", "") }() logger.Info("listening", "web", webSrv.Addr, "agents", agentSrv.Addr) + var serveErr error select { case <-ctx.Done(): logger.Info("shutdown_signal") case err := <-errCh: if !errors.Is(err, http.ErrServerClosed) { - return err + serveErr = err } } shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() _ = webSrv.Shutdown(shutdownCtx) _ = agentSrv.Shutdown(shutdownCtx) - return nil + // The poller writes to the store, which is closed when run returns. + stopPolling() + <-pollDone + return serveErr } diff --git a/internal/master/config/config.go b/internal/master/config/config.go index 77acd6f..50ce6bc 100644 --- a/internal/master/config/config.go +++ b/internal/master/config/config.go @@ -28,11 +28,17 @@ type Config struct { // UpdateTimeout bounds how long one webhook request waits for its auto // jobs (UPDATE_TIMEOUT, default 5m). UpdateTimeout time.Duration + // PollInterval is the time between registry polling cycles + // (POLL_INTERVAL, default 6h). + PollInterval time.Duration } // defaultUpdateTimeout is used when UPDATE_TIMEOUT is unset. const defaultUpdateTimeout = 5 * time.Minute +// defaultPollInterval is used when POLL_INTERVAL is unset. +const defaultPollInterval = 6 * time.Hour + // Load reads and validates the master configuration, failing fast on // anything missing so a half-configured master never starts. func Load() (*Config, error) { @@ -50,6 +56,7 @@ func Load() (*Config, error) { LogLevel: getenvDefault("LOG_LEVEL", "info"), UpdaterAPIKey: os.Getenv("UPDATER_API_KEY"), UpdateTimeout: defaultUpdateTimeout, + PollInterval: defaultPollInterval, } for _, r := range []struct{ name, value string }{ {"DATABASE_URL", cfg.DatabaseURL}, @@ -77,6 +84,13 @@ func Load() (*Config, error) { } cfg.UpdateTimeout = d } + if v := os.Getenv("POLL_INTERVAL"); v != "" { + d, err := time.ParseDuration(v) + if err != nil || d <= 0 { + return nil, fmt.Errorf("POLL_INTERVAL must be a positive duration such as 6h, got %q", v) + } + cfg.PollInterval = d + } return cfg, nil } diff --git a/internal/master/config/config_test.go b/internal/master/config/config_test.go index 977ee67..9b5a942 100644 --- a/internal/master/config/config_test.go +++ b/internal/master/config/config_test.go @@ -22,7 +22,7 @@ func setAll(t *testing.T) { for k, v := range required { t.Setenv(k, v) } - for _, k := range []string{"PORT", "AGENT_PORT", "AGENT_IMAGE", "LOG_LEVEL", "UPDATER_API_KEY", "UPDATE_TIMEOUT"} { + for _, k := range []string{"PORT", "AGENT_PORT", "AGENT_IMAGE", "LOG_LEVEL", "UPDATER_API_KEY", "UPDATE_TIMEOUT", "POLL_INTERVAL"} { t.Setenv(k, "") } } @@ -38,6 +38,26 @@ func TestLoad_Defaults(t *testing.T) { assert.Equal(t, "updater-admins", cfg.AllowedGroup) assert.Empty(t, cfg.UpdaterAPIKey) assert.Equal(t, 5*time.Minute, cfg.UpdateTimeout) + assert.Equal(t, 6*time.Hour, cfg.PollInterval) +} + +func TestLoad_PollInterval(t *testing.T) { + setAll(t) + t.Setenv("POLL_INTERVAL", "30m") + cfg, err := Load() + require.NoError(t, err) + assert.Equal(t, 30*time.Minute, cfg.PollInterval) +} + +func TestLoad_InvalidPollInterval(t *testing.T) { + for _, v := range []string{"abc", "0s", "-1h", "6"} { + t.Run(v, func(t *testing.T) { + setAll(t) + t.Setenv("POLL_INTERVAL", v) + _, err := Load() + assert.ErrorContains(t, err, "POLL_INTERVAL must be a positive duration such as 6h, got") + }) + } } func TestLoad_WebhookSettings(t *testing.T) { diff --git a/internal/master/web/server.go b/internal/master/web/server.go index 58434c1..4bf6a39 100644 --- a/internal/master/web/server.go +++ b/internal/master/web/server.go @@ -26,12 +26,18 @@ var assets embed.FS // serviceJobsLimit is how many jobs the service page lists. const serviceJobsLimit = 20 +// versionsLimit is how many image versions the service page lists. +const versionsLimit = 10 + var agentNamePattern = regexp.MustCompile(`^[a-z0-9][a-z0-9-]{0,62}$`) // Store is the storage the UI reads and writes. *store.Store satisfies it. type Store interface { ListServices(ctx context.Context) ([]model.Service, error) GetService(ctx context.Context, id int64) (model.Service, error) + // RecentVersions returns up to limit versions of repo the master has + // seen, most recently first seen first. + RecentVersions(ctx context.Context, repo string, limit int) ([]model.ImageVersion, error) // SetUpdateMode returns model.ErrNotFound for an unknown or removed service. SetUpdateMode(ctx context.Context, id int64, mode string) error GetJob(ctx context.Context, id int64) (model.Job, error) @@ -54,6 +60,9 @@ type Jobs interface { Updatable(svc model.Service) error AutoAllowed(svc model.Service) error Trigger(ctx context.Context, serviceID int64, requestedBy string) (model.Job, error) + // TriggerVersion updates to exactly the image version versionID of the + // service's repo; model.ErrNotFound for an unknown service or version. + TriggerVersion(ctx context.Context, serviceID, versionID int64, requestedBy string) (model.Job, error) } // Broker delivers change signals to open streams and accepts signals from @@ -63,11 +72,12 @@ type Broker interface { Publish(t events.Topic) } -// Config holds values shown in the "add agent" snippet. +// Config holds values shown in the "add agent" snippet and on the service page. type Config struct { - AgentURL string // what agents dial, e.g. wss://10.0.0.5:8090/agent - MasterKey string // the master's fingerprint - AgentImage string + AgentURL string // what agents dial, e.g. wss://10.0.0.5:8090/agent + MasterKey string // the master's fingerprint + AgentImage string + PollInterval time.Duration // shown as "every " on the service page } // Server renders the UI. @@ -85,7 +95,7 @@ type Server struct { // New parses the templates. Each page is its own template set so every page // can define its own "content" block. func New(store Store, hub Hub, jobs Jobs, broker Broker, cfg Config, log *slog.Logger) (*Server, error) { - funcs := template.FuncMap{"shortRef": shortRef, "duration": jobDuration} + funcs := template.FuncMap{"shortRef": shortRef, "duration": jobDuration, "followsNoTag": followsNoTag} pages := map[string]*template.Template{} for _, page := range []string{"overview.html", "agents.html", "agent_created.html", "service.html", "jobs.html", "job.html"} { t, err := template.New(page).Funcs(funcs).ParseFS(assets, "templates/layout.html", "templates/"+page) @@ -132,6 +142,9 @@ type pageData struct { ConfigFiles []string Reason string // why Service cannot be updated; "" when it can AutoBlocked string // why Service cannot be set to auto; "" when it can + Versions []versionRow + Polling string // what the service page says about registry polling + PollFailed bool // Polling is the last registry error Jobs []model.Job Job model.Job } @@ -222,6 +235,12 @@ func (s *Server) serviceDetail(r *http.Request, id int64) (pageData, error) { return data, err } data.AutoBlocked = s.autoBlocked(svc) + versions, err := s.store.RecentVersions(r.Context(), svc.ImageRepo, versionsLimit) + if err != nil { + return data, err + } + data.Versions = versionRows(versions, svc.ActualDigest) + data.Polling, data.PollFailed = pollingStatus(svc, s.cfg.PollInterval) data.Jobs, err = s.store.ListJobs(r.Context(), model.JobFilter{ServiceID: id, Limit: serviceJobsLimit}) return data, err } @@ -260,7 +279,18 @@ func (s *Server) updateService(w http.ResponseWriter, r *http.Request) { return } sess, _ := webauth.SessionFrom(r.Context()) - job, err := s.jobs.Trigger(r.Context(), id, sess.Subject) + var job model.Job + var err error + if v := r.PostFormValue("version"); v == "" { + job, err = s.jobs.Trigger(r.Context(), id, sess.Subject) + } else { + versionID, perr := strconv.ParseInt(v, 10, 64) + if perr != nil { + http.Error(w, "invalid version", http.StatusBadRequest) + return + } + job, err = s.jobs.TriggerVersion(r.Context(), id, versionID, sess.Subject) + } var nu *model.NotUpdatableError switch { case err == nil || errors.Is(err, model.ErrJobActive): diff --git a/internal/master/web/server_test.go b/internal/master/web/server_test.go index 52e40b4..68645c1 100644 --- a/internal/master/web/server_test.go +++ b/internal/master/web/server_test.go @@ -34,6 +34,17 @@ type fakeStore struct { created map[string]string // name → token hash revokedID int64 revokeFailed []int64 // job IDs RevokeAgent reports as failed + versions []model.ImageVersion +} + +func (f *fakeStore) RecentVersions(_ context.Context, repo string, limit int) ([]model.ImageVersion, error) { + var out []model.ImageVersion + for _, v := range f.versions { + if v.Repo == repo && len(out) < limit { + out = append(out, v) + } + } + return out, nil } func (f *fakeStore) ListServices(context.Context) ([]model.Service, error) { return f.services, nil } @@ -128,6 +139,9 @@ type fakeJobs struct { mu sync.Mutex requestedBy string + triggered []int64 // version IDs passed to TriggerVersion + triggerSvc []int64 // service IDs passed to TriggerVersion + plainCalls int // Trigger calls } func (f *fakeJobs) Updatable(model.Service) error { return f.updatable } @@ -138,6 +152,16 @@ func (f *fakeJobs) Trigger(_ context.Context, _ int64, by string) (model.Job, er f.mu.Lock() defer f.mu.Unlock() f.requestedBy = by + f.plainCalls++ + return f.job, f.err +} + +func (f *fakeJobs) TriggerVersion(_ context.Context, serviceID, versionID int64, by string) (model.Job, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.requestedBy = by + f.triggerSvc = append(f.triggerSvc, serviceID) + f.triggered = append(f.triggered, versionID) return f.job, f.err } @@ -177,7 +201,9 @@ func newTestServer(t *testing.T) (*httptest.Server, *fakeStore, *fakeHub, *event return srv, store, hub, broker } -func newTestEnv(t *testing.T) (*httptest.Server, *fakeStore, *fakeHub, *events.Broker, *fakeJobs) { +// newTestEnv starts the UI on fake storage. pollInterval is what the service +// page shows as the polling interval; it defaults to zero. +func newTestEnv(t *testing.T, pollInterval ...time.Duration) (*httptest.Server, *fakeStore, *fakeHub, *events.Broker, *fakeJobs) { t.Helper() agentID := int64(2) store := &fakeStore{ @@ -201,6 +227,7 @@ func newTestEnv(t *testing.T) (*httptest.Server, *fakeStore, *fakeHub, *events.B jobs := &fakeJobs{} s, err := New(store, hub, jobs, broker, Config{ AgentURL: "wss://10.0.0.5:8090/agent", MasterKey: "sha256:abc", AgentImage: "gitea.shcizo.se/shcizo/package-updater:latest", + PollInterval: append(pollInterval, 0)[0], }, slog.New(slog.NewTextHandler(io.Discard, nil))) require.NoError(t, err) mux := http.NewServeMux() diff --git a/internal/master/web/static/app.css b/internal/master/web/static/app.css index e50510a..28cf749 100644 --- a/internal/master/web/static/app.css +++ b/internal/master/web/static/app.css @@ -29,5 +29,6 @@ table.details { width: auto; } table.details th { padding-right: 24px; } pre.log { white-space: pre-wrap; } .badge-update { color: var(--warn); border-color: var(--warn); } +.badge-error { color: var(--bad); border-color: var(--bad); } form.mode { display: inline; margin-left: 8px; } form.mode button { font-size: 12px; padding: 0 6px; } diff --git a/internal/master/web/templates/overview.html b/internal/master/web/templates/overview.html index 3ff03b4..78ce293 100644 --- a/internal/master/web/templates/overview.html +++ b/internal/master/web/templates/overview.html @@ -14,7 +14,7 @@ {{range .Services}} - {{.DisplayName}}{{if .Self}} managed outside{{end}}{{if .AvailableRef}} update available{{end}} + {{.DisplayName}}{{if .Self}} managed outside{{end}}{{if .AvailableRef}} update available{{end}}{{if followsNoTag .Service}} follows no tag{{end}}{{if .LastPollError}} poll failed{{end}} {{shortRef .ActualRef}} {{.Status}} {{.UpdateMode}} diff --git a/internal/master/web/templates/service.html b/internal/master/web/templates/service.html index 68c0082..8709bb1 100644 --- a/internal/master/web/templates/service.html +++ b/internal/master/web/templates/service.html @@ -10,6 +10,7 @@ {{if .Service.DesiredRef}}Desired image{{shortRef .Service.DesiredRef}}{{end}} {{if .Service.AvailableRef}}Available{{.Service.AvailableRef}} via {{.Service.AvailableSource}}{{with .Service.AvailableAt}}, {{.Format "2006-01-02 15:04"}}{{end}}{{end}} Mode{{.Service.UpdateMode}} + Polling{{.Polling}} {{if .ConfigFiles}}Config files{{range .ConfigFiles}}{{.}}
{{end}}{{end}} {{if .Reason}} @@ -31,6 +32,28 @@

{{if .AutoBlocked}}Auto unavailable: {{.AutoBlocked}}.{{else}}The webhook will update this service as soon as CI pushes a new image.{{end}}

{{end}} +

Versions

+{{if eq .Service.UpdateMode "auto"}}

This service is in auto mode: a version picked here is replaced again when a newer version is announced or found by polling.

{{end}} + + + + {{range .Versions}} + + + + + + + {{else}} + + {{end}} + +
VersionSeen viaFirst seen
{{shortRef .Ref}}{{if .Running}} running{{end}}{{if not .Tag}} the service will follow no tag afterwards{{end}}{{.Source}}{{.FirstSeenAt.Format "2006-01-02 15:04"}}{{if not $.AutoBlocked}} +
+ + + +
{{end}}
No versions seen yet.

Update history

{{template "history" .}} diff --git a/internal/master/web/versions_test.go b/internal/master/web/versions_test.go new file mode 100644 index 0000000..838ce5c --- /dev/null +++ b/internal/master/web/versions_test.go @@ -0,0 +1,221 @@ +package web + +import ( + "io" + "net/http" + "net/http/httptest" + "net/url" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/shcizo/package-updater/internal/master/model" +) + +const sluskRepo = "gitea.shcizo.se/shcizo/slusk" + +func TestServicePage_VersionPicker(t *testing.T) { + srv, store, _, _, _ := newTestEnv(t) + store.services[0].ImageRepo = sluskRepo + store.services[0].ActualDigest = "sha256:cur" + first := time.Date(2026, 10, 12, 9, 5, 0, 0, time.UTC) + store.versions = []model.ImageVersion{ + {ID: 31, Repo: sluskRepo, Tag: "1.5.0", Digest: "sha256:" + strings.Repeat("a", 40), Source: "poll", FirstSeenAt: first}, + {ID: 32, Repo: sluskRepo, Tag: "1.4.2", Digest: "sha256:cur", Source: "running", FirstSeenAt: first}, + {ID: 33, Repo: sluskRepo, Digest: "sha256:" + strings.Repeat("b", 40), Source: "running", FirstSeenAt: first}, + {ID: 34, Repo: sluskRepo, Tag: "main", Source: "webhook", FirstSeenAt: first}, + {ID: 40, Repo: "other/repo", Tag: "x", Source: "poll", FirstSeenAt: first}, + } + + _, body := get(t, srv, "/services/1") + + for _, id := range []string{"31", "32", "33", "34"} { + assert.Contains(t, body, ``, id) + } + assert.NotContains(t, body, `name="version" value="40"`, "versions of other repos are not listed") + assert.Equal(t, 4, strings.Count(body, `name="version"`)) + assert.Contains(t, body, `name="csrf_token" value="csrf"`) + assert.Contains(t, body, `title="`+sluskRepo+`:1.5.0@sha256:`+strings.Repeat("a", 40)+`"`, "full ref as title") + assert.Contains(t, body, sluskRepo+":1.5.0@sha256:"+strings.Repeat("a", 12), "short ref as text") + assert.Equal(t, 1, strings.Count(body, `running`), "only the running digest is marked") + assert.Equal(t, 1, strings.Count(body, "the service will follow no tag afterwards"), "only the tagless row") + assert.Contains(t, body, "2026-10-12 09:05") + assert.NotContains(t, body, "auto mode", "no auto warning for a manual service") +} + +func TestServicePage_VersionPickerShowsAtMostTenRows(t *testing.T) { + srv, store, _, _, _ := newTestEnv(t) + store.services[0].ImageRepo = sluskRepo + for i := int64(1); i <= 12; i++ { + store.versions = append(store.versions, model.ImageVersion{ID: i, Repo: sluskRepo, Tag: "t", Digest: "sha256:d", Source: "poll"}) + } + + _, body := get(t, srv, "/services/1") + + assert.Equal(t, 10, strings.Count(body, `name="version"`)) +} + +func TestServicePage_VersionPickerWarnsForAutoServices(t *testing.T) { + srv, store, _, _, _ := newTestEnv(t) + store.services[0].UpdateMode = "auto" + + _, body := get(t, srv, "/services/1") + + assert.Contains(t, body, "This service is in auto mode") +} + +func TestServicePage_VersionPickerHiddenWhenServiceCannotBeUpdated(t *testing.T) { + for _, reason := range []string{model.ReasonSelf, model.ReasonContainer, model.ReasonRemoved, model.ReasonBuilt} { + t.Run(reason, func(t *testing.T) { + srv, store, _, _, jobs := newTestEnv(t) + store.services[0].ImageRepo = sluskRepo + store.versions = []model.ImageVersion{{ID: 31, Repo: sluskRepo, Tag: "1.5.0", Source: "poll"}} + jobs.autoErr = &model.NotUpdatableError{Reason: reason} + + _, body := get(t, srv, "/services/1") + + assert.Contains(t, body, "

Versions

") + assert.Contains(t, body, "1.5.0", "the history is still shown") + assert.NotContains(t, body, `name="version"`) + }) + } +} + +func TestServicePage_VersionPickerOffersVersionsToAServiceThatFollowsNoTag(t *testing.T) { + srv, store, _, _, jobs := newTestEnv(t) + store.services[0].ImageRepo = sluskRepo + store.versions = []model.ImageVersion{{ID: 31, Repo: sluskRepo, Tag: "1.5.0", Digest: "sha256:a", Source: "poll"}} + jobs.updatable = &model.NotUpdatableError{Reason: model.ReasonNoTag} // autoErr stays nil: an explicit ref is fine + + _, body := get(t, srv, "/services/1") + + assert.Contains(t, body, `name="version" value="31"`) + assert.NotContains(t, body, ``, "the plain Update button stays hidden") +} + +func TestServicePage_PollingRow(t *testing.T) { + tests := []struct { + name string + edit func(*model.Service) + want string + failed bool + }{ + {"polled", func(s *model.Service) { s.ActualDigest = "sha256:cur" }, "every 6h", false}, + {"poll error", func(s *model.Service) { s.ActualDigest = "sha256:cur"; s.LastPollError = "head x: 401 " }, "last check failed: head x: 401 <b>", true}, + {"built", func(s *model.Service) { s.ActualDigest = "sha256:cur"; s.Built = true }, "not polled: built from source", false}, + {"no digest", func(s *model.Service) {}, "not polled: no registry digest", false}, + {"no tag", func(s *model.Service) { s.ActualRef = "r@sha256:a"; s.ActualDigest = "sha256:a" }, "not polled: follows no tag", false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + srv, store, _, _, _ := newTestEnv(t, 6*time.Hour) + tt.edit(&store.services[0]) + + _, body := get(t, srv, "/services/1") + + assert.Contains(t, body, "Polling`+tt.want)) + }) + } +} + +func TestShortDuration(t *testing.T) { + tests := []struct { + d time.Duration + want string + }{ + {6 * time.Hour, "6h"}, + {90 * time.Minute, "1h30m"}, + {30 * time.Minute, "30m"}, + {90 * time.Second, "1m30s"}, + {45 * time.Second, "45s"}, + {6*time.Hour + 5*time.Second, "6h0m5s"}, + } + for _, tt := range tests { + t.Run(tt.want, func(t *testing.T) { + assert.Equal(t, tt.want, shortDuration(tt.d)) + }) + } +} + +func postVersion(t *testing.T, srv *httptest.Server, version string) (*http.Response, string) { + t.Helper() + resp, err := noRedirectClient().PostForm(srv.URL+"/services/1/update", url.Values{"csrf_token": {"csrf"}, "version": {version}}) + require.NoError(t, err) + defer resp.Body.Close() + body, _ := io.ReadAll(resp.Body) + return resp, string(body) +} + +func TestUpdateService_VersionStartsThatVersion(t *testing.T) { + srv, _, _, _, jobs := newTestEnv(t) + jobs.job = model.Job{ID: 42} + + resp, _ := postVersion(t, srv, "31") + + assert.Equal(t, http.StatusSeeOther, resp.StatusCode) + assert.Equal(t, "/jobs/42", resp.Header.Get("Location")) + assert.Equal(t, []int64{31}, jobs.triggered) + assert.Equal(t, []int64{1}, jobs.triggerSvc) + assert.Equal(t, "alice", jobs.triggeredBy()) + assert.Zero(t, jobs.plainCalls) +} + +func TestUpdateService_WithoutVersionTriggersTheFollowedTag(t *testing.T) { + srv, _, _, _, jobs := newTestEnv(t) + jobs.job = model.Job{ID: 42} + + resp, _ := postVersion(t, srv, "") + + assert.Equal(t, http.StatusSeeOther, resp.StatusCode) + assert.Equal(t, 1, jobs.plainCalls) + assert.Empty(t, jobs.triggered) +} + +func TestUpdateService_VersionErrors(t *testing.T) { + tests := []struct { + name string + version string + err error + wantCode int + wantBody string + wantJob string // Location header + }{ + {"not a number", "abc", nil, http.StatusBadRequest, "invalid version", ""}, + {"a ref instead of an id", "repo:evil@sha256:x", nil, http.StatusBadRequest, "invalid version", ""}, + {"unknown version or other repo", "77", model.ErrNotFound, http.StatusNotFound, "not found", ""}, + {"not updatable", "31", &model.NotUpdatableError{Reason: model.ReasonBuilt}, http.StatusConflict, "Cannot update: " + model.ReasonBuilt + ".", ""}, + {"job already active", "31", model.ErrJobActive, http.StatusSeeOther, "", "/jobs/41"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + srv, _, _, _, jobs := newTestEnv(t) + jobs.err, jobs.job = tt.err, model.Job{ID: 41} + + resp, body := postVersion(t, srv, tt.version) + + assert.Equal(t, tt.wantCode, resp.StatusCode) + assert.Contains(t, body, tt.wantBody) + assert.Equal(t, tt.wantJob, resp.Header.Get("Location")) + if tt.wantCode == http.StatusBadRequest { + assert.Empty(t, jobs.triggered, "no job for a malformed version") + } + }) + } +} + +func TestOverview_ShowsFollowsNoTagAndPollFailedBadges(t *testing.T) { + srv, store, _, _, _ := newTestEnv(t) + store.services[0].LastPollError = `head x: "boom"