shcizo/package-updater
1
0
Fork 0
Code Issues 8 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat(web): restore, accept and merge endpoints for drifted and moved services (#10)

Browse Source
This commit is contained in:
shcizo
2026-10-13 10:28:34 +02:00
parent d0bf3e09d9
commit 19a21da7bf
3 changed files with 466 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/master/web/drift_test.go
+287
View File
@@ -0,0 +1,287 @@
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"}})
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"}, "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
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: "not drifted", path: "/services/1/accept", form: url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.4.2"}},
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"}},
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 meanwhile", path: "/services/1/accept", form: url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.4.2"}},
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)
}
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)
})
}
}
internal/master/web/server.go
+167
View File
@@ -18,6 +18,7 @@ import (
"github.com/shcizo/package-updater/internal/master/events"
"github.com/shcizo/package-updater/internal/master/model"
"github.com/shcizo/package-updater/internal/master/webauth"
"github.com/shcizo/package-updater/internal/protocol"
)
//go:embed templates/*.html static/*
@@ -40,6 +41,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)
@@ -63,6 +74,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
@@ -118,6 +133,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)
@@ -145,6 +163,12 @@ type pageData struct {
Versions []versionRow
Polling string // what the service page says about registry polling
PollFailed bool // Polling is the last registry error
// 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
}
@@ -235,6 +259,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
@@ -245,6 +277,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 {
@@ -308,6 +361,120 @@ func (s *Server) updateService(w http.ResponseWriter, r *http.Request) {
}
}
// 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 ref the user saw; 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 {
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.ActualRef {
s.renderServiceError(w, r, id, http.StatusConflict, "The running version changed; check it again.")
return
}
desired := model.RunningRef(svc.ActualRef, svc.ActualDigest)
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", from, "to", 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.render(w, status, "service.html", "layout", 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) {
internal/master/web/server_test.go
+10
View File
@@ -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 }