diff --git a/internal/e2e/agent_master_test.go b/internal/e2e/agent_master_test.go index 44c65e1..13cfa9e 100644 --- a/internal/e2e/agent_master_test.go +++ b/internal/e2e/agent_master_test.go @@ -245,6 +245,7 @@ func (m *memStore) FinishJob(_ context.Context, agentID, jobID int64, f model.Jo if f.Status == model.JobSucceeded { svc := m.svcRows[j.ServiceID] svc.DesiredRef = f.Ref + svc.AwaitingInventory = true if _, d, _ := strings.Cut(f.Ref, "@"); !strings.Contains(requestedRef, "@") && digestPattern.MatchString(d) { svc.LastTagDigest = d } diff --git a/internal/imageref/imageref.go b/internal/imageref/imageref.go index cd268b1..43f98fd 100644 --- a/internal/imageref/imageref.go +++ b/internal/imageref/imageref.go @@ -2,7 +2,10 @@ // and digest. package imageref -import "strings" +import ( + "regexp" + "strings" +) // Ref is an image reference split into its parts. Tag and Digest are empty // when the reference has none. @@ -41,3 +44,8 @@ func (r Ref) String() string { } return s } + +var digestPattern = regexp.MustCompile(`^sha256:[0-9a-f]{64}$`) + +// IsDigest reports whether d is a well-formed sha256 digest. +func IsDigest(d string) bool { return digestPattern.MatchString(d) } diff --git a/internal/imageref/imageref_test.go b/internal/imageref/imageref_test.go index c86414b..7ce6685 100644 --- a/internal/imageref/imageref_test.go +++ b/internal/imageref/imageref_test.go @@ -1,6 +1,7 @@ package imageref import ( + "strings" "testing" "github.com/stretchr/testify/assert" @@ -27,3 +28,19 @@ func TestParse(t *testing.T) { }) } } + +func TestIsDigest(t *testing.T) { + cases := []struct { + in string + want bool + }{ + {"sha256:" + strings.Repeat("a", 64), true}, + {"sha256:" + strings.Repeat("a", 63), false}, + {"sha256:" + strings.Repeat("A", 64), false}, + {"sha512:" + strings.Repeat("a", 64), false}, + {"", false}, + } + for _, c := range cases { + assert.Equal(t, c.want, IsDigest(c.in), c.in) + } +} diff --git a/internal/master/model/job.go b/internal/master/model/job.go index ea0f24f..cbbade5 100644 --- a/internal/master/model/job.go +++ b/internal/master/model/job.go @@ -44,6 +44,27 @@ const ( ReasonNoTag = "follows no tag" ReasonRemoved = "service no longer exists" ReasonBuilt = "built from source (build:)" + + ReasonDrifted = "drifted: restore or accept the running version first" + ReasonNotDrifted = "runs its desired version" +) + +// MergeRefusedError says why two services cannot be merged. Reason is shown +// in the UI as is. +type MergeRefusedError struct { + Reason string +} + +func (e *MergeRefusedError) Error() string { return "cannot merge: " + e.Reason } + +// Reasons two services cannot be merged. ReasonSelf is reused for the +// updater itself. +const ( + ReasonMergeSame = "a service cannot be merged into itself" + ReasonMergeLiveOld = "the old service still exists" + ReasonMergeGoneNew = "the new service no longer exists" + ReasonMergeKind = "only services of the same kind can be merged" + ReasonMergeContainer = "plain containers cannot be merged" ) // JobStatus is the state of an update job. diff --git a/internal/master/model/model.go b/internal/master/model/model.go index 7ca11b7..a92b10d 100644 --- a/internal/master/model/model.go +++ b/internal/master/model/model.go @@ -68,9 +68,13 @@ type Service struct { // not start another job for the same digest, so a rollback sticks until the // registry shows a different one. LastTagDigest string - FirstSeenAt time.Time - UpdatedAt time.Time - RemovedAt *time.Time + // AwaitingInventory is true from a successful job until the next inventory + // write for the row. Until then DesiredRef is newer than ActualRef without + // the service having drifted. + AwaitingInventory bool + FirstSeenAt time.Time + UpdatedAt time.Time + RemovedAt *time.Time } // Update modes of a service. @@ -113,6 +117,63 @@ func (v ImageVersion) Ref() string { return imageref.Ref{Repo: v.Repo, Tag: v.Tag, Digest: v.Digest}.String() } +// Drifted reports whether the service runs something other than the version +// the last job deployed. A service the updater never touched, a removed one +// and one whose inventory has not caught up with the last job do not drift. +func (s Service) Drifted() bool { + return s.DesiredRef != "" && s.RemovedAt == nil && !s.AwaitingInventory && + !RunsRef(s.DesiredRef, s.ActualRef, s.ActualDigest) +} + +// RunsRef reports whether a service reporting actualRef and actualDigest runs +// the desired ref. The digest decides when both sides know one, so +// "repo:tag@D" is run by "repo:tag" reporting digest D, and by "repo@D". +// Otherwise the repo (compared exactly, without registry normalisation) and +// the followed tag must be equal. +func RunsRef(desired, actualRef, actualDigest string) bool { + d, a := imageref.Parse(desired), imageref.Parse(actualRef) + if d.Repo != a.Repo { + return false + } + ad := a.Digest + if ad == "" { + ad = actualDigest + } + if d.Digest != "" && ad != "" { + return d.Digest == ad + } + return followedTag(d) == followedTag(a) +} + +// followedTag is the tag a ref follows: a bare image follows "latest" and a +// ref with a digest but no tag follows nothing. +func followedTag(r imageref.Ref) string { + if r.Tag == "" && r.Digest == "" { + return "latest" + } + return r.Tag +} + +// RunningRef is the ref to record as the desired one when the running version +// is accepted: actualRef itself if it carries a digest, "repo:tag@digest" if +// only actualDigest is known (a bare image is "latest"), else actualRef. +func RunningRef(actualRef, actualDigest string) string { + r := imageref.Parse(actualRef) + if r.Digest != "" || !imageref.IsDigest(actualDigest) { + return actualRef + } + r.Tag, r.Digest = followedTag(r), actualDigest + return r.String() +} + +// Move says that the removed service FromID has probably moved to the live +// service ToID on another agent. The names are the agents' own; a revoked +// agent keeps its name. +type Move struct { + FromID, ToID int64 + FromAgent, ToAgent 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 96cb8af..5f3ca2b 100644 --- a/internal/master/model/model_test.go +++ b/internal/master/model/model_test.go @@ -2,6 +2,7 @@ package model import ( "testing" + "time" "github.com/stretchr/testify/assert" @@ -58,6 +59,84 @@ func TestImageVersionRef(t *testing.T) { } } +const ( + dA = "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + dB = "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" +) + +func TestRunsRef(t *testing.T) { + cases := []struct { + name string + desired, actualRef, actualDig string + want bool + }{ + {"same string", "repo:v1@" + dA, "repo:v1@" + dA, "", true}, + {"tag and digest, actual reports it separately", "repo:tag@" + dA, "repo:tag", dA, true}, + {"bare image reports latest digest", "postgres:latest@" + dA, "postgres", dA, true}, + {"tagless actual with the same digest", "repo:tag@" + dA, "repo@" + dA, "", true}, + {"same digest under another tag", "repo:v1@" + dA, "repo:v2@" + dA, "", true}, + {"same digest, actual tag differs, digest reported", "repo:v1@" + dA, "repo:v2", dA, true}, + {"same tag, actual digest unknown", "repo:tag@" + dA, "repo:tag", "", true}, + {"desired without digest, same tag", "repo:tag", "repo:tag@" + dA, "", true}, + {"bare desired, latest actual", "repo", "repo:latest", "", true}, + {"registry with port", "localhost:5000/foo:v1@" + dA, "localhost:5000/foo:v1", dA, true}, + {"registry with port, other digest", "localhost:5000/foo:v1@" + dA, "localhost:5000/foo:v1", dB, false}, + {"same tag, other digest", "repo:tag@" + dA, "repo:tag", dB, false}, + {"other digest in the ref", "repo:tag@" + dA, "repo:tag@" + dB, "", false}, + {"other tag, digest unknown", "repo:v1@" + dA, "repo:v2", "", false}, + {"tagless desired, tagged actual, digest unknown", "repo@" + dA, "repo:v1", "", false}, + {"other repo", "repo:tag@" + dA, "other:tag", dA, false}, + {"no docker.io normalisation", "nginx:1@" + dA, "docker.io/library/nginx:1", dA, false}, + {"case-sensitive repo", "Repo:1@" + dA, "repo:1", dA, false}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + assert.Equal(t, c.want, RunsRef(c.desired, c.actualRef, c.actualDig)) + }) + } +} + +func TestDrifted(t *testing.T) { + now := time.Now() + drifted := Service{DesiredRef: "repo:tag@" + dA, ActualRef: "repo:tag", ActualDigest: dB} + cases := []struct { + name string + mutate func(*Service) + want bool + }{ + {"runs another digest", func(*Service) {}, true}, + {"never updated", func(s *Service) { s.DesiredRef = "" }, false}, + {"removed", func(s *Service) { s.RemovedAt = &now }, false}, + {"awaiting inventory", func(s *Service) { s.AwaitingInventory = true }, false}, + {"runs the desired digest", func(s *Service) { s.ActualDigest = dA }, false}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + svc := drifted + c.mutate(&svc) + assert.Equal(t, c.want, svc.Drifted()) + }) + } +} + +func TestRunningRef(t *testing.T) { + cases := []struct { + name, actualRef, actualDigest, want string + }{ + {"ref with digest is kept", "repo:v1@" + dB, dA, "repo:v1@" + dB}, + {"tag plus reported digest", "repo:v1", dA, "repo:v1@" + dA}, + {"bare image becomes latest", "postgres", dA, "postgres:latest@" + dA}, + {"registry with port", "localhost:5000/foo:v1", dA, "localhost:5000/foo:v1@" + dA}, + {"no digest known", "repo:v1", "", "repo:v1"}, + {"malformed digest", "repo:v1", "sha256:abc", "repo:v1"}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + assert.Equal(t, c.want, RunningRef(c.actualRef, c.actualDigest)) + }) + } +} + func TestJob_Active(t *testing.T) { cases := []struct { status JobStatus diff --git a/internal/master/store/drift_merge_integration_test.go b/internal/master/store/drift_merge_integration_test.go new file mode 100644 index 0000000..1433920 --- /dev/null +++ b/internal/master/store/drift_merge_integration_test.go @@ -0,0 +1,550 @@ +//go:build integration + +package store + +import ( + "context" + "errors" + "strings" + "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" +) + +var ( + digNew = "sha256:" + strings.Repeat("e", 64) + digOld = "sha256:" + strings.Repeat("f", 64) + sluskDrift = "gitea.shcizo.se/shcizo/slusk" +) + +// serviceID returns the ID of the service the agent reports under svc's key. +func serviceID(t *testing.T, s *Store, agentID int64, svc protocol.Service) int64 { + t.Helper() + var id int64 + require.NoError(t, s.pool.QueryRow(context.Background(), + `SELECT id FROM services WHERE location_key = $1`, model.LocationKey(agentID, "", svc.Key())).Scan(&id)) + return id +} + +// finishJob runs a job for the service to success with the deployed ref. +func finishJob(t *testing.T, s *Store, agentID, svcID int64, toRef, deployed string) { + t.Helper() + ctx := context.Background() + nj := newJob(svcID, agentID) + nj.ToRef = toRef + j, err := s.CreateJob(ctx, nj) + require.NoError(t, err) + ok, err := s.FinishJob(ctx, agentID, j.ID, model.JobFinish{Status: model.JobSucceeded, Ref: deployed, StartedAt: t0, FinishedAt: t0}) + require.NoError(t, err) + require.True(t, ok) +} + +func mustExec(t *testing.T, s *Store, sql string, args ...any) { + t.Helper() + _, err := s.pool.Exec(context.Background(), sql, args...) + require.NoError(t, err) +} + +func TestFinishJob_SetsAwaitingInventoryUntilTheNextUpsert(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.False(t, got.AwaitingInventory) + + finishJob(t, s, agent, svc, "r:1", "r:1@"+digNew) + got, err = s.GetService(ctx, svc) + require.NoError(t, err) + assert.True(t, got.AwaitingInventory, "set by the successful job") + assert.False(t, got.Drifted(), "desired is ahead of actual only until the inventory arrives") + + require.NoError(t, s.UpsertService(ctx, agent, "", slusk, t0.Add(time.Minute))) + got, err = s.GetService(ctx, svc) + require.NoError(t, err) + assert.False(t, got.AwaitingInventory, "cleared by the agent's upsert") + assert.True(t, got.Drifted(), "the agent still reports the old version") +} + +func TestFinishJob_FailedJobDoesNotSetAwaitingInventory(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + agent, svc := jobFixture(t, s, "a") + j, err := s.CreateJob(ctx, newJob(svc, agent)) + require.NoError(t, err) + + ok, err := s.FinishJob(ctx, agent, j.ID, model.JobFinish{Status: model.JobFailed, Error: "boom", StartedAt: t0, FinishedAt: t0}) + require.NoError(t, err) + require.True(t, ok) + + got, err := s.GetService(ctx, svc) + require.NoError(t, err) + assert.False(t, got.AwaitingInventory) +} + +func TestCreateJob_UnknownServiceIsNotFound(t *testing.T) { + s := openTestStore(t) + agent := newAgent(t, s, "a") + + _, err := s.CreateJob(context.Background(), newJob(9999, agent)) + + assert.ErrorIs(t, err, model.ErrNotFound) +} + +// driftedFixture is a service updated to digNew whose agent now reports digD1. +func driftedFixture(t *testing.T, s *Store) (agent, svc int64) { + t.Helper() + agent, svc = jobFixture(t, s, "a") + finishJob(t, s, agent, svc, "r:1", sluskDrift+":1.4.2@"+digNew) + require.NoError(t, s.UpsertService(context.Background(), agent, "", slusk, t0.Add(time.Minute))) + return agent, svc +} + +func TestAcceptRunning_WritesTheDesiredRef(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + _, svc := driftedFixture(t, s) + require.NoError(t, s.SetLastTagDigest(ctx, svc, "sha256:ltd")) + require.NoError(t, s.SetAvailable(ctx, svc, model.Available{Ref: sluskDrift + ":1.5.0", Source: model.SourcePoll, At: t0})) + require.NoError(t, s.SetUpdateMode(ctx, svc, model.ModeAuto)) + before, err := s.GetService(ctx, svc) + require.NoError(t, err) + require.True(t, before.Drifted()) + + running := model.RunningRef(before.ActualRef, before.ActualDigest) + require.NoError(t, s.AcceptRunning(ctx, svc, before.ActualRef, before.ActualDigest, running)) + + got, err := s.GetService(ctx, svc) + require.NoError(t, err) + assert.Equal(t, sluskDrift+":1.4.2@"+digD1, got.DesiredRef) + assert.False(t, got.Drifted()) + assert.False(t, got.AwaitingInventory) + assert.Equal(t, "sha256:ltd", got.LastTagDigest, "last_tag_digest untouched") + assert.Equal(t, sluskDrift+":1.5.0", got.AvailableRef, "available suggestion untouched") + assert.Equal(t, model.ModeAuto, got.UpdateMode) +} + +func TestAcceptRunning_Refusals(t *testing.T) { + tests := []struct { + name string + prepare func(t *testing.T, s *Store, agent, svc int64) (id int64, actualRef, actualDigest string) + wantError error + }{ + {"running version changed", func(t *testing.T, s *Store, agent, svc int64) (int64, string, string) { + return svc, sluskDrift + ":1.4.2", digOld + }, model.ErrConflict}, + {"actual ref changed", func(t *testing.T, s *Store, agent, svc int64) (int64, string, string) { + return svc, sluskDrift + ":1.3.0", digD1 + }, model.ErrConflict}, + {"never updated", func(t *testing.T, s *Store, agent, svc int64) (int64, string, string) { + mustExec(t, s, `UPDATE services SET desired_ref = '' WHERE id = $1`, svc) + return svc, sluskDrift + ":1.4.2", digD1 + }, model.ErrConflict}, + {"removed", func(t *testing.T, s *Store, agent, svc int64) (int64, string, string) { + require.NoError(t, s.RemoveService(context.Background(), agent, "", slusk.Key(), t0.Add(time.Hour))) + return svc, sluskDrift + ":1.4.2", digD1 + }, model.ErrNotFound}, + {"unknown", func(t *testing.T, s *Store, agent, svc int64) (int64, string, string) { + return 9999, sluskDrift + ":1.4.2", digD1 + }, model.ErrNotFound}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + s := openTestStore(t) + agent, svc := driftedFixture(t, s) + id, ref, digest := tt.prepare(t, s, agent, svc) + + err := s.AcceptRunning(context.Background(), id, ref, digest, "x:1@"+digOld) + + assert.ErrorIs(t, err, tt.wantError) + got, gerr := s.GetService(context.Background(), svc) + require.NoError(t, gerr) + assert.NotEqual(t, "x:1@"+digOld, got.DesiredRef, "nothing written") + }) + } +} + +// moveFixture has agent "a" with slusk removed after it was updated (history, +// auto mode) and agent "b" reporting a new slusk row. +type moveFixture struct { + a, b int64 + from, to int64 +} + +func newMoveFixture(t *testing.T, s *Store) moveFixture { + t.Helper() + ctx := context.Background() + f := moveFixture{a: newAgent(t, s, "a"), b: newAgent(t, s, "b")} + require.NoError(t, s.ApplySnapshot(ctx, f.a, worker, []protocol.Service{slusk}, t0)) + f.from = serviceID(t, s, f.a, slusk) + finishJob(t, s, f.a, f.from, "r:1", sluskDrift+":1.4.2@"+digNew) + require.NoError(t, s.SetUpdateMode(ctx, f.from, model.ModeAuto)) + require.NoError(t, s.ApplySnapshot(ctx, f.a, worker, nil, t0.Add(time.Hour))) + require.NoError(t, s.ApplySnapshot(ctx, f.b, worker, []protocol.Service{slusk}, t0.Add(2*time.Hour))) + f.to = serviceID(t, s, f.b, slusk) + return f +} + +func TestMoveCandidates(t *testing.T) { + tests := []struct { + name string + mutate func(t *testing.T, s *Store, f moveFixture) + want bool + from string // agent name of the removed row + }{ + {"one removed and one live", func(*testing.T, *Store, moveFixture) {}, true, "a"}, + {"removed row's agent was revoked", func(t *testing.T, s *Store, f moveFixture) { + _, err := s.RevokeAgent(context.Background(), f.a, t0.Add(3*time.Hour)) + require.NoError(t, err) + }, true, "a"}, + {"two live targets", func(t *testing.T, s *Store, f moveFixture) { + c := newAgent(t, s, "c") + require.NoError(t, s.ApplySnapshot(context.Background(), c, worker, []protocol.Service{slusk}, t0.Add(3*time.Hour))) + }, false, ""}, + {"two removed sources", func(t *testing.T, s *Store, f moveFixture) { + ctx := context.Background() + d := newAgent(t, s, "d") + require.NoError(t, s.ApplySnapshot(ctx, d, worker, []protocol.Service{slusk}, t0)) + require.NoError(t, s.ApplySnapshot(ctx, d, worker, nil, t0.Add(time.Hour))) + }, false, ""}, + {"target has history", func(t *testing.T, s *Store, f moveFixture) { + finishJob(t, s, f.b, f.to, "r:1", sluskDrift+":1.4.2@"+digNew) + }, false, ""}, + {"target has a desired ref", func(t *testing.T, s *Store, f moveFixture) { + mustExec(t, s, `UPDATE services SET desired_ref = 'r:1@x' WHERE id = $1`, f.to) + }, false, ""}, + {"source has no history", func(t *testing.T, s *Store, f moveFixture) { + mustExec(t, s, `DELETE FROM update_jobs WHERE service_id = $1`, f.from) + mustExec(t, s, `UPDATE services SET desired_ref = '', update_mode = 'manual' WHERE id = $1`, f.from) + }, false, ""}, + {"source only has auto mode", func(t *testing.T, s *Store, f moveFixture) { + mustExec(t, s, `DELETE FROM update_jobs WHERE service_id = $1`, f.from) + mustExec(t, s, `UPDATE services SET desired_ref = '' WHERE id = $1`, f.from) + }, true, "a"}, + {"source only has jobs", func(t *testing.T, s *Store, f moveFixture) { + mustExec(t, s, `UPDATE services SET desired_ref = '', update_mode = 'manual' WHERE id = $1`, f.from) + }, true, "a"}, + {"target is older than the source", func(t *testing.T, s *Store, f moveFixture) { + mustExec(t, s, `UPDATE services SET first_seen_at = $2 WHERE id = $1`, f.to, t0.Add(-time.Hour)) + }, false, ""}, + {"plain containers", func(t *testing.T, s *Store, f moveFixture) { + mustExec(t, s, `UPDATE services SET kind = 'container' WHERE id IN ($1, $2)`, f.from, f.to) + }, false, ""}, + {"swarm services", func(t *testing.T, s *Store, f moveFixture) { + mustExec(t, s, `UPDATE services SET kind = 'swarm' WHERE id IN ($1, $2)`, f.from, f.to) + }, false, ""}, + {"the updater itself (target)", func(t *testing.T, s *Store, f moveFixture) { + mustExec(t, s, `UPDATE services SET self = true WHERE id = $1`, f.to) + }, false, ""}, + {"the updater itself (source)", func(t *testing.T, s *Store, f moveFixture) { + mustExec(t, s, `UPDATE services SET self = true WHERE id = $1`, f.from) + }, false, ""}, + {"live row also removed", func(t *testing.T, s *Store, f moveFixture) { + require.NoError(t, s.RemoveService(context.Background(), f.b, "", slusk.Key(), t0.Add(3*time.Hour))) + }, false, ""}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + s := openTestStore(t) + f := newMoveFixture(t, s) + tt.mutate(t, s, f) + + got, err := s.MoveCandidates(context.Background()) + + require.NoError(t, err) + if !tt.want { + assert.Empty(t, got) + return + } + assert.Equal(t, []model.Move{{FromID: f.from, ToID: f.to, FromAgent: tt.from, ToAgent: "b"}}, got) + }) + } +} + +func TestMergeService_MovesHistoryModeAndDesiredToTheNewLocation(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + f := newMoveFixture(t, s) + finishJob(t, s, f.b, f.to, "r:2", sluskDrift+":2.0.0@"+digOld) + require.NoError(t, s.SetLastTagDigest(ctx, f.to, "sha256:fromto")) + require.NoError(t, s.SetAvailable(ctx, f.to, model.Available{Ref: sluskDrift + ":3", Digest: digNew, Source: model.SourcePoll, At: t0})) + require.NoError(t, s.SetPollError(ctx, []int64{f.to}, "registry down")) + // A job on the old row is finished, and awaiting_inventory is set on both. + before, err := s.GetService(ctx, f.from) + require.NoError(t, err) + + require.NoError(t, s.MergeService(ctx, f.from, f.to)) + + got, err := s.GetService(ctx, f.from) + require.NoError(t, err) + assert.Equal(t, f.from, got.ID, "the old id survives") + assert.Nil(t, got.RemovedAt) + assert.False(t, got.AwaitingInventory) + assert.Equal(t, "b", got.AgentName) + assert.Equal(t, model.LocationKey(f.b, "", slusk.Key()), got.LocationKey) + assert.Equal(t, before.FirstSeenAt, got.FirstSeenAt, "first_seen_at kept") + assert.Equal(t, model.ModeAuto, got.UpdateMode, "mode kept") + assert.Equal(t, before.DesiredRef, got.DesiredRef, "desired ref kept") + assert.Equal(t, slusk.Image, got.ActualRef, "inventory from the duplicate") + assert.Equal(t, digD1, got.ActualDigest) + assert.Equal(t, "running", got.Status) + assert.Equal(t, sluskDrift+":3", got.AvailableRef) + assert.Equal(t, digNew, got.AvailableDigest) + assert.Equal(t, "registry down", got.LastPollError) + assert.Equal(t, before.LastTagDigest, got.LastTagDigest, "an existing last_tag_digest wins") + + _, err = s.GetService(ctx, f.to) + assert.ErrorIs(t, err, model.ErrNotFound, "the duplicate is deleted") + jobs, err := s.ListJobs(ctx, model.JobFilter{ServiceID: f.from}) + require.NoError(t, err) + assert.Len(t, jobs, 2, "the duplicate's job moved to the survivor") + live, err := s.ListServices(ctx) + require.NoError(t, err) + assert.Equal(t, []string{"slusk/web"}, names(live)) +} + +func TestMergeService_TakesLastTagDigestOfTheDuplicateWhenTheOldRowHasNone(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + f := newMoveFixture(t, s) + mustExec(t, s, `UPDATE services SET last_tag_digest = '' WHERE id = $1`, f.from) + require.NoError(t, s.SetLastTagDigest(ctx, f.to, "sha256:fromto")) + + require.NoError(t, s.MergeService(ctx, f.from, f.to)) + + got, err := s.GetService(ctx, f.from) + require.NoError(t, err) + assert.Equal(t, "sha256:fromto", got.LastTagDigest) +} + +func TestMergeService_ManualMergeOfAnyRemovedServiceOfTheSameKind(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + f := newMoveFixture(t, s) + other := protocol.Service{Kind: protocol.KindCompose, ComposeProject: "renamed", ComposeService: "app", Image: "x/y:1", Status: "running"} + require.NoError(t, s.UpsertService(ctx, f.b, "", other, t0.Add(3*time.Hour))) + target := serviceID(t, s, f.b, other) + + require.NoError(t, s.MergeService(ctx, f.from, target)) + + got, err := s.GetService(ctx, f.from) + require.NoError(t, err) + assert.Equal(t, "renamed/app", got.DisplayName(), "names may differ") +} + +func TestMergeService_SameAgentIsAllowed(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + f := newMoveFixture(t, s) + other := protocol.Service{Kind: protocol.KindCompose, ComposeProject: "renamed", ComposeService: "app", Image: "x/y:1", Status: "running"} + require.NoError(t, s.UpsertService(ctx, f.a, "", other, t0.Add(3*time.Hour))) + + require.NoError(t, s.MergeService(ctx, f.from, serviceID(t, s, f.a, other))) + + got, err := s.GetService(ctx, f.from) + require.NoError(t, err) + assert.Equal(t, "a", got.AgentName) +} + +func TestMergeService_Swarm(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + m1 := newAgent(t, s, "m1") + require.NoError(t, s.ApplySnapshot(ctx, m1, manager, []protocol.Service{gitea}, t0)) + oldID := serviceIDByName(t, s, "gitea_gitea") + require.NoError(t, s.ApplySnapshot(ctx, m1, manager, nil, t0.Add(time.Hour))) + renamed := gitea + renamed.SwarmService = "gitea2_gitea" + require.NoError(t, s.ApplySnapshot(ctx, m1, manager, []protocol.Service{renamed}, t0.Add(2*time.Hour))) + newID := serviceIDByName(t, s, "gitea2_gitea") + + require.NoError(t, s.MergeService(ctx, oldID, newID)) + + got, err := s.GetService(ctx, oldID) + require.NoError(t, err) + assert.Equal(t, "gitea2_gitea", got.SwarmService) + assert.Nil(t, got.RemovedAt) +} + +func serviceIDByName(t *testing.T, s *Store, swarmService string) int64 { + t.Helper() + var id int64 + require.NoError(t, s.pool.QueryRow(context.Background(), + `SELECT id FROM services WHERE swarm_service = $1`, swarmService).Scan(&id)) + return id +} + +func TestMergeService_Refusals(t *testing.T) { + refused := func(reason string) func(*testing.T, error) { + return func(t *testing.T, err error) { + var mr *model.MergeRefusedError + require.True(t, errors.As(err, &mr), "got %v", err) + assert.Equal(t, reason, mr.Reason) + } + } + is := func(want error) func(*testing.T, error) { + return func(t *testing.T, err error) { assert.ErrorIs(t, err, want) } + } + tests := []struct { + name string + run func(t *testing.T, s *Store, f moveFixture) error + assert func(*testing.T, error) + }{ + {"into itself", func(t *testing.T, s *Store, f moveFixture) error { + return s.MergeService(context.Background(), f.to, f.to) + }, refused(model.ReasonMergeSame)}, + {"the old service still exists", func(t *testing.T, s *Store, f moveFixture) error { + mustExec(t, s, `UPDATE services SET removed_at = NULL WHERE id = $1`, f.from) + return s.MergeService(context.Background(), f.from, f.to) + }, refused(model.ReasonMergeLiveOld)}, + {"the new service was removed", func(t *testing.T, s *Store, f moveFixture) error { + require.NoError(t, s.RemoveService(context.Background(), f.b, "", slusk.Key(), t0.Add(3*time.Hour))) + return s.MergeService(context.Background(), f.from, f.to) + }, refused(model.ReasonMergeGoneNew)}, + {"different kinds", func(t *testing.T, s *Store, f moveFixture) error { + mustExec(t, s, `UPDATE services SET kind = 'swarm' WHERE id = $1`, f.to) + return s.MergeService(context.Background(), f.from, f.to) + }, refused(model.ReasonMergeKind)}, + {"plain containers", func(t *testing.T, s *Store, f moveFixture) error { + mustExec(t, s, `UPDATE services SET kind = 'container' WHERE id IN ($1, $2)`, f.from, f.to) + return s.MergeService(context.Background(), f.from, f.to) + }, refused(model.ReasonMergeContainer)}, + {"the updater itself (new)", func(t *testing.T, s *Store, f moveFixture) error { + mustExec(t, s, `UPDATE services SET self = true WHERE id = $1`, f.to) + return s.MergeService(context.Background(), f.from, f.to) + }, refused(model.ReasonSelf)}, + {"the updater itself (old)", func(t *testing.T, s *Store, f moveFixture) error { + mustExec(t, s, `UPDATE services SET self = true WHERE id = $1`, f.from) + return s.MergeService(context.Background(), f.from, f.to) + }, refused(model.ReasonSelf)}, + {"active job on the old service", func(t *testing.T, s *Store, f moveFixture) error { + _, err := s.CreateJob(context.Background(), newJob(f.from, f.a)) + require.NoError(t, err) + return s.MergeService(context.Background(), f.from, f.to) + }, is(model.ErrJobActive)}, + {"active job on the new service", func(t *testing.T, s *Store, f moveFixture) error { + _, err := s.CreateJob(context.Background(), newJob(f.to, f.b)) + require.NoError(t, err) + return s.MergeService(context.Background(), f.from, f.to) + }, is(model.ErrJobActive)}, + {"unknown old service", func(t *testing.T, s *Store, f moveFixture) error { + return s.MergeService(context.Background(), 9999, f.to) + }, is(model.ErrNotFound)}, + {"unknown new service", func(t *testing.T, s *Store, f moveFixture) error { + return s.MergeService(context.Background(), f.from, 9999) + }, is(model.ErrNotFound)}, + {"the new service's agent is revoked", func(t *testing.T, s *Store, f moveFixture) error { + _, err := s.RevokeAgent(context.Background(), f.b, t0.Add(3*time.Hour)) + require.NoError(t, err) + return s.MergeService(context.Background(), f.from, f.to) + }, is(model.ErrRevoked)}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + s := openTestStore(t) + f := newMoveFixture(t, s) + var rowsBefore int + require.NoError(t, s.pool.QueryRow(context.Background(), `SELECT count(*) FROM services`).Scan(&rowsBefore)) + + err := tt.run(t, s, f) + + tt.assert(t, err) + var rowsAfter int + require.NoError(t, s.pool.QueryRow(context.Background(), `SELECT count(*) FROM services`).Scan(&rowsAfter)) + assert.Equal(t, rowsBefore, rowsAfter, "nothing is deleted when the merge is refused") + }) + } +} + +func TestMergeService_InventoryAfterTheMerge(t *testing.T) { + setup := func(t *testing.T) (*Store, moveFixture) { + s := openTestStore(t) + f := newMoveFixture(t, s) + require.NoError(t, s.MergeService(context.Background(), f.from, f.to)) + return s, f + } + + t.Run("an upsert for the new location updates the survivor", func(t *testing.T) { + s, f := setup(t) + ctx := context.Background() + moved := slusk + moved.Digest = digOld + require.NoError(t, s.UpsertService(ctx, f.b, "", moved, t0.Add(4*time.Hour))) + + live, err := s.ListServices(ctx) + require.NoError(t, err) + require.Len(t, live, 1) + assert.Equal(t, f.from, live[0].ID) + assert.Equal(t, digOld, live[0].ActualDigest) + }) + + t.Run("a snapshot keeps the survivor", func(t *testing.T) { + s, f := setup(t) + ctx := context.Background() + require.NoError(t, s.ApplySnapshot(ctx, f.b, worker, []protocol.Service{slusk}, t0.Add(4*time.Hour))) + + got, err := s.GetService(ctx, f.from) + require.NoError(t, err) + assert.Nil(t, got.RemovedAt) + live, err := s.ListServices(ctx) + require.NoError(t, err) + assert.Len(t, live, 1) + }) + + t.Run("the old agent returning with the stack creates a new row", func(t *testing.T) { + s, f := setup(t) + ctx := context.Background() + require.NoError(t, s.ApplySnapshot(ctx, f.a, worker, []protocol.Service{slusk}, t0.Add(4*time.Hour))) + + returned := serviceID(t, s, f.a, slusk) + assert.NotEqual(t, f.from, returned) + got, err := s.GetService(ctx, returned) + require.NoError(t, err) + assert.Equal(t, model.ModeManual, got.UpdateMode) + assert.Empty(t, got.DesiredRef) + live, err := s.ListServices(ctx) + require.NoError(t, err) + assert.Len(t, live, 2) + }) +} + +func TestMergeSources(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + f := newMoveFixture(t, s) + // A removed service with another name, removed later than the matching one. + other := protocol.Service{Kind: protocol.KindCompose, ComposeProject: "renamed", ComposeService: "app", Image: "x/y:1", Status: "running"} + require.NoError(t, s.UpsertService(ctx, f.a, "", other, t0)) + require.NoError(t, s.RemoveService(ctx, f.a, "", other.Key(), t0.Add(5*time.Hour))) + otherID := serviceID(t, s, f.a, other) + // Other kinds, the updater itself and a live service never show up. + require.NoError(t, s.ApplySnapshot(ctx, f.a, worker, []protocol.Service{adhoc}, t0.Add(6*time.Hour))) + require.NoError(t, s.ApplySnapshot(ctx, f.a, worker, nil, t0.Add(7*time.Hour))) + selfSvc := protocol.Service{Kind: protocol.KindCompose, ComposeProject: "pu", ComposeService: "agent", Image: "pu:1", Status: "running", Self: true} + require.NoError(t, s.UpsertService(ctx, f.a, "", selfSvc, t0)) + require.NoError(t, s.RemoveService(ctx, f.a, "", selfSvc.Key(), t0.Add(8*time.Hour))) + + got, err := s.MergeSources(ctx, f.to) + + require.NoError(t, err) + var ids []int64 + for _, m := range got { + ids = append(ids, m.ID) + } + assert.Equal(t, []int64{f.from, otherID}, ids, "same name first, then the most recently removed") + assert.Equal(t, "a", got[0].AgentName) + + _, err = s.MergeSources(ctx, 9999) + assert.ErrorIs(t, err, model.ErrNotFound) + + containerID := serviceID(t, s, f.a, adhoc) + got, err = s.MergeSources(ctx, containerID) + require.NoError(t, err) + assert.Empty(t, got, "plain containers cannot be merged") +} diff --git a/internal/master/store/jobs.go b/internal/master/store/jobs.go index 1917cf2..6d99ea1 100644 --- a/internal/master/store/jobs.go +++ b/internal/master/store/jobs.go @@ -10,6 +10,7 @@ import ( "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgconn" + "github.com/shcizo/package-updater/internal/imageref" "github.com/shcizo/package-updater/internal/master/model" "github.com/shcizo/package-updater/internal/protocol" ) @@ -90,6 +91,9 @@ func (s *Store) CreateJob(ctx context.Context, j model.NewJob) (model.Job, error } return active, model.ErrJobActive } + if errors.As(err, &pgErr) && pgErr.Code == "23503" && pgErr.ConstraintName == "update_jobs_service_id_fkey" { + return model.Job{}, model.ErrNotFound + } if err != nil { return model.Job{}, fmt.Errorf("create job: %w", err) } @@ -232,11 +236,11 @@ func (s *Store) FinishJob(ctx context.Context, agentID, jobID int64, f model.Job // A plain tag job, whoever started it, deployed the tag at the digest // the agent pinned; remember it. Pinned and tagless picks do not. tagDigest := "" - if _, d, _ := strings.Cut(f.Ref, "@"); requestedDigest == "" && digestPattern.MatchString(d) { + if _, d, _ := strings.Cut(f.Ref, "@"); requestedDigest == "" && imageref.IsDigest(d) { tagDigest = d } if _, err := tx.Exec(ctx, ` - UPDATE services SET desired_ref = $2, + UPDATE services SET desired_ref = $2, awaiting_inventory = true, last_tag_digest = CASE WHEN $6 <> '' THEN $6 ELSE last_tag_digest END, 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, diff --git a/internal/master/store/merge.go b/internal/master/store/merge.go new file mode 100644 index 0000000..0a4f0a7 --- /dev/null +++ b/internal/master/store/merge.go @@ -0,0 +1,249 @@ +package store + +import ( + "context" + "errors" + "fmt" + + "github.com/jackc/pgx/v5" + + "github.com/shcizo/package-updater/internal/master/model" + "github.com/shcizo/package-updater/internal/protocol" +) + +// moveCandidatesSQL pairs a removed Compose service with a live one of the +// same project and service name on another agent. A pair counts only when it +// is the single one for both rows, and the live row is new (newer than the +// removed one, never updated, no jobs) while the removed one has something +// worth moving (a desired ref, auto mode or jobs). +const moveCandidatesSQL = ` +WITH pairs AS ( + SELECT f.id AS from_id, t.id AS to_id + FROM services f + JOIN services t ON t.kind = 'compose' AND t.removed_at IS NULL AND NOT t.self + AND t.compose_project = f.compose_project AND t.compose_service = f.compose_service + AND t.agent_id IS DISTINCT FROM f.agent_id + WHERE f.kind = 'compose' AND f.removed_at IS NOT NULL AND NOT f.self +), uniq AS ( + SELECT from_id, to_id FROM pairs p + WHERE (SELECT count(*) FROM pairs q WHERE q.from_id = p.from_id) = 1 + AND (SELECT count(*) FROM pairs q WHERE q.to_id = p.to_id) = 1 +) +SELECT u.from_id, u.to_id, COALESCE(fa.name, ''), COALESCE(ta.name, '') +FROM uniq u +JOIN services f ON f.id = u.from_id +JOIN services t ON t.id = u.to_id +LEFT JOIN agents fa ON fa.id = f.agent_id +LEFT JOIN agents ta ON ta.id = t.agent_id +WHERE t.first_seen_at > f.first_seen_at + AND t.desired_ref = '' + AND NOT EXISTS (SELECT 1 FROM update_jobs j WHERE j.service_id = t.id) + AND (f.desired_ref <> '' OR f.update_mode = 'auto' + OR EXISTS (SELECT 1 FROM update_jobs j WHERE j.service_id = f.id)) +ORDER BY u.to_id` + +// MoveCandidates returns the removed Compose services that look like they +// moved to another agent; see model.Move. It only suggests: nothing is merged. +func (s *Store) MoveCandidates(ctx context.Context) ([]model.Move, error) { + rows, err := s.pool.Query(ctx, moveCandidatesSQL) + if err != nil { + return nil, fmt.Errorf("move candidates: %w", err) + } + defer rows.Close() + var out []model.Move + for rows.Next() { + var m model.Move + if err := rows.Scan(&m.FromID, &m.ToID, &m.FromAgent, &m.ToAgent); err != nil { + return nil, fmt.Errorf("scan move: %w", err) + } + out = append(out, m) + } + return out, rows.Err() +} + +// MergeSources returns up to 50 removed services that could be merged into +// the service toID: the same kind (Compose or Swarm), not the updater itself, +// those with the same name first, then the most recently removed. Returns +// model.ErrNotFound if toID does not exist. +func (s *Store) MergeSources(ctx context.Context, toID int64) ([]model.Service, error) { + to, err := s.GetService(ctx, toID) + if err != nil { + return nil, err + } + if to.Kind == protocol.KindContainer { + return nil, nil + } + rows, err := s.pool.Query(ctx, ` + SELECT `+serviceColumns+` + FROM services s LEFT JOIN agents a ON a.id = s.agent_id + WHERE s.removed_at IS NOT NULL AND s.kind = $1 AND NOT s.self AND s.id <> $2 + ORDER BY (s.compose_project, s.compose_service, s.swarm_service) = ($3, $4, $5) DESC, + s.removed_at DESC, s.id DESC + LIMIT 50`, + string(to.Kind), toID, to.ComposeProject, to.ComposeService, to.SwarmService) + if err != nil { + return nil, fmt.Errorf("merge sources: %w", err) + } + defer rows.Close() + var out []model.Service + for rows.Next() { + m, err := scanService(rows) + if err != nil { + return nil, fmt.Errorf("scan service: %w", err) + } + out = append(out, m) + } + return out, rows.Err() +} + +// mergeRow is what MergeService needs to know about one of the two rows. +type mergeRow struct { + kind string + agentID *int64 + removed bool + self bool +} + +// MergeService moves the live service toID onto the removed service fromID: +// the old row keeps its id, first_seen_at, update mode and desired ref, takes +// the location and inventory of the live row (and its available suggestion), +// becomes live, and the live row is deleted. The jobs of the live row move to +// the old one. This is the only place a services row is deleted. +// +// It returns *model.MergeRefusedError when the pair cannot be merged, +// model.ErrJobActive when either row has a queued or running job, +// model.ErrRevoked when the live row's agent is revoked, model.ErrNotFound +// when a row does not exist, and model.ErrConflict when the live row moved to +// another agent while the merge started. +// +// Locks are taken in the order inventory writes use: the agent row first, +// then the service rows by id. The live row is deleted before the old one is +// updated because location_key is unique. +func (s *Store) MergeService(ctx context.Context, fromID, toID int64) error { + if fromID == toID { + return &model.MergeRefusedError{Reason: model.ReasonMergeSame} + } + tx, err := s.pool.Begin(ctx) + if err != nil { + return fmt.Errorf("begin merge: %w", err) + } + defer func() { _ = tx.Rollback(ctx) }() + + var toAgent *int64 + err = tx.QueryRow(ctx, `SELECT agent_id FROM services WHERE id = $1`, toID).Scan(&toAgent) + if errors.Is(err, pgx.ErrNoRows) { + return model.ErrNotFound + } + if err != nil { + return fmt.Errorf("merge: find new service: %w", err) + } + if toAgent != nil { + if err := lockActiveAgent(ctx, tx, *toAgent); err != nil { + return err + } + } + + rows, err := tx.Query(ctx, ` + SELECT id, kind, agent_id, removed_at IS NOT NULL, self FROM services + WHERE id IN ($1, $2) ORDER BY id FOR UPDATE`, fromID, toID) + if err != nil { + return fmt.Errorf("merge: lock services: %w", err) + } + locked := map[int64]mergeRow{} + for rows.Next() { + var id int64 + var r mergeRow + if err := rows.Scan(&id, &r.kind, &r.agentID, &r.removed, &r.self); err != nil { + rows.Close() + return fmt.Errorf("merge: scan service: %w", err) + } + locked[id] = r + } + rows.Close() + if err := rows.Err(); err != nil { + return fmt.Errorf("merge: lock services: %w", err) + } + from, okFrom := locked[fromID] + to, okTo := locked[toID] + if !okFrom || !okTo { + return model.ErrNotFound + } + if (to.agentID == nil) != (toAgent == nil) || (toAgent != nil && *to.agentID != *toAgent) { + return model.ErrConflict + } + if reason := mergeRefusal(from, to); reason != "" { + return &model.MergeRefusedError{Reason: reason} + } + + var active bool + if err := tx.QueryRow(ctx, ` + SELECT EXISTS (SELECT 1 FROM update_jobs + WHERE service_id IN ($1, $2) AND status IN ('queued', 'running'))`, + fromID, toID).Scan(&active); err != nil { + return fmt.Errorf("merge: check active jobs: %w", err) + } + if active { + return model.ErrJobActive + } + + if _, err := tx.Exec(ctx, `UPDATE update_jobs SET service_id = $1 WHERE service_id = $2`, fromID, toID); err != nil { + return fmt.Errorf("merge: move jobs: %w", err) + } + var t model.Service + var lastTagDigest string + err = tx.QueryRow(ctx, ` + SELECT agent_id, location_key, compose_project, compose_service, working_dir, config_files, built, + swarm_cluster_id, swarm_service, container_name, image_repo, actual_ref, actual_digest, status, self, + updated_at, available_ref, available_digest, available_source, available_at, last_poll_error, + last_tag_digest + FROM services WHERE id = $1`, toID).Scan( + &t.AgentID, &t.LocationKey, &t.ComposeProject, &t.ComposeService, &t.WorkingDir, &t.ConfigFiles, &t.Built, + &t.SwarmClusterID, &t.SwarmService, &t.ContainerName, &t.ImageRepo, &t.ActualRef, &t.ActualDigest, &t.Status, &t.Self, + &t.UpdatedAt, &t.AvailableRef, &t.AvailableDigest, &t.AvailableSource, &t.AvailableAt, &t.LastPollError, + &lastTagDigest) + if err != nil { + return fmt.Errorf("merge: read new service: %w", err) + } + if _, err := tx.Exec(ctx, `DELETE FROM services WHERE id = $1`, toID); err != nil { + return fmt.Errorf("merge: delete new service: %w", err) + } + if _, err := tx.Exec(ctx, ` + UPDATE services SET + agent_id = $2, location_key = $3, compose_project = $4, compose_service = $5, working_dir = $6, + config_files = $7, built = $8, swarm_cluster_id = $9, swarm_service = $10, container_name = $11, + image_repo = $12, actual_ref = $13, actual_digest = $14, status = $15, self = $16, updated_at = $17, + available_ref = $18, available_digest = $19, available_source = $20, available_at = $21, + last_poll_error = $22, + last_tag_digest = CASE WHEN last_tag_digest <> '' THEN last_tag_digest ELSE $23 END, + removed_at = NULL, awaiting_inventory = false + WHERE id = $1`, + fromID, t.AgentID, t.LocationKey, t.ComposeProject, t.ComposeService, t.WorkingDir, + t.ConfigFiles, t.Built, t.SwarmClusterID, t.SwarmService, t.ContainerName, + t.ImageRepo, t.ActualRef, t.ActualDigest, t.Status, t.Self, t.UpdatedAt, + t.AvailableRef, t.AvailableDigest, t.AvailableSource, t.AvailableAt, + t.LastPollError, lastTagDigest); err != nil { + return fmt.Errorf("merge: update old service: %w", err) + } + if err := tx.Commit(ctx); err != nil { + return fmt.Errorf("commit merge: %w", err) + } + return nil +} + +// mergeRefusal returns why from (the old row) cannot be merged with to (the +// live one), or "". +func mergeRefusal(from, to mergeRow) string { + switch { + case !from.removed: + return model.ReasonMergeLiveOld + case to.removed: + return model.ReasonMergeGoneNew + case from.kind != to.kind: + return model.ReasonMergeKind + case from.kind == string(protocol.KindContainer): + return model.ReasonMergeContainer + case from.self || to.self: + return model.ReasonSelf + } + return "" +} diff --git a/internal/master/store/migrations/0005_drift_merge.sql b/internal/master/store/migrations/0005_drift_merge.sql new file mode 100644 index 0000000..75f0fcd --- /dev/null +++ b/internal/master/store/migrations/0005_drift_merge.sql @@ -0,0 +1,7 @@ +-- true from a successful job until the next inventory write for the row: the +-- agent's upsert arrives a debounce interval after job.result, and until then +-- desired_ref is newer than actual_ref without the service having drifted. +ALTER TABLE services ADD COLUMN awaiting_inventory boolean NOT NULL DEFAULT false; + +-- move suggestions look up Compose rows by project + service across agents +CREATE INDEX services_by_compose_name ON services (compose_project, compose_service) WHERE kind = 'compose'; diff --git a/internal/master/store/services.go b/internal/master/store/services.go index ea94419..4a58044 100644 --- a/internal/master/store/services.go +++ b/internal/master/store/services.go @@ -4,7 +4,6 @@ import ( "context" "errors" "fmt" - "regexp" "time" "github.com/jackc/pgx/v5" @@ -55,10 +54,8 @@ ON CONFLICT (location_key) DO UPDATE SET status = EXCLUDED.status, self = EXCLUDED.self, 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}$`) + removed_at = NULL, + awaiting_inventory = false` func upsertService(ctx context.Context, db execer, agentID int64, clusterID string, svc protocol.Service, at time.Time) error { image := imageref.Parse(svc.Image) @@ -75,7 +72,7 @@ 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 digestPattern.MatchString(svc.Digest) { + if imageref.IsDigest(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 == "" { @@ -178,7 +175,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.last_tag_digest, s.first_seen_at, s.updated_at, s.removed_at` + s.last_poll_error, s.last_tag_digest, s.awaiting_inventory, s.first_seen_at, s.updated_at, s.removed_at` func scanService(row pgx.Row) (model.Service, error) { var m model.Service @@ -187,7 +184,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.LastTagDigest, &m.FirstSeenAt, &m.UpdatedAt, &m.RemovedAt) + &m.LastPollError, &m.LastTagDigest, &m.AwaitingInventory, &m.FirstSeenAt, &m.UpdatedAt, &m.RemovedAt) m.Kind = protocol.Kind(kind) return m, err } @@ -331,6 +328,34 @@ func (s *Store) SetUpdateMode(ctx context.Context, id int64, mode string) error return nil } +// AcceptRunning makes the running version the desired one: desired_ref = +// desired, provided the service is live, has been updated before and still +// reports actualRef and actualDigest (what the caller saw when it decided). +// It starts no job and leaves last_tag_digest, the available suggestion and +// the update mode alone. Returns model.ErrNotFound for an unknown or removed +// service and model.ErrConflict when the service was never updated or the +// running version changed meanwhile. +func (s *Store) AcceptRunning(ctx context.Context, id int64, actualRef, actualDigest, desired string) error { + tag, err := s.pool.Exec(ctx, ` + UPDATE services SET desired_ref = $4, awaiting_inventory = false + WHERE id = $1 AND removed_at IS NULL AND desired_ref <> '' + AND actual_ref = $2 AND actual_digest = $3`, id, actualRef, actualDigest, desired) + if err != nil { + return fmt.Errorf("accept running: %w", err) + } + if tag.RowsAffected() == 1 { + return nil + } + svc, err := s.GetService(ctx, id) + if err != nil { + return err + } + if svc.RemovedAt != nil { + return model.ErrNotFound + } + return model.ErrConflict +} + // GetService returns one service, including a removed one (RemovedAt set). // Returns model.ErrNotFound if the ID does not exist. func (s *Store) GetService(ctx context.Context, id int64) (model.Service, error) {