diff --git a/internal/e2e/agent_master_test.go b/internal/e2e/agent_master_test.go index bb737ef..658e822 100644 --- a/internal/e2e/agent_master_test.go +++ b/internal/e2e/agent_master_test.go @@ -91,6 +91,18 @@ func (m *memStore) SetPollError(_ context.Context, ids []int64, msg string) erro return nil } +func (m *memStore) SetLastAutoDigest(_ context.Context, id int64, digest string) error { + m.mu.Lock() + defer m.mu.Unlock() + svc, ok := m.svcRows[id] + if !ok { + return model.ErrNotFound + } + svc.LastAutoDigest = digest + m.svcRows[id] = svc + return nil +} + // seedService adds a service row that jobs can be triggered for. func (m *memStore) seedService(svc model.Service) { m.mu.Lock() diff --git a/internal/e2e/versions_test.go b/internal/e2e/versions_test.go index a48bb9b..675055a 100644 --- a/internal/e2e/versions_test.go +++ b/internal/e2e/versions_test.go @@ -73,7 +73,12 @@ func TestPollStartsAutoJob(t *testing.T) { p.Cycle(context.Background()) - assert.Equal(t, slusk+":1.4.2", <-env.exec.started, "the agent gets repo:tag and resolves the digest itself") + select { + case ref := <-env.exec.started: + assert.Equal(t, slusk+":1.4.2", ref, "the agent gets repo:tag and resolves the digest itself") + case <-time.After(waitFor): + t.Fatal("the agent never received the job") + } require.Equal(t, 1, env.store.jobCount()) job := env.job(t, 1) assert.Equal(t, model.TriggerPoll, job.Trigger) @@ -83,6 +88,7 @@ func TestPollStartsAutoJob(t *testing.T) { assert.Equal(t, slusk+":1.4.2", svc.AvailableRef) assert.Equal(t, "sha256:new", svc.AvailableDigest) assert.Equal(t, model.SourcePoll, svc.AvailableSource) + assert.Equal(t, "sha256:new", svc.LastAutoDigest) p.Cycle(context.Background()) diff --git a/internal/master/model/model.go b/internal/master/model/model.go index 975538c..a084405 100644 --- a/internal/master/model/model.go +++ b/internal/master/model/model.go @@ -61,9 +61,13 @@ type Service struct { 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 + // LastAutoDigest is the registry digest the poller last started an + // automatic update for; "" if never. The poller does not start another + // one for the same digest, so a rollback sticks until a new digest appears. + LastAutoDigest string + FirstSeenAt time.Time + UpdatedAt time.Time + RemovedAt *time.Time } // Update modes of a service. diff --git a/internal/master/poller/poller.go b/internal/master/poller/poller.go index 83fb16c..3b9eacf 100644 --- a/internal/master/poller/poller.go +++ b/internal/master/poller/poller.go @@ -25,7 +25,9 @@ const maxErrorBytes = 500 // Store is the storage the poller needs. *store.Store satisfies it. type Store interface { ListServices(ctx context.Context) ([]model.Service, error) + GetService(ctx context.Context, id int64) (model.Service, error) SetAvailable(ctx context.Context, id int64, a model.Available) error + SetLastAutoDigest(ctx context.Context, id int64, digest string) error SetPollError(ctx context.Context, ids []int64, msg string) error } @@ -74,7 +76,7 @@ func New(st Store, j Jobs, reg Registry, pub Publisher, cfg Config, log *slog.Lo func SkipReason(svc model.Service) string { switch { case jobs.FollowedRef(svc.ActualRef) == "": - return "follows no tag" + return model.ReasonNoTag case svc.Built: return "built from source" case svc.ActualDigest == "": @@ -208,35 +210,74 @@ func (p *Poller) record(ctx context.Context, l lookup, svcs []model.Service, sta 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) { + if !p.suggest(ctx, svc, l, stats) { 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 + if autoWanted(svc, l.ref, l.digest) { + p.startAuto(ctx, svc.ID, l.ref, l.digest, stats) } - 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}) +// suggest marks l.digest as the available version of svc unless the service +// already runs it or already has exactly that suggestion. It returns false +// only when writing failed. +func (p *Poller) suggest(ctx context.Context, svc model.Service, l lookup, stats *cycleStats) bool { + if svc.ActualDigest == l.digest || (svc.AvailableRef == l.ref && svc.AvailableDigest == l.digest) { + return true + } + err := p.store.SetAvailable(ctx, svc.ID, model.Available{Ref: l.ref, Digest: l.digest, Source: model.SourcePoll, At: p.now()}) + if err != nil { + p.log.Error("poll_set_available_failed", "service_id", svc.ID, "ref", l.ref, "error", err) + return false + } + stats.wrote = true + stats.suggested++ + return true +} + +// autoWanted reports whether an update of svc to ref at digest should be +// started automatically: the service is in auto mode, not removed, does not +// already run that digest, was not already updated automatically to it +// (otherwise a rollback to an older version would be undone at the next +// cycle), and its last job did not already deploy it (desired_ref holds the +// digest until the agent reports the new inventory). +func autoWanted(svc model.Service, ref, digest string) bool { + return svc.RemovedAt == nil && + svc.UpdateMode == model.ModeAuto && + svc.ActualDigest != digest && + svc.LastAutoDigest != digest && + svc.DesiredRef != ref+"@"+digest +} + +// startAuto starts the update job of an auto service to ref and remembers +// digest as services.last_auto_digest, so the same digest is not started +// again, for instance after the user rolled back. It re-reads the service +// first, because a cycle can run for a long time and the user may have +// switched the service to manual meanwhile. A service that already has a +// job or cannot be updated is not an error for the cycle, and last_auto_digest +// is left alone so the next cycle tries again. +func (p *Poller) startAuto(ctx context.Context, id int64, ref, digest string, stats *cycleStats) { + svc, err := p.store.GetService(ctx, id) + if err != nil { + p.log.Warn("poll_get_service_failed", "service_id", id, "error", err) + return + } + if !autoWanted(svc, ref, digest) { + return + } + _, err = p.jobs.Start(ctx, svc, jobs.StartRequest{Trigger: model.TriggerPoll, Ref: ref}) var notUpdatable *model.NotUpdatableError switch { case err == nil: stats.jobs++ + if err := p.store.SetLastAutoDigest(ctx, id, digest); err != nil { + p.log.Error("poll_set_last_auto_digest_failed", "service_id", id, "error", err) + } 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()) + p.log.Debug("poll_job_not_started", "service_id", id, "ref", ref, "reason", err.Error()) default: - p.log.Warn("poll_job_failed", "service_id", svc.ID, "ref", ref, "error", err) + p.log.Warn("poll_job_failed", "service_id", id, "ref", ref, "error", err) } } diff --git a/internal/master/poller/poller_test.go b/internal/master/poller/poller_test.go index 23a4597..e91e734 100644 --- a/internal/master/poller/poller_test.go +++ b/internal/master/poller/poller_test.go @@ -30,9 +30,16 @@ type pollErrorCall struct { msg string } +type lastAutoCall struct { + id int64 + digest string +} + type fakeStore struct { mu sync.Mutex services []model.Service + fresh map[int64]model.Service // what GetService returns instead of the listed row + lastAuto []lastAutoCall listErr error available []availableCall pollErrors []pollErrorCall @@ -43,6 +50,34 @@ func (f *fakeStore) ListServices(context.Context) ([]model.Service, error) { return f.services, f.listErr } +func (f *fakeStore) GetService(_ context.Context, id int64) (model.Service, error) { + f.mu.Lock() + defer f.mu.Unlock() + if s, ok := f.fresh[id]; ok { + return s, nil + } + for _, s := range f.services { + if s.ID == id { + return s, nil + } + } + return model.Service{}, model.ErrNotFound +} + +// SetLastAutoDigest records the call and, like the database, changes the +// row the next cycle lists. +func (f *fakeStore) SetLastAutoDigest(_ context.Context, id int64, digest string) error { + f.mu.Lock() + defer f.mu.Unlock() + f.lastAuto = append(f.lastAuto, lastAutoCall{id, digest}) + for i := range f.services { + if f.services[i].ID == id { + f.services[i].LastAutoDigest = digest + } + } + return nil +} + func (f *fakeStore) SetAvailable(ctx context.Context, id int64, a model.Available) error { f.mu.Lock() defer f.mu.Unlock() @@ -175,7 +210,7 @@ func TestSkipReason(t *testing.T) { }{ {"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"}, + {"digest only follows no tag", model.Service{ActualRef: "r@sha256:a", ActualDigest: "sha256:a"}, model.ReasonNoTag}, {"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"}, ""}, @@ -252,8 +287,10 @@ func TestNewDigest(t *testing.T) { 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) + assert.Equal(t, []lastAutoCall{{1, "sha256:new"}}, st.lastAuto) } else { assert.Empty(t, j.starts) + assert.Empty(t, st.lastAuto) } assert.Equal(t, []events.Topic{events.Services}, pub.topics) }) @@ -268,6 +305,7 @@ func TestSameSuggestionIsIdempotent(t *testing.T) { suggested := svc(2, ref) suggested.UpdateMode = model.ModeAuto suggested.AvailableRef, suggested.AvailableDigest = ref, "sha256:new" + suggested.LastAutoDigest = "sha256:new" // its job was started in an earlier cycle staleSuggestion := svc(3, ref) staleSuggestion.AvailableRef, staleSuggestion.AvailableDigest = ref, "sha256:older" st := &fakeStore{services: []model.Service{running, suggested, staleSuggestion}} @@ -303,6 +341,89 @@ func TestStartErrorsDoNotAbortCycle(t *testing.T) { assert.Len(t, st.available, 3) assert.Len(t, j.starts, 3) + assert.Empty(t, st.lastAuto, "a job that was not created is retried next cycle") +} + +func TestRollbackOfAutoServiceIsNotUndone(t *testing.T) { + ref := repoA + ":main" + // Updated to sha256:new automatically, then rolled back to sha256:old by hand. + s := svc(1, ref+"@sha256:old") + s.UpdateMode = model.ModeAuto + s.LastAutoDigest = "sha256:new" + s.DesiredRef = ref + "@sha256:old" + st := &fakeStore{services: []model.Service{s}} + reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}} + j := &fakeJobs{} + p, _ := newPoller(st, j, reg, Config{}) + + p.Cycle(context.Background()) + p.Cycle(context.Background()) + + assert.Empty(t, j.starts, "same registry digest: the rollback stays") + assert.Equal(t, "sha256:new", st.available[0].a.Digest, "the newer version is still offered") + + reg.digests[ref] = "sha256:newer" + p.Cycle(context.Background()) + + assert.Equal(t, []startCall{{1, jobs.StartRequest{Trigger: model.TriggerPoll, Ref: ref}}}, j.starts, "a new digest moves it again") + assert.Equal(t, []lastAutoCall{{1, "sha256:newer"}}, st.lastAuto) +} + +func TestActiveJobIsRetriedNextCycle(t *testing.T) { + ref := repoA + ":1.4" + s := svc(1, ref) + s.UpdateMode = model.ModeAuto + st := &fakeStore{services: []model.Service{s}} + reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}} + j := &fakeJobs{errFor: map[int64]error{1: model.ErrJobActive}} + p, _ := newPoller(st, j, reg, Config{}) + + p.Cycle(context.Background()) + require.Len(t, j.starts, 1) + assert.Empty(t, st.lastAuto) + + j.errFor = nil + p.Cycle(context.Background()) + + assert.Len(t, j.starts, 2) + assert.Equal(t, []lastAutoCall{{1, "sha256:new"}}, st.lastAuto) + + p.Cycle(context.Background()) + assert.Len(t, j.starts, 2, "started once the job exists") +} + +func TestJobThatAlreadyDeployedTheDigestIsNotStartedAgain(t *testing.T) { + ref := repoA + ":1.4" + // FinishJob set desired_ref; the agent has not reported the new digest yet. + s := svc(1, ref+"@sha256:old") + s.UpdateMode = model.ModeAuto + s.DesiredRef = ref + "@sha256:new" + st := &fakeStore{services: []model.Service{s}} + reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}} + j := &fakeJobs{} + p, _ := newPoller(st, j, reg, Config{}) + + p.Cycle(context.Background()) + + assert.Empty(t, j.starts) +} + +func TestServiceSwitchedToManualDuringTheCycleIsNotStarted(t *testing.T) { + ref := repoA + ":1.4" + s := svc(1, ref) + s.UpdateMode = model.ModeAuto + manual := s + manual.UpdateMode = model.ModeManual + st := &fakeStore{services: []model.Service{s}, fresh: map[int64]model.Service{1: manual}} + reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}} + j := &fakeJobs{} + p, _ := newPoller(st, j, reg, Config{}) + + p.Cycle(context.Background()) + + assert.Len(t, st.available, 1, "still suggested") + assert.Empty(t, j.starts) + assert.Empty(t, st.lastAuto) } func TestErrorIsPerRef(t *testing.T) { @@ -384,18 +505,37 @@ func TestListErrorEndsTheCycle(t *testing.T) { } func TestConcurrencyLimit(t *testing.T) { + const workers = 3 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}) + reg := &fakeRegistry{release: make(chan struct{}), started: make(chan struct{}, 8)} + p, _ := newPoller(st, &fakeJobs{}, reg, Config{Workers: workers}) + done := make(chan struct{}) + go func() { p.Cycle(context.Background()); close(done) }() - p.Cycle(context.Background()) + for i := 0; i < workers; i++ { + select { + case <-reg.started: + case <-time.After(2 * time.Second): + t.Fatalf("only %d lookups started, want %d", i, workers) + } + } + select { + case <-reg.started: + t.Fatal("more lookups in flight than Workers") + case <-time.After(100 * time.Millisecond): + } + close(reg.release) + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("Cycle did not finish") + } assert.Len(t, reg.calls, 8) - assert.Equal(t, 3, reg.maxSeen) } func TestCycleStopsOnCancel(t *testing.T) { diff --git a/internal/master/store/migrations/0004_polling.sql b/internal/master/store/migrations/0004_polling.sql index a1da48d..1ea2f78 100644 --- a/internal/master/store/migrations/0004_polling.sql +++ b/internal/master/store/migrations/0004_polling.sql @@ -3,6 +3,7 @@ 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 ''; +ALTER TABLE services ADD COLUMN last_auto_digest text NOT NULL DEFAULT ''; -- Every version the master has seen. '' means "no tag" / "digest unknown"; -- never NULL, so the unique constraint deduplicates. diff --git a/internal/master/store/services.go b/internal/master/store/services.go index 0649086..e6f1a74 100644 --- a/internal/master/store/services.go +++ b/internal/master/store/services.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "regexp" "time" "github.com/jackc/pgx/v5" @@ -56,7 +57,11 @@ ON CONFLICT (location_key) DO UPDATE SET updated_at = EXCLUDED.updated_at, removed_at = NULL` +// digestPattern is the form of a digest worth recording as a version. +var digestPattern = regexp.MustCompile(`^sha256:[0-9a-f]{64}$`) + func upsertService(ctx context.Context, db execer, agentID int64, clusterID string, svc protocol.Service, at time.Time) error { + image := imageref.Parse(svc.Image) clusterCol := "" if svc.Kind == protocol.KindSwarm { clusterCol = clusterID @@ -65,14 +70,18 @@ func upsertService(ctx context.Context, db execer, agentID int64, clusterID stri string(svc.Kind), model.LocationKey(agentID, clusterID, svc.Key()), agentID, svc.ComposeProject, svc.ComposeService, svc.WorkingDir, svc.ConfigFiles, svc.Built, clusterCol, svc.SwarmService, svc.ContainerName, - imageref.Parse(svc.Image).Repo, svc.Image, svc.Digest, + image.Repo, svc.Image, svc.Digest, svc.Status, svc.Self, at) 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 { + if digestPattern.MatchString(svc.Digest) { + // A bare image runs "latest", as jobs.FollowedRef sees it, so it + // shares its row with the polled one. repo@digest has no tag. + if image.Tag == "" && image.Digest == "" { + image.Tag = "latest" + } + if err := insertVersion(ctx, db, image.Repo, image.Tag, svc.Digest, model.SourceRunning, at); err != nil { return err } } @@ -87,7 +96,7 @@ func insertVersion(ctx context.Context, db execer, repo, tag, digest, source str 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 fmt.Errorf("record version %s: %w", imageref.Ref{Repo: repo, Tag: tag, Digest: digest}, err) } return nil } @@ -169,7 +178,7 @@ const serviceColumns = `s.id, s.kind, s.location_key, s.agent_id, COALESCE(a.nam 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_digest, s.available_source, s.available_at, - s.last_poll_error, s.first_seen_at, s.updated_at, s.removed_at` + s.last_poll_error, s.last_auto_digest, s.first_seen_at, s.updated_at, s.removed_at` func scanService(row pgx.Row) (model.Service, error) { var m model.Service @@ -178,7 +187,7 @@ func scanService(row pgx.Row) (model.Service, error) { &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.AvailableDigest, &m.AvailableSource, &m.AvailableAt, - &m.LastPollError, &m.FirstSeenAt, &m.UpdatedAt, &m.RemovedAt) + &m.LastPollError, &m.LastAutoDigest, &m.FirstSeenAt, &m.UpdatedAt, &m.RemovedAt) m.Kind = protocol.Kind(kind) return m, err } @@ -252,16 +261,27 @@ func (s *Store) SetPollError(ctx context.Context, ids []int64, msg string) error return nil } +// SetLastAutoDigest records the digest an automatic update was last started +// for; see model.Service.LastAutoDigest. Inventory writes never change it. +func (s *Store) SetLastAutoDigest(ctx context.Context, id int64, digest string) error { + if _, err := s.pool.Exec(ctx, + `UPDATE services SET last_auto_digest = $2 WHERE id = $1`, id, digest); err != nil { + return fmt.Errorf("set last auto digest: %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. +// same tag and a digest was first seen at or after it. 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 <> '')) + WHERE d.repo = v.repo AND d.tag = v.tag AND d.digest <> '' + AND d.first_seen_at >= v.first_seen_at)) ORDER BY v.first_seen_at DESC, v.id DESC LIMIT $2`, repo, limit) if err != nil { diff --git a/internal/master/store/services_integration_test.go b/internal/master/store/services_integration_test.go index e0dc4db..f8a4c86 100644 --- a/internal/master/store/services_integration_test.go +++ b/internal/master/store/services_integration_test.go @@ -17,9 +17,9 @@ import ( var ( t0 = time.Date(2026, 10, 12, 12, 0, 0, 0, time.UTC) - slusk = protocol.Service{Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web", WorkingDir: "/srv/slusk", Image: "gitea.shcizo.se/shcizo/slusk:1.4.2", Digest: "sha256:d1", Status: "running"} + slusk = protocol.Service{Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web", WorkingDir: "/srv/slusk", Image: "gitea.shcizo.se/shcizo/slusk:1.4.2", Digest: digD1, Status: "running"} adhoc = protocol.Service{Kind: protocol.KindContainer, ContainerName: "adhoc", Image: "redis:alpine", Status: "stopped"} - gitea = protocol.Service{Kind: protocol.KindSwarm, SwarmService: "gitea_gitea", Image: "gitea/gitea:1.27.3@sha256:87a6", Digest: "sha256:87a6", Status: "running"} + gitea = protocol.Service{Kind: protocol.KindSwarm, SwarmService: "gitea_gitea", Image: "gitea/gitea:1.27.3@sha256:87a6", Digest: digGitea, Status: "running"} worker = protocol.Hello{Hostname: "node-2"} manager = protocol.Hello{Hostname: "swarm-mgr", SwarmManager: true, SwarmClusterID: "c1"} ) @@ -53,7 +53,7 @@ func TestApplySnapshot_InsertsAsManualWithAgentName(t *testing.T) { assert.Equal(t, "node-2", got[0].AgentName) assert.Equal(t, "gitea.shcizo.se/shcizo/slusk", got[0].ImageRepo) assert.Equal(t, "gitea.shcizo.se/shcizo/slusk:1.4.2", got[0].ActualRef) - assert.Equal(t, "sha256:d1", got[0].ActualDigest) + assert.Equal(t, digD1, got[0].ActualDigest) assert.Equal(t, "compose|"+itoa(id)+"|slusk|web", got[0].LocationKey) } diff --git a/internal/master/store/versions_integration_test.go b/internal/master/store/versions_integration_test.go index 5d86c98..fd1d3f0 100644 --- a/internal/master/store/versions_integration_test.go +++ b/internal/master/store/versions_integration_test.go @@ -4,6 +4,7 @@ package store import ( "context" + "strings" "testing" "time" @@ -16,6 +17,13 @@ import ( const sluskRepo = "gitea.shcizo.se/shcizo/slusk" +// Digests of the fixture services; only well-formed ones are recorded as versions. +var ( + digD1 = "sha256:" + strings.Repeat("d", 63) + "1" + digGitea = "sha256:" + strings.Repeat("a", 63) + "1" + digPg = "sha256:" + strings.Repeat("b", 63) + "1" +) + func versionsOf(t *testing.T, s *Store, repo string) []model.ImageVersion { t.Helper() got, err := s.RecentVersions(context.Background(), repo, 100) @@ -59,8 +67,8 @@ 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"} + tagless := protocol.Service{Kind: protocol.KindSwarm, SwarmService: "g_g", Image: "gitea/gitea@" + digGitea, Digest: digGitea, Status: "running"} + bare := protocol.Service{Kind: protocol.KindContainer, ContainerName: "pg", Image: "postgres", Digest: digPg, 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} @@ -71,19 +79,31 @@ func TestSnapshotRecordsRunningVersions(t *testing.T) { 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, digD1, 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) + assert.Equal(t, digGitea, 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.Equal(t, "latest", b[0].Tag, "a bare image runs latest, like the followed ref") assert.Empty(t, versionsOf(t, s, "nodigest")) } +func TestSnapshotRecordsOnlyWellFormedDigests(t *testing.T) { + s := openTestStore(t) + id := newAgent(t, s, "node-2") + bad := protocol.Service{Kind: protocol.KindContainer, ContainerName: "bad", Image: "bad:1", Digest: "sha256:short", Status: "running"} + upper := protocol.Service{Kind: protocol.KindContainer, ContainerName: "up", Image: "up:1", Digest: "sha256:" + strings.Repeat("A", 64), Status: "running"} + + require.NoError(t, s.ApplySnapshot(context.Background(), id, protocol.Hello{Hostname: "node-2"}, []protocol.Service{bad, upper}, t0)) + + assert.Empty(t, versionsOf(t, s, "bad")) + assert.Empty(t, versionsOf(t, s, "up")) +} + func TestRecentVersions(t *testing.T) { ctx := context.Background() t.Run("order, limit and repo filter", func(t *testing.T) { @@ -123,7 +143,23 @@ func TestRecentVersions(t *testing.T) { 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) + assert.Equal(t, []string{sluskRepo + ":main@sha256:m1", sluskRepo + ":solo", sluskRepo + ":1.4.2@" + digD1}, refs) + }) + + t.Run("digest-less row newer than the digest row of its tag is shown", 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", "sha256:m1", t0.Add(time.Minute)) + add("main", "", t0.Add(2*time.Minute)) + + var refs []string + for _, v := range versionsOf(t, s, sluskRepo) { + refs = append(refs, v.Ref()) + } + assert.Equal(t, []string{sluskRepo + ":main", sluskRepo + ":main@sha256:m1", sluskRepo + ":1.4.2@" + digD1}, refs) }) } @@ -158,3 +194,23 @@ func TestSetPollError(t *testing.T) { require.NoError(t, err) assert.Empty(t, got.LastPollError) } + +func TestSetLastAutoDigest(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + agent, svc := jobFixture(t, s, "a") + got, err := s.GetService(ctx, svc) + require.NoError(t, err) + assert.Empty(t, got.LastAutoDigest) + + require.NoError(t, s.SetLastAutoDigest(ctx, svc, "sha256:n1")) + 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, "sha256:n1", got.LastAutoDigest, "inventory writes do not touch it") + list, err := s.ListServices(ctx) + require.NoError(t, err) + require.Len(t, list, 1) + assert.Equal(t, "sha256:n1", list[0].LastAutoDigest) +}