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

feat(master,agent): deploy digest-pinned refs and route them to protocol 3 agents (#9)

Browse Source
This commit is contained in:
shcizo
2026-10-13 08:07:45 +02:00
parent af61f17dee
commit d0726ca992
15 changed files with 303 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/agent/updater/compose.go
+2 -2
View File
@@ -91,7 +91,7 @@ func NewComposeExecutor(cli ComposeDocker, resolve *Resolver, selfID string, run
return &ComposeExecutor{cli: cli, resolve: resolve, selfID: selfID, run: run, build: NewBuildChecker(cli, selfID, runOut)}
}
// Execute pins job.Ref to its current digest, writes it to the stack's
// Execute pins job.Ref to its current digest (a ref that has one is used as is), writes it to the stack's
// override file and recreates the service. Working directory and config
// files come from the service's container labels on this machine, never from
// the job. It returns the pinned ref.
@@ -150,7 +150,7 @@ func (e *ComposeExecutor) Execute(ctx context.Context, job protocol.JobStart, ou
if err != nil {
return "", err
}
fmt.Fprintf(out, "resolved %s\n", pinned)
fmt.Fprintln(out, resolveNote(job.Ref, pinned))
if _, err := WriteOverride(dir, service, pinned); err != nil {
return "", err
internal/agent/updater/compose_test.go
+18
View File
@@ -108,6 +108,24 @@ func TestComposeExecutor(t *testing.T) {
assert.Equal(t, pinned, serviceImage(doc, "web"))
assert.Contains(t, out.String(), "resolved "+pinned)
})
t.Run("a pinned ref is written verbatim and not resolved", func(t *testing.T) {
t.Cleanup(cleanOverride)
for _, ref := range []string{"gitea.shcizo.se/shcizo/slusk:1.3@" + d, "gitea.shcizo.se/shcizo/slusk@" + d} {
rec := &recorder{}
dist := &fakeDist{digest: "sha256:other"}
var out strings.Builder
pinnedJob := job
pinnedJob.Ref = ref
got, err := setup(fakeCompose{rw, []container.Summary{ct}}, dist, rec).Execute(context.Background(), pinnedJob, &out)
require.NoError(t, err)
assert.Equal(t, ref, got)
assert.Empty(t, dist.refs, "no registry lookup")
_, doc := readOverride(t, stack)
assert.Equal(t, ref, serviceImage(doc, "web"))
assert.Contains(t, out.String(), "pinned "+ref)
assert.NotContains(t, out.String(), "resolved")
}
})
t.Run("built from source is refused before anything is written", func(t *testing.T) {
t.Cleanup(cleanOverride)
rec := &recorder{}
internal/agent/updater/resolve.go
+17 -4
View File
@@ -16,7 +16,8 @@ type DistributionClient interface {
}
// Resolver turns "repo:tag" into "repo:tag@digest" by asking the local
// daemon, which contacts the registry.
// daemon, which contacts the registry. A ref that already has a digest is
// passed through.
type Resolver struct {
cli DistributionClient
auth *RegistryAuth
@@ -27,13 +28,15 @@ func NewResolver(cli DistributionClient, auth *RegistryAuth) *Resolver {
return &Resolver{cli: cli, auth: auth}
}
// Resolve returns ref pinned to the digest its tag currently points at. It
// fails for a ref without a tag or with a digest already.
// Resolve returns ref pinned to the digest its tag currently points at. A ref
// that already has a digest ("repo:tag@digest" or "repo@digest") is returned
// unchanged without asking the daemon. It fails for a ref with neither tag nor
// digest.
func (r *Resolver) Resolve(ctx context.Context, ref string) (string, error) {
p := imageref.Parse(ref)
switch {
case p.Digest != "":
return "", fmt.Errorf("image %q already has a digest", ref)
return ref, nil
case p.Tag == "":
return "", fmt.Errorf("image %q has no tag to resolve", ref)
}
@@ -46,3 +49,13 @@ func (r *Resolver) Resolve(ctx context.Context, ref string) (string, error) {
}
return ref + "@" + info.Descriptor.Digest.String(), nil
}
// resolveNote is the job-output line that says which ref is deployed:
// "resolved" when pinned came from resolving a tag, "pinned" when ref already
// carried its digest.
func resolveNote(ref, pinned string) string {
if pinned == ref {
return "pinned " + pinned
}
return "resolved " + pinned
}
internal/agent/updater/resolve_test.go
+11 -4
View File
@@ -21,14 +21,21 @@ func TestResolver(t *testing.T) {
assert.Equal(t, []string{"gitea.shcizo.se/shcizo/slusk:1.4"}, dist.refs)
assert.Equal(t, []string{"enc-gitea"}, dist.auths)
})
t.Run("digest or missing tag is refused without asking the daemon", func(t *testing.T) {
for _, ref := range []string{"repo@sha256:x", "repo:1@sha256:x", "repo"} {
t.Run("a ref with a digest passes through without asking the daemon", func(t *testing.T) {
for _, ref := range []string{"repo@sha256:x", "repo:1@sha256:x", "localhost:5000/repo:1@sha256:x"} {
dist := &fakeDist{digest: d}
_, err := NewResolver(dist, nil).Resolve(context.Background(), ref)
assert.Error(t, err, ref)
got, err := NewResolver(dist, nil).Resolve(context.Background(), ref)
require.NoError(t, err, ref)
assert.Equal(t, ref, got)
assert.Empty(t, dist.refs)
}
})
t.Run("a bare ref is refused without asking the daemon", func(t *testing.T) {
dist := &fakeDist{digest: d}
_, err := NewResolver(dist, nil).Resolve(context.Background(), "repo")
assert.ErrorContains(t, err, "has no tag to resolve")
assert.Empty(t, dist.refs)
})
t.Run("daemon error is wrapped", func(t *testing.T) {
cause := errors.New("unauthorized")
_, err := NewResolver(&fakeDist{err: cause}, nil).Resolve(context.Background(), "repo:1")
internal/agent/updater/swarm.go
+2 -2
View File
@@ -35,7 +35,7 @@ func NewSwarmExecutor(cli SwarmClient, resolve *Resolver, auth *RegistryAuth) *S
}
// Execute checks that the local daemon is a manager of the job's cluster,
// pins job.Ref to its current digest and updates the service to it. The
// pins job.Ref to its current digest (a ref that has one is used as is) and updates the service to it. The
// update is addressed by the ID and version of the service that was
// inspected, so it changes exactly that object. Success means the daemon
// accepted the new spec, not that the tasks have converged. It returns the
@@ -71,7 +71,7 @@ func (e *SwarmExecutor) Execute(ctx context.Context, job protocol.JobStart, out
if err != nil {
return "", err
}
fmt.Fprintf(out, "resolved %s\n", pinned)
fmt.Fprintln(out, resolveNote(job.Ref, pinned))
spec := svc.Spec
spec.TaskTemplate.ContainerSpec.Image = pinned
internal/agent/updater/swarm_test.go
+16
View File
@@ -77,6 +77,22 @@ func swarmSvc(name string) swarm.Service {
}
}
func TestSwarmUsesPinnedRef(t *testing.T) {
const ref = "gitea/gitea:1.26@sha256:abababababababababababababababababababababababababababababababab"
job := protocol.JobStart{Kind: protocol.KindSwarm, SwarmClusterID: "c1", SwarmService: "gitea_gitea", Ref: ref}
fake := &fakeSwarm{info: managerOf("c1"), svc: swarmSvc("gitea_gitea")}
dist := &fakeDist{digest: "sha256:other"}
var out bytes.Buffer
got, err := NewSwarmExecutor(fake, NewResolver(dist, nil), nil).Execute(context.Background(), job, &out)
require.NoError(t, err)
assert.Equal(t, ref, got)
assert.Equal(t, ref, fake.updSpec.TaskTemplate.ContainerSpec.Image)
assert.Empty(t, dist.refs, "no registry lookup")
assert.Contains(t, out.String(), "pinned "+ref)
}
func TestSwarmExecutor(t *testing.T) {
job := protocol.JobStart{Kind: protocol.KindSwarm, SwarmClusterID: "c1", SwarmService: "gitea_gitea", Ref: "gitea/gitea:1.27"}
workerInfo := managerOf("c1")
internal/e2e/agent_master_test.go
+3 -1
View File
@@ -328,7 +328,9 @@ type currentHub struct{ p atomic.Pointer[hub.Hub] }
func (c *currentHub) ServeHTTP(w http.ResponseWriter, r *http.Request) { c.p.Load().ServeHTTP(w, r) }
func (c *currentHub) Disconnect(id int64) { c.p.Load().Disconnect(id) }
func (c *currentHub) Route(svc model.Service) (model.Route, error) { return c.p.Load().Route(svc) }
func (c *currentHub) Route(svc model.Service, minProtocol int) (model.Route, error) {
return c.p.Load().Route(svc, minProtocol)
}
func (c *currentHub) StartJob(r model.Route, id int64, s protocol.JobStart) error {
return c.p.Load().StartJob(r, id, s)
}
internal/e2e/jobs_test.go
+2 -2
View File
@@ -119,7 +119,7 @@ func (e *jobEnv) waitRoutable(t *testing.T) {
t.Helper()
svc, err := e.store.GetService(context.Background(), 1)
require.NoError(t, err)
require.Eventually(t, func() bool { _, err := e.m.hub.Route(svc); return err == nil }, waitFor, pollEvery)
require.Eventually(t, func() bool { _, err := e.m.hub.Route(svc, protocol.MinJobVersion); return err == nil }, waitFor, pollEvery)
}
func (e *jobEnv) job(t *testing.T, id int64) model.Job {
@@ -240,7 +240,7 @@ func TestPhaseOneAgentGetsNoJobs(t *testing.T) {
require.NoError(t, wsjson.Write(ctx, c, snap))
require.Eventually(t, func() bool {
_, err := m.hub.Route(svcRow)
_, err := m.hub.Route(svcRow, protocol.MinJobVersion)
return errors.Is(err, model.ErrAgentTooOld)
}, waitFor, pollEvery)
internal/e2e/webhook_test.go
+2 -1
View File
@@ -16,6 +16,7 @@ import (
"github.com/shcizo/package-updater/internal/master/model"
"github.com/shcizo/package-updater/internal/master/webhook"
"github.com/shcizo/package-updater/internal/protocol"
)
// webhookServer mounts the webhook handler on a test server, wired to the
@@ -111,7 +112,7 @@ func TestWebhookFailedAutoJobIs500(t *testing.T) {
e.agent.stopConn() // no agent connection: the job fails at once with "agent offline"
require.Eventually(t, func() bool {
svc, _ := e.store.GetService(context.Background(), 1)
_, err := e.m.hub.Route(svc)
_, err := e.m.hub.Route(svc, protocol.MinJobVersion)
return err != nil
}, waitFor, pollEvery)
internal/master/hub/jobs_test.go
+2 -2
View File
@@ -43,7 +43,7 @@ func waitRoute(t *testing.T, h *Hub) model.Route {
var r model.Route
require.Eventually(t, func() bool {
var err error
r, err = h.Route(svc)
r, err = h.Route(svc, protocol.MinJobVersion)
return err == nil
}, 2*time.Second, 5*time.Millisecond)
return r
@@ -61,7 +61,7 @@ func TestHub_ReconcilesBeforeRoutable(t *testing.T) {
send(t, c, protocol.TypeHello, protocol.Hello{Hostname: "node-2", ProtocolVersion: 2, KnownJobs: []string{"7", "9", "x"}})
svc := model.Service{Kind: protocol.KindCompose, AgentID: ptr(node2.ID)}
time.Sleep(50 * time.Millisecond) // hello is now blocked inside FailLostJobs
_, err = h.Route(svc)
_, err = h.Route(svc, protocol.MinJobVersion)
assert.ErrorIs(t, err, model.ErrAgentOffline)
close(store.lostGate)
internal/master/hub/route.go
+7 -6
View File
@@ -14,9 +14,10 @@ type candidate struct {
hello protocol.Hello
}
// Route picks the connection a job for svc goes to. It returns
// model.ErrAgentOffline or model.ErrAgentTooOld when there is none.
func (h *Hub) Route(svc model.Service) (model.Route, error) {
// Route picks the connection a job for svc goes to, among agents that speak
// at least protocol version minProtocol. It returns model.ErrAgentOffline or
// model.ErrAgentTooOld when there is none.
func (h *Hub) Route(svc model.Service, minProtocol int) (model.Route, error) {
h.mu.Lock()
conns := make([]candidate, 0, len(h.conns))
for id, c := range h.conns {
@@ -25,7 +26,7 @@ func (h *Hub) Route(svc model.Service) (model.Route, error) {
}
}
h.mu.Unlock()
return route(svc, conns)
return route(svc, conns, minProtocol)
}
// route picks the connection a job for svc goes to. Compose services and
@@ -33,7 +34,7 @@ func (h *Hub) Route(svc model.Service) (model.Route, error) {
// run on the reporting agent if it qualifies, else on any other connected
// manager of the same cluster (lowest agent ID first). Too-old wins over
// offline only when a connection would otherwise have qualified.
func route(svc model.Service, conns []candidate) (model.Route, error) {
func route(svc model.Service, conns []candidate, minProtocol int) (model.Route, error) {
sort.Slice(conns, func(a, b int) bool { return conns[a].agentID < conns[b].agentID })
tooOld := false
try := func(c candidate) bool {
@@ -42,7 +43,7 @@ func route(svc model.Service, conns []candidate) (model.Route, error) {
return false
}
}
if c.hello.ProtocolVersion < protocol.MinJobVersion {
if c.hello.ProtocolVersion < minProtocol {
tooOld = true
return false
}
internal/master/hub/route_test.go
+24 -15
View File
@@ -12,37 +12,46 @@ import (
func TestRoute(t *testing.T) {
composeSvc := model.Service{Kind: protocol.KindCompose, AgentID: ptr(int64(2))}
swarmSvc := model.Service{Kind: protocol.KindSwarm, AgentID: ptr(int64(1)), SwarmClusterID: "c1"}
v2 := protocol.Hello{ProtocolVersion: protocol.Version}
mgrV2 := protocol.Hello{ProtocolVersion: protocol.Version, SwarmManager: true, SwarmClusterID: "c1"}
v2 := protocol.Hello{ProtocolVersion: 2}
v3 := protocol.Hello{ProtocolVersion: 3}
mgrV2 := protocol.Hello{ProtocolVersion: 2, SwarmManager: true, SwarmClusterID: "c1"}
mgrV3 := protocol.Hello{ProtocolVersion: 3, SwarmManager: true, SwarmClusterID: "c1"}
jobs, pinned := protocol.MinJobVersion, protocol.MinPinnedRefVersion
tests := []struct {
name string
svc model.Service
conns []candidate
min int
want model.Route
err error
}{
{"compose on own agent", composeSvc, []candidate{{2, 20, v2}}, model.Route{AgentID: 2, ConnID: 20}, nil},
{"compose, own agent too old", composeSvc, []candidate{{2, 20, protocol.Hello{}}}, model.Route{}, model.ErrAgentTooOld},
{"compose, own agent missing", composeSvc, nil, model.Route{}, model.ErrAgentOffline},
{"compose, only another agent online", composeSvc, []candidate{{5, 50, v2}}, model.Route{}, model.ErrAgentOffline},
{"swarm on reporting agent", swarmSvc, []candidate{{1, 10, mgrV2}, {3, 30, mgrV2}}, model.Route{AgentID: 1, ConnID: 10}, nil},
{"swarm, reporting agent offline, other manager", swarmSvc, []candidate{{3, 30, mgrV2}}, model.Route{AgentID: 3, ConnID: 30}, nil},
{"swarm, lowest agent id first", swarmSvc, []candidate{{4, 40, mgrV2}, {3, 30, mgrV2}}, model.Route{AgentID: 3, ConnID: 30}, nil},
{"compose on own agent", composeSvc, []candidate{{2, 20, v2}}, jobs, model.Route{AgentID: 2, ConnID: 20}, nil},
{"compose, own agent too old", composeSvc, []candidate{{2, 20, protocol.Hello{}}}, jobs, model.Route{}, model.ErrAgentTooOld},
{"compose, own agent missing", composeSvc, nil, jobs, model.Route{}, model.ErrAgentOffline},
{"compose, only another agent online", composeSvc, []candidate{{5, 50, v2}}, jobs, model.Route{}, model.ErrAgentOffline},
{"compose, v2 agent and a pinned ref", composeSvc, []candidate{{2, 20, v2}}, pinned, model.Route{}, model.ErrAgentTooOld},
{"compose, v3 agent and a pinned ref", composeSvc, []candidate{{2, 20, v3}}, pinned, model.Route{AgentID: 2, ConnID: 20}, nil},
{"swarm on reporting agent", swarmSvc, []candidate{{1, 10, mgrV2}, {3, 30, mgrV2}}, jobs, model.Route{AgentID: 1, ConnID: 10}, nil},
{"swarm, reporting agent offline, other manager", swarmSvc, []candidate{{3, 30, mgrV2}}, jobs, model.Route{AgentID: 3, ConnID: 30}, nil},
{"swarm, lowest agent id first", swarmSvc, []candidate{{4, 40, mgrV2}, {3, 30, mgrV2}}, jobs, model.Route{AgentID: 3, ConnID: 30}, nil},
{"swarm, other manager in another cluster", swarmSvc,
[]candidate{{3, 30, protocol.Hello{ProtocolVersion: 2, SwarmManager: true, SwarmClusterID: "c2"}}}, model.Route{}, model.ErrAgentOffline},
[]candidate{{3, 30, protocol.Hello{ProtocolVersion: 2, SwarmManager: true, SwarmClusterID: "c2"}}}, jobs, model.Route{}, model.ErrAgentOffline},
{"swarm, only a v0 manager of the cluster", swarmSvc,
[]candidate{{1, 10, protocol.Hello{SwarmManager: true, SwarmClusterID: "c1"}}}, model.Route{}, model.ErrAgentTooOld},
[]candidate{{1, 10, protocol.Hello{SwarmManager: true, SwarmClusterID: "c1"}}}, jobs, model.Route{}, model.ErrAgentTooOld},
{"swarm, v0 manager but a v2 manager exists", swarmSvc,
[]candidate{{1, 10, protocol.Hello{SwarmManager: true, SwarmClusterID: "c1"}}, {3, 30, mgrV2}}, model.Route{AgentID: 3, ConnID: 30}, nil},
[]candidate{{1, 10, protocol.Hello{SwarmManager: true, SwarmClusterID: "c1"}}, {3, 30, mgrV2}}, jobs, model.Route{AgentID: 3, ConnID: 30}, nil},
{"swarm, worker of the cluster", swarmSvc,
[]candidate{{1, 10, protocol.Hello{ProtocolVersion: 2, SwarmClusterID: "c1"}}}, model.Route{}, model.ErrAgentOffline},
[]candidate{{1, 10, protocol.Hello{ProtocolVersion: 2, SwarmClusterID: "c1"}}}, jobs, model.Route{}, model.ErrAgentOffline},
{"swarm, manager with empty cluster id", model.Service{Kind: protocol.KindSwarm},
[]candidate{{1, 10, protocol.Hello{ProtocolVersion: 2, SwarmManager: true}}}, model.Route{}, model.ErrAgentOffline},
[]candidate{{1, 10, protocol.Hello{ProtocolVersion: 2, SwarmManager: true}}}, jobs, model.Route{}, model.ErrAgentOffline},
{"swarm, pinned ref falls back to a v3 manager when the reporting one is v2", swarmSvc,
[]candidate{{1, 10, mgrV2}, {3, 30, mgrV3}}, pinned, model.Route{AgentID: 3, ConnID: 30}, nil},
{"swarm, pinned ref and only v2 managers", swarmSvc, []candidate{{1, 10, mgrV2}}, pinned, model.Route{}, model.ErrAgentTooOld},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got, err := route(tt.svc, tt.conns)
got, err := route(tt.svc, tt.conns, tt.min)
assert.ErrorIs(t, err, tt.err)
assert.Equal(t, tt.want, got)
})
internal/master/jobs/jobs.go
+35 -6
View File
@@ -20,13 +20,15 @@ type Store interface {
CreateJob(ctx context.Context, j model.NewJob) (model.Job, error)
FailJob(ctx context.Context, id int64, msg string, at time.Time) (bool, error)
GetJob(ctx context.Context, id int64) (model.Job, error)
GetImageVersion(ctx context.Context, id int64) (model.ImageVersion, error)
}
// Hub routes jobs to agent connections. *hub.Hub satisfies it.
type Hub interface {
// Route picks the connection for svc: model.ErrAgentOffline or
// Route picks the connection for svc among agents speaking at least
// protocol version minProtocol: model.ErrAgentOffline or
// model.ErrAgentTooOld when there is none.
Route(svc model.Service) (model.Route, error)
Route(svc model.Service, minProtocol int) (model.Route, error)
// StartJob sends job.start over exactly the routed connection.
// model.ErrAgentOffline if that connection is gone or was replaced.
StartJob(r model.Route, jobID int64, start protocol.JobStart) error
@@ -116,8 +118,9 @@ func (s *Service) updatable(svc model.Service, explicitRef bool) error {
type StartRequest struct {
Trigger model.JobTrigger
RequestedBy string // OIDC subject for manual jobs; "" otherwise
// Ref is the ref to deploy ("repo:tag"). Empty means the service's
// followed tag, derived from its actual ref.
// Ref is the ref to deploy ("repo:tag", "repo:tag@digest" or
// "repo@digest"). Empty means the service's followed tag, derived from
// its actual ref.
Ref string
}
@@ -131,8 +134,30 @@ func (s *Service) Trigger(ctx context.Context, serviceID int64, requestedBy stri
return s.Start(ctx, svc, StartRequest{Trigger: model.TriggerManual, RequestedBy: requestedBy})
}
// TriggerVersion starts a manual update of the service to exactly the image
// version versionID, which must be a version of the service's own repo; the
// ref is built from the stored version, never taken from the caller. It
// returns model.ErrNotFound for an unknown service, an unknown version or a
// version of another repo, and otherwise what Start returns. A version
// without a tag leaves the service following no tag.
func (s *Service) TriggerVersion(ctx context.Context, serviceID, versionID int64, requestedBy string) (model.Job, error) {
svc, err := s.store.GetService(ctx, serviceID)
if err != nil {
return model.Job{}, err
}
v, err := s.store.GetImageVersion(ctx, versionID)
if err != nil {
return model.Job{}, err
}
if v.Repo != svc.ImageRepo {
return model.Job{}, model.ErrNotFound
}
return s.Start(ctx, svc, StartRequest{Trigger: model.TriggerManual, RequestedBy: requestedBy, Ref: v.Ref()})
}
// Start starts an update of svc. It returns the job, which may already be
// failed ("agent offline", "agent too old for updates"); a
// failed ("agent offline", "agent too old for updates", the latter also when
// the ref has a digest or no tag and the agent speaks protocol version 2); a
// *model.NotUpdatableError when the service cannot be updated (no job
// created); the active job and model.ErrJobActive when one is queued or
// running.
@@ -145,7 +170,11 @@ func (s *Service) Start(ctx context.Context, svc model.Service, req StartRequest
ref = FollowedRef(svc.ActualRef) // updatable checked the tag
}
r, routeErr := s.hub.Route(svc)
minProtocol := protocol.MinJobVersion
if p := imageref.Parse(ref); p.Digest != "" || p.Tag == "" {
minProtocol = protocol.MinPinnedRefVersion
}
r, routeErr := s.hub.Route(svc, minProtocol)
nj := model.NewJob{
ServiceID: svc.ID, FromRef: svc.ActualRef, ToRef: ref,
Trigger: req.Trigger, RequestedBy: req.RequestedBy, CreatedAt: s.now(),
internal/master/jobs/jobs_test.go
+152 -5
View File
@@ -21,6 +21,7 @@ type fakeStore struct {
jobs map[int64]model.Job
created []model.NewJob
active *model.Job // returned with ErrJobActive when set
versions map[int64]model.ImageVersion
revokedAgents map[int64]bool // CreateJob returns ErrAgentOffline for these
}
@@ -54,6 +55,14 @@ func (f *fakeStore) FailJob(_ context.Context, id int64, msg string, at time.Tim
return true, nil
}
func (f *fakeStore) GetImageVersion(_ context.Context, id int64) (model.ImageVersion, error) {
v, ok := f.versions[id]
if !ok {
return model.ImageVersion{}, model.ErrNotFound
}
return v, nil
}
func (f *fakeStore) GetJob(_ context.Context, id int64) (model.Job, error) { return f.jobs[id], nil }
type startCall struct {
@@ -63,13 +72,21 @@ type startCall struct {
}
type fakeHub struct {
route model.Route
routeErr error
startErr error
starts []startCall
route model.Route
routeErr error
startErr error
starts []startCall
agentProtocol int // when set, Route fails with ErrAgentTooOld for a higher minProtocol
mins []int // minProtocol of every Route call
}
func (h *fakeHub) Route(model.Service) (model.Route, error) { return h.route, h.routeErr }
func (h *fakeHub) Route(_ model.Service, minProtocol int) (model.Route, error) {
h.mins = append(h.mins, minProtocol)
if h.agentProtocol != 0 && minProtocol > h.agentProtocol {
return model.Route{}, model.ErrAgentTooOld
}
return h.route, h.routeErr
}
func (h *fakeHub) StartJob(r model.Route, id int64, s protocol.JobStart) error {
h.starts = append(h.starts, startCall{r, id, s})
@@ -341,3 +358,133 @@ func TestFollowedRef(t *testing.T) {
assert.Equal(t, "localhost:5000/foo:latest", FollowedRef("localhost:5000/foo"))
assert.Empty(t, FollowedRef("redis@sha256:abc"))
}
func TestStart_RoutesWithTheMinimumProtocolOfTheRef(t *testing.T) {
tests := []struct {
name string
ref string // "" means the followed tag
want int
}{
{"followed tag", "", protocol.MinJobVersion},
{"repo:tag", "x/y:2", protocol.MinJobVersion},
{"repo:tag@digest", "x/y:2@sha256:a", protocol.MinPinnedRefVersion},
{"repo@digest", "x/y@sha256:a", protocol.MinPinnedRefVersion},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
st := &fakeStore{services: map[int64]model.Service{1: composeSvc}, jobs: map[int64]model.Job{}}
h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}}
s, _ := newService(st, h)
_, err := s.Start(context.Background(), composeSvc, StartRequest{Trigger: model.TriggerManual, Ref: tt.ref})
require.NoError(t, err)
assert.Equal(t, []int{tt.want}, h.mins)
})
}
}
func TestTriggerVersion_Ref(t *testing.T) {
const repo = "gitea.shcizo.se/shcizo/slusk"
noTag := model.Service{ID: 4, Kind: protocol.KindCompose, ComposeProject: "n", ComposeService: "n",
AgentID: ptr(int64(7)), ImageRepo: repo, ActualRef: repo + "@sha256:d1"}
svc := composeSvc
svc.ImageRepo = repo
tests := []struct {
name string
svc model.Service
version model.ImageVersion
wantRef string
wantMin int
}{
{"tag and digest", svc, model.ImageVersion{ID: 10, Repo: repo, Tag: "1.3.0", Digest: "sha256:old"}, repo + ":1.3.0@sha256:old", protocol.MinPinnedRefVersion},
{"tag only", svc, model.ImageVersion{ID: 10, Repo: repo, Tag: "main"}, repo + ":main", protocol.MinJobVersion},
{"digest only", svc, model.ImageVersion{ID: 10, Repo: repo, Digest: "sha256:old"}, repo + "@sha256:old", protocol.MinPinnedRefVersion},
{"service that follows no tag", noTag, model.ImageVersion{ID: 10, Repo: repo, Tag: "1.3.0", Digest: "sha256:old"}, repo + ":1.3.0@sha256:old", protocol.MinPinnedRefVersion},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
st := &fakeStore{
services: map[int64]model.Service{tt.svc.ID: tt.svc},
jobs: map[int64]model.Job{},
versions: map[int64]model.ImageVersion{10: tt.version},
}
h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}}
s, _ := newService(st, h)
job, err := s.TriggerVersion(context.Background(), tt.svc.ID, 10, "sub-1")
require.NoError(t, err)
assert.Equal(t, model.JobQueued, job.Status)
require.Len(t, st.created, 1)
assert.Equal(t, model.TriggerManual, st.created[0].Trigger)
assert.Equal(t, "sub-1", st.created[0].RequestedBy)
assert.Equal(t, tt.wantRef, st.created[0].ToRef)
require.Len(t, h.starts, 1)
assert.Equal(t, tt.wantRef, h.starts[0].start.Ref)
assert.Equal(t, []int{tt.wantMin}, h.mins)
})
}
}
func TestTriggerVersion_NotFound(t *testing.T) {
const repo = "gitea.shcizo.se/shcizo/slusk"
svc := composeSvc
svc.ImageRepo = repo
tests := []struct {
name string
serviceID int64
versionID int64
}{
{"unknown service", 9, 10},
{"unknown version", 1, 99},
{"version of another repo", 1, 11},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
st := &fakeStore{
services: map[int64]model.Service{1: svc},
jobs: map[int64]model.Job{},
versions: map[int64]model.ImageVersion{
10: {ID: 10, Repo: repo, Tag: "1"},
11: {ID: 11, Repo: "evil/other", Tag: "1"},
},
}
h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}}
s, pub := newService(st, h)
_, err := s.TriggerVersion(context.Background(), tt.serviceID, tt.versionID, "sub-1")
assert.ErrorIs(t, err, model.ErrNotFound)
assert.Empty(t, st.created, "no job")
assert.Empty(t, h.starts)
assert.Empty(t, pub.topics)
})
}
}
func TestTriggerVersion_RequiresPinnedProtocol(t *testing.T) {
const repo = "gitea.shcizo.se/shcizo/slusk"
svc := composeSvc
svc.ImageRepo = repo
st := &fakeStore{
services: map[int64]model.Service{1: svc},
jobs: map[int64]model.Job{},
versions: map[int64]model.ImageVersion{
10: {ID: 10, Repo: repo, Tag: "1.3.0", Digest: "sha256:old"},
11: {ID: 11, Repo: repo, Tag: "main"},
},
}
h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}, agentProtocol: 2}
s, _ := newService(st, h)
pinned, err := s.TriggerVersion(context.Background(), 1, 10, "sub-1")
require.NoError(t, err)
assert.Equal(t, model.JobFailed, pinned.Status)
assert.Equal(t, "agent too old for updates", pinned.Error)
assert.Empty(t, h.starts, "nothing sent to the old agent")
plain, err := s.TriggerVersion(context.Background(), 1, 11, "sub-1")
require.NoError(t, err)
assert.Equal(t, model.JobQueued, plain.Status)
}
internal/protocol/protocol.go
+10 -3
View File
@@ -16,11 +16,16 @@ import (
// Version is the protocol version this build speaks. Agents send it in
// Hello. Phase-1 agents do not send the field, which reads as 0.
const Version = 2
const Version = 3
// MinJobVersion is the lowest agent protocol version the master sends jobs to.
const MinJobVersion = 2
// MinPinnedRefVersion is the lowest agent protocol version the master sends
// jobs whose Ref has a digest or no tag to. Version-2 agents refuse a Ref
// with a digest.
const MinPinnedRefVersion = 3
// MaxJobOutput caps the job output an agent keeps and the master stores, in
// bytes. Only the tail is kept. With JSON escaping (worst case 6 bytes per
// byte) a JobResult carrying this much output still fits the 1 MiB read limit.
@@ -123,8 +128,10 @@ type Remove struct {
Key string `json:"key"`
}
// JobStart asks the agent to deploy Ref to one service. Ref is "repo:tag";
// the agent resolves the tag's current digest and deploys "repo:tag@digest".
// JobStart asks the agent to deploy Ref to one service. Ref is "repo:tag",
// "repo:tag@digest" or "repo@digest". For "repo:tag" the agent resolves the
// tag's current digest and deploys "repo:tag@digest"; a Ref with a digest is
// deployed as is.
// The agent reads the Compose working directory and config files from its
// own Docker daemon, not from this message.
type JobStart struct {