fix: address review findings for self-update (#15)

This commit is contained in:
2026-10-13 12:06:49 +02:00
parent fe4aca2883
commit aec21d23d9
25 changed files with 365 additions and 126 deletions
+10 -11
View File
@@ -58,14 +58,12 @@ DATABASE_URL=postgres://... go test -tags integration ./internal/master/store/
the agent's own container whose source equals its destination.** Mounts are read per the agent's own container whose source equals its destination.** Mounts are read per
job (`ContainerInspect(SELF_ID)`) and checked after symlink resolution. This replaces job (`ContainerInspect(SELF_ID)`) and checked after symlink resolution. This replaces
the old `STACKS_ROOT`. the old `STACKS_ROOT`.
- **Self-update**: an agent's own Compose service is an ordinary target. Pull runs in the - **Self-update** (an agent's own Compose service): only `up -d` runs in the helper
old agent; `up -d` runs in a helper container made from the agent's image (same mounts), container; the new agent resumes the job (`updater.ResumeSelfUpdates`) before the runner
labelled `se.shcizo.package-updater.self-update.*`. The new agent finds the helper at and connection start; the old agent records no result once SIGTERM cancels it. Jobs for
startup (`updater.ResumeSelfUpdates`, before runner and connection start) and reports the agent's own service run last on the agent and the master starts them last in a
the result under the old job ID. The old agent records no result once SIGTERM cancels batch. An agent's own Swarm service is refused (`ReasonSelfSwarm`). Protocol 4 is needed,
its context. Self jobs (`JobStart.Self`) run after every other queued job. An agent's also for any service with the `AGENT_IMAGE` repo.
own Swarm service is refused (`ReasonSelfSwarm`). Needs protocol 4. The master updates
itself with no special code: the agent on its host resends the result.
- **Unknown message types are logged and ignored; malformed messages close the connection.** - **Unknown message types are logged and ignored; malformed messages close the connection.**
- **Reconcile before routable**: the hub fails a reconnecting agent's jobs it does not - **Reconcile before routable**: the hub fails a reconnecting agent's jobs it does not
list in `known_jobs`, and only then accepts jobs for that connection. list in `known_jobs`, and only then accepts jobs for that connection.
@@ -131,8 +129,9 @@ DATABASE_URL=postgres://... go test -tags integration ./internal/master/store/
- **`protocol.Service.ConfigFiles` is the raw comma-joined label string**, so `Service` - **`protocol.Service.ConfigFiles` is the raw comma-joined label string**, so `Service`
stays comparable with `==` (`inventory.Diff` depends on it). stays comparable with `==` (`inventory.Diff` depends on it).
- **Helper containers carry `se.shcizo.package-updater.self-update.*` labels** and are - **Helper containers carry `se.shcizo.package-updater.self-update.*` labels** and are
not inventoried (the collector skips any container with the `.job` label). The labels not inventoried (the collector skips any container with the `.job` label). The label
are constants in `internal/protocol`. strings are constants in `internal/agent/updater/selfupdate.go`, and the `.job` one is
repeated in `internal/agent/inventory`.
- **Swarm "succeeded" means the daemon accepted the new spec**, not that tasks converged. - **Swarm "succeeded" means the daemon accepted the new spec**, not that tasks converged.
- **Swarm updates need manager-node API access** for the agent doing the update. - **Swarm updates need manager-node API access** for the agent doing the update.
- **The image's default entrypoint is `updater-agent`.** The master overrides it with - **The image's default entrypoint is `updater-agent`.** The master overrides it with
@@ -145,7 +144,7 @@ DATABASE_URL=postgres://... go test -tags integration ./internal/master/store/
knows about; only the waiting stops. The master never fails or cancels a timed-out job. knows about; only the waiting stops. The master never fails or cancels a timed-out job.
- **A webhook job's target is `<image_repo>:<request tag>`**, built from the service's own - **A webhook job's target is `<image_repo>:<request tag>`**, built from the service's own
repo, never from the request's image string. With an explicit target the "follows no tag" repo, never from the request's image string. With an explicit target the "follows no tag"
refusal does not apply; removed, built, container and Swarm self still do. refusal does not apply; removed, built, container and an agent's own Swarm service still do.
- **`available_*` is cleared by `FinishJob`** on success only when the job's `to_ref` - **`available_*` is cleared by `FinishJob`** on success only when the job's `to_ref`
(as created, without `@digest`) equals `available_ref`, `available_at <= job.created_at` (as created, without `@digest`) equals `available_ref`, `available_at <= job.created_at`
(that version is now running) and, when the job's `to_ref` carries a digest, that digest (that version is now running) and, when the job's `to_ref` carries a digest, that digest
+8 -2
View File
@@ -62,7 +62,7 @@ Content-Type: application/json
in the request are ignored for matching; matching is case-sensitive). `tag` is optional. in the request are ignored for matching; matching is case-sensitive). `tag` is optional.
For every match the master records the version as *available* (the overview shows an For every match the master records the version as *available* (the overview shows an
"update available" badge), including services that cannot be updated automatically "update available" badge), including services that cannot be updated automatically
(updater's own, `build:` and plain containers), so the badge shows for them too. Services (`build:` services, plain containers and an agent's own Swarm service), so the badge shows for them too. Services
in **manual** mode stop there. Services in **auto** in **manual** mode stop there. Services in **auto**
mode also get an update job, and the request waits for those jobs to finish, at most mode also get an update job, and the request waits for those jobs to finish, at most
`UPDATE_TIMEOUT`. `UPDATE_TIMEOUT`.
@@ -299,7 +299,13 @@ any other service; the overview marks the agent's row "This agent".
upgraded to a version with protocol 4 by hand once. upgraded to a version with protocol 4 by hand once.
- **Order.** A job for an agent's own service runs after every other queued job on that - **Order.** A job for an agent's own service runs after every other queued job on that
agent. A master and an agent on the same host share an image, so one push starts both agent. A master and an agent on the same host share an image, so one push starts both
jobs, and the master's job has to finish before the agent replaces itself. jobs, and the master's job has to finish before the agent replaces itself. The master
also starts a job for an agent's own service after the other jobs of the same poll
cycle or webhook request. Any other job that reaches an agent while it replaces itself
is marked lost when the new agent connects without it.
- **Docker socket.** The helper runs on the default network, so `DOCKER_HOST` only works
with a `docker.sock` bind mount. A TCP socket proxy that is reachable only on a Compose
network does not work for self-update.
- **Swarm.** An agent's own Swarm service is refused: the agent cannot update the service - **Swarm.** An agent's own Swarm service is refused: the agent cannot update the service
it runs in. it runs in.
+1 -1
View File
@@ -104,7 +104,7 @@ func run() error {
broker := events.NewBroker() broker := events.NewBroker()
agentHub := hub.New(st, broker, logger) agentHub := hub.New(st, broker, logger)
jobService := jobs.New(st, agentHub, broker, logger) jobService := jobs.New(st, agentHub, broker, cfg.AgentImage, logger)
provider, err := webauth.NewOIDCProvider(ctx, webauth.OIDCConfig{ provider, err := webauth.NewOIDCProvider(ctx, webauth.OIDCConfig{
Issuer: cfg.OIDCIssuer, ClientID: cfg.OIDCClientID, ClientSecret: cfg.OIDCClientSecret, Issuer: cfg.OIDCIssuer, ClientID: cfg.OIDCClientID, ClientSecret: cfg.OIDCClientSecret,
@@ -229,7 +229,7 @@ Online/offline is held in memory only.
| Master restarts mid-job | Jobs in `running` are reconciled when the agent reconnects; jobs the agent does not know are marked `failed` ("lost"). | | Master restarts mid-job | Jobs in `running` are reconciled when the agent reconnects; jobs the agent does not know are marked `failed` ("lost"). |
| New trigger while a job for the same service is queued or running | No new job; only `available_*` is updated. Two jobs never race on one service. | | New trigger while a job for the same service is queued or running | No new job; only `available_*` is updated. Two jobs never race on one service. |
| Job exceeds `UPDATE_TIMEOUT` | The agent cancels it, as today. | | Job exceeds `UPDATE_TIMEOUT` | The agent cancels it, as today. |
| Agent updates itself | The old agent runs `up -d` in a helper container and waits. When the helper recreates the agent, the old agent is stopped and records nothing. The new agent finds the helper by its labels before it connects, so its hello lists the job; it reports the helper's outcome as the job's result. If the helper fails or the new agent never starts, the job stays `running` until the agent reconnects without it ("lost") or `UPDATE_TIMEOUT` ends the wait. | | Agent updates itself | The old agent runs `up -d` in a helper container and waits. If the helper exits while the old agent still runs, the old agent reports the outcome. If the helper stops the old agent, the old agent records nothing; the new agent finds the helper by its labels before it connects, so its hello lists the job, and it reports the helper's outcome. If no new agent resumes the job, it is marked lost when the agent reconnects without it. If `UPDATE_TIMEOUT` passes first, the old agent reports a timeout and leaves the helper running; a later success is then not recorded. |
| Polling fails for one image | Error stored in `last_poll_error` and shown; other images continue. | | Polling fails for one image | Error stored in `last_poll_error` and shown; other images continue. |
### Self-update ### Self-update
@@ -258,11 +258,17 @@ update itself").
- **Stopping.** When the agent is stopped (SIGTERM) while a job runs, the runner drops the - **Stopping.** When the agent is stopped (SIGTERM) while a job runs, the runner drops the
job and records no result, so a "context canceled" result cannot overtake the new job and records no result, so a "context canceled" result cannot overtake the new
agent's. agent's.
- **Order.** `job.start` carries `self`; the agent runs a self job only when no other job - **Order.** `job.start` carries `self`; the agent runs a job for its own service only
is queued (the master's job on a shared host has to finish first). Whether to use a when no other job is queued (the master's job on a shared host has to finish first).
helper is decided by the agent from its own container ID, not from the flag. The master starts such jobs after the other jobs of the same poll cycle or webhook
- **Protocol 4.** The master routes a self job only to an agent speaking protocol 4 request; messages to one agent arrive in order, so the others are queued first. A job
(`agent too old to update itself`). that reaches an agent while it replaces itself is marked lost when the new agent
connects without it. Whether to use a helper is decided by the agent from its own
container ID, not from the flag.
- **Protocol 4.** The master routes a job for an agent's own service, and any job for a
service with the `AGENT_IMAGE` repo (an older agent does not report itself), only to an
agent speaking protocol 4 (`agent too old to update itself`). The helper runs on the
default network, so `DOCKER_HOST` must be a `docker.sock` bind mount.
## Registry polling ## Registry polling
+50 -6
View File
@@ -28,6 +28,8 @@ const (
labelConfigFiles = "com.docker.compose.project.config_files" labelConfigFiles = "com.docker.compose.project.config_files"
labelEnvFile = "com.docker.compose.project.environment_file" labelEnvFile = "com.docker.compose.project.environment_file"
labelSwarmTask = "com.docker.swarm.service.id" labelSwarmTask = "com.docker.swarm.service.id"
labelSwarmName = "com.docker.swarm.service.name"
labelSelfUpdateJob = "se.shcizo.package-updater.self-update.job"
) )
// DockerClient is the subset of the Docker SDK the collector needs. // DockerClient is the subset of the Docker SDK the collector needs.
@@ -61,6 +63,7 @@ type Collector struct {
now func() time.Time now func() time.Time
mu sync.Mutex mu sync.Mutex
self *selfInfo // the agent's own container, once inspected
cache map[string]buildEntry // by Compose project cache map[string]buildEntry // by Compose project
warned map[string]bool // logged warnings: "<project>" (check failed) and "<project>/<service>" (service not in config) warned map[string]bool // logged warnings: "<project>" (check failed) and "<project>/<service>" (service not in config)
} }
@@ -77,8 +80,15 @@ type buildEntry struct {
// stackLabels are the Compose labels a build check needs. // stackLabels are the Compose labels a build check needs.
type stackLabels struct{ workingDir, configFiles, envFile string } type stackLabels struct{ workingDir, configFiles, envFile string }
// New returns a Collector. selfID is a prefix of the agent's own container // selfInfo is what the inspect of the agent's own container tells.
// ID (inside a container, the hostname); that container is marked Self. type selfInfo struct {
id string // full container ID
swarmService string // name of the Swarm service the container is a task of, if any
}
// New returns a Collector. selfID is the agent's own container: a prefix of
// its ID (inside a container, the hostname) or its name. That container is
// marked Self, and so is the Swarm service it is a task of.
func New(cli DockerClient, selfID string) *Collector { func New(cli DockerClient, selfID string) *Collector {
return &Collector{cli: cli, selfID: selfID, now: time.Now, cache: map[string]buildEntry{}, warned: map[string]bool{}} return &Collector{cli: cli, selfID: selfID, now: time.Now, cache: map[string]buildEntry{}, warned: map[string]bool{}}
} }
@@ -92,6 +102,29 @@ func (c *Collector) WithBuildCheck(check BuildCheck, log *slog.Logger) *Collecto
return c return c
} }
// resolveSelf inspects the agent's own container once and remembers the
// answer. It returns nil while that fails (or selfID is empty), and the
// caller then compares container IDs with the selfID prefix.
func (c *Collector) resolveSelf(ctx context.Context) *selfInfo {
if c.selfID == "" {
return nil
}
c.mu.Lock()
defer c.mu.Unlock()
if c.self != nil {
return c.self
}
insp, err := c.cli.ContainerInspect(ctx, c.selfID)
if err != nil || insp.ContainerJSONBase == nil || insp.ID == "" {
return nil
}
c.self = &selfInfo{id: insp.ID}
if insp.Config != nil {
c.self.swarmService = insp.Config.Labels[labelSwarmName]
}
return c.self
}
// Hello returns the machine facts sent on connect. AgentVersion is left for // Hello returns the machine facts sent on connect. AgentVersion is left for
// the caller to fill in. // the caller to fill in.
func (c *Collector) Hello(ctx context.Context) (protocol.Hello, error) { func (c *Collector) Hello(ctx context.Context) (protocol.Hello, error) {
@@ -125,20 +158,21 @@ func (c *Collector) Snapshot(ctx context.Context) ([]protocol.Service, error) {
repoDigests[img.ID] = img.RepoDigests repoDigests[img.ID] = img.RepoDigests
} }
self := c.resolveSelf(ctx)
byKey := make(map[string]protocol.Service) byKey := make(map[string]protocol.Service)
stacks := make(map[string]stackLabels) stacks := make(map[string]stackLabels)
for _, ct := range containers { for _, ct := range containers {
if _, isTask := ct.Labels[labelSwarmTask]; isTask { if _, isTask := ct.Labels[labelSwarmTask]; isTask {
continue continue
} }
if _, isHelper := ct.Labels[protocol.LabelSelfUpdateJob]; isHelper { if _, isHelper := ct.Labels[labelSelfUpdateJob]; isHelper {
continue continue
} }
ref, err := c.imageRef(ctx, ct) ref, err := c.imageRef(ctx, ct)
if err != nil { if err != nil {
return nil, err return nil, err
} }
svc := c.fromContainer(ct, ref, repoDigests[ct.ImageID]) svc := c.fromContainer(ct, ref, repoDigests[ct.ImageID], self)
if svc.Kind == protocol.KindCompose { if svc.Kind == protocol.KindCompose {
if _, ok := stacks[svc.ComposeProject]; !ok { if _, ok := stacks[svc.ComposeProject]; !ok {
stacks[svc.ComposeProject] = stackLabels{svc.WorkingDir, svc.ConfigFiles, ct.Labels[labelEnvFile]} stacks[svc.ComposeProject] = stackLabels{svc.WorkingDir, svc.ConfigFiles, ct.Labels[labelEnvFile]}
@@ -162,6 +196,7 @@ func (c *Collector) Snapshot(ctx context.Context) ([]protocol.Service, error) {
continue continue
} }
svc := fromSwarmService(s) svc := fromSwarmService(s)
svc.Self = self != nil && self.swarmService != "" && svc.SwarmService == self.swarmService
byKey[svc.Key()] = svc byKey[svc.Key()] = svc
} }
} }
@@ -192,12 +227,12 @@ func (c *Collector) imageRef(ctx context.Context, ct container.Summary) (string,
return insp.Config.Image, nil return insp.Config.Image, nil
} }
func (c *Collector) fromContainer(ct container.Summary, ref string, repoDigests []string) protocol.Service { func (c *Collector) fromContainer(ct container.Summary, ref string, repoDigests []string, self *selfInfo) protocol.Service {
svc := protocol.Service{ svc := protocol.Service{
Image: ref, Image: ref,
Digest: digestFor(ref, repoDigests), Digest: digestFor(ref, repoDigests),
Status: containerStatus(ct.State), Status: containerStatus(ct.State),
Self: c.selfID != "" && strings.HasPrefix(ct.ID, c.selfID), Self: c.isSelf(ct.ID, self),
} }
project, service := ct.Labels[labelProject], ct.Labels[labelService] project, service := ct.Labels[labelProject], ct.Labels[labelService]
if project != "" && service != "" { if project != "" && service != "" {
@@ -216,6 +251,15 @@ func (c *Collector) fromContainer(ct container.Summary, ref string, repoDigests
return svc return svc
} }
// isSelf reports whether id is the agent's own container: by the inspected
// ID when there is one, else by the selfID prefix.
func (c *Collector) isSelf(id string, self *selfInfo) bool {
if self != nil {
return id == self.id
}
return c.selfID != "" && strings.HasPrefix(id, c.selfID)
}
func fromSwarmService(s swarm.Service) protocol.Service { func fromSwarmService(s swarm.Service) protocol.Service {
ref := s.Spec.TaskTemplate.ContainerSpec.Image ref := s.Spec.TaskTemplate.ContainerSpec.Image
return protocol.Service{ return protocol.Service{
+58 -1
View File
@@ -138,7 +138,7 @@ func TestSnapshot_PlainContainerAndSwarmTaskSkipped(t *testing.T) {
func TestSnapshot_SelfUpdateHelperSkipped(t *testing.T) { func TestSnapshot_SelfUpdateHelperSkipped(t *testing.T) {
cli := &fakeDocker{containers: []container.Summary{ cli := &fakeDocker{containers: []container.Summary{
{ID: "h1", Names: []string{"/quirky_helper"}, Image: "sha256:agentimage", State: "running", {ID: "h1", Names: []string{"/quirky_helper"}, Image: "sha256:agentimage", State: "running",
Labels: map[string]string{protocol.LabelSelfUpdateJob: "7"}}, Labels: map[string]string{labelSelfUpdateJob: "7"}},
{ID: "p1", Names: []string{"/adhoc"}, Image: "redis:alpine", State: "exited"}, {ID: "p1", Names: []string{"/adhoc"}, Image: "redis:alpine", State: "exited"},
}} }}
got, err := New(cli, "").Snapshot(context.Background()) got, err := New(cli, "").Snapshot(context.Background())
@@ -158,6 +158,63 @@ func TestSnapshot_SelfFlag(t *testing.T) {
assert.True(t, got[0].Self) assert.True(t, got[0].Self)
} }
func TestSnapshot_SelfFlagByContainerName(t *testing.T) {
cli := &fakeDocker{
containers: []container.Summary{
composeContainer("0123456789abcdef", "updater-agent", "agent", "pu:1", "sha256:i", "running"),
composeContainer("fedcba9876543210", "slusk", "web", "slusk:1", "sha256:j", "running"),
},
inspect: map[string]container.InspectResponse{
"updater-agent-agent-1": {ContainerJSONBase: &container.ContainerJSONBase{ID: "0123456789abcdef"}},
},
}
got, err := New(cli, "updater-agent-agent-1").Snapshot(context.Background())
require.NoError(t, err)
require.Len(t, got, 2)
assert.False(t, got[0].Self, "slusk")
assert.True(t, got[1].Self, "updater-agent")
}
func TestSnapshot_SelfFlagFallsBackToIDPrefixWhenInspectFails(t *testing.T) {
cli := &fakeDocker{
containers: []container.Summary{composeContainer("0123456789abcdef", "updater-agent", "agent", "pu:1", "sha256:i", "running")},
inspectErr: errors.New("boom"),
}
got, err := New(cli, "0123456789ab").Snapshot(context.Background())
require.NoError(t, err)
require.Len(t, got, 1)
assert.True(t, got[0].Self)
}
func TestSnapshot_AgentsOwnSwarmServiceIsSelf(t *testing.T) {
swarmSvc := func(name string) swarm.Service {
return swarm.Service{
Spec: swarm.ServiceSpec{
Annotations: swarm.Annotations{Name: name},
TaskTemplate: swarm.TaskSpec{ContainerSpec: &swarm.ContainerSpec{Image: "pu:1"}},
},
ServiceStatus: &swarm.ServiceStatus{RunningTasks: 1, DesiredTasks: 1},
}
}
cli := &fakeDocker{
info: managerInfo(),
services: []swarm.Service{swarmSvc("pu_agent"), swarmSvc("pu_master")},
inspect: map[string]container.InspectResponse{
"abc123": {
ContainerJSONBase: &container.ContainerJSONBase{ID: "abc123full"},
Config: &container.Config{Labels: map[string]string{labelSwarmName: "pu_agent"}},
},
},
}
got, err := New(cli, "abc123").Snapshot(context.Background())
require.NoError(t, err)
require.Len(t, got, 2)
assert.Equal(t, "pu_agent", got[0].SwarmService)
assert.True(t, got[0].Self)
assert.Equal(t, "pu_master", got[1].SwarmService)
assert.False(t, got[1].Self)
}
func TestSnapshot_SwarmServicesOnlyOnManager(t *testing.T) { func TestSnapshot_SwarmServicesOnlyOnManager(t *testing.T) {
svc := swarm.Service{ svc := swarm.Service{
Spec: swarm.ServiceSpec{ Spec: swarm.ServiceSpec{
+33 -20
View File
@@ -6,6 +6,7 @@ import (
"io" "io"
"os" "os"
"os/exec" "os/exec"
"slices"
"strings" "strings"
"time" "time"
@@ -98,10 +99,6 @@ func NewComposeExecutor(cli ComposeDocker, resolve *Resolver, selfID string, run
// When the service is the agent's own container, the recreate runs in a // When the service is the agent's own container, the recreate runs in a
// helper container instead (see selfUpdate), because it stops this process. // helper container instead (see selfUpdate), because it stops this process.
func (e *ComposeExecutor) Execute(ctx context.Context, job Job, out io.Writer) (string, error) { func (e *ComposeExecutor) Execute(ctx context.Context, job Job, out io.Writer) (string, error) {
// What the job has printed so far goes into the helper's labels.
soFar := &tailBuffer{max: selfUpdateOutputMax}
out = io.MultiWriter(out, tailWriter{soFar})
project, service := job.ComposeProject, job.ComposeService project, service := job.ComposeProject, job.ComposeService
cts, err := e.cli.ContainerList(ctx, container.ListOptions{ cts, err := e.cli.ContainerList(ctx, container.ListOptions{
All: true, All: true,
@@ -116,14 +113,23 @@ func (e *ComposeExecutor) Execute(ctx context.Context, job Job, out io.Writer) (
if len(cts) == 0 { if len(cts) == 0 {
return "", fmt.Errorf("no container for compose service %s/%s on this machine", project, service) return "", fmt.Errorf("no container for compose service %s/%s on this machine", project, service)
} }
workingDir := cts[0].Labels[labelWorkingDir]
configFiles := cts[0].Labels[labelConfigFiles]
envFileLabel := cts[0].Labels[labelEnvFile]
self, err := inspectSelf(ctx, e.cli, e.selfID) self, err := inspectSelf(ctx, e.cli, e.selfID)
if err != nil { if err != nil {
return "", err return "", err
} }
isSelf := self.ContainerJSONBase != nil && self.ID != "" &&
slices.ContainsFunc(cts, func(ct container.Summary) bool { return ct.ID == self.ID })
// What an update of the agent's own service has printed so far goes into
// the helper's labels.
var soFar *tailBuffer
if isSelf {
soFar = &tailBuffer{max: selfUpdateOutputMax}
out = io.MultiWriter(out, tailWriter{soFar})
}
workingDir := cts[0].Labels[labelWorkingDir]
configFiles := cts[0].Labels[labelConfigFiles]
envFileLabel := cts[0].Labels[labelEnvFile]
roots := rootsOf(self) roots := rootsOf(self)
files, dir, err := composeFiles(roots, configFiles) files, dir, err := composeFiles(roots, configFiles)
if err != nil { if err != nil {
@@ -170,22 +176,29 @@ func (e *ComposeExecutor) Execute(ctx context.Context, job Job, out io.Writer) (
} }
base := composeBase(project, workingDir, files, envs) base := composeBase(project, workingDir, files, envs)
isSelf := false run := func(args []string) error {
if self.ContainerJSONBase != nil && self.ID != "" {
for _, ct := range cts {
isSelf = isSelf || ct.ID == self.ID
}
}
for _, step := range [][]string{{"pull", service}, {"up", "-d", service}} {
args := append(append([]string(nil), base...), step...)
if isSelf && step[0] == "up" {
return e.selfUpdate(ctx, job, self, workingDir, args, pinned, out, soFar)
}
fmt.Fprintf(out, "$ docker %s\n", strings.Join(args, " ")) fmt.Fprintf(out, "$ docker %s\n", strings.Join(args, " "))
tail := &tailBuffer{max: errorOutputTail} tail := &tailBuffer{max: errorOutputTail}
if err := e.run(ctx, workingDir, io.MultiWriter(out, tailWriter{tail}), args...); err != nil { if err := e.run(ctx, workingDir, io.MultiWriter(out, tailWriter{tail}), args...); err != nil {
return "", fmt.Errorf("docker compose %s: %w (output: %s)", step[0], err, strings.TrimSpace(tail.String())) return composeError(args[len(base)], err, tail)
} }
return nil
}
if err := run(append(append([]string(nil), base...), "pull", service)); err != nil {
return "", err
}
up := append(append([]string(nil), base...), "up", "-d", service)
if isSelf {
return e.selfUpdate(ctx, job, self, workingDir, up, pinned, out, soFar)
}
if err := run(up); err != nil {
return "", err
} }
return pinned, nil return pinned, nil
} }
// composeError describes a failed `docker compose <step>` call with the tail
// of its output.
func composeError(step string, err error, tail *tailBuffer) error {
return fmt.Errorf("docker compose %s: %w (output: %s)", step, err, strings.TrimSpace(tail.String()))
}
+24 -7
View File
@@ -64,6 +64,7 @@ type fakeHelper struct {
hold bool // ContainerWait never reports an exit hold bool // ContainerWait never reports an exit
logs string // what the helper printed logs string // what the helper printed
onStart func() onStart func()
state container.ContainerState // what an inspect reports; exited when empty
created []createdHelper created []createdHelper
started []string started []string
@@ -79,6 +80,14 @@ func (h *fakeHelper) ContainerCreate(_ context.Context, cfg *container.Config, h
return container.CreateResponse{ID: "helper1"}, nil return container.CreateResponse{ID: "helper1"}, nil
} }
func (h *fakeHelper) ContainerInspect(context.Context, string) (container.InspectResponse, error) {
state := h.state
if state == "" {
state = container.StateExited
}
return container.InspectResponse{ContainerJSONBase: &container.ContainerJSONBase{State: &container.State{Status: state}}}, nil
}
func (h *fakeHelper) ContainerStart(_ context.Context, id string, _ container.StartOptions) error { func (h *fakeHelper) ContainerStart(_ context.Context, id string, _ container.StartOptions) error {
h.started = append(h.started, id) h.started = append(h.started, id)
if h.onStart != nil { if h.onStart != nil {
@@ -116,6 +125,14 @@ type fakeHelperCompose struct {
h *fakeHelper h *fakeHelper
} }
// ContainerInspect answers for the agent's own container ("me") like fakeCompose
// and for any other ID like the helper.
func (f fakeHelperCompose) ContainerInspect(ctx context.Context, id string) (container.InspectResponse, error) {
if id == "me" {
return f.fakeCompose.ContainerInspect(ctx, id)
}
return f.h.ContainerInspect(ctx, id)
}
func (f fakeHelperCompose) ContainerCreate(ctx context.Context, cfg *container.Config, host *container.HostConfig, n *network.NetworkingConfig, p *ocispec.Platform, name string) (container.CreateResponse, error) { func (f fakeHelperCompose) ContainerCreate(ctx context.Context, cfg *container.Config, host *container.HostConfig, n *network.NetworkingConfig, p *ocispec.Platform, name string) (container.CreateResponse, error) {
return f.h.ContainerCreate(ctx, cfg, host, n, p, name) return f.h.ContainerCreate(ctx, cfg, host, n, p, name)
} }
@@ -528,13 +545,13 @@ func TestComposeSelfUpdate(t *testing.T) {
assert.Equal(t, container.WaitConditionNextExit, c.helper.waitCond) assert.Equal(t, container.WaitConditionNextExit, c.helper.waitCond)
labels := cfg.Labels labels := cfg.Labels
assert.Len(t, labels, 6, "no compose labels") assert.Len(t, labels, 6, "no compose labels")
assert.Equal(t, "7", labels[protocol.LabelSelfUpdateJob]) assert.Equal(t, "7", labels[labelSelfUpdateJob])
assert.Equal(t, pinned, labels[protocol.LabelSelfUpdateRef]) assert.Equal(t, pinned, labels[labelSelfUpdateRef])
assert.Equal(t, "2026-10-13T09:00:00Z", labels[protocol.LabelSelfUpdateStarted]) assert.Equal(t, "2026-10-13T09:00:00Z", labels[labelSelfUpdateStarted])
assert.Equal(t, "updater", labels[protocol.LabelSelfUpdateProject]) assert.Equal(t, "updater", labels[labelSelfUpdateProject])
assert.Equal(t, "agent", labels[protocol.LabelSelfUpdateService]) assert.Equal(t, "agent", labels[labelSelfUpdateService])
assert.Contains(t, labels[protocol.LabelSelfUpdateOutput], "resolved "+pinned) assert.Contains(t, labels[labelSelfUpdateOutput], "resolved "+pinned)
assert.Contains(t, labels[protocol.LabelSelfUpdateOutput], "$ docker "+strings.Join(withStep("pull", "agent"), " ")) assert.Contains(t, labels[labelSelfUpdateOutput], "$ docker "+strings.Join(withStep("pull", "agent"), " "))
}) })
} }
+1 -1
View File
@@ -260,7 +260,7 @@ func TestRunner_SelfJobsRunLast(t *testing.T) {
assert.Equal(t, []string{"a:2", "a:4", "self:1", "self:3"}, order) assert.Equal(t, []string{"a:2", "a:4", "self:1", "self:3"}, order)
} }
func TestRunner_SelfJobRunsWhenAloneAndNonSelfSubmittedLater(t *testing.T) { func TestRunner_SelfJobRunsWhenAlone(t *testing.T) {
fe := newFakeExec() fe := newFakeExec()
r := startRunner(t, fe, time.Minute) r := startRunner(t, fe, time.Minute)
r.Submit("1", protocol.JobStart{Ref: "self:1", Self: true}) r.Submit("1", protocol.JobStart{Ref: "self:1", Self: true})
+36 -15
View File
@@ -15,8 +15,17 @@ import (
"github.com/docker/docker/api/types/network" "github.com/docker/docker/api/types/network"
"github.com/docker/docker/pkg/stdcopy" "github.com/docker/docker/pkg/stdcopy"
ocispec "github.com/opencontainers/image-spec/specs-go/v1" ocispec "github.com/opencontainers/image-spec/specs-go/v1"
)
"github.com/shcizo/package-updater/internal/protocol" // Labels on a self-update helper container. The new agent finds the helper
// by them, and the inventory skips containers that carry the job label.
const (
labelSelfUpdateJob = "se.shcizo.package-updater.self-update.job"
labelSelfUpdateRef = "se.shcizo.package-updater.self-update.ref"
labelSelfUpdateStarted = "se.shcizo.package-updater.self-update.started"
labelSelfUpdateProject = "se.shcizo.package-updater.self-update.project"
labelSelfUpdateService = "se.shcizo.package-updater.self-update.service"
labelSelfUpdateOutput = "se.shcizo.package-updater.self-update.output"
) )
// selfUpdateOutputMax caps the job output stored in the helper container's // selfUpdateOutputMax caps the job output stored in the helper container's
@@ -32,6 +41,7 @@ const resumeSeparator = "--- agent restarted; output of the update helper ---\n"
// HelperDocker is the subset of the Docker SDK the self-update helper // HelperDocker is the subset of the Docker SDK the self-update helper
// container needs. *client.Client satisfies it. // container needs. *client.Client satisfies it.
type HelperDocker interface { type HelperDocker interface {
ContainerInspect(ctx context.Context, containerID string) (container.InspectResponse, error)
ContainerCreate(ctx context.Context, config *container.Config, hostConfig *container.HostConfig, networkingConfig *network.NetworkingConfig, platform *ocispec.Platform, containerName string) (container.CreateResponse, error) ContainerCreate(ctx context.Context, config *container.Config, hostConfig *container.HostConfig, networkingConfig *network.NetworkingConfig, platform *ocispec.Platform, containerName string) (container.CreateResponse, error)
ContainerStart(ctx context.Context, containerID string, options container.StartOptions) error ContainerStart(ctx context.Context, containerID string, options container.StartOptions) error
ContainerWait(ctx context.Context, containerID string, condition container.WaitCondition) (<-chan container.WaitResponse, <-chan error) ContainerWait(ctx context.Context, containerID string, condition container.WaitCondition) (<-chan container.WaitResponse, <-chan error)
@@ -57,12 +67,12 @@ func (e *ComposeExecutor) selfUpdate(ctx context.Context, job Job, self containe
WorkingDir: workingDir, WorkingDir: workingDir,
Env: dockerEnv(), Env: dockerEnv(),
Labels: map[string]string{ Labels: map[string]string{
protocol.LabelSelfUpdateJob: job.ID, labelSelfUpdateJob: job.ID,
protocol.LabelSelfUpdateRef: pinned, labelSelfUpdateRef: pinned,
protocol.LabelSelfUpdateStarted: job.Started.UTC().Format(time.RFC3339Nano), labelSelfUpdateStarted: job.Started.UTC().Format(time.RFC3339Nano),
protocol.LabelSelfUpdateProject: job.ComposeProject, labelSelfUpdateProject: job.ComposeProject,
protocol.LabelSelfUpdateService: job.ComposeService, labelSelfUpdateService: job.ComposeService,
protocol.LabelSelfUpdateOutput: outputLabel(soFar), labelSelfUpdateOutput: outputLabel(soFar),
}, },
} }
created, err := e.cli.ContainerCreate(ctx, cfg, &container.HostConfig{Mounts: helperMounts(self)}, nil, nil, "") created, err := e.cli.ContainerCreate(ctx, cfg, &container.HostConfig{Mounts: helperMounts(self)}, nil, nil, "")
@@ -72,7 +82,9 @@ func (e *ComposeExecutor) selfUpdate(ctx context.Context, job Job, self containe
// Wait is registered before the start so a fast exit is not missed. // Wait is registered before the start so a fast exit is not missed.
waitCh, errCh := e.cli.ContainerWait(ctx, created.ID, container.WaitConditionNextExit) waitCh, errCh := e.cli.ContainerWait(ctx, created.ID, container.WaitConditionNextExit)
if err := e.cli.ContainerStart(ctx, created.ID, container.StartOptions{}); err != nil { if err := e.cli.ContainerStart(ctx, created.ID, container.StartOptions{}); err != nil {
e.cli.ContainerRemove(context.WithoutCancel(ctx), created.ID, container.RemoveOptions{Force: true}) if rmErr := e.cli.ContainerRemove(context.WithoutCancel(ctx), created.ID, container.RemoveOptions{Force: true}); rmErr != nil {
fmt.Fprintf(out, "warning: remove the update helper: %v\n", rmErr)
}
return "", fmt.Errorf("start update helper: %w", err) return "", fmt.Errorf("start update helper: %w", err)
} }
code, err := awaitExit(ctx, waitCh, errCh) code, err := awaitExit(ctx, waitCh, errCh)
@@ -154,7 +166,7 @@ func finishHelper(ctx context.Context, cli HelperDocker, id string, w io.Writer)
// exitError describes a failed helper like a failed `docker compose` call. // exitError describes a failed helper like a failed `docker compose` call.
func exitError(code int64, tail *tailBuffer) error { func exitError(code int64, tail *tailBuffer) error {
return fmt.Errorf("docker compose up: exit status %d (output: %s)", code, strings.TrimSpace(tail.String())) return composeError("up", fmt.Errorf("exit status %d", code), tail)
} }
// ResumeDocker is the subset of the Docker SDK ResumeSelfUpdates needs. // ResumeDocker is the subset of the Docker SDK ResumeSelfUpdates needs.
@@ -186,16 +198,16 @@ func ResumeSelfUpdates(ctx context.Context, cli ResumeDocker, selfID string, run
} }
cts, err := cli.ContainerList(ctx, container.ListOptions{ cts, err := cli.ContainerList(ctx, container.ListOptions{
All: true, All: true,
Filters: filters.NewArgs(filters.Arg("label", protocol.LabelSelfUpdateJob)), Filters: filters.NewArgs(filters.Arg("label", labelSelfUpdateJob)),
}) })
if err != nil { if err != nil {
return fmt.Errorf("list update helpers: %w", err) return fmt.Errorf("list update helpers: %w", err)
} }
for _, ct := range cts { for _, ct := range cts {
if ct.Labels[protocol.LabelSelfUpdateProject] != project || ct.Labels[protocol.LabelSelfUpdateService] != service { if ct.Labels[labelSelfUpdateProject] != project || ct.Labels[labelSelfUpdateService] != service {
continue continue
} }
jobID := ct.Labels[protocol.LabelSelfUpdateJob] jobID := ct.Labels[labelSelfUpdateJob]
if jobID == "" { if jobID == "" {
log.Warn("self_update_helper_without_job", "helper", ct.ID) log.Warn("self_update_helper_without_job", "helper", ct.ID)
continue continue
@@ -205,7 +217,7 @@ func ResumeSelfUpdates(ctx context.Context, cli ResumeDocker, selfID string, run
started = time.Now() started = time.Now()
} }
log.Info("self_update_resumed", "job_id", jobID, "helper", ct.ID) log.Info("self_update_resumed", "job_id", jobID, "helper", ct.ID)
runner.Resume(jobID, started, resumeFunc(cli, ct.ID, ct.Labels[protocol.LabelSelfUpdateOutput], ref, labelErr)) runner.Resume(jobID, started, resumeFunc(cli, ct.ID, ct.Labels[labelSelfUpdateOutput], ref, labelErr))
} }
return nil return nil
} }
@@ -214,11 +226,11 @@ func ResumeSelfUpdates(ctx context.Context, cli ResumeDocker, selfID string, run
// labels. started is zero when it is missing or unreadable, and err then says // labels. started is zero when it is missing or unreadable, and err then says
// what is wrong. // what is wrong.
func parseHelperLabels(labels map[string]string) (ref string, started time.Time, err error) { func parseHelperLabels(labels map[string]string) (ref string, started time.Time, err error) {
ref = labels[protocol.LabelSelfUpdateRef] ref = labels[labelSelfUpdateRef]
if ref == "" { if ref == "" {
err = errors.New("update helper has no ref label") err = errors.New("update helper has no ref label")
} }
raw := labels[protocol.LabelSelfUpdateStarted] raw := labels[labelSelfUpdateStarted]
if t, perr := time.Parse(time.RFC3339Nano, raw); perr == nil { if t, perr := time.Parse(time.RFC3339Nano, raw); perr == nil {
started = t started = t
} else if err == nil { } else if err == nil {
@@ -233,6 +245,15 @@ func parseHelperLabels(labels map[string]string) (ref string, started time.Time,
// failed or its labels were unusable. // failed or its labels were unusable.
func resumeFunc(cli HelperDocker, helperID, earlier, ref string, labelErr error) ResumeFunc { func resumeFunc(cli HelperDocker, helperID, earlier, ref string, labelErr error) ResumeFunc {
return func(ctx context.Context, out io.Writer) (string, error) { return func(ctx context.Context, out io.Writer) (string, error) {
// A helper that was created but never started would read as exited
// with code 0, so it must not be mistaken for a finished update.
if insp, err := cli.ContainerInspect(ctx, helperID); err == nil && insp.ContainerJSONBase != nil &&
insp.State != nil && insp.State.Status == container.StateCreated {
if err := cli.ContainerRemove(context.WithoutCancel(ctx), helperID, container.RemoveOptions{Force: true}); err != nil {
fmt.Fprintf(out, "warning: remove the update helper: %v\n", err)
}
return "", errors.New("update helper was never started")
}
waitCh, errCh := cli.ContainerWait(ctx, helperID, container.WaitConditionNotRunning) waitCh, errCh := cli.ContainerWait(ctx, helperID, container.WaitConditionNotRunning)
code, err := awaitExit(ctx, waitCh, errCh) code, err := awaitExit(ctx, waitCh, errCh)
if err != nil { if err != nil {
+18 -11
View File
@@ -22,12 +22,12 @@ func TestResumeSelfUpdates(t *testing.T) {
helper := func(id, project, service string, extra map[string]string) container.Summary { helper := func(id, project, service string, extra map[string]string) container.Summary {
labels := map[string]string{ labels := map[string]string{
protocol.LabelSelfUpdateJob: "j-" + id, labelSelfUpdateJob: "j-" + id,
protocol.LabelSelfUpdateRef: pinned, labelSelfUpdateRef: pinned,
protocol.LabelSelfUpdateStarted: started.Format(time.RFC3339Nano), labelSelfUpdateStarted: started.Format(time.RFC3339Nano),
protocol.LabelSelfUpdateProject: project, labelSelfUpdateProject: project,
protocol.LabelSelfUpdateService: service, labelSelfUpdateService: service,
protocol.LabelSelfUpdateOutput: "resolved " + pinned + "\n$ docker compose pull agent\n", labelSelfUpdateOutput: "resolved " + pinned + "\n$ docker compose pull agent\n",
} }
for k, v := range extra { for k, v := range extra {
if v == "" { if v == "" {
@@ -44,6 +44,8 @@ func TestResumeSelfUpdates(t *testing.T) {
name string name string
helper container.Summary helper container.Summary
exit int64 exit int64
state container.ContainerState
noStarted bool // the old agent's start time is lost with the label
wantStatus string wantStatus string
wantErr string wantErr string
wantOutput string wantOutput string
@@ -52,16 +54,19 @@ func TestResumeSelfUpdates(t *testing.T) {
wantOutput: "resolved " + pinned + "\n$ docker compose pull agent\n" + separator + "Recreated\n"}, wantOutput: "resolved " + pinned + "\n$ docker compose pull agent\n" + separator + "Recreated\n"},
{name: "helper failed", helper: helper("h1", "updater", "agent", nil), exit: 1, wantStatus: protocol.ResultFailed, {name: "helper failed", helper: helper("h1", "updater", "agent", nil), exit: 1, wantStatus: protocol.ResultFailed,
wantErr: "docker compose up: exit status 1 (output: Recreated)"}, wantErr: "docker compose up: exit status 1 (output: Recreated)"},
{name: "no ref label", helper: helper("h1", "updater", "agent", map[string]string{protocol.LabelSelfUpdateRef: ""}), {name: "no ref label", helper: helper("h1", "updater", "agent", map[string]string{labelSelfUpdateRef: ""}),
wantStatus: protocol.ResultFailed, wantErr: "update helper has no ref label"}, wantStatus: protocol.ResultFailed, wantErr: "update helper has no ref label"},
{name: "no started label", helper: helper("h1", "updater", "agent", map[string]string{protocol.LabelSelfUpdateStarted: ""}), {name: "no started label", helper: helper("h1", "updater", "agent", map[string]string{labelSelfUpdateStarted: ""}),
noStarted: true,
wantStatus: protocol.ResultFailed, wantErr: `update helper has no usable started label ""`}, wantStatus: protocol.ResultFailed, wantErr: `update helper has no usable started label ""`},
{name: "no output label", helper: helper("h1", "updater", "agent", map[string]string{protocol.LabelSelfUpdateOutput: ""}), {name: "helper never started", helper: helper("h1", "updater", "agent", nil), state: container.StateCreated,
wantStatus: protocol.ResultFailed, wantErr: "update helper was never started"},
{name: "no output label", helper: helper("h1", "updater", "agent", map[string]string{labelSelfUpdateOutput: ""}),
wantStatus: protocol.ResultSucceeded, wantOutput: separator + "Recreated\n"}, wantStatus: protocol.ResultSucceeded, wantOutput: separator + "Recreated\n"},
} }
for _, c := range cases { for _, c := range cases {
t.Run(c.name, func(t *testing.T) { t.Run(c.name, func(t *testing.T) {
h := &fakeHelper{exit: c.exit, logs: "Recreated\n"} h := &fakeHelper{exit: c.exit, state: c.state, logs: "Recreated\n"}
unrelated := []container.Summary{ unrelated := []container.Summary{
helper("other-project", "other", "agent", nil), helper("other-project", "other", "agent", nil),
helper("other-service", "updater", "web", nil), helper("other-service", "updater", "web", nil),
@@ -86,7 +91,9 @@ func TestResumeSelfUpdates(t *testing.T) {
assert.Equal(t, "j-h1", got[0].ID) assert.Equal(t, "j-h1", got[0].ID)
res := got[0].Result res := got[0].Result
assert.Equal(t, c.wantStatus, res.Status) assert.Equal(t, c.wantStatus, res.Status)
if c.state != container.StateCreated {
assert.Equal(t, container.WaitConditionNotRunning, h.waitCond) assert.Equal(t, container.WaitConditionNotRunning, h.waitCond)
}
assert.Equal(t, []string{"h1"}, h.removed, "only the matching helper is touched") assert.Equal(t, []string{"h1"}, h.removed, "only the matching helper is touched")
if c.wantStatus == protocol.ResultSucceeded { if c.wantStatus == protocol.ResultSucceeded {
assert.Equal(t, pinned, res.Ref) assert.Equal(t, pinned, res.Ref)
@@ -94,7 +101,7 @@ func TestResumeSelfUpdates(t *testing.T) {
} else { } else {
assert.Equal(t, c.wantErr, res.Error) assert.Equal(t, c.wantErr, res.Error)
} }
if c.name != "no started label" { if !c.noStarted {
assert.True(t, started.Equal(res.StartedAt), "the old agent's start time is kept") assert.True(t, started.Equal(res.StartedAt), "the old agent's start time is kept")
} }
}) })
+2 -2
View File
@@ -119,7 +119,7 @@ func newJobEnv(t *testing.T) *jobEnv {
m := startMaster(t, store) m := startMaster(t, store)
exec := newGateExecutor() exec := newGateExecutor()
env := &jobEnv{store: store, m: m, exec: exec, env := &jobEnv{store: store, m: m, exec: exec,
jobs: jobs.New(store, m.hub, m.broker, slog.New(slog.NewTextHandler(io.Discard, nil))), jobs: jobs.New(store, m.hub, m.broker, "gitea.shcizo.se/shcizo/package-updater:latest", slog.New(slog.NewTextHandler(io.Discard, nil))),
agent: startJobAgent(t, m, "pua_ok", exec)} agent: startJobAgent(t, m, "pua_ok", exec)}
env.waitRoutable(t) env.waitRoutable(t)
return env return env
@@ -282,7 +282,7 @@ func TestPhaseOneAgentGetsNoJobs(t *testing.T) {
AgentID: ptr(agentID), ActualRef: slusk + ":1.4.2@sha256:old"} AgentID: ptr(agentID), ActualRef: slusk + ":1.4.2@sha256:old"}
store.seedService(svcRow) store.seedService(svcRow)
m := startMaster(t, store) m := startMaster(t, store)
js := jobs.New(store, m.hub, m.broker, slog.New(slog.NewTextHandler(io.Discard, nil))) js := jobs.New(store, m.hub, m.broker, "gitea.shcizo.se/shcizo/package-updater:latest", slog.New(slog.NewTextHandler(io.Discard, nil)))
// A raw client speaking the phase-1 protocol: its hello has no // A raw client speaking the phase-1 protocol: its hello has no
// protocol_version. // protocol_version.
+12 -4
View File
@@ -43,13 +43,21 @@ type Service struct {
store Store store Store
hub Hub hub Hub
pub Publisher pub Publisher
selfRepo string
log *slog.Logger log *slog.Logger
now func() time.Time now func() time.Time
} }
// New returns a Service. // New returns a Service. agentImage is the master's AGENT_IMAGE. A service
func New(store Store, hub Hub, pub Publisher, log *slog.Logger) *Service { // whose image repo equals its repo may be an agent that reports itself as
return &Service{store: store, hub: hub, pub: pub, log: log, now: time.Now} // another service (an agent older than protocol version 4 does not tell), so
// such a job is only sent to an agent that speaks protocol version
// protocol.MinSelfUpdateVersion.
func New(store Store, hub Hub, pub Publisher, agentImage string, log *slog.Logger) *Service {
return &Service{
store: store, hub: hub, pub: pub, log: log, now: time.Now,
selfRepo: imageref.Parse(agentImage).Repo,
}
} }
// target splits a service's actual ref into the repo and tag to deploy. A bare // target splits a service's actual ref into the repo and tag to deploy. A bare
@@ -245,7 +253,7 @@ func (s *Service) Start(ctx context.Context, svc model.Service, req StartRequest
if p := imageref.Parse(ref); p.Digest != "" || p.Tag == "" { if p := imageref.Parse(ref); p.Digest != "" || p.Tag == "" {
minProtocol = protocol.MinPinnedRefVersion minProtocol = protocol.MinPinnedRefVersion
} }
if svc.Self { if svc.Self || imageref.Parse(svc.ActualRef).Repo == s.selfRepo {
minProtocol = protocol.MinSelfUpdateVersion minProtocol = protocol.MinSelfUpdateVersion
} }
r, routeErr := s.hub.Route(svc, minProtocol) r, routeErr := s.hub.Route(svc, minProtocol)
+17 -1
View File
@@ -16,6 +16,8 @@ import (
"github.com/shcizo/package-updater/internal/protocol" "github.com/shcizo/package-updater/internal/protocol"
) )
const selfImage = "gitea.shcizo.se/shcizo/package-updater:latest"
type fakeStore struct { type fakeStore struct {
services map[int64]model.Service services map[int64]model.Service
jobs map[int64]model.Job jobs map[int64]model.Job
@@ -125,7 +127,7 @@ func ptr[T any](v T) *T { return &v }
func newService(st *fakeStore, h *fakeHub) (*Service, *fakePub) { func newService(st *fakeStore, h *fakeHub) (*Service, *fakePub) {
pub := &fakePub{} pub := &fakePub{}
return New(st, h, pub, slog.New(slog.NewTextHandler(io.Discard, nil))), pub return New(st, h, pub, selfImage, slog.New(slog.NewTextHandler(io.Discard, nil))), pub
} }
func TestUpdatable(t *testing.T) { func TestUpdatable(t *testing.T) {
@@ -417,6 +419,20 @@ func TestStart_SelfServiceNeedsASelfUpdatingAgent(t *testing.T) {
require.Len(t, h.starts, 1) require.Len(t, h.starts, 1)
assert.True(t, h.starts[0].start.Self) assert.True(t, h.starts[0].start.Self)
}) })
t.Run("a service with the agent image repo needs a self-updating agent but is not marked", func(t *testing.T) {
svc := composeSvc
svc.ActualRef = selfImage
st := &fakeStore{services: map[int64]model.Service{1: svc}, jobs: map[int64]model.Job{}}
h := &fakeHub{route: ok}
s, _ := newService(st, h)
_, err := s.Start(context.Background(), svc, StartRequest{Trigger: model.TriggerManual})
require.NoError(t, err)
assert.Equal(t, []int{protocol.MinSelfUpdateVersion}, h.mins)
require.Len(t, h.starts, 1)
assert.False(t, h.starts[0].start.Self)
})
t.Run("a service that is not the agent's own is not marked", func(t *testing.T) { t.Run("a service that is not the agent's own is not marked", func(t *testing.T) {
st := &fakeStore{services: map[int64]model.Service{1: composeSvc}, jobs: map[int64]model.Job{}} st := &fakeStore{services: map[int64]model.Service{1: composeSvc}, jobs: map[int64]model.Job{}}
h := &fakeHub{route: ok} h := &fakeHub{route: ok}
+2 -3
View File
@@ -49,7 +49,6 @@ func (e *NotUpdatableError) Error() string { return "not updatable: " + e.Reason
// Reasons a service cannot be updated. // Reasons a service cannot be updated.
const ( const (
ReasonSelf = "updated manually outside the system"
ReasonContainer = "plain containers cannot be updated" ReasonContainer = "plain containers cannot be updated"
ReasonNoTag = "follows no tag" ReasonNoTag = "follows no tag"
ReasonRemoved = "service no longer exists" ReasonRemoved = "service no longer exists"
@@ -70,14 +69,14 @@ type MergeRefusedError struct {
func (e *MergeRefusedError) Error() string { return "cannot merge: " + e.Reason } func (e *MergeRefusedError) Error() string { return "cannot merge: " + e.Reason }
// Reasons two services cannot be merged. ReasonSelf is reused for the // Reasons two services cannot be merged.
// updater itself.
const ( const (
ReasonMergeSame = "a service cannot be merged into itself" ReasonMergeSame = "a service cannot be merged into itself"
ReasonMergeLiveOld = "the old service still exists" ReasonMergeLiveOld = "the old service still exists"
ReasonMergeGoneNew = "the new service no longer exists" ReasonMergeGoneNew = "the new service no longer exists"
ReasonMergeKind = "only services of the same kind can be merged" ReasonMergeKind = "only services of the same kind can be merged"
ReasonMergeContainer = "plain containers cannot be merged" ReasonMergeContainer = "plain containers cannot be merged"
ReasonMergeSelf = "an agent's own service cannot be merged"
) )
// JobStatus is the state of an update job. // JobStatus is the state of an update job.
+16
View File
@@ -7,6 +7,7 @@ import (
"context" "context"
"errors" "errors"
"log/slog" "log/slog"
"slices"
"strings" "strings"
"sync" "sync"
"time" "time"
@@ -209,14 +210,29 @@ func (p *Poller) record(ctx context.Context, l lookup, svcs []model.Service, sta
} }
stats.ok++ stats.ok++
p.setPollError(ctx, svcs, "", stats, func(s model.Service) bool { return s.LastPollError != "" }) p.setPollError(ctx, svcs, "", stats, func(s model.Service) bool { return s.LastPollError != "" })
var auto []model.Service
for _, svc := range svcs { for _, svc := range svcs {
if !p.suggest(ctx, svc, l, stats) { if !p.suggest(ctx, svc, l, stats) {
continue continue
} }
if autoWanted(svc, l.ref, l.digest) { if autoWanted(svc, l.ref, l.digest) {
auto = append(auto, svc)
}
}
// An agent's own service goes last: jobs that reach an agent while it
// replaces itself are lost, and messages to one agent arrive in order.
slices.SortStableFunc(auto, func(a, b model.Service) int { return selfRank(a) - selfRank(b) })
for _, svc := range auto {
p.startAuto(ctx, svc.ID, l.ref, l.digest, stats) p.startAuto(ctx, svc.ID, l.ref, l.digest, stats)
} }
}
// selfRank sorts an agent's own service after all others.
func selfRank(svc model.Service) int {
if svc.Self {
return 1
} }
return 0
} }
// suggest marks l.digest as the available version of svc unless the service // suggest marks l.digest as the available version of svc unless the service
+18
View File
@@ -297,6 +297,24 @@ func TestNewDigest(t *testing.T) {
} }
} }
func TestAgentsOwnServiceIsStartedLast(t *testing.T) {
ref := repoA + ":1.4"
own := svc(1, ref+"@sha256:old")
own.UpdateMode, own.Self = model.ModeAuto, true
other := svc(2, ref+"@sha256:old")
other.UpdateMode = model.ModeAuto
st := &fakeStore{services: []model.Service{own, other}}
reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}}
j := &fakeJobs{}
p, _ := newPoller(st, j, reg, Config{})
p.Cycle(context.Background())
require.Len(t, j.starts, 2)
assert.Equal(t, int64(2), j.starts[0].svcID)
assert.Equal(t, int64(1), j.starts[1].svcID)
}
func TestSameSuggestionIsIdempotent(t *testing.T) { func TestSameSuggestionIsIdempotent(t *testing.T) {
ref := repoA + ":1.4" ref := repoA + ":1.4"
running := svc(1, ref+"@sha256:old") running := svc(1, ref+"@sha256:old")
@@ -418,11 +418,11 @@ func TestMergeService_Refusals(t *testing.T) {
{"the updater itself (new)", func(t *testing.T, s *Store, f moveFixture) error { {"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) mustExec(t, s, `UPDATE services SET self = true WHERE id = $1`, f.to)
return s.MergeService(context.Background(), f.from, f.to) return s.MergeService(context.Background(), f.from, f.to)
}, refused(model.ReasonSelf)}, }, refused(model.ReasonMergeSelf)},
{"the updater itself (old)", func(t *testing.T, s *Store, f moveFixture) error { {"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) mustExec(t, s, `UPDATE services SET self = true WHERE id = $1`, f.from)
return s.MergeService(context.Background(), f.from, f.to) return s.MergeService(context.Background(), f.from, f.to)
}, refused(model.ReasonSelf)}, }, refused(model.ReasonMergeSelf)},
{"active job on the old service", func(t *testing.T, s *Store, f moveFixture) error { {"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)) _, err := s.CreateJob(context.Background(), newJob(f.from, f.a))
require.NoError(t, err) require.NoError(t, err)
+1 -1
View File
@@ -243,7 +243,7 @@ func mergeRefusal(from, to mergeRow) string {
case from.kind == string(protocol.KindContainer): case from.kind == string(protocol.KindContainer):
return model.ReasonMergeContainer return model.ReasonMergeContainer
case from.self || to.self: case from.self || to.self:
return model.ReasonSelf return model.ReasonMergeSelf
} }
return "" return ""
} }
+2 -2
View File
@@ -273,8 +273,8 @@ func TestServiceDetail_DriftAndMoveData(t *testing.T) {
{name: "swarm service can merge, no suggestion", id: 3, wantSources: sources}, {name: "swarm service can merge, no suggestion", id: 3, wantSources: sources},
{name: "container cannot merge", id: 4}, {name: "container cannot merge", id: 4},
{name: "self cannot merge", id: 5}, {name: "self cannot merge", id: 5},
{name: "restore refusal is passed on", id: 1, restorable: &model.NotUpdatableError{Reason: model.ReasonSelf}, {name: "restore refusal is passed on", id: 1, restorable: &model.NotUpdatableError{Reason: model.ReasonContainer},
wantBlocked: model.ReasonSelf, wantFrom: &move, wantSources: sources}, wantBlocked: model.ReasonContainer, wantFrom: &move, wantSources: sources},
} }
for _, tt := range tests { for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) { t.Run(tt.name, func(t *testing.T) {
+1 -1
View File
@@ -539,7 +539,7 @@ func TestServicePage_ShowsUpdateButton(t *testing.T) {
} }
func TestServicePage_DisabledReason(t *testing.T) { func TestServicePage_DisabledReason(t *testing.T) {
for _, reason := range []string{model.ReasonSelf, model.ReasonContainer, model.ReasonNoTag, model.ReasonRemoved, model.ReasonBuilt} { for _, reason := range []string{model.ReasonContainer, model.ReasonNoTag, model.ReasonRemoved, model.ReasonBuilt} {
t.Run(reason, func(t *testing.T) { t.Run(reason, func(t *testing.T) {
srv, _, _, _, jobs := newTestEnv(t) srv, _, _, _, jobs := newTestEnv(t)
jobs.updatable = &model.NotUpdatableError{Reason: reason} jobs.updatable = &model.NotUpdatableError{Reason: reason}
+1 -1
View File
@@ -68,7 +68,7 @@ func TestServicePage_VersionPickerWarnsForAutoServices(t *testing.T) {
} }
func TestServicePage_VersionPickerHiddenWhenServiceCannotBeUpdated(t *testing.T) { func TestServicePage_VersionPickerHiddenWhenServiceCannotBeUpdated(t *testing.T) {
for _, reason := range []string{model.ReasonSelf, model.ReasonContainer, model.ReasonRemoved, model.ReasonBuilt} { for _, reason := range []string{model.ReasonContainer, model.ReasonRemoved, model.ReasonBuilt} {
t.Run(reason, func(t *testing.T) { t.Run(reason, func(t *testing.T) {
srv, store, _, _, jobs := newTestEnv(t) srv, store, _, _, jobs := newTestEnv(t)
store.services[0].ImageRepo = sluskRepo store.services[0].ImageRepo = sluskRepo
+12
View File
@@ -15,6 +15,7 @@ import (
"log/slog" "log/slog"
"net/http" "net/http"
"regexp" "regexp"
"slices"
"strconv" "strconv"
"strings" "strings"
"time" "time"
@@ -181,6 +182,9 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
resp := response{RequestID: requestID, Image: req.Image, Tag: req.Tag, Matched: len(services)} resp := response{RequestID: requestID, Image: req.Image, Tag: req.Tag, Matched: len(services)}
pending := map[int]int64{} // index into resp.Results -> job ID pending := map[int]int64{} // index into resp.Results -> job ID
// An agent's own service goes last: jobs that reach an agent while it
// replaces itself are lost, and messages to one agent arrive in order.
slices.SortStableFunc(services, func(a, b model.Service) int { return selfRank(a) - selfRank(b) })
for _, svc := range services { for _, svc := range services {
res, job := h.handle(work, svc, req.Tag, started, log) res, job := h.handle(work, svc, req.Tag, started, log)
resp.Results = append(resp.Results, res) resp.Results = append(resp.Results, res)
@@ -252,6 +256,14 @@ func (h *Handler) handle(ctx context.Context, svc model.Service, tag string, sta
return res, nil return res, nil
} }
// selfRank sorts an agent's own service after all others.
func selfRank(svc model.Service) int {
if svc.Self {
return 1
}
return 0
}
func (h *Handler) setJob(res *result, job model.Job) { func (h *Handler) setJob(res *result, job model.Job) {
res.JobID = job.ID res.JobID = job.ID
res.JobURL = h.cfg.PublicURL + "/jobs/" + strconv.FormatInt(job.ID, 10) res.JobURL = h.cfg.PublicURL + "/jobs/" + strconv.FormatInt(job.ID, 10)
+15
View File
@@ -500,6 +500,21 @@ func TestMixedAutoAndManual(t *testing.T) {
assert.Len(t, e.store.avail, 3) assert.Len(t, e.store.avail, 3)
} }
func TestAgentsOwnServiceIsStartedLast(t *testing.T) {
own := compose(1, model.ModeAuto, slusk+":1")
own.Self = true
e := newEnv(t, own, compose(2, model.ModeAuto, slusk+":1"))
e.jobs.jobFor[1] = model.Job{Status: model.JobSucceeded}
e.jobs.jobFor[2] = model.Job{Status: model.JobSucceeded}
e.post(`{"image":"` + slusk + `","tag":"2"}`)
calls := e.jobs.startCalls()
require.Len(t, calls, 2)
assert.Equal(t, int64(2), calls[0].svc.ID)
assert.Equal(t, int64(1), calls[1].svc.ID)
}
func TestClientCancelStopsWaitingButNotStarting(t *testing.T) { func TestClientCancelStopsWaitingButNotStarting(t *testing.T) {
e := newEnv(t, compose(1, model.ModeAuto, slusk+":1.4.2")) e := newEnv(t, compose(1, model.ModeAuto, slusk+":1.4.2"))
e.h.cfg.Timeout = time.Minute e.h.cfg.Timeout = time.Minute
-15
View File
@@ -31,21 +31,6 @@ const MinPinnedRefVersion = 3
// update in a helper container nor resume it after restarting. // update in a helper container nor resume it after restarting.
const MinSelfUpdateVersion = 4 const MinSelfUpdateVersion = 4
// SelfUpdateLabelPrefix starts the names of the labels on the helper
// container that finishes an agent's self-update. The new agent finds the
// helper by them, and the inventory skips containers that carry them.
const SelfUpdateLabelPrefix = "se.shcizo.package-updater.self-update."
// Labels on a self-update helper container.
const (
LabelSelfUpdateJob = SelfUpdateLabelPrefix + "job"
LabelSelfUpdateRef = SelfUpdateLabelPrefix + "ref"
LabelSelfUpdateStarted = SelfUpdateLabelPrefix + "started"
LabelSelfUpdateProject = SelfUpdateLabelPrefix + "project"
LabelSelfUpdateService = SelfUpdateLabelPrefix + "service"
LabelSelfUpdateOutput = SelfUpdateLabelPrefix + "output"
)
// MaxJobOutput caps the job output an agent keeps and the master stores, in // 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 // 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. // byte) a JobResult carrying this much output still fits the 1 MiB read limit.