Merge origin/main (phase 5, drift and merge) into feat/ledger-ui-services-agents-18
Conflicts: - CLAUDE.md: keep both sets of gotchas (drift/merge from main, SSE ping and swarm_cluster_id from this branch). - internal/imageref/imageref_test.go: keep both TestIsDigest (main) and TestIsRegistryHost (this branch). - internal/master/web/server.go: take main's renderServiceError and make it render through s.page, so service error pages get the sidebar nav data. Also renumber this branch's migration 0005_agent_cluster.sql to 0006_agent_cluster.sql, after main's 0005_drift_merge.sql. Claude-Session: https://claude.ai/code/session_01GAZsp3P7Qto6JnhipxWUJQ
This commit is contained in:
@@ -11,7 +11,6 @@ import (
|
||||
"net"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"regexp"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
@@ -25,6 +24,7 @@ import (
|
||||
"github.com/shcizo/package-updater/internal/agent/conn"
|
||||
"github.com/shcizo/package-updater/internal/agent/updater"
|
||||
"github.com/shcizo/package-updater/internal/agenttoken"
|
||||
"github.com/shcizo/package-updater/internal/imageref"
|
||||
"github.com/shcizo/package-updater/internal/master/events"
|
||||
"github.com/shcizo/package-updater/internal/master/hub"
|
||||
"github.com/shcizo/package-updater/internal/master/model"
|
||||
@@ -32,9 +32,6 @@ import (
|
||||
"github.com/shcizo/package-updater/internal/protocol"
|
||||
)
|
||||
|
||||
// digestPattern is the digest format the real store records.
|
||||
var digestPattern = regexp.MustCompile(`^sha256:[0-9a-f]{64}$`)
|
||||
|
||||
type memStore struct {
|
||||
mu sync.Mutex
|
||||
agents map[string]model.Agent
|
||||
@@ -245,7 +242,8 @@ 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
|
||||
if _, d, _ := strings.Cut(f.Ref, "@"); !strings.Contains(requestedRef, "@") && digestPattern.MatchString(d) {
|
||||
svc.AwaitingInventory = true
|
||||
if _, d, _ := strings.Cut(f.Ref, "@"); !strings.Contains(requestedRef, "@") && imageref.IsDigest(d) {
|
||||
svc.LastTagDigest = d
|
||||
}
|
||||
// Same rule as the SQL: the job was created for the suggested ref after it was reported.
|
||||
|
||||
@@ -0,0 +1,113 @@
|
||||
package e2e
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"log/slog"
|
||||
"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/master/poller"
|
||||
"github.com/shcizo/package-updater/internal/protocol"
|
||||
)
|
||||
|
||||
const (
|
||||
driftDesired = "sha256:" + "d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1"
|
||||
driftOther = "sha256:" + "0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a"
|
||||
)
|
||||
|
||||
// seedDrifted replaces the environment's service with an auto service that was
|
||||
// updated to slusk:1.4.2 at driftDesired but runs driftOther now.
|
||||
func (e *jobEnv) seedDrifted() {
|
||||
e.store.seedService(model.Service{ID: 1, Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web",
|
||||
AgentID: ptr(agentID), ImageRepo: slusk, UpdateMode: model.ModeAuto,
|
||||
DesiredRef: slusk + ":1.4.2@" + driftDesired,
|
||||
ActualRef: slusk + ":1.4.2", ActualDigest: driftOther})
|
||||
}
|
||||
|
||||
// inventoryWrite is what the store does for an agent's upsert of the service.
|
||||
func (e *jobEnv) inventoryWrite(id int64, actualRef, actualDigest string) {
|
||||
e.store.mu.Lock()
|
||||
defer e.store.mu.Unlock()
|
||||
svc := e.store.svcRows[id]
|
||||
svc.ActualRef, svc.ActualDigest, svc.AwaitingInventory = actualRef, actualDigest, false
|
||||
e.store.svcRows[id] = svc
|
||||
}
|
||||
|
||||
func TestDriftedAutoServiceIsNotUpdatedByPollOrWebhook(t *testing.T) {
|
||||
e := newJobEnv(t)
|
||||
e.seedDrifted()
|
||||
reg := &fixedRegistry{digests: map[string]string{slusk + ":1.4.2": "sha256:new"}}
|
||||
p := poller.New(e.store, e.jobs, reg, e.m.broker, poller.Config{Workers: 1}, slog.New(slog.NewTextHandler(io.Discard, nil)))
|
||||
|
||||
p.Cycle(context.Background())
|
||||
|
||||
assert.Equal(t, 1, reg.calls)
|
||||
assert.Zero(t, e.store.jobCount(), "polling starts no job for a drifted service")
|
||||
svc, err := e.store.GetService(context.Background(), 1)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, slusk+":1.4.2", svc.AvailableRef, "the suggestion is still recorded")
|
||||
|
||||
code, body := callWebhook(t, e.webhookServer(t), `{"image":"`+slusk+`:ignored","tag":"2.0.0"}`)
|
||||
|
||||
assert.Equal(t, 200, code)
|
||||
require.Len(t, body.Results, 1)
|
||||
assert.Equal(t, "refused", body.Results[0].Status)
|
||||
assert.Contains(t, body.Results[0].Error, "drifted")
|
||||
assert.Zero(t, e.store.jobCount(), "the webhook starts no job either")
|
||||
}
|
||||
|
||||
func TestRestoreDeploysThePinnedRefOnTheAgent(t *testing.T) {
|
||||
e := newJobEnv(t)
|
||||
e.seedDrifted()
|
||||
close(e.exec.release)
|
||||
|
||||
job, err := e.jobs.RestoreDesired(context.Background(), 1, "alice")
|
||||
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, model.TriggerManual, job.Trigger)
|
||||
select {
|
||||
case ref := <-e.exec.started:
|
||||
assert.Equal(t, slusk+":1.4.2@"+driftDesired, ref, "the agent receives exactly the desired ref")
|
||||
case <-time.After(waitFor):
|
||||
t.Fatal("the agent never received the job")
|
||||
}
|
||||
done := e.waitStatus(t, job.ID, model.JobSucceeded)
|
||||
assert.Equal(t, slusk+":1.4.2@"+driftDesired, done.ToRef)
|
||||
svc, err := e.store.GetService(context.Background(), 1)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, driftOther, svc.LastTagDigest, "the drifted digest is remembered so polling does not redeploy it")
|
||||
assert.False(t, svc.Drifted(), "the inventory has not caught up yet")
|
||||
|
||||
e.inventoryWrite(1, slusk+":1.4.2", driftDesired)
|
||||
svc, err = e.store.GetService(context.Background(), 1)
|
||||
require.NoError(t, err)
|
||||
assert.False(t, svc.Drifted())
|
||||
}
|
||||
|
||||
func TestNoDriftAfterASuccessfulJob(t *testing.T) {
|
||||
e := newJobEnv(t)
|
||||
close(e.exec.release)
|
||||
|
||||
job, err := e.jobs.Trigger(context.Background(), 1, "alice")
|
||||
|
||||
require.NoError(t, err)
|
||||
e.waitStatus(t, job.ID, model.JobSucceeded)
|
||||
svc, err := e.store.GetService(context.Background(), 1)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, slusk+":1.4.2@"+e2eDigest, svc.DesiredRef)
|
||||
assert.Equal(t, slusk+":1.4.2@sha256:old", svc.ActualRef, "the agent's inventory write has not arrived")
|
||||
assert.False(t, svc.Drifted(), "between job.result and the inventory write the row is not drifted")
|
||||
|
||||
e.inventoryWrite(1, slusk+":1.4.2", e2eDigest)
|
||||
svc, _ = e.store.GetService(context.Background(), 1)
|
||||
assert.False(t, svc.Drifted())
|
||||
|
||||
e.inventoryWrite(1, slusk+":1.4.2", driftOther)
|
||||
svc, _ = e.store.GetService(context.Background(), 1)
|
||||
assert.True(t, svc.Drifted(), "a later inventory write of another digest is drift")
|
||||
}
|
||||
@@ -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.
|
||||
@@ -47,3 +50,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) }
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package imageref
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
@@ -47,3 +48,19 @@ func TestIsRegistryHost(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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,237 @@
|
||||
package jobs
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"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 (
|
||||
driftRepo = "gitea.shcizo.se/shcizo/slusk"
|
||||
driftD1 = "sha256:1111111111111111111111111111111111111111111111111111111111111111"
|
||||
driftD2 = "sha256:2222222222222222222222222222222222222222222222222222222222222222"
|
||||
driftD3 = "sha256:3333333333333333333333333333333333333333333333333333333333333333"
|
||||
)
|
||||
|
||||
// driftedSvc was updated to 1.4.2 at D1 but now runs 1.4.2 at D2.
|
||||
func driftedSvc() model.Service {
|
||||
svc := composeSvc
|
||||
svc.ImageRepo = driftRepo
|
||||
svc.DesiredRef = driftRepo + ":1.4.2@" + driftD1
|
||||
svc.ActualRef = driftRepo + ":1.4.2"
|
||||
svc.ActualDigest = driftD2
|
||||
return svc
|
||||
}
|
||||
|
||||
func notUpdatableReason(t *testing.T, err error) string {
|
||||
t.Helper()
|
||||
var nu *model.NotUpdatableError
|
||||
require.True(t, errors.As(err, &nu), "got %v", err)
|
||||
return nu.Reason
|
||||
}
|
||||
|
||||
func TestRestorable(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
mutate func(*model.Service)
|
||||
reason string // "" means restorable
|
||||
}{
|
||||
{"drifted", func(*model.Service) {}, ""},
|
||||
{"drifted service that follows no tag", func(s *model.Service) {
|
||||
s.ActualRef = driftRepo + "@" + driftD2
|
||||
}, ""},
|
||||
{"not drifted", func(s *model.Service) { s.ActualDigest = driftD1 }, model.ReasonNotDrifted},
|
||||
{"never updated", func(s *model.Service) { s.DesiredRef = "" }, model.ReasonNotDrifted},
|
||||
{"awaiting inventory", func(s *model.Service) { s.AwaitingInventory = true }, model.ReasonNotDrifted},
|
||||
{"removed", func(s *model.Service) { s.RemovedAt = ptr(time.Now()) }, model.ReasonRemoved},
|
||||
{"flagged self", func(s *model.Service) { s.Self = true }, model.ReasonSelf},
|
||||
{"built", func(s *model.Service) { s.Built = true }, model.ReasonBuilt},
|
||||
{"container", func(s *model.Service) { s.Kind = protocol.KindContainer }, model.ReasonContainer},
|
||||
{"desired ref is the agent image", func(s *model.Service) {
|
||||
s.DesiredRef = selfImage + "@" + driftD1
|
||||
}, model.ReasonSelf},
|
||||
}
|
||||
s, _ := newService(&fakeStore{}, &fakeHub{})
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
svc := driftedSvc()
|
||||
tt.mutate(&svc)
|
||||
err := s.Restorable(svc)
|
||||
if tt.reason == "" {
|
||||
assert.NoError(t, err)
|
||||
return
|
||||
}
|
||||
assert.Equal(t, tt.reason, notUpdatableReason(t, err))
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestRestoreDesired_StartsAManualJobToTheDesiredRef(t *testing.T) {
|
||||
svc := driftedSvc()
|
||||
st := &fakeStore{services: map[int64]model.Service{1: svc}, jobs: map[int64]model.Job{}}
|
||||
h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}}
|
||||
s, _ := newService(st, h)
|
||||
|
||||
job, err := s.RestoreDesired(context.Background(), 1, "sub-1")
|
||||
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, model.JobQueued, job.Status)
|
||||
require.Len(t, st.created, 1)
|
||||
assert.Equal(t, svc.DesiredRef, st.created[0].ToRef)
|
||||
assert.Equal(t, svc.ActualRef, st.created[0].FromRef)
|
||||
assert.Equal(t, "sub-1", st.created[0].RequestedBy)
|
||||
assert.Equal(t, model.TriggerManual, st.created[0].Trigger)
|
||||
assert.Equal(t, []int{protocol.MinPinnedRefVersion}, h.mins)
|
||||
require.Len(t, h.starts, 1)
|
||||
assert.Equal(t, svc.DesiredRef, h.starts[0].start.Ref)
|
||||
}
|
||||
|
||||
func TestRestoreDesired_Refusals(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
mutate func(*model.Service)
|
||||
reason string
|
||||
}{
|
||||
{"not drifted", func(s *model.Service) { s.ActualDigest = driftD1 }, model.ReasonNotDrifted},
|
||||
{"self", func(s *model.Service) { s.Self = true }, model.ReasonSelf},
|
||||
{"built", func(s *model.Service) { s.Built = true }, model.ReasonBuilt},
|
||||
{"container", func(s *model.Service) { s.Kind = protocol.KindContainer }, model.ReasonContainer},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
svc := driftedSvc()
|
||||
tt.mutate(&svc)
|
||||
st := &fakeStore{services: map[int64]model.Service{1: svc}, jobs: map[int64]model.Job{}}
|
||||
h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}}
|
||||
s, pub := newService(st, h)
|
||||
|
||||
_, err := s.RestoreDesired(context.Background(), 1, "sub-1")
|
||||
|
||||
assert.Equal(t, tt.reason, notUpdatableReason(t, err))
|
||||
assert.Empty(t, st.created)
|
||||
assert.Empty(t, h.starts)
|
||||
assert.Empty(t, pub.topics)
|
||||
assert.Empty(t, st.lastTag, "last_tag_digest untouched")
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestRestoreDesired_UnknownService(t *testing.T) {
|
||||
s, _ := newService(&fakeStore{services: map[int64]model.Service{}}, &fakeHub{})
|
||||
|
||||
_, err := s.RestoreDesired(context.Background(), 9, "sub-1")
|
||||
|
||||
assert.ErrorIs(t, err, model.ErrNotFound)
|
||||
}
|
||||
|
||||
func TestRestoreDesired_ActiveJobReturnsItAndMarksNothing(t *testing.T) {
|
||||
activeJob := model.Job{ID: 99, Status: model.JobRunning}
|
||||
st := &fakeStore{services: map[int64]model.Service{1: driftedSvc()}, jobs: map[int64]model.Job{99: activeJob}, active: &activeJob}
|
||||
s, _ := newService(st, &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}})
|
||||
|
||||
job, err := s.RestoreDesired(context.Background(), 1, "sub-1")
|
||||
|
||||
assert.ErrorIs(t, err, model.ErrJobActive)
|
||||
assert.Equal(t, int64(99), job.ID)
|
||||
assert.Empty(t, st.lastTag)
|
||||
}
|
||||
|
||||
func TestRestoreDesired_OldAgentGivesAFailedJob(t *testing.T) {
|
||||
st := &fakeStore{services: map[int64]model.Service{1: driftedSvc()}, jobs: map[int64]model.Job{}}
|
||||
h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}, agentProtocol: 2}
|
||||
s, _ := newService(st, h)
|
||||
|
||||
job, err := s.RestoreDesired(context.Background(), 1, "sub-1")
|
||||
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, model.JobFailed, job.Status)
|
||||
assert.Equal(t, model.ErrAgentTooOldForPinned.Error(), job.Error)
|
||||
assert.Empty(t, h.starts)
|
||||
}
|
||||
|
||||
func TestRestoreDesired_LastTagDigest(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
mutate func(*model.Service)
|
||||
want string // "" means SetLastTagDigest is not called
|
||||
}{
|
||||
{"drifted to another digest of the same tag", func(*model.Service) {}, driftD2},
|
||||
{"digest in the actual ref", func(s *model.Service) {
|
||||
s.ActualRef, s.ActualDigest = driftRepo+":1.4.2@"+driftD3, ""
|
||||
}, driftD3},
|
||||
{"last tag digest holds back a newer one", func(s *model.Service) {
|
||||
s.DesiredRef = driftRepo + ":1.4.2@" + driftD1
|
||||
s.LastTagDigest = driftD3
|
||||
}, ""},
|
||||
{"last tag digest equals the desired digest", func(s *model.Service) { s.LastTagDigest = driftD1 }, driftD2},
|
||||
{"already remembered", func(s *model.Service) { s.LastTagDigest = driftD2 }, ""},
|
||||
{"different tag", func(s *model.Service) { s.ActualRef = driftRepo + ":1.5.0" }, ""},
|
||||
{"actual digest malformed", func(s *model.Service) { s.ActualDigest = "sha256:abc" }, ""},
|
||||
{"desired follows no tag", func(s *model.Service) { s.DesiredRef = driftRepo + "@" + driftD1 }, ""},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
svc := driftedSvc()
|
||||
tt.mutate(&svc)
|
||||
st := &fakeStore{services: map[int64]model.Service{1: svc}, jobs: map[int64]model.Job{}}
|
||||
s, _ := newService(st, &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}})
|
||||
|
||||
_, err := s.RestoreDesired(context.Background(), 1, "sub-1")
|
||||
|
||||
require.NoError(t, err)
|
||||
if tt.want == "" {
|
||||
assert.Empty(t, st.lastTag)
|
||||
return
|
||||
}
|
||||
assert.Equal(t, map[int64]string{1: tt.want}, st.lastTag)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestStart_RefusesPollAndWebhookWhileDrifted(t *testing.T) {
|
||||
tests := []struct {
|
||||
trigger model.JobTrigger
|
||||
refused bool
|
||||
}{
|
||||
{model.TriggerPoll, true},
|
||||
{model.TriggerWebhook, true},
|
||||
{model.TriggerManual, false},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(string(tt.trigger), func(t *testing.T) {
|
||||
svc := driftedSvc()
|
||||
st := &fakeStore{services: map[int64]model.Service{1: svc}, jobs: map[int64]model.Job{}}
|
||||
h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}}
|
||||
s, _ := newService(st, h)
|
||||
|
||||
_, err := s.Start(context.Background(), svc, StartRequest{Trigger: tt.trigger, Ref: driftRepo + ":2.0.0"})
|
||||
|
||||
if tt.refused {
|
||||
assert.Equal(t, model.ReasonDrifted, notUpdatableReason(t, err))
|
||||
assert.Empty(t, st.created)
|
||||
return
|
||||
}
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, st.created, 1)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestStart_PollAllowedWhenNotDrifted(t *testing.T) {
|
||||
svc := driftedSvc()
|
||||
svc.AwaitingInventory = true
|
||||
st := &fakeStore{services: map[int64]model.Service{1: svc}, jobs: map[int64]model.Job{}}
|
||||
s, _ := newService(st, &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}})
|
||||
|
||||
_, err := s.Start(context.Background(), svc, StartRequest{Trigger: model.TriggerPoll})
|
||||
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, st.created, 1)
|
||||
}
|
||||
@@ -21,6 +21,7 @@ type Store interface {
|
||||
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)
|
||||
SetLastTagDigest(ctx context.Context, id int64, digest string) error
|
||||
}
|
||||
|
||||
// Hub routes jobs to agent connections. *hub.Hub satisfies it.
|
||||
@@ -114,6 +115,81 @@ func (s *Service) updatable(svc model.Service, explicitRef bool) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Restorable returns nil or a *model.NotUpdatableError saying why the service
|
||||
// cannot be restored to its desired version: it must have drifted, and the
|
||||
// desired version is deployed like an explicit ref, so a service that follows
|
||||
// no tag is fine. The service page calls it for every render.
|
||||
func (s *Service) Restorable(svc model.Service) error {
|
||||
if svc.RemovedAt != nil {
|
||||
return &model.NotUpdatableError{Reason: model.ReasonRemoved}
|
||||
}
|
||||
if !svc.Drifted() {
|
||||
return &model.NotUpdatableError{Reason: model.ReasonNotDrifted}
|
||||
}
|
||||
if err := s.updatable(svc, true); err != nil {
|
||||
return err
|
||||
}
|
||||
if imageref.Parse(svc.DesiredRef).Repo == s.selfRepo {
|
||||
return &model.NotUpdatableError{Reason: model.ReasonSelf}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// RestoreDesired starts a manual job that deploys the drifted service's
|
||||
// desired ref again. The ref is read from the row, never from the caller. It
|
||||
// returns model.ErrNotFound for an unknown service and otherwise what Start
|
||||
// returns, with a *model.NotUpdatableError from Restorable if it cannot be
|
||||
// restored.
|
||||
//
|
||||
// When the service had drifted to another digest of the same tag, that digest
|
||||
// becomes last_tag_digest, so the poller does not deploy it again before the
|
||||
// registry shows a newer one (the same rule as for any rollback). It is not
|
||||
// written when last_tag_digest holds back a newer digest, that is when it is
|
||||
// set and differs from the digest the desired ref names.
|
||||
func (s *Service) RestoreDesired(ctx context.Context, serviceID int64, requestedBy string) (model.Job, error) {
|
||||
svc, err := s.store.GetService(ctx, serviceID)
|
||||
if err != nil {
|
||||
return model.Job{}, err
|
||||
}
|
||||
if err := s.Restorable(svc); err != nil {
|
||||
return model.Job{}, err
|
||||
}
|
||||
job, err := s.Start(ctx, svc, StartRequest{Trigger: model.TriggerManual, RequestedBy: requestedBy, Ref: svc.DesiredRef})
|
||||
if err != nil {
|
||||
return job, err
|
||||
}
|
||||
if d := driftedTagDigest(svc); d != "" && d != svc.LastTagDigest && holdsNothingNewer(svc) {
|
||||
if err := s.store.SetLastTagDigest(ctx, svc.ID, d); err != nil {
|
||||
s.log.Warn("last_tag_digest_not_set", "service_id", svc.ID, "error", err)
|
||||
}
|
||||
}
|
||||
return job, nil
|
||||
}
|
||||
|
||||
// holdsNothingNewer reports whether last_tag_digest may be overwritten by a
|
||||
// restore: it is empty or equals the digest the desired ref names. Anything
|
||||
// else is a digest the poller must keep holding back.
|
||||
func holdsNothingNewer(svc model.Service) bool {
|
||||
return svc.LastTagDigest == "" || svc.LastTagDigest == imageref.Parse(svc.DesiredRef).Digest
|
||||
}
|
||||
|
||||
// driftedTagDigest returns the digest a drifted service runs when it runs the
|
||||
// same tag as desired at another digest, else "".
|
||||
func driftedTagDigest(svc model.Service) string {
|
||||
tag := FollowedRef(svc.DesiredRef)
|
||||
if tag == "" || tag != FollowedRef(svc.ActualRef) {
|
||||
return ""
|
||||
}
|
||||
d := imageref.Parse(svc.ActualRef).Digest
|
||||
if d == "" {
|
||||
d = svc.ActualDigest
|
||||
}
|
||||
if !imageref.IsDigest(d) || d == imageref.Parse(svc.DesiredRef).Digest {
|
||||
return ""
|
||||
}
|
||||
return d
|
||||
}
|
||||
|
||||
// StartRequest says who asks for an update and what to deploy.
|
||||
type StartRequest struct {
|
||||
Trigger model.JobTrigger
|
||||
@@ -161,8 +237,12 @@ func (s *Service) TriggerVersion(ctx context.Context, serviceID, versionID int64
|
||||
// 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.
|
||||
// running. Poll and webhook jobs are refused for a drifted service; only a
|
||||
// person starts one then.
|
||||
func (s *Service) Start(ctx context.Context, svc model.Service, req StartRequest) (model.Job, error) {
|
||||
if req.Trigger != model.TriggerManual && svc.Drifted() {
|
||||
return model.Job{}, &model.NotUpdatableError{Reason: model.ReasonDrifted}
|
||||
}
|
||||
if err := s.updatable(svc, req.Ref != ""); err != nil {
|
||||
return model.Job{}, err
|
||||
}
|
||||
|
||||
@@ -23,7 +23,8 @@ type fakeStore struct {
|
||||
active *model.Job // returned with ErrJobActive when set
|
||||
versions map[int64]model.ImageVersion
|
||||
|
||||
revokedAgents map[int64]bool // CreateJob returns ErrAgentOffline for these
|
||||
revokedAgents map[int64]bool // CreateJob returns ErrAgentOffline for these
|
||||
lastTag map[int64]string // SetLastTagDigest calls: service ID to digest
|
||||
}
|
||||
|
||||
func (f *fakeStore) GetService(_ context.Context, id int64) (model.Service, error) {
|
||||
@@ -63,6 +64,14 @@ func (f *fakeStore) GetImageVersion(_ context.Context, id int64) (model.ImageVer
|
||||
return v, nil
|
||||
}
|
||||
|
||||
func (f *fakeStore) SetLastTagDigest(_ context.Context, id int64, digest string) error {
|
||||
if f.lastTag == nil {
|
||||
f.lastTag = map[int64]string{}
|
||||
}
|
||||
f.lastTag[id] = digest
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f *fakeStore) GetJob(_ context.Context, id int64) (model.Job, error) { return f.jobs[id], nil }
|
||||
|
||||
type startCall struct {
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -15,6 +15,8 @@ import (
|
||||
// Sentinel errors returned by storage.
|
||||
var (
|
||||
ErrNotFound = errors.New("not found")
|
||||
// ErrConflict means the row already exists, or that it changed meanwhile
|
||||
// (AcceptRunning, MergeService).
|
||||
ErrConflict = errors.New("already exists")
|
||||
ErrRevoked = errors.New("agent revoked")
|
||||
)
|
||||
@@ -71,9 +73,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.
|
||||
@@ -116,6 +122,68 @@ 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()
|
||||
}
|
||||
|
||||
// RunningRef returns the ref accepting the running version records as the
|
||||
// desired one; see the function RunningRef. Templates post it as the guard of
|
||||
// the accept form.
|
||||
func (s Service) RunningRef() string { return RunningRef(s.ActualRef, s.ActualDigest) }
|
||||
|
||||
// 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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
@@ -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,
|
||||
|
||||
@@ -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 ""
|
||||
}
|
||||
@@ -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 and service across agents.
|
||||
CREATE INDEX services_by_compose_name ON services (compose_project, compose_service) WHERE kind = 'compose';
|
||||
@@ -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
|
||||
}
|
||||
@@ -262,8 +259,8 @@ func (s *Store) SetPollError(ctx context.Context, ids []int64, msg string) error
|
||||
}
|
||||
|
||||
// SetLastTagDigest sets last_tag_digest; see model.Service.LastTagDigest. The
|
||||
// poller calls it, and FinishJob sets the same column after a tag job.
|
||||
// Inventory writes never change it.
|
||||
// poller and jobs.RestoreDesired call it, and FinishJob sets the same column
|
||||
// after a tag job. Inventory writes never change it.
|
||||
func (s *Store) SetLastTagDigest(ctx context.Context, id int64, digest string) error {
|
||||
if _, err := s.pool.Exec(ctx,
|
||||
`UPDATE services SET last_tag_digest = $2 WHERE id = $1`, id, digest); err != nil {
|
||||
@@ -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) {
|
||||
|
||||
@@ -0,0 +1,297 @@
|
||||
package web
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"net/url"
|
||||
"strconv"
|
||||
"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/model"
|
||||
"github.com/shcizo/package-updater/internal/protocol"
|
||||
)
|
||||
|
||||
const (
|
||||
driftDesired = "sha256:1111111111111111111111111111111111111111111111111111111111111111"
|
||||
driftActual = "sha256:2222222222222222222222222222222222222222222222222222222222222222"
|
||||
)
|
||||
|
||||
type acceptCall struct {
|
||||
id int64
|
||||
actualRef, actualDigest, desiredRef string
|
||||
}
|
||||
|
||||
func (f *fakeStore) AcceptRunning(_ context.Context, id int64, actualRef, actualDigest, desired string) error {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.acceptCalls = append(f.acceptCalls, acceptCall{id, actualRef, actualDigest, desired})
|
||||
return f.acceptErr
|
||||
}
|
||||
|
||||
func (f *fakeStore) MoveCandidates(context.Context) ([]model.Move, error) { return f.moves, nil }
|
||||
|
||||
func (f *fakeStore) MergeSources(context.Context, int64) ([]model.Service, error) {
|
||||
return f.sources, nil
|
||||
}
|
||||
|
||||
func (f *fakeStore) MergeService(_ context.Context, from, to int64) error {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.merged = append(f.merged, [2]int64{from, to})
|
||||
return f.mergeErr
|
||||
}
|
||||
|
||||
func (f *fakeJobs) Restorable(model.Service) error { return f.restorable }
|
||||
|
||||
func (f *fakeJobs) RestoreDesired(_ context.Context, serviceID int64, by string) (model.Job, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.requestedBy = by
|
||||
f.restored = append(f.restored, serviceID)
|
||||
return f.job, f.err
|
||||
}
|
||||
|
||||
// driftService makes the first fixture service a drifted one: updated to
|
||||
// 1.4.2 at driftDesired, now running 1.4.2 at driftActual.
|
||||
func driftService(store *fakeStore) {
|
||||
store.services[0].DesiredRef = "gitea.shcizo.se/shcizo/slusk:1.4.2@" + driftDesired
|
||||
store.services[0].ActualDigest = driftActual
|
||||
}
|
||||
|
||||
func postTo(t *testing.T, srv *httptest.Server, path string, form url.Values) (*http.Response, string) {
|
||||
t.Helper()
|
||||
form.Set("csrf_token", "csrf")
|
||||
resp, err := noRedirectClient().PostForm(srv.URL+path, form)
|
||||
require.NoError(t, err)
|
||||
defer resp.Body.Close()
|
||||
body, _ := io.ReadAll(resp.Body)
|
||||
return resp, string(body)
|
||||
}
|
||||
|
||||
func TestRestoreDesired(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
job model.Job
|
||||
err error
|
||||
path string
|
||||
wantCode int
|
||||
wantLocation string
|
||||
wantBody string
|
||||
}{
|
||||
{name: "started", job: model.Job{ID: 42}, path: "/services/1/restore",
|
||||
wantCode: http.StatusSeeOther, wantLocation: "/jobs/42"},
|
||||
{name: "already running", job: model.Job{ID: 41}, err: model.ErrJobActive, path: "/services/1/restore",
|
||||
wantCode: http.StatusSeeOther, wantLocation: "/jobs/41"},
|
||||
{name: "refused", err: &model.NotUpdatableError{Reason: model.ReasonNotDrifted}, path: "/services/1/restore",
|
||||
wantCode: http.StatusConflict, wantBody: "Cannot restore: runs its desired version."},
|
||||
{name: "unknown service", err: model.ErrNotFound, path: "/services/1/restore", wantCode: http.StatusNotFound},
|
||||
{name: "invalid id", path: "/services/x/restore", wantCode: http.StatusBadRequest},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
srv, _, _, _, jobs := newTestEnv(t)
|
||||
jobs.job, jobs.err = tt.job, tt.err
|
||||
|
||||
resp, body := postTo(t, srv, tt.path, url.Values{})
|
||||
|
||||
assert.Equal(t, tt.wantCode, resp.StatusCode)
|
||||
assert.Equal(t, tt.wantLocation, resp.Header.Get("Location"))
|
||||
assert.Contains(t, body, tt.wantBody)
|
||||
if tt.wantCode != http.StatusBadRequest {
|
||||
assert.Equal(t, "alice", jobs.triggeredBy())
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestAcceptRunning_WritesRunningRefAndPublishes(t *testing.T) {
|
||||
srv, store, _, broker, _ := newTestEnv(t)
|
||||
driftService(store)
|
||||
ch, cancel := broker.Subscribe()
|
||||
defer cancel()
|
||||
|
||||
resp, _ := postTo(t, srv, "/services/1/accept", url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.4.2@" + driftActual}})
|
||||
|
||||
assert.Equal(t, http.StatusSeeOther, resp.StatusCode)
|
||||
assert.Equal(t, "/services/1", resp.Header.Get("Location"))
|
||||
assert.Equal(t, []acceptCall{{
|
||||
id: 1, actualRef: "gitea.shcizo.se/shcizo/slusk:1.4.2", actualDigest: driftActual,
|
||||
desiredRef: "gitea.shcizo.se/shcizo/slusk:1.4.2@" + driftActual,
|
||||
}}, store.acceptCalls, "the value written comes from the row, not the form")
|
||||
select {
|
||||
case topic := <-ch:
|
||||
assert.Equal(t, events.Services, topic)
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("no services signal")
|
||||
}
|
||||
}
|
||||
|
||||
func TestAcceptRunning_ReturnToOverview(t *testing.T) {
|
||||
srv, store, _, _, _ := newTestEnv(t)
|
||||
driftService(store)
|
||||
|
||||
resp, _ := postTo(t, srv, "/services/1/accept",
|
||||
url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.4.2@" + driftActual}, "return": {"overview"}})
|
||||
|
||||
assert.Equal(t, http.StatusSeeOther, resp.StatusCode)
|
||||
assert.Equal(t, "/", resp.Header.Get("Location"))
|
||||
}
|
||||
|
||||
func TestAcceptRunning_Refusals(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
path string
|
||||
form url.Values
|
||||
notDrifted bool
|
||||
removed bool
|
||||
acceptErr error
|
||||
wantCode int
|
||||
wantBody string
|
||||
}{
|
||||
{name: "missing actual", path: "/services/1/accept", form: url.Values{}, wantCode: http.StatusBadRequest},
|
||||
{name: "stale actual", path: "/services/1/accept", form: url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.3.0"}},
|
||||
wantCode: http.StatusConflict, wantBody: "The running version changed; check it again."},
|
||||
{name: "same ref, other digest", path: "/services/1/accept",
|
||||
form: url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.4.2@" + driftDesired}},
|
||||
wantCode: http.StatusConflict, wantBody: "The running version changed; check it again."},
|
||||
{name: "not drifted", path: "/services/1/accept", form: url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.4.2@" + driftActual}},
|
||||
notDrifted: true, wantCode: http.StatusConflict, wantBody: "Cannot accept: the service runs its desired version."},
|
||||
{name: "changed under our feet", path: "/services/1/accept", form: url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.4.2@" + driftActual}},
|
||||
acceptErr: model.ErrConflict, wantCode: http.StatusConflict, wantBody: "The running version changed; check it again."},
|
||||
{name: "unknown service", path: "/services/9/accept", form: url.Values{"actual": {"x"}}, wantCode: http.StatusNotFound},
|
||||
{name: "removed", path: "/services/1/accept", form: url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.4.2@" + driftActual}},
|
||||
removed: true, wantCode: http.StatusNotFound},
|
||||
{name: "removed meanwhile", path: "/services/1/accept", form: url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.4.2@" + driftActual}},
|
||||
acceptErr: model.ErrNotFound, wantCode: http.StatusNotFound},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
srv, store, _, _, _ := newTestEnv(t)
|
||||
if !tt.notDrifted {
|
||||
driftService(store)
|
||||
}
|
||||
if tt.removed {
|
||||
removedAt := time.Now()
|
||||
store.services[0].RemovedAt = &removedAt
|
||||
}
|
||||
store.acceptErr = tt.acceptErr
|
||||
|
||||
resp, body := postTo(t, srv, tt.path, tt.form)
|
||||
|
||||
assert.Equal(t, tt.wantCode, resp.StatusCode)
|
||||
assert.Contains(t, body, tt.wantBody)
|
||||
if tt.acceptErr == nil {
|
||||
assert.Empty(t, store.acceptCalls)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestMergeService_RedirectsToTheSurvivorAndPublishes(t *testing.T) {
|
||||
srv, store, _, broker, _ := newTestEnv(t)
|
||||
ch, cancel := broker.Subscribe()
|
||||
defer cancel()
|
||||
|
||||
resp, _ := postTo(t, srv, "/services/1/merge", url.Values{"from": {"17"}})
|
||||
|
||||
assert.Equal(t, http.StatusSeeOther, resp.StatusCode)
|
||||
assert.Equal(t, "/services/17", resp.Header.Get("Location"))
|
||||
assert.Equal(t, [][2]int64{{17, 1}}, store.merged)
|
||||
var got []events.Topic
|
||||
for len(got) < 2 {
|
||||
select {
|
||||
case topic := <-ch:
|
||||
got = append(got, topic)
|
||||
case <-time.After(time.Second):
|
||||
t.Fatalf("signals so far: %v", got)
|
||||
}
|
||||
}
|
||||
assert.ElementsMatch(t, []events.Topic{events.Services, events.Jobs}, got)
|
||||
}
|
||||
|
||||
func TestMergeService_Refusals(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
id, from string
|
||||
err error
|
||||
wantCode int
|
||||
wantBody string
|
||||
}{
|
||||
{name: "invalid from", id: "1", from: "x", wantCode: http.StatusBadRequest},
|
||||
{name: "empty from", id: "1", from: "", wantCode: http.StatusBadRequest},
|
||||
{name: "refused", id: "1", from: "17", err: &model.MergeRefusedError{Reason: model.ReasonMergeLiveOld},
|
||||
wantCode: http.StatusConflict, wantBody: "Cannot merge: the old service still exists."},
|
||||
{name: "job active", id: "1", from: "17", err: model.ErrJobActive,
|
||||
wantCode: http.StatusConflict, wantBody: "Cannot merge: an update is already in progress for this service."},
|
||||
{name: "agent revoked", id: "1", from: "17", err: model.ErrRevoked, wantCode: http.StatusConflict, wantBody: "Cannot merge:"},
|
||||
{name: "not found", id: "1", from: "17", err: model.ErrNotFound, wantCode: http.StatusNotFound},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
srv, store, _, _, _ := newTestEnv(t)
|
||||
store.mergeErr = tt.err
|
||||
|
||||
resp, body := postTo(t, srv, "/services/"+tt.id+"/merge", url.Values{"from": {tt.from}})
|
||||
|
||||
assert.Equal(t, tt.wantCode, resp.StatusCode)
|
||||
assert.Contains(t, body, tt.wantBody)
|
||||
assert.Empty(t, resp.Header.Get("Location"))
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// serviceDetail's data is what the service template reads.
|
||||
func TestServiceDetail_DriftAndMoveData(t *testing.T) {
|
||||
removed := time.Now()
|
||||
live := model.Service{ID: 1, Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web", ActualRef: "slusk:1"}
|
||||
gone := model.Service{ID: 2, Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web", ActualRef: "slusk:1", RemovedAt: &removed}
|
||||
swarm := model.Service{ID: 3, Kind: protocol.KindSwarm, SwarmService: "gitea", ActualRef: "gitea:1"}
|
||||
container := model.Service{ID: 4, Kind: protocol.KindContainer, ContainerName: "adhoc", ActualRef: "redis:7"}
|
||||
self := model.Service{ID: 5, Kind: protocol.KindCompose, ComposeProject: "pu", ComposeService: "agent", ActualRef: "pu:1", Self: true}
|
||||
move := model.Move{FromID: 2, ToID: 1, FromAgent: "old", ToAgent: "new"}
|
||||
sources := []model.Service{gone}
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
id int64
|
||||
restorable error
|
||||
wantBlocked string
|
||||
wantFrom *model.Move
|
||||
wantTo *model.Move
|
||||
wantSources []model.Service
|
||||
}{
|
||||
{name: "live row gets the suggestion and merge choices", id: 1, wantFrom: &move, wantSources: sources},
|
||||
{name: "removed row gets where it went", id: 2, wantTo: &move},
|
||||
{name: "swarm service can merge, no suggestion", id: 3, wantSources: sources},
|
||||
{name: "container cannot merge", id: 4},
|
||||
{name: "self cannot merge", id: 5},
|
||||
{name: "restore refusal is passed on", id: 1, restorable: &model.NotUpdatableError{Reason: model.ReasonSelf},
|
||||
wantBlocked: model.ReasonSelf, wantFrom: &move, wantSources: sources},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
store := &fakeStore{services: []model.Service{live, gone, swarm, container, self},
|
||||
moves: []model.Move{move}, sources: sources}
|
||||
jobs := &fakeJobs{restorable: tt.restorable}
|
||||
s, err := New(store, &fakeHub{}, jobs, events.NewBroker(), Config{}, slog.New(slog.NewTextHandler(io.Discard, nil)))
|
||||
require.NoError(t, err)
|
||||
req := httptest.NewRequest(http.MethodGet, "/services/"+strconv.FormatInt(tt.id, 10), nil)
|
||||
|
||||
data, err := s.serviceDetail(req, tt.id)
|
||||
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, tt.wantBlocked, data.RestoreBlocked)
|
||||
assert.Equal(t, tt.wantFrom, data.MoveFrom)
|
||||
assert.Equal(t, tt.wantTo, data.MoveTo)
|
||||
assert.Equal(t, tt.wantSources, data.MergeSources)
|
||||
})
|
||||
}
|
||||
}
|
||||
+175
-16
@@ -20,6 +20,7 @@ import (
|
||||
"github.com/shcizo/package-updater/internal/master/model"
|
||||
"github.com/shcizo/package-updater/internal/master/web/icons"
|
||||
"github.com/shcizo/package-updater/internal/master/webauth"
|
||||
"github.com/shcizo/package-updater/internal/protocol"
|
||||
)
|
||||
|
||||
//go:embed templates/*.html static/*
|
||||
@@ -42,6 +43,16 @@ type Store interface {
|
||||
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
|
||||
// AcceptRunning sets the desired ref of a drifted service; model.ErrNotFound
|
||||
// for an unknown or removed service, model.ErrConflict when the running
|
||||
// version is no longer actualRef@actualDigest.
|
||||
AcceptRunning(ctx context.Context, id int64, actualRef, actualDigest, desired string) error
|
||||
// MoveCandidates returns the suggested moves of removed services.
|
||||
MoveCandidates(ctx context.Context) ([]model.Move, error)
|
||||
// MergeSources returns the removed services that could be merged into toID.
|
||||
MergeSources(ctx context.Context, toID int64) ([]model.Service, error)
|
||||
// MergeService moves the live service toID onto the removed fromID.
|
||||
MergeService(ctx context.Context, fromID, toID int64) error
|
||||
GetJob(ctx context.Context, id int64) (model.Job, error)
|
||||
ListJobs(ctx context.Context, f model.JobFilter) ([]model.Job, error)
|
||||
ListAgents(ctx context.Context) ([]model.Agent, error)
|
||||
@@ -65,6 +76,10 @@ type Jobs interface {
|
||||
// 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)
|
||||
// Restorable explains why a drifted service cannot be restored.
|
||||
Restorable(svc model.Service) error
|
||||
// RestoreDesired deploys the service's desired ref again.
|
||||
RestoreDesired(ctx context.Context, serviceID int64, requestedBy string) (model.Job, error)
|
||||
}
|
||||
|
||||
// Broker delivers change signals to open streams and accepts signals from
|
||||
@@ -122,6 +137,9 @@ func (s *Server) Register(mux *http.ServeMux, protect func(http.Handler) http.Ha
|
||||
route("GET /fragments/services/{id}/jobs", s.serviceJobsFragment)
|
||||
route("POST /services/{id}/update", s.updateService)
|
||||
route("POST /services/{id}/mode", s.setMode)
|
||||
route("POST /services/{id}/restore", s.restoreDesired)
|
||||
route("POST /services/{id}/accept", s.acceptRunning)
|
||||
route("POST /services/{id}/merge", s.mergeService)
|
||||
route("GET /jobs", s.jobsPage)
|
||||
route("GET /fragments/jobs", s.jobsFragment)
|
||||
route("GET /jobs/{id}", s.jobPage)
|
||||
@@ -186,8 +204,14 @@ type pageData struct {
|
||||
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
|
||||
// RestoreBlocked is why a drifted Service cannot be restored; "" when it
|
||||
// can. Only meaningful when Service.Drifted().
|
||||
RestoreBlocked string
|
||||
MoveFrom *model.Move // a live Service: the removed service it probably replaces
|
||||
MoveTo *model.Move // a removed Service: the live service it probably moved to
|
||||
MergeSources []model.Service // a live Compose or Swarm Service: removed ones to merge in
|
||||
Jobs []model.Job
|
||||
Job model.Job
|
||||
}
|
||||
|
||||
func (s *Server) base(r *http.Request, title string) pageData {
|
||||
@@ -352,6 +376,14 @@ func (s *Server) serviceDetail(r *http.Request, id int64) (pageData, error) {
|
||||
return data, err
|
||||
}
|
||||
data.AutoBlocked = s.autoBlocked(svc)
|
||||
if err := s.jobs.Restorable(svc); errors.As(err, &nu) {
|
||||
data.RestoreBlocked = nu.Reason
|
||||
} else if err != nil {
|
||||
return data, err
|
||||
}
|
||||
if err := s.moveData(r, &data); err != nil {
|
||||
return data, err
|
||||
}
|
||||
versions, err := s.store.RecentVersions(r.Context(), svc.ImageRepo, versionsLimit)
|
||||
if err != nil {
|
||||
return data, err
|
||||
@@ -362,6 +394,27 @@ func (s *Server) serviceDetail(r *http.Request, id int64) (pageData, error) {
|
||||
return data, err
|
||||
}
|
||||
|
||||
// moveData fills the move suggestion and the merge choices of the service page.
|
||||
func (s *Server) moveData(r *http.Request, data *pageData) error {
|
||||
svc := data.Service
|
||||
moves, err := s.store.MoveCandidates(r.Context())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for i, m := range moves {
|
||||
switch svc.ID {
|
||||
case m.ToID:
|
||||
data.MoveFrom = &moves[i]
|
||||
case m.FromID:
|
||||
data.MoveTo = &moves[i]
|
||||
}
|
||||
}
|
||||
if svc.RemovedAt == nil && !svc.Self && svc.Kind != protocol.KindContainer {
|
||||
data.MergeSources, err = s.store.MergeSources(r.Context(), svc.ID)
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
func (s *Server) servicePage(w http.ResponseWriter, r *http.Request) {
|
||||
id, ok := pathID(w, r)
|
||||
if !ok {
|
||||
@@ -413,18 +466,130 @@ func (s *Server) updateService(w http.ResponseWriter, r *http.Request) {
|
||||
case err == nil || errors.Is(err, model.ErrJobActive):
|
||||
http.Redirect(w, r, "/jobs/"+strconv.FormatInt(job.ID, 10), http.StatusSeeOther)
|
||||
case errors.As(err, &nu):
|
||||
data, derr := s.serviceDetail(r, id)
|
||||
if derr != nil {
|
||||
s.failLookup(w, derr)
|
||||
return
|
||||
}
|
||||
data.Error = "Cannot update: " + nu.Reason + "."
|
||||
s.page(w, r, http.StatusConflict, "service.html", data)
|
||||
s.renderServiceError(w, r, id, http.StatusConflict, "Cannot update: "+nu.Reason+".")
|
||||
default:
|
||||
s.failLookup(w, err)
|
||||
}
|
||||
}
|
||||
|
||||
// restoreDesired deploys the desired ref of a drifted service again.
|
||||
func (s *Server) restoreDesired(w http.ResponseWriter, r *http.Request) {
|
||||
id, ok := pathID(w, r)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
sess, _ := webauth.SessionFrom(r.Context())
|
||||
job, err := s.jobs.RestoreDesired(r.Context(), id, sess.Subject)
|
||||
var nu *model.NotUpdatableError
|
||||
switch {
|
||||
case err == nil || errors.Is(err, model.ErrJobActive):
|
||||
http.Redirect(w, r, "/jobs/"+strconv.FormatInt(job.ID, 10), http.StatusSeeOther)
|
||||
case errors.As(err, &nu):
|
||||
s.renderServiceError(w, r, id, http.StatusConflict, "Cannot restore: "+nu.Reason+".")
|
||||
default:
|
||||
s.failLookup(w, err)
|
||||
}
|
||||
}
|
||||
|
||||
// acceptRunning makes the running version of a drifted service the desired
|
||||
// one. The form's actual field is the running ref the user saw (Service.RunningRef);
|
||||
// it only guards against the service having changed since, the value written
|
||||
// comes from the row.
|
||||
func (s *Server) acceptRunning(w http.ResponseWriter, r *http.Request) {
|
||||
id, ok := pathID(w, r)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
actual := r.PostFormValue("actual")
|
||||
if actual == "" {
|
||||
http.Error(w, "missing actual", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
svc, err := s.store.GetService(r.Context(), id)
|
||||
if err == nil && svc.RemovedAt != nil {
|
||||
err = model.ErrNotFound
|
||||
}
|
||||
if err != nil {
|
||||
s.failLookup(w, err)
|
||||
return
|
||||
}
|
||||
if !svc.Drifted() {
|
||||
s.renderServiceError(w, r, id, http.StatusConflict, "Cannot accept: the service runs its desired version.")
|
||||
return
|
||||
}
|
||||
if actual != svc.RunningRef() {
|
||||
s.renderServiceError(w, r, id, http.StatusConflict, "The running version changed; check it again.")
|
||||
return
|
||||
}
|
||||
desired := svc.RunningRef()
|
||||
err = s.store.AcceptRunning(r.Context(), id, svc.ActualRef, svc.ActualDigest, desired)
|
||||
switch {
|
||||
case errors.Is(err, model.ErrConflict):
|
||||
s.renderServiceError(w, r, id, http.StatusConflict, "The running version changed; check it again.")
|
||||
return
|
||||
case err != nil:
|
||||
s.failLookup(w, err)
|
||||
return
|
||||
}
|
||||
sess, _ := webauth.SessionFrom(r.Context())
|
||||
s.log.Info("drift_accepted", "service_id", id, "subject", sess.Subject, "desired", desired)
|
||||
s.broker.Publish(events.Services)
|
||||
if r.PostFormValue("return") == "overview" {
|
||||
http.Redirect(w, r, "/", http.StatusSeeOther)
|
||||
return
|
||||
}
|
||||
http.Redirect(w, r, "/services/"+strconv.FormatInt(id, 10), http.StatusSeeOther)
|
||||
}
|
||||
|
||||
// mergeService moves the live service {id} onto the removed service in the
|
||||
// form field "from", which keeps its history, mode and desired ref.
|
||||
func (s *Server) mergeService(w http.ResponseWriter, r *http.Request) {
|
||||
id, ok := pathID(w, r)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
from, err := strconv.ParseInt(r.PostFormValue("from"), 10, 64)
|
||||
if err != nil {
|
||||
http.Error(w, "invalid service", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
err = s.store.MergeService(r.Context(), from, id)
|
||||
var refused *model.MergeRefusedError
|
||||
switch {
|
||||
case errors.As(err, &refused):
|
||||
s.renderServiceError(w, r, id, http.StatusConflict, "Cannot merge: "+refused.Reason+".")
|
||||
return
|
||||
case errors.Is(err, model.ErrJobActive):
|
||||
s.renderServiceError(w, r, id, http.StatusConflict, "Cannot merge: "+model.ErrJobActive.Error()+".")
|
||||
return
|
||||
case errors.Is(err, model.ErrRevoked):
|
||||
s.renderServiceError(w, r, id, http.StatusConflict, "Cannot merge: the agent is revoked.")
|
||||
return
|
||||
case errors.Is(err, model.ErrConflict):
|
||||
s.renderServiceError(w, r, id, http.StatusConflict, "Cannot merge: the service changed; check it again.")
|
||||
return
|
||||
case err != nil:
|
||||
s.failLookup(w, err)
|
||||
return
|
||||
}
|
||||
sess, _ := webauth.SessionFrom(r.Context())
|
||||
s.log.Info("service_merged", "from_id", from, "to_id", id, "subject", sess.Subject)
|
||||
s.broker.Publish(events.Services)
|
||||
s.broker.Publish(events.Jobs)
|
||||
http.Redirect(w, r, "/services/"+strconv.FormatInt(from, 10), http.StatusSeeOther)
|
||||
}
|
||||
|
||||
// renderServiceError renders the service page of id with msg as its error.
|
||||
func (s *Server) renderServiceError(w http.ResponseWriter, r *http.Request, id int64, status int, msg string) {
|
||||
data, err := s.serviceDetail(r, id)
|
||||
if err != nil {
|
||||
s.failLookup(w, err)
|
||||
return
|
||||
}
|
||||
data.Error = msg
|
||||
s.page(w, r, status, "service.html", data)
|
||||
}
|
||||
|
||||
// setMode switches a service between manual and auto. Manual is always
|
||||
// allowed; auto only when jobs.AutoAllowed says so.
|
||||
func (s *Server) setMode(w http.ResponseWriter, r *http.Request) {
|
||||
@@ -448,13 +613,7 @@ func (s *Server) setMode(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
var nu *model.NotUpdatableError
|
||||
if err := s.jobs.AutoAllowed(svc); errors.As(err, &nu) {
|
||||
data, derr := s.serviceDetail(r, id)
|
||||
if derr != nil {
|
||||
s.failLookup(w, derr)
|
||||
return
|
||||
}
|
||||
data.Error = "Cannot enable auto: " + nu.Reason + "."
|
||||
s.page(w, r, http.StatusConflict, "service.html", data)
|
||||
s.renderServiceError(w, r, id, http.StatusConflict, "Cannot enable auto: "+nu.Reason+".")
|
||||
return
|
||||
} else if err != nil {
|
||||
s.fail(w, err)
|
||||
|
||||
@@ -35,6 +35,13 @@ type fakeStore struct {
|
||||
revokedID int64
|
||||
revokeFailed []int64 // job IDs RevokeAgent reports as failed
|
||||
versions []model.ImageVersion
|
||||
|
||||
acceptCalls []acceptCall
|
||||
acceptErr error
|
||||
moves []model.Move
|
||||
sources []model.Service
|
||||
merged [][2]int64 // from, to
|
||||
mergeErr error
|
||||
}
|
||||
|
||||
func (f *fakeStore) RecentVersions(_ context.Context, repo string, limit int) ([]model.ImageVersion, error) {
|
||||
@@ -142,6 +149,9 @@ type fakeJobs struct {
|
||||
triggered []int64 // version IDs passed to TriggerVersion
|
||||
triggerSvc []int64 // service IDs passed to TriggerVersion
|
||||
plainCalls int // Trigger calls
|
||||
|
||||
restorable error // returned by Restorable
|
||||
restored []int64 // service IDs passed to RestoreDesired
|
||||
}
|
||||
|
||||
func (f *fakeJobs) Updatable(model.Service) error { return f.updatable }
|
||||
|
||||
Reference in New Issue
Block a user