From fa833938098a95fa98bca67ff0b15ee982417200 Mon Sep 17 00:00:00 2001 From: Ivan Liang Date: Wed, 14 Oct 2026 19:04:30 -0400 Subject: [PATCH 1/4] refactor(images): name a challenge's image repository in one place instanceImageName built the repository inline. Preloading needs it on its own, to scope worker cleanup to one challenge's tags, so it is now imageRepository, and instanceImageName is that plus the content tag. The names it produces are unchanged. --- cork/docker.go | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/cork/docker.go b/cork/docker.go index bea4101..d4f591a 100644 --- a/cork/docker.go +++ b/cork/docker.go @@ -699,11 +699,15 @@ func (m *Manager) retagLegacyImages(db *sqlx.DB, id BuildId, challenge string, s // pushed after building and pulled before launching; cork and corkd must be // configured with the same registry value for the derived names to agree. func (m *Manager) instanceImageName(challenge ChallengeId, bMeta *BuildMetadata, image Image) string { - name := fmt.Sprintf("%s:%s", challenge, bMeta.dockerId(image)) + return fmt.Sprintf("%s:%s", m.imageRepository(challenge), bMeta.dockerId(image)) +} + +// imageRepository is the repository every image of the challenge is tagged in. +func (m *Manager) imageRepository(challenge ChallengeId) string { if m.challengeRegistry != "" { - name = fmt.Sprintf("%s/%s", m.challengeRegistry, name) + return fmt.Sprintf("%s/%s", m.challengeRegistry, challenge) } - return name + return string(challenge) } // cacheRefsFor picks the BuildKit cache sources for one image build. From 794b8699178d089b1f56414709fbc0eb4bfa022f Mon Sep 17 00:00:00 2001 From: Ivan Liang Date: Fri, 16 Oct 2026 15:15:49 -0400 Subject: [PATCH 2/4] feat(preload): hold a schema's images on every worker ahead of its launches A worker pulls an image the first time it launches it. That is the right mode for a catalogue too large to fit on one box, and costs a pull on each worker's first launch of each image: a warm launch never contacts the registry at all (the tag is checked on the worker's own daemon), so the cost is exactly those cold pulls, and it lands hardest when many players launch at once. A set that does fit on the workers need not pay it at all. A preloaded schema (the preloads table; AddPreload, RemovePreload, ListPreloads, put in the API in a later commit) has every image it launches pulled onto every worker ahead of time. It is state, not a one-shot, so it covers workers added later (worker-add, recovery, the autoscaler) and follows redeploys. Each worker runs a pass when it joins placement, when a schema is preloaded, as each build lands, and every five minutes as a backstop. A pass that fails -- a pull, the worker's image list, the database -- goes again after 30 seconds, doubling while it keeps failing, up to the sweep. A pass pulls what the worker lacks, one image at a time, outside the launch slots and under the restart pull ceiling, and never holds the worker out of placement: a box added under load serves at once, and its launches pull as they always have until the pass catches up. Each worker pulls in its own random order, and at most CORK_PRELOAD_PULLS (default 8) preload pulls run at once across the fleet. Before each pull the pass reads afresh whether the image is still wanted, so a preload removed, or an image superseded, while it runs is not pulled; a read that fails pulls nothing and fails the pass. That is also the difference in cleanup. An on-demand schema leaves it out of band: corkd removes no images, and docker-reaper evicts what goes unused. While a challenge is preloaded its cleanup is corkd's, and opportunistic: after its pulls, each worker deletes the tags in the challenge's repository that no current build launches, in any schema. An old version waits for a successor that landed too late for the pass to try, which the next pass pulls first: deleting it before would take the layers the two share, for that pull to fetch again. A failed pull holds nothing back, since on a full disk the delete is what makes room for the retry. The delete is never forced, so dockerd itself refuses it while a container uses the image -- an instance the redeploy has yet to stop or restart on the new version, a launch it overtook -- and corkd tags that image :stale- instead, a reference-store write, for a later pass, an operator or a reaper to delete once nothing uses it. Nothing is held or paced around these deletes. Measured on Docker 29.9 with dockerd's IO held to gp3's baseline, deleting a whole challenge image's own layers takes about 1s for a typical one and up to 6s for the heaviest, while the daemon's creates, image checks and lists and container removals wait; but a redeploy's old version shares most of its layers with the new one, which the worker by then holds, so the delete that follows it is small. Removals read the database afresh and take no lock: a rollback that lands meanwhile can at worst have an unused image deleted that has just become current again, which the next launch or pass pulls back. A launch caught in such a transition may pull cold or fail as retryable, which is preferred to any machinery that would protect it. A delete or stale tag that times out ends the worker's removals until its next pass, and none of the preloader's timeouts count as the worker's stalls. A failed delete waits for the five-minute sweep rather than the 30-second retry: deletes come last, and the usual failure is a busy daemon; the removal phase's own database read, failing, fails the pass like any other. What an operation takes out of preloading -- the preload removed, the schema deleted, a challenge dropped -- may stay on the workers rather than being tracked down: a stale image costs disk that a preloaded worker has room for and an operator can vet, where a wrong delete costs launches. corkd logs the images that may remain, read from its database without asking any worker (releasedImages, against what preloading covered when the operation began), and RemovePreload returns them. A preload ends with RemovePreload or DeleteSchema only: an update that leaves the schema empty, such as a converge that failed, keeps it for the schema's return, and the renumbering refusal of a hand-over says to preload-add again after the remove-schema it asks for. Every operation still takes the one plain updateMu, now through lockUpdates/unlockUpdates, which note what preloading covered before it and keep the preloading flag current. initWorkers now registers its workers under workersMu, since each conn's goroutine starts as it is made, and Build draws its schema name under randMu, which the preloader's shuffle now shares. cork can see counts, not bytes: sizing the workers' image store for the set, plus a second version of what one update redeploys, stays the operator's, as does turning docker-reaper's image eviction off on those workers and back on after the preload is removed. Unit tests cover the planning, including a tag an on-demand schema launches in a preloaded challenge's repository, which stays, and an old version that waits for a successor landing mid-pass but not for a failed pull; the sets read from the database; add, remove and delete, and what each releases; and passes against a fake daemon: a worker admitted to placement starting its preloader, the pull cap, the worker re-checked after the wait for a fleet slot, a pass that stops pulling what is no longer wanted, one that cannot read the database before a pull or before its removals and so fails, the retry, its backoff and its reset, unforced deletes and the stale tag on an image in use, timeouts on either that neither eject the worker nor go on, an unreachable worker left alone, kicks from preload-add and landing builds only, and a preload that outlives an emptied schema. The fake answers as dockerd 29.9.1 was found to where the code depends on it: 404 for a missing tag, 409 for an image's last tag while a container uses it, the tag gone before a slow delete answers, and a tag that moves. Each guard's test fails with the guard removed, and the cork package passes under the race detector. --- cork/api.go | 31 +- cork/buildplane.go | 1 + cork/database.go | 6 + cork/database_builds.go | 5 + cork/env.go | 2 +- cork/handover.go | 10 +- cork/preload.go | 676 +++++++++++++++++++++++++ cork/preload_test.go | 1068 +++++++++++++++++++++++++++++++++++++++ cork/types.go | 20 +- cork/workers.go | 26 +- 10 files changed, 1819 insertions(+), 26 deletions(-) create mode 100644 cork/preload.go create mode 100644 cork/preload_test.go diff --git a/cork/api.go b/cork/api.go index 0883f4a..48ec78f 100644 --- a/cork/api.go +++ b/cork/api.go @@ -52,6 +52,7 @@ func NewManager(logLevel LogLevel, options ...ManagerOption) *Manager { } mgr.workerTiming = mgr.workerTimingFromEnv() + mgr.preloadPulls = mgr.preloadPullSlots() if err := mgr.setDirectories(); err != nil { return nil @@ -68,6 +69,10 @@ func NewManager(logLevel LogLevel, options ...ManagerOption) *Manager { if err := mgr.initDatabase(); err != nil { return nil } + if err := mgr.notePreloading(); err != nil { + mgr.log.errorf("could not read whether any schema is preloaded: %s", err) + return nil + } if err := mgr.checkExternalBuildPlane(); err != nil { return nil @@ -308,8 +313,8 @@ func (m *Manager) UpdateWithOptions(fp string, options UpdateOptions) *Challenge // ports twice, and taking the network the other had just created for a // leftover of an earlier generation (startNetwork). An update is an // operator action, so the second one waits rather than being refused. - m.updateMu.Lock() - defer m.updateMu.Unlock() + m.lockUpdates() + defer m.unlockUpdates() cu := m.DetectChanges(fp) errs := m.addChallenges(cu.Added) @@ -350,7 +355,9 @@ func (m *Manager) Build(challenge ChallengeId, seeds []int, flagFormat string) ( if m.externalBuildPlane { return nil, ErrExternalBuildPlane } + m.randMu.Lock() schema := fmt.Sprintf("%s%x", manualSchemaPrefix, m.rand.Int63()) + m.randMu.Unlock() instanceCount := -1 builds := make([]*BuildMetadata, len(seeds)) @@ -648,8 +655,8 @@ func (m *Manager) ListSchemas() ([]string, error) { // boundary, and never inside: the unlocked helpers below are what the // operations call each other through. func (m *Manager) CreateSchema(schema *Schema) []error { - m.updateMu.Lock() - defer m.updateMu.Unlock() + m.lockUpdates() + defer m.unlockUpdates() return m.createSchema(schema) } @@ -671,8 +678,8 @@ func (m *Manager) createSchema(schema *Schema) []error { // state. Serialized with rebuilds and other schema operations; see // CreateSchema. func (m *Manager) UpdateSchema(schema *Schema) []error { - m.updateMu.Lock() - defer m.updateMu.Unlock() + m.lockUpdates() + defer m.unlockUpdates() exists, err := m.schemaExists(schema.Name) if err != nil { @@ -831,15 +838,21 @@ func (m *Manager) convergeBuildInstances(buildMeta *BuildMetadata, cMeta *Challe // tag is content-addressed so the orchestrator taking it resolves the same // one. func (m *Manager) DeleteSchema(name string, retire bool) error { - m.updateMu.Lock() - defer m.updateMu.Unlock() + m.lockUpdates() + defer m.unlockUpdates() err := m.lockSchema(name) if err != nil { return err } - return m.cleanupSchemaResources(name, retire) + if err := m.cleanupSchemaResources(name, retire); err != nil { + return err + } + // A preload goes with its schema, here only: one an update leaves empty + // (a converge that failed) is still preloaded when it comes back. + _, err = m.db.Exec("DELETE FROM preloads WHERE schema = ?;", name) + return err } // cleanupSchemaResources stops what a schema runs and takes its builds diff --git a/cork/buildplane.go b/cork/buildplane.go index 562e86b..e5001e6 100644 --- a/cork/buildplane.go +++ b/cork/buildplane.go @@ -94,6 +94,7 @@ func (m *Manager) initBuildPlane() error { // or reclaiming them. var orchestratorOnlySettings = []string{ CONCURRENT_LAUNCHES_ENV, + PRELOAD_PULLS_ENV, PORTS_ENV, IFACE_ENV, DISK_QUOTA_ENV, diff --git a/cork/database.go b/cork/database.go index 10bfd22..e1e275c 100644 --- a/cork/database.go +++ b/cork/database.go @@ -170,6 +170,12 @@ const schemaQuery string = ` public TEXT NOT NULL DEFAULT '' ); + -- Schemas every worker holds the images of (see preload.go). Not a + -- foreign key: a schema is the builds that name it, with no row of its own. + CREATE TABLE IF NOT EXISTS preloads ( + schema TEXT NOT NULL PRIMARY KEY + ); + CREATE TABLE IF NOT EXISTS containers ( instance INTEGER NOT NULL, id TEXT NOT NULL PRIMARY KEY, diff --git a/cork/database_builds.go b/cork/database_builds.go index cdb3741..b8f20b8 100644 --- a/cork/database_builds.go +++ b/cork/database_builds.go @@ -280,6 +280,11 @@ func (m *Manager) finalizeBuild(build *BuildMetadata) error { err = txn.Commit() if err != nil { // It's undocumented what this means... m.log.error(err) + } else if m.preloading.Load() { + // A new generation goes out to the workers as it lands, not when the + // operation it is part of ends, which for an update of many challenges + // can be minutes later. + m.kickPreloads() } return err diff --git a/cork/env.go b/cork/env.go index 23c8871..1a2a277 100644 --- a/cork/env.go +++ b/cork/env.go @@ -53,7 +53,7 @@ func Getenv(name string) string { var settingNames = []string{ DB_ENV, DIR_ENV, ARTIFACT_DIR_ENV, REGISTRY_ENV, REGISTRY_USER_ENV, REGISTRY_TOKEN_ENV, REGISTRY_CERT_DIR_ENV, LOGGING_ENV, IFACE_ENV, - PORTS_ENV, CONCURRENT_LAUNCHES_ENV, DISK_QUOTA_ENV, PRUNE_AGE_ENV, + PORTS_ENV, CONCURRENT_LAUNCHES_ENV, PRELOAD_PULLS_ENV, DISK_QUOTA_ENV, PRUNE_AGE_ENV, DB_WAL_ENV, WORKER_POLL_INTERVAL_ENV, WORKER_POLL_TIMEOUT_ENV, WORKER_PING_TIMEOUT_ENV, WORKER_DEEP_PROBE_ENV, WORKER_MAX_MISSES_ENV, WORKER_LOAD_MISSES_ENV, diff --git a/cork/handover.go b/cork/handover.go index 6a87cea..00c7c1e 100644 --- a/cork/handover.go +++ b/cork/handover.go @@ -106,8 +106,8 @@ func (m *Manager) HandOverChallenge(id ChallengeId, handOver *HandOver, options // Serialized with updates, schema operations and other hand-overs: a // hand-over restarts and relaunches the instances of what it replaces, // as a rebuild does, and would work over any of them. - m.updateMu.Lock() - defer m.updateMu.Unlock() + m.lockUpdates() + defer m.unlockUpdates() current, err := m.lookupChallengeMetadata(id) if err != nil { @@ -394,7 +394,7 @@ func (m *Manager) commitHandedOverBuild(cMeta *ChallengeMetadata, delivered *Bui // bundle. if delivered.Id != 0 && row.Id != delivered.Id { return nil, []error{fmt.Errorf( - "%s is on record as build %d and was handed over as build %d: the build plane has renumbered it, and its artifact bundle is now named for the new id; release the schema (remove-schema) and hand it over again", + "%s is on record as build %d and was handed over as build %d: the build plane has renumbered it, and its artifact bundle is now named for the new id; release the schema (remove-schema) and hand it over again, then preload-add it again if it was preloaded", handOverBuildName(index, cMeta.Id, delivered), row.Id, delivered.Id)} } // A row with no generation on it -- opened now, or by a hand-over that @@ -467,8 +467,8 @@ func (m *Manager) commitHandedOverBuild(cMeta *ChallengeMetadata, delivered *Bui // there the tree decides: the next update finds the challenge again unless // it is gone from the tree as well. func (m *Manager) RemoveChallenge(id ChallengeId) error { - m.updateMu.Lock() - defer m.updateMu.Unlock() + m.lockUpdates() + defer m.unlockUpdates() meta, err := m.lookupChallengeMetadata(id) if err != nil { diff --git a/cork/preload.go b/cork/preload.go new file mode 100644 index 0000000..6f5e00a --- /dev/null +++ b/cork/preload.go @@ -0,0 +1,676 @@ +package cork + +import ( + "context" + "errors" + "maps" + "sort" + "strings" + "sync" + "time" + + "github.com/containerd/errdefs" + "github.com/moby/moby/api/types/image" + "github.com/moby/moby/client" +) + +// A preloaded schema has every image it launches held by every worker, so no +// launch waits on a pull: the mode for a set that fits on the workers, where +// pulling on demand (the default) suits one that does not. It covers workers +// added later, and follows redeploys: the new generation is pulled everywhere. +// +// Cleanup is opportunistic, for as long as a challenge is preloaded: after its +// pulls, each worker deletes the images of it that no current build launches +// (superseded), unless a container still uses one, which is tagged stale +// (staleTag) instead and deleted by a later pass once nothing uses it. What an +// operation takes out of preloading -- the schema's preload removed, the schema +// deleted, a challenge dropped -- may stay on the workers, and corkd lists the +// images that may remain (releasedImages) rather than track them down. + +var ( + // ErrPreloadNeedsRegistry: workers preload from the challenge registry. + ErrPreloadNeedsRegistry = errors.New("preloading needs a challenge registry (CORK_REGISTRY)") +) + +// preloadSweepInterval is the pass every worker gets without being asked: the +// backstop for removals put off, and the slowest retry of a failing pass. +const preloadSweepInterval = 5 * time.Minute + +// preloadRetryAfter is how soon a pass that failed goes again, doubling while +// it keeps failing (up to the sweep). A variable so tests can shorten it. +var preloadRetryAfter = 30 * time.Second + +// AddPreload preloads the schema on every worker. Idempotent. +func (m *Manager) AddPreload(schema string) (err error) { + if m.challengeRegistry == "" { + return ErrPreloadNeedsRegistry + } + // Deferred first, so it runs once the lock is let go. + defer func() { + if err == nil { + m.kickPreloads() + } + }() + m.lockUpdates() + defer m.unlockUpdates() + + exists, err := m.schemaExists(schema) + if err != nil { + return err + } + if !exists { + return &UnknownIdentifierError{Type: "schema", Name: schema} + } + _, err = m.db.Exec("INSERT INTO preloads(schema) VALUES (?) ON CONFLICT(schema) DO NOTHING;", schema) + return err +} + +// RemovePreload returns the schema to pulling on demand. Its images stay on the +// workers, warm, and leave corkd's cleanup unless another preloaded schema +// wants them; it returns those (releasedImages), for an operator to know what +// eviction, or a hand, has to take from here. +func (m *Manager) RemovePreload(schema string) ([]string, error) { + // Serialized with updates, so that what it reports is what it changed. + m.lockUpdates() + defer m.unlockUpdates() + // Read here rather than taken from lockUpdates, which reads only while + // it believes something is preloaded: an answer of "nothing released" + // must mean that, not that the reading was missing. + before, err := m.preloadSets() + if err != nil { + return nil, err + } + res, err := m.db.Exec("DELETE FROM preloads WHERE schema = ?;", schema) + if err != nil { + return nil, err + } + if n, _ := res.RowsAffected(); n == 0 { + return nil, &UnknownIdentifierError{Type: "preload", Name: schema} + } + after, err := m.preloadSets() + if err != nil { + return nil, err + } + return releasedImages(before, after), nil +} + +// PreloadInfo is one preloaded schema and how far each worker has got with it. +type PreloadInfo struct { + Schema string `json:"schema"` + Images int `json:"images"` + Workers []PreloadWorker `json:"workers"` +} + +type PreloadWorker struct { + IP string `json:"ip"` + Reachable string `json:"reachable"` + // Held counts the schema's images the worker was last seen to hold. + Held int `json:"held"` + Pulling string `json:"pulling,omitempty"` + Failed []PreloadFailure `json:"failed,omitempty"` + LastPass *time.Time `json:"last_pass,omitempty"` +} + +type PreloadFailure struct { + Image string `json:"image"` + Error string `json:"error"` +} + +// ListPreloads reports every preloaded schema, worker by worker. +func (m *Manager) ListPreloads() ([]PreloadInfo, error) { + sets, err := m.preloadSets() + if err != nil { + return nil, err + } + + type snapshot struct { + ip string + reachable workerReachable + status *preloadStatus + } + m.workersMu.RLock() + snaps := make([]snapshot, 0, len(m.workerOrder)) + for _, ip := range m.workerOrder { + w := m.workers[ip] + snaps = append(snaps, snapshot{ip, workerReachable(w.reachable.Load()), &w.preload}) + } + m.workersMu.RUnlock() + + infos := []PreloadInfo{} + for _, schema := range sets.schemaNames() { + wanted := sets.bySchema[schema] + info := PreloadInfo{Schema: schema, Images: len(wanted), Workers: []PreloadWorker{}} + for _, snap := range snaps { + info.Workers = append(info.Workers, snap.status.report(snap.ip, snap.reachable, wanted)) + } + infos = append(infos, info) + } + return infos, nil +} + +// preloadStatus is what a worker's preloader last established. A replaced conn +// starts empty and its first pass fills it in. +type preloadStatus struct { + mu sync.Mutex + held map[string]bool // images the worker holds + failed map[string]string // wanted images whose last pull failed, and why + pulling string + lastPass time.Time +} + +func (s *preloadStatus) begin(held map[string]bool) { + s.mu.Lock() + defer s.mu.Unlock() + s.held = maps.Clone(held) + s.failed = map[string]string{} +} + +func (s *preloadStatus) pull(image string) { + s.mu.Lock() + defer s.mu.Unlock() + s.pulling = image +} + +func (s *preloadStatus) pulled(image string, err error) { + s.mu.Lock() + defer s.mu.Unlock() + s.pulling = "" + if err != nil { + s.failed[image] = err.Error() + return + } + s.held[image] = true +} + +func (s *preloadStatus) removed(image string) { + s.mu.Lock() + defer s.mu.Unlock() + delete(s.held, image) +} + +func (s *preloadStatus) finish() { + s.mu.Lock() + defer s.mu.Unlock() + s.pulling = "" + s.lastPass = time.Now() +} + +func (s *preloadStatus) report(ip string, reachable workerReachable, wanted []string) PreloadWorker { + s.mu.Lock() + defer s.mu.Unlock() + r := PreloadWorker{IP: ip, Reachable: reachable.String()} + for _, image := range wanted { + if image == s.pulling { + r.Pulling = image // only this schema's: another's pull says nothing of it + } + if s.held[image] { + r.Held++ + } + if e, ok := s.failed[image]; ok { + r.Failed = append(r.Failed, PreloadFailure{Image: image, Error: e}) + } + } + if !s.lastPass.IsZero() { + at := s.lastPass + r.LastPass = &at + } + return r +} + +// kickPreloads asks every worker's preloader for a pass now. +func (m *Manager) kickPreloads() { + m.workersMu.RLock() + defer m.workersMu.RUnlock() + for _, w := range m.workers { + select { + case w.preloadNow <- struct{}{}: + default: + } + } +} + +// preloadLoop runs a worker's passes, from its admission until its conn is +// replaced or removed. +func (m *Manager) preloadLoop(w *workerConn) { + sweep := time.NewTicker(preloadSweepInterval) + defer sweep.Stop() + retryIn := preloadRetryAfter + for { + // A failed pass goes again soon, getting the new images out being + // what comes first, but less often while it keeps failing. + var retry <-chan time.Time + var timer *time.Timer + if m.preloadPass(w) { + timer = time.NewTimer(retryIn) + retry = timer.C + retryIn = min(2*retryIn, preloadSweepInterval) + } else { + retryIn = preloadRetryAfter + } + select { + case <-w.done: + case <-w.preloadNow: + case <-sweep.C: + case <-retry: + } + if timer != nil { + timer.Stop() + } + if w.gone() { + return + } + } +} + +// preloadPass pulls what the worker lacks, one image at a time so a pass under +// live traffic takes one core for extraction rather than every one, then +// removes what the preloaded schemas have superseded. It reports whether +// anything failed that a pass soon should try again. +func (m *Manager) preloadPass(w *workerConn) (failed bool) { + if !m.preloadable(w) { + return false + } + // No lock: what lands while the pass runs kicks the next one. + sets, err := m.preloadSets() + if err != nil { + m.log.errorf("worker %s: could not read what to preload: %s", w.ip, err) + return true + } + if len(sets.schemas) == 0 { + return false + } + images, err := m.workerImages(w) + if err != nil { + return true + } + defer w.preload.finish() + held := heldTags(images) + w.preload.begin(held) + + pulls := missing(sets.wanted, held) + // Each worker in its own order, so that a fleet starting together does + // not ask the registry for the same image all at once. + m.shuffle(pulls) + pullTimeout := m.restartLimits().pullTimeout + pulled, pullsFailed := 0, 0 + tried := map[string]bool{} + for _, name := range pulls { + release, ok := m.acquirePreloadPull(w) + if !ok { + return false + } + // Checked with the slot held, since waiting for one may take long. + if !m.preloadable(w) { + release() + m.log.warnf("worker %s: preload pass stopped after %d pull(s): the worker is no longer reachable", w.ip, pulled+pullsFailed) + return false + } + // A pass is long: its preload may have been removed since, or the + // image superseded. Unread, nothing is pulled, and the pass goes again. + if now, err := m.preloadSets(); err != nil || !now.wanted[name] { + release() + if err != nil { + m.log.errorf("worker %s: could not read what to preload: %s", w.ip, err) + return true + } + continue + } + w.preload.pull(name) + err := m.pullImage(w.cli, name, pullTimeout) + release() + tried[name] = true + if err != nil { + pullsFailed++ + m.notePreloadError(w, err) + } else { + pulled++ + } + w.preload.pulled(name, err) + } + if pulled+pullsFailed > 0 { + m.log.infof("worker %s: preloaded %d image(s), %d failed", w.ip, pulled, pullsFailed) + } + removalsUnread := m.removeSuperseded(w, held, tried) + return pullsFailed > 0 || removalsUnread +} + +// defaultPreloadPulls is how many preload pulls run at once across the fleet +// (CORK_PRELOAD_PULLS): each worker pulls one at a time, and a fleet starting +// together would otherwise put every worker on the registry at once. +const defaultPreloadPulls = 8 + +func (m *Manager) preloadPullSlots() chan struct{} { + n := defaultPreloadPulls + m.envCount(PRELOAD_PULLS_ENV, &n) + return make(chan struct{}, n) +} + +// acquirePreloadPull takes one of the fleet's preload pull slots, unless the +// conn goes first. A Manager built without them (tests) has no cap. +func (m *Manager) acquirePreloadPull(w *workerConn) (release func(), ok bool) { + if m.preloadPulls == nil { + return func() {}, true + } + select { + case m.preloadPulls <- struct{}{}: + return func() { <-m.preloadPulls }, true + case <-w.done: + return nil, false + } +} + +func (m *Manager) shuffle(names []string) { + m.randMu.Lock() + defer m.randMu.Unlock() + if m.rand != nil { + m.rand.Shuffle(len(names), func(i, j int) { names[i], names[j] = names[j], names[i] }) + } +} + +func (m *Manager) preloadable(w *workerConn) bool { + return !w.gone() && workerReachable(w.reachable.Load()) == workerReachableOk +} + +// removeSuperseded deletes, of the tags the worker held, those in preloaded +// challenges' repositories that no current build launches. The delete is not +// forced, so dockerd refuses it while a container uses the image, which is +// tagged stale instead (markStale). Read afresh, without a lock: at worst a +// rollback that lands meanwhile has an unused image deleted, which the next +// launch or pass pulls back. It reports whether it could not read the +// database, which the pass retries; a failed delete waits for the sweep, as +// deletes come last and the usual failure is a busy daemon. +func (m *Manager) removeSuperseded(w *workerConn, held, tried map[string]bool) (unread bool) { + // held is the pass's listing, tried what it pulled or failed to: a tag + // gone since is not found, which counts as removed, and one that came + // since waits for a later pass. + sets, err := m.preloadSets() + if err != nil { + m.log.errorf("worker %s: could not read what preloading supersedes: %s", w.ip, err) + return true + } + for _, tag := range superseded(sets, held, tried) { + if !m.preloadable(w) { + return false + } + err := m.removeWorkerImage(w, tag) + if errdefs.IsConflict(err) { + err = m.markStale(w, tag) + } else if err == nil { + w.preload.removed(tag) + m.log.infof("worker %s: removed superseded image %s", w.ip, tag) + } else { + m.log.warnf("worker %s: could not remove superseded image %s: %s", w.ip, tag, err) + } + if errors.Is(err, context.DeadlineExceeded) { + // The daemon is slow; more deletes would only queue behind it. + m.log.warnf("worker %s: its other removals wait for a later pass", w.ip) + return false + } + } + return false +} + +// staleTagPrefix marks a superseded image a container still used when corkd +// went to delete it, for a later pass, an operator or a reaper to delete once +// nothing does. A reference filter's * stops at a /, so list them with +// docker images --format '{{.Repository}}:{{.Tag}}' | grep ':stale-'. +const staleTagPrefix = "stale-" + +func staleTag(ref string) string { + repo := imageRepo(ref) + return repo + ":" + staleTagPrefix + strings.TrimPrefix(ref, repo+":") +} + +func isStale(ref string) bool { + return strings.HasPrefix(strings.TrimPrefix(ref, imageRepo(ref)+":"), staleTagPrefix) +} + +// markStale gives an image in use the stale tag in place of its superseded +// one, which, with another tag on the image, a delete only takes off. Both are +// reference-store writes, quick whatever the image's size. +func (m *Manager) markStale(w *workerConn, tag string) error { + if isStale(tag) { + return nil // still in use; a later pass tries again + } + stale := staleTag(tag) + ctx, cancel := m.controlCtx() + _, err := w.cli.ImageTag(ctx, client.ImageTagOptions{Source: tag, Target: stale}) + cancel() + if err == nil { + err = m.removeWorkerImage(w, tag) + } else { + m.notePreloadError(w, err) + } + if err != nil { + m.log.warnf("worker %s: superseded image %s is still in use, and could not be marked %s: %s", w.ip, tag, stale, err) + return err + } + w.preload.removed(tag) + m.log.infof("worker %s: superseded image %s is still in use; kept as %s, for a later pass to remove once nothing uses it", w.ip, tag, stale) + return nil +} + +// lockUpdates takes updateMu for an operation that changes builds and, while +// anything is preloaded, notes what preloading covers before it +// (unlockUpdates). Every caller goes through it. +func (m *Manager) lockUpdates() { + m.updateMu.Lock() + m.updateBefore = nil + if m.preloading.Load() { + if before, err := m.preloadSets(); err == nil { + m.updateBefore = before + } + } +} + +// unlockUpdates ends such an operation: it logs what preloading no longer +// covers, then releases updateMu. It kicks no one: the builds an operation +// lands kick the workers as they do. +func (m *Manager) unlockUpdates() { + after, err := m.preloadSets() + if err != nil { + m.log.errorf("could not read what is preloaded after an update: %s", err) + } else { + m.preloading.Store(len(after.schemas) > 0) + if m.updateBefore != nil { + if released := releasedImages(m.updateBefore, after); len(released) > 0 { + m.log.warnf("preloading no longer covers %d image(s), which may remain on the workers, as may older ones of their repositories, until eviction or an operator takes them: %s", + len(released), strings.Join(released, ", ")) + } + } + } + m.updateBefore = nil + m.updateMu.Unlock() +} + +// releasedImages lists the images preloading wanted before an operation that it +// neither wants nor removes after it. Read from the database alone, not from +// any worker: a worker may hold fewer of them, and older ones of the same +// repositories that removal had yet to reach. +func releasedImages(before, after *preloadSets) []string { + released := []string{} + for name := range before.wanted { + if after.wanted[name] || (!after.named[name] && after.repos[imageRepo(name)]) { + continue + } + released = append(released, name) + } + sort.Strings(released) + return released +} + +// notePreloadError passes a failed preload call to the worker's health only +// when its daemon is gone (refused, reset). A timeout is the daemon busy with +// the launches it serves, often behind preloading's own delete, and counting +// it as a stall could eject a worker that is serving. +func (m *Manager) notePreloadError(w *workerConn, err error) { + if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, ErrPullTimeout) { + return + } + m.noteWorkerTransportError(w.ip, w.cli, err) +} + +// notePreloading reads whether any schema is preloaded (m.preloading), for a +// Manager starting up; operations keep it current from there (unlockUpdates). +func (m *Manager) notePreloading() error { + var n int + if err := m.db.Get(&n, "SELECT COUNT(1) FROM preloads;"); err != nil { + return err + } + m.preloading.Store(n > 0) + return nil +} + +func (m *Manager) workerImages(w *workerConn) ([]image.Summary, error) { + ctx, cancel := m.controlCtx() + defer cancel() + list, err := w.cli.ImageList(ctx, client.ImageListOptions{}) + if err != nil { + m.log.errorf("worker %s: could not list images to preload: %s", w.ip, err) + m.notePreloadError(w, err) + return nil, err + } + return list.Items, nil +} + +func (m *Manager) removeWorkerImage(w *workerConn, ref string) error { + ctx, cancel := m.controlCtx() + defer cancel() + _, err := w.cli.ImageRemove(ctx, ref, client.ImageRemoveOptions{PruneChildren: true}) + if errdefs.IsNotFound(err) { + return nil + } + if err != nil { + m.notePreloadError(w, err) + } + return err +} + +// preloadSets is what the database says preloading involves. Image names are +// as instanceImageName gives them, which is how a worker lists what it pulled. +type preloadSets struct { + schemas map[string]bool + bySchema map[string][]string // schema -> images it preloads, sorted + wanted map[string]bool // every image a preloaded schema launches + named map[string]bool // every image any current build launches + repos map[string]bool // repositories of preloaded challenges +} + +func (s *preloadSets) schemaNames() []string { + names := make([]string, 0, len(s.schemas)) + for name := range s.schemas { + names = append(names, name) + } + sort.Strings(names) + return names +} + +func (m *Manager) preloadSets() (*preloadSets, error) { + sets := &preloadSets{ + schemas: map[string]bool{}, + bySchema: map[string][]string{}, + wanted: map[string]bool{}, + named: map[string]bool{}, + repos: map[string]bool{}, + } + var schemas []string + if err := m.db.Select(&schemas, "SELECT schema FROM preloads;"); err != nil { + return nil, err + } + if len(schemas) == 0 { + return sets, nil + } + for _, s := range schemas { + sets.schemas[s] = true + sets.bySchema[s] = []string{} + } + + // Built rows only: an unbuilt one launches nothing, and one whose rebuild + // failed still serves the generation it records. + var rows []struct { + Schema string `db:"schema"` + Challenge ChallengeId `db:"challenge"` + Seed int `db:"seed"` + Checksum uint32 `db:"checksum"` + Host string `db:"host"` + ChallengeType string `db:"challengetype"` + PortCount int `db:"portcount"` + } + err := m.db.Select(&rows, `SELECT b.schema, b.challenge, b.seed, b.checksum, i.host, c.challengetype, + (SELECT COUNT(*) FROM portNames WHERE portNames.challenge = b.challenge) AS portcount + FROM builds AS b + JOIN images AS i ON i.build = b.id + JOIN challenges AS c ON c.id = b.challenge + WHERE b.checksum != 0 AND b.flag != '' AND i.host != 'builder';`) + if err != nil { + return nil, err + } + for _, r := range rows { + name := m.instanceImageName(r.Challenge, &BuildMetadata{Seed: r.Seed, Checksum: r.Checksum}, Image{Host: r.Host}) + sets.named[name] = true + if !sets.schemas[r.Schema] || deriveDeliveryType(r.ChallengeType, r.PortCount) != DeliveryService { + continue + } + sets.wanted[name] = true + sets.bySchema[r.Schema] = append(sets.bySchema[r.Schema], name) + sets.repos[m.imageRepository(r.Challenge)] = true + } + for s := range sets.bySchema { + sort.Strings(sets.bySchema[s]) + } + return sets, nil +} + +// heldTags is every tag a worker listed. +func heldTags(images []image.Summary) map[string]bool { + held := map[string]bool{} + for _, img := range images { + for _, tag := range img.RepoTags { + held[tag] = true + } + } + return held +} + +// missing lists, sorted, the wanted images not in have. +func missing(wanted, have map[string]bool) []string { + var names []string + for name := range wanted { + if !have[name] { + names = append(names, name) + } + } + sort.Strings(names) + return names +} + +// superseded lists, sorted, the held tags in a preloaded challenge's repository +// that no current build launches, preloaded or not. A repository waits while +// one of its wanted images landed too late for the pass to try (the next pass +// pulls it first): deleting the old version before would take the layers the +// two share. A failed pull holds nothing back, since the delete may be what +// makes room for it. +func superseded(sets *preloadSets, held, tried map[string]bool) []string { + waiting := map[string]bool{} + for name := range sets.wanted { + if !held[name] && !tried[name] { + waiting[imageRepo(name)] = true + } + } + var tags []string + for tag := range held { + if repo := imageRepo(tag); !sets.named[tag] && sets.repos[repo] && !waiting[repo] { + tags = append(tags, tag) + } + } + sort.Strings(tags) + return tags +} + +// imageRepo strips the tag from an image name; a registry port is not a tag. +func imageRepo(name string) string { + if i := strings.LastIndex(name, ":"); i > strings.LastIndex(name, "/") { + return name[:i] + } + return name +} diff --git a/cork/preload_test.go b/cork/preload_test.go new file mode 100644 index 0000000..4afd646 --- /dev/null +++ b/cork/preload_test.go @@ -0,0 +1,1068 @@ +package cork + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "reflect" + "slices" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/moby/moby/api/types/image" + "github.com/moby/moby/client" +) + +const preloadTestRegistry = "registry.test:5000" + +func TestImageRepo(t *testing.T) { + for name, want := range map[string]string{ + "registry.test:5000/test/web:s1-1111-challenge": "registry.test:5000/test/web", + "registry.test:5000/test/web": "registry.test:5000/test/web", + "test/web:s1-1111-challenge": "test/web", + "web": "web", + } { + if got := imageRepo(name); got != want { + t.Errorf("imageRepo(%q) = %q, want %q", name, got, want) + } + } +} + +// What a pass pulls and removes. A tag of a preloaded challenge that an +// on-demand schema still launches stays: what any current build launches is +// kept, not only what preloaded schemas want. +func TestPlanPreload(t *testing.T) { + sets := &preloadSets{ + wanted: map[string]bool{"r/web:new": true, "r/web:held": true}, + named: map[string]bool{"r/web:new": true, "r/web:held": true, "r/web:ondemand": true, "r/other:current": true}, + repos: map[string]bool{"r/web": true}, + } + held := heldTags([]image.Summary{ + {RepoTags: []string{"r/web:held"}}, + {RepoTags: []string{"r/web:old", "r/web:older"}}, + {RepoTags: []string{"r/web:ondemand"}}, + {RepoTags: []string{"r/other:current", "r/other:old"}}, // not preloaded: never removed + }) + if got, want := missing(sets.wanted, held), []string{"r/web:new"}; !reflect.DeepEqual(got, want) { + t.Errorf("pulls = %v, want %v", got, want) + } + // The old versions wait for a new one the pass has yet to try, whose + // layers they share, but not for one whose pull failed. + if got := superseded(sets, held, nil); len(got) > 0 { + t.Errorf("removals before the new image is tried = %v, want none", got) + } + want := []string{"r/web:old", "r/web:older"} + if got := superseded(sets, held, map[string]bool{"r/web:new": true}); !reflect.DeepEqual(got, want) { + t.Errorf("removals after the new image failed to pull = %v, want %v", got, want) + } + held["r/web:new"] = true + if got := superseded(sets, held, nil); !reflect.DeepEqual(got, want) { + t.Errorf("removals = %v, want %v", got, want) + } +} + +// Released is what preloading wanted that it neither wants nor removes now. +func TestReleasedImages(t *testing.T) { + before := &preloadSets{wanted: map[string]bool{ + "r/web:shared": true, "r/web:dropped": true, "r/pwn:launched": true, "r/gone:old": true, + }} + after := &preloadSets{ + wanted: map[string]bool{"r/web:shared": true}, // another preloaded schema's + named: map[string]bool{"r/web:shared": true, "r/pwn:launched": true}, // r/web:dropped is removed + repos: map[string]bool{"r/web": true}, + } + if got, want := releasedImages(before, after), []string{"r/gone:old", "r/pwn:launched"}; !reflect.DeepEqual(got, want) { + t.Errorf("released = %v, want %v", got, want) + } +} + +// preloadFixture is an external orchestrator with three challenges: test/web +// (a service, preloaded through "event"), test/docs (artifact-only, also in +// "event") and test/other (a service in "archive"). +type preloadFixture struct { + m *Manager + web, other BuildId + webName string // the image event launches + otherName string + unbuiltWebSeed int +} + +func newPreloadFixture(t *testing.T) *preloadFixture { + t.Helper() + m := setupExternalTestManager(t) + m.challengeRegistry = preloadTestRegistry + f := &preloadFixture{m: m, unbuiltWebSeed: 2} + + docs := retentionTestChallenge("test/docs", 0) + if errs := m.addChallenges([]*ChallengeMetadata{testChallenge("test/web", 0), testChallenge("test/other", 0), docs}); len(errs) > 0 { + t.Fatalf("addChallenges: %v", errs) + } + build := func(schema, challenge string, seed int, checksum uint32, hosts ...string) *BuildMetadata { + t.Helper() + b := &BuildMetadata{ + Id: insertTestBuild(t, m, schema, challenge, "flag{%s}", seed, checksum), Flag: "flag{x}", + Challenge: ChallengeId(challenge), Seed: seed, Checksum: checksum, + } + for _, host := range hosts { + b.Images = append(b.Images, Image{Host: host, Ports: []string{}}) + } + if err := m.finalizeBuild(b); err != nil { + t.Fatalf("finalizeBuild: %s", err) + } + return b + } + web := build("event", "test/web", 1, 0x1111, "challenge", "builder") + build("event", "test/docs", 1, 0x3333, "challenge") + other := build("archive", "test/other", 1, 0x2222, "challenge") + insertTestBuildRow(t, m, "", "event", "test/web", "flag{%s}", f.unbuiltWebSeed, 0) + + f.web, f.other = web.Id, other.Id + f.webName = m.instanceImageName(web.Challenge, web, Image{Host: "challenge"}) + f.otherName = m.instanceImageName(other.Challenge, other, Image{Host: "challenge"}) + if err := m.AddPreload("event"); err != nil { + t.Fatalf("AddPreload: %s", err) + } + return f +} + +func TestPreloadSets(t *testing.T) { + f := newPreloadFixture(t) + sets, err := f.m.preloadSets() + if err != nil { + t.Fatalf("preloadSets: %s", err) + } + // Only the built service image: not the builder host, not the + // artifact-only challenge, not the unbuilt seed, not another schema. + if want := map[string]bool{f.webName: true}; !reflect.DeepEqual(sets.wanted, want) { + t.Errorf("wanted = %v, want %v", sets.wanted, want) + } + if !sets.named[f.webName] || !sets.named[f.otherName] { + t.Errorf("named = %v, want both service images", sets.named) + } + // Cleanup covers the challenges whose images the workers hold. + if want := map[string]bool{preloadTestRegistry + "/test/web": true}; !reflect.DeepEqual(sets.repos, want) { + t.Errorf("repos = %v, want %v", sets.repos, want) + } + if want := []string{f.webName}; !reflect.DeepEqual(sets.bySchema["event"], want) { + t.Errorf("bySchema[event] = %v, want %v", sets.bySchema["event"], want) + } +} + +func TestAddAndRemovePreload(t *testing.T) { + f := newPreloadFixture(t) + m := f.m + + if err := m.AddPreload("event"); err != nil { + t.Errorf("a second AddPreload: %s", err) + } + var unknown *UnknownIdentifierError + if err := m.AddPreload("nonesuch"); !errors.As(err, &unknown) { + t.Errorf("AddPreload of an unknown schema = %v, want an unknown identifier", err) + } + if _, err := m.RemovePreload("event"); err != nil { + t.Fatalf("RemovePreload: %s", err) + } + if _, err := m.RemovePreload("event"); !errors.As(err, &unknown) { + t.Errorf("a second RemovePreload = %v, want an unknown identifier", err) + } + + m.challengeRegistry = "" + if err := m.AddPreload("event"); !errors.Is(err, ErrPreloadNeedsRegistry) { + t.Errorf("AddPreload without a registry = %v, want ErrPreloadNeedsRegistry", err) + } +} + +// Deleting a preloaded schema leaves what it pulled on the workers, for an +// operator to vet, and says which images those are. Another schema stays +// preloaded, so the pass after the delete does run. +func TestDeletingAPreloadedSchemaLeavesItsImages(t *testing.T) { + f := newPreloadFixture(t) + if err := f.m.AddPreload("archive"); err != nil { + t.Fatalf("AddPreload: %s", err) + } + var logged bytes.Buffer + f.m.log = captureLog(&logged) + old := preloadTestRegistry + "/test/web:s1-aaaa-challenge" + d := newFakeImageDaemon(t, + image.Summary{ID: "web", RepoTags: []string{f.webName}}, + image.Summary{ID: "old-web", RepoTags: []string{old}}, + image.Summary{ID: "other", RepoTags: []string{f.otherName}}, + ) + w := d.conn(t, f.m) + if err := f.m.DeleteSchema("event", false); err != nil { + t.Fatalf("DeleteSchema: %s", err) + } + var schemas []string + if err := f.m.db.Select(&schemas, "SELECT schema FROM preloads;"); err != nil || !reflect.DeepEqual(schemas, []string{"archive"}) { + t.Errorf("preloads after the delete = %v (err %v), want only archive", schemas, err) + } + if !strings.Contains(logged.String(), "no longer covers 1 image(s)") || !strings.Contains(logged.String(), f.webName) { + t.Errorf("the delete did not log what preloading left behind:\n%s", logged.String()) + } + f.m.preloadPass(w) + if pulls, removes := d.calls(); len(pulls)+len(removes) > 0 { + t.Errorf("after the delete, a pass pulled %v and removed %v", pulls, removes) + } + if w.preload.lastPass.IsZero() { + t.Error("the pass after the delete did not run") + } +} + +// preload-remove returns what the schema leaves to on-demand cleanup: what +// another preloaded schema wants stays covered. +func TestRemovePreloadReportsWhatItReleases(t *testing.T) { + f := newPreloadFixture(t) + second := &BuildMetadata{ + Id: insertTestBuild(t, f.m, "second", "test/web", "flag{%s}", 9, 0x4444), Flag: "flag{x}", + Challenge: "test/web", Seed: 9, Checksum: 0x4444, Images: []Image{{Host: "challenge", Ports: []string{}}}, + } + if err := f.m.finalizeBuild(second); err != nil { + t.Fatalf("finalizeBuild: %s", err) + } + secondName := f.m.instanceImageName(second.Challenge, second, Image{Host: "challenge"}) + shared := &BuildMetadata{ // event's web image, wanted by a third schema too + Id: insertTestBuild(t, f.m, "shared", "test/web", "flag{%s}", 1, 0x1111), Flag: "flag{x}", + Challenge: "test/web", Seed: 1, Checksum: 0x1111, Images: []Image{{Host: "challenge", Ports: []string{}}}, + } + if err := f.m.finalizeBuild(shared); err != nil { + t.Fatalf("finalizeBuild: %s", err) + } + for _, s := range []string{"second", "shared"} { + if err := f.m.AddPreload(s); err != nil { + t.Fatalf("AddPreload(%s): %s", s, err) + } + } + for _, tc := range []struct { + schema string + want []string + }{ + {"event", []string{}}, + {"shared", []string{f.webName}}, + {"second", []string{secondName}}, + } { + got, err := f.m.RemovePreload(tc.schema) + if err != nil { + t.Fatalf("RemovePreload(%s): %s", tc.schema, err) + } + if !reflect.DeepEqual(got, tc.want) { + t.Errorf("RemovePreload(%s) released %v, want %v", tc.schema, got, tc.want) + } + } +} + +// fakeImageDaemon serves the image calls a preload pass makes, keeping the +// store they act on. +type fakeImageDaemon struct { + srv *httptest.Server + mu sync.Mutex + images []image.Summary + pulls []string + removes []string // " force=<0|1>" + tagged []string // " " + onDelete func() // runs once a delete has untagged, before it answers, outside the lock + onPull func() // runs as a pull arrives, outside the lock + onList func() // runs as an image list arrives, outside the lock + requests []string // every request, " " + inUse map[string]bool // image ids a container uses + pullFails bool // answer pulls with a stream error, as a registry missing the tag does + listFails bool // answer image lists with a 500 +} + +func newFakeImageDaemon(t *testing.T, images ...image.Summary) *fakeImageDaemon { + t.Helper() + d := &fakeImageDaemon{images: images, inUse: map[string]bool{}} + d.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + d.mu.Lock() + delHook, pullHook, listHook := d.onDelete, d.onPull, d.onList + d.mu.Unlock() + if r.Method == "DELETE" { + status, body := d.delete(r) + // As dockerd: the tag is gone at once, and the delete's work + // takes its time after it. + if status == http.StatusOK && delHook != nil { + delHook() + } + if status != http.StatusOK { + http.Error(w, body.(string), status) + return + } + json.NewEncoder(w).Encode(body) + return + } + if r.Method == "POST" && strings.HasSuffix(r.URL.Path, "/images/create") && pullHook != nil { + pullHook() + } + if r.Method == "GET" && strings.HasSuffix(r.URL.Path, "/images/json") && listHook != nil { + listHook() + } + d.mu.Lock() + defer d.mu.Unlock() + d.requests = append(d.requests, r.Method+" "+r.URL.Path) + _, ref, _ := strings.Cut(r.URL.Path, "/images/") + switch { + // What admitting a worker asks of it (runWorker), for an empty one. + case strings.HasSuffix(r.URL.Path, "/_ping"): + w.Header().Set("Api-Version", "1.44") + w.Header().Set("Ostype", "linux") + case strings.HasSuffix(r.URL.Path, "/containers/json"), strings.HasSuffix(r.URL.Path, "/networks"): + io.WriteString(w, "[]") + case r.Method == "GET" && ref == "json" && d.listFails: + http.Error(w, "the daemon is unwell", http.StatusInternalServerError) + case r.Method == "GET" && ref == "json": + json.NewEncoder(w).Encode(d.images) + case r.Method == "POST" && strings.HasSuffix(ref, "/tag"): + source := strings.TrimSuffix(ref, "/tag") + target := r.URL.Query().Get("repo") + ":" + r.URL.Query().Get("tag") + i := slices.IndexFunc(d.images, func(img image.Summary) bool { + return img.ID == source || slices.Contains(img.RepoTags, source) + }) + if i < 0 { + http.Error(w, "No such image: "+source, http.StatusNotFound) + return + } + // As dockerd: a target held elsewhere moves, leaving that image + // untagged rather than gone; one already here stays one. + for j := range d.images { + d.images[j].RepoTags = slices.DeleteFunc(slices.Clone(d.images[j].RepoTags), func(t string) bool { return t == target }) + } + d.images[i].RepoTags = append(d.images[i].RepoTags, target) + d.tagged = append(d.tagged, source+" "+target) + w.WriteHeader(http.StatusCreated) + case r.Method == "GET" && strings.HasSuffix(ref, "/json"): + name := strings.TrimSuffix(ref, "/json") + for _, img := range d.images { + if img.ID == name || slices.Contains(img.RepoTags, name) { + json.NewEncoder(w).Encode(image.InspectResponse{ID: img.ID}) + return + } + } + http.Error(w, "No such image: "+name, http.StatusNotFound) + case r.Method == "POST" && ref == "create" && d.pullFails: + io.WriteString(w, `{"errorDetail":{"message":"manifest unknown"},"error":"manifest unknown"}`+"\n") + case r.Method == "POST" && ref == "create": + name := r.URL.Query().Get("fromImage") + ":" + r.URL.Query().Get("tag") + d.pulls = append(d.pulls, name) + d.images = append(d.images, image.Summary{ID: "pulled-" + name, RepoTags: []string{name}}) + io.WriteString(w, `{"status":"Pull complete"}`+"\n") + default: + http.Error(w, "unexpected "+r.Method+" "+r.URL.Path, http.StatusNotFound) + } + })) + t.Cleanup(d.srv.Close) + return d +} + +// delete serves an image delete, answering a status and either the deletes' +// records or an error message. +func (d *fakeImageDaemon) delete(r *http.Request) (int, any) { + d.mu.Lock() + defer d.mu.Unlock() + d.requests = append(d.requests, r.Method+" "+r.URL.Path) + _, ref, _ := strings.Cut(r.URL.Path, "/images/") + force := "0" + if r.URL.Query().Get("force") == "1" { + force = "1" + } + d.removes = append(d.removes, ref+" force="+force) + i := slices.IndexFunc(d.images, func(img image.Summary) bool { + return img.ID == ref || slices.Contains(img.RepoTags, ref) + }) + if i < 0 { + return http.StatusNotFound, "No such image: " + ref + } + img := d.images[i] + // As dockerd: unforced, the last tag of an image a container uses is + // refused; one of several is only taken off. + if force == "0" && d.inUse[img.ID] && len(img.RepoTags) <= 1 { + return http.StatusConflict, "conflict: unable to remove repository reference " + ref + " (must force) - container is using its referenced image" + } + d.images[i].RepoTags = slices.DeleteFunc(slices.Clone(img.RepoTags), func(t string) bool { return t == ref }) + records := []image.DeleteResponse{{Untagged: ref}} + if img.ID == ref || len(d.images[i].RepoTags) == 0 { + d.images = slices.Delete(d.images, i, i+1) + records = append(records, image.DeleteResponse{Deleted: img.ID}) + } + return http.StatusOK, records +} + +// tags is what a pass would have listed: every tag the daemon holds now. +func (d *fakeImageDaemon) tags() map[string]bool { + d.mu.Lock() + defer d.mu.Unlock() + return heldTags(d.images) +} + +func (d *fakeImageDaemon) calls() (pulls, removes []string) { + d.mu.Lock() + defer d.mu.Unlock() + return append([]string(nil), d.pulls...), append([]string(nil), d.removes...) +} + +func (d *fakeImageDaemon) conn(t *testing.T, m *Manager) *workerConn { + t.Helper() + cli, err := client.NewClientWithOpts( + client.WithHost(strings.Replace(d.srv.URL, "http://", "tcp://", 1)), + client.WithVersion("1.44"), + ) + if err != nil { + t.Fatalf("building a client for the fake daemon: %s", err) + } + w := &workerConn{ + ip: "127.0.0.1", cli: cli, queue: newDaemonQueue(1), + unreachCh: make(chan struct{}), done: make(chan struct{}), + probeNow: make(chan struct{}, 1), preloadNow: make(chan struct{}, 1), + } + w.reachable.Store(int32(workerReachableOk)) + w.reconciled.Store(true) + m.workers = map[string]*workerConn{w.ip: w} + m.workerOrder = []string{w.ip} + t.Cleanup(func() { close(w.done) }) + return w +} + +func TestPreloadPass(t *testing.T) { + f := newPreloadFixture(t) + m := f.m + repo := preloadTestRegistry + "/test/" + d := newFakeImageDaemon(t, + image.Summary{ID: "old-web", RepoTags: []string{repo + "web:s1-aaaa-challenge"}}, + image.Summary{ID: "old-other", RepoTags: []string{repo + "other:s1-bbbb-challenge"}}, + image.Summary{ID: "other", RepoTags: []string{f.otherName}}, + ) + w := d.conn(t, m) + + m.preloadPass(w) + pulls, removes := d.calls() + if want := []string{f.webName}; !reflect.DeepEqual(pulls, want) { + t.Errorf("pulls = %v, want %v", pulls, want) + } + // Only the preloaded challenge's old tag: test/other is not preloaded. + if want := []string{repo + "web:s1-aaaa-challenge force=0"}; !reflect.DeepEqual(removes, want) { + t.Errorf("removes = %v, want %v", removes, want) + } + + m.preloadPass(w) + if again, removedAgain := d.calls(); len(again) != 1 || len(removedAgain) != 1 { + t.Errorf("a second pass pulled %v and removed %v, want nothing new", again, removedAgain) + } + + infos, err := m.ListPreloads() + if err != nil { + t.Fatalf("ListPreloads: %s", err) + } + if len(infos) != 1 || infos[0].Schema != "event" || infos[0].Images != 1 || len(infos[0].Workers) != 1 { + t.Fatalf("ListPreloads = %+v", infos) + } + if got := infos[0].Workers[0]; got.Held != 1 || got.LastPass == nil || got.Reachable != "ok" || len(got.Failed) > 0 { + t.Errorf("worker status = %+v, want 1 held, checked, nothing failed", got) + } +} + +// A schema nobody preloads costs a worker one query a pass and no daemon call. +func TestPreloadPassWithoutPreloads(t *testing.T) { + f := newPreloadFixture(t) + if _, err := f.m.RemovePreload("event"); err != nil { + t.Fatalf("RemovePreload: %s", err) + } + d := newFakeImageDaemon(t, image.Summary{ID: "old-web", RepoTags: []string{preloadTestRegistry + "/test/web:s1-aaaa-challenge"}}) + f.m.preloadPass(d.conn(t, f.m)) + if pulls, removes := d.calls(); len(pulls)+len(removes) > 0 { + t.Errorf("a pass with nothing preloaded pulled %v and removed %v", pulls, removes) + } + d.mu.Lock() + defer d.mu.Unlock() + if len(d.requests) > 0 { + t.Errorf("a pass with nothing preloaded called the daemon: %v", d.requests) + } +} + +func TestKickPreloadsNeverBlocks(t *testing.T) { + f := newPreloadFixture(t) + d := newFakeImageDaemon(t) + d.conn(t, f.m) + done := make(chan struct{}) + go func() { + f.m.kickPreloads() + f.m.kickPreloads() // the buffer is full: dropped, not waited on + close(done) + }() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("kickPreloads blocked on a worker nobody is draining") + } +} + +// Deletes that outlast the control timeout are a busy daemon, not a dead one: +// they must not count against the worker, and the phase stops at the first. +func TestPreloadTimeoutsDoNotEjectTheWorker(t *testing.T) { + f := newPreloadFixture(t) + timing := defaultWorkerTiming + timing.controlTimeout = 50 * time.Millisecond + f.m.workerTiming = timing + old := []image.Summary{{ID: "web", RepoTags: []string{f.webName}}} + for j := 0; j < 3; j++ { + old = append(old, image.Summary{ID: fmt.Sprintf("old-%d", j), + RepoTags: []string{fmt.Sprintf("%s/test/web:s1-%d-challenge", preloadTestRegistry, j)}}) + } + d := newFakeImageDaemon(t, old...) + var attempts atomic.Int32 + d.mu.Lock() + d.onDelete = func() { attempts.Add(1); time.Sleep(300 * time.Millisecond) } + d.mu.Unlock() + w := d.conn(t, f.m) + + f.m.removeSuperseded(w, d.tags(), nil) + time.Sleep(50 * time.Millisecond) + if n := attempts.Load(); n != 1 { + t.Errorf("%d delete(s) attempted, want the phase to stop after the first timed out", n) + } + if r := workerReachable(w.reachable.Load()); r != workerReachableOk { + t.Errorf("worker is %s after preload deletes timed out, want ok", r) + } +} + +// A preload ends with preload-remove or the schema's deletion, not with an +// update that leaves the schema empty (a converge that failed): the schema +// comes back preloaded, and meanwhile preload-list shows it holding nothing. +func TestAPreloadOutlivesAnEmptiedSchema(t *testing.T) { + f := newPreloadFixture(t) + if errs := f.m.UpdateSchema(&Schema{Name: "event", FlagFormat: "flag{%s}", Challenges: map[ChallengeId]BuildSpecification{}}); len(errs) > 0 { + t.Fatalf("UpdateSchema: %v", errs) + } + infos, err := f.m.ListPreloads() + if err != nil { + t.Fatalf("ListPreloads: %s", err) + } + if len(infos) != 1 || infos[0].Schema != "event" || infos[0].Images != 0 { + t.Errorf("preloads after an update emptied the schema = %+v, want event with no images", infos) + } + if !f.m.preloading.Load() { + t.Error("an emptied schema's preload left the fleet marked as preloading nothing") + } +} + +// At most CORK_PRELOAD_PULLS pulls run at once across the fleet. +func TestPreloadPullsAreCappedAcrossTheFleet(t *testing.T) { + f := newPreloadFixture(t) + f.m.preloadPulls = make(chan struct{}, 1) + var inFlight, most atomic.Int32 + hook := func() { + n := inFlight.Add(1) + for m := most.Load(); n > m && !most.CompareAndSwap(m, n); m = most.Load() { + } + time.Sleep(20 * time.Millisecond) + inFlight.Add(-1) + } + var conns []*workerConn + for i := 0; i < 3; i++ { + d := newFakeImageDaemon(t) + d.mu.Lock() + d.onPull = hook + d.mu.Unlock() + conns = append(conns, d.conn(t, f.m)) + } + var wg sync.WaitGroup + for _, w := range conns { + wg.Add(1) + go func(w *workerConn) { defer wg.Done(); f.m.preloadPass(w) }(w) + } + wg.Wait() + if got := most.Load(); got != 1 { + t.Errorf("%d preload pulls ran at once with a cap of 1", got) + } +} + +// A pull that waited for a fleet slot goes ahead only if its worker is still +// placeable once it has one. +func TestAPreloadPullChecksItsWorkerAfterTheSlotWait(t *testing.T) { + f := newPreloadFixture(t) + f.m.preloadPulls = make(chan struct{}, 1) + f.m.preloadPulls <- struct{}{} // another worker's pull + d := newFakeImageDaemon(t) + w := d.conn(t, f.m) + done := make(chan struct{}) + go func() { f.m.preloadPass(w); close(done) }() + eventually(t, time.Second, "the pass to list the worker", func() bool { + d.mu.Lock() + defer d.mu.Unlock() + return len(d.requests) > 0 + }) + time.Sleep(20 * time.Millisecond) // into the slot wait + w.reachable.Store(int32(workerUnresponsive)) + <-f.m.preloadPulls + <-done + if pulls, _ := d.calls(); len(pulls) > 0 { + t.Errorf("pulled %v onto a worker that left placement during the slot wait", pulls) + } +} + +// preload-list shows a pull under the schema it is for, not under every one. +func TestPreloadStatusShowsOnlyItsSchemasPull(t *testing.T) { + var s preloadStatus + s.begin(map[string]bool{}) + s.pull("r/other:1") + if got := s.report("w", workerReachableOk, []string{"r/web:1"}); got.Pulling != "" { + t.Errorf("a schema not pulling anything shows %q", got.Pulling) + } + if got := s.report("w", workerReachableOk, []string{"r/other:1"}); got.Pulling != "r/other:1" { + t.Errorf("the schema being pulled for shows %q", got.Pulling) + } +} + +// Startup registers its workers under the lock: each conn's goroutine starts +// as it is made, and may read the map. +func TestInitWorkersRegistersUnderTheLock(t *testing.T) { + m := setupTestManager(t) + t.Cleanup(func() { m.db.Close() }) + m.ctx = context.Background() + m.workerTiming = probeTiming() + for i := 1; i <= 50; i++ { + m.db.MustExec("INSERT INTO workers(ip, public) VALUES (?, '');", fmt.Sprintf("127.0.0.%d", i)) + } + stop := make(chan struct{}) + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case <-stop: + return + default: + m.kickPreloads() // iterates the workers, as a build landing would + } + } + }() + if err := m.initWorkers(); err != nil { + t.Fatalf("initWorkers: %s", err) + } + close(stop) + wg.Wait() + m.workersMu.Lock() + for _, w := range m.workers { + close(w.done) + } + m.workersMu.Unlock() +} + +// A build's new generation goes out as it lands, while preloading is in play, +// rather than when the operation it belongs to ends. +func TestABuildLandingKicksTheWorkers(t *testing.T) { + f := newPreloadFixture(t) + d := newFakeImageDaemon(t) + w := d.conn(t, f.m) + build, err := f.m.lookupBuildMetadata(f.web) + if err != nil { + t.Fatalf("lookupBuildMetadata: %s", err) + } + build.Checksum++ + if err := f.m.finalizeBuild(build); err != nil { + t.Fatalf("finalizeBuild: %s", err) + } + if len(w.preloadNow) != 1 { + t.Error("a preloaded build landed without kicking the workers") + } + <-w.preloadNow + if _, err := f.m.RemovePreload("event"); err != nil { + t.Fatalf("RemovePreload: %s", err) + } + build.Checksum++ + if err := f.m.finalizeBuild(build); err != nil { + t.Fatalf("finalizeBuild: %s", err) + } + if len(w.preloadNow) != 0 { + t.Error("a build landed on a fleet preloading nothing and kicked the workers") + } +} + +func TestStaleTag(t *testing.T) { + ref := preloadTestRegistry + "/test/web:s1-aaaa-challenge" + stale := staleTag(ref) + if want := preloadTestRegistry + "/test/web:stale-s1-aaaa-challenge"; stale != want { + t.Errorf("staleTag(%q) = %q, want %q", ref, stale, want) + } + if isStale(ref) || !isStale(stale) { + t.Errorf("isStale: %q = %v, %q = %v", ref, isStale(ref), stale, isStale(stale)) + } +} + +// A superseded image a container still uses is not forced off: dockerd refuses +// the unforced delete, and it is kept under its stale tag instead, which a +// later pass deletes once nothing uses it. +func TestAnImageInUseIsKeptAsStale(t *testing.T) { + f := newPreloadFixture(t) + old := preloadTestRegistry + "/test/web:s1-aaaa-challenge" + stale := staleTag(old) + d := newFakeImageDaemon(t, + image.Summary{ID: "web", RepoTags: []string{f.webName}}, + image.Summary{ID: "old-web", RepoTags: []string{old}}, + ) + d.inUse["old-web"] = true + w := d.conn(t, f.m) + + f.m.removeSuperseded(w, d.tags(), nil) + if got, want := d.tags(), map[string]bool{f.webName: true, stale: true}; !reflect.DeepEqual(got, want) { + t.Fatalf("tags after the refused delete = %v, want %v", got, want) + } + f.m.removeSuperseded(w, d.tags(), nil) // still in use + if got, want := d.tags(), map[string]bool{f.webName: true, stale: true}; !reflect.DeepEqual(got, want) { + t.Errorf("tags while still in use = %v, want %v", got, want) + } + d.mu.Lock() + tagged := slices.Clone(d.tagged) + d.inUse["old-web"] = false + d.mu.Unlock() + if len(tagged) != 1 { + t.Errorf("tagged %v, want the one stale tag", tagged) + } + f.m.removeSuperseded(w, d.tags(), nil) + if got, want := d.tags(), map[string]bool{f.webName: true}; !reflect.DeepEqual(got, want) { + t.Errorf("tags once nothing uses the image = %v, want %v", got, want) + } + _, removes := d.calls() + for _, r := range removes { + if strings.HasSuffix(r, "force=1") { + t.Errorf("forced a removal: %s", r) + } + } +} + +// A pass that fails goes again soon, and less often while it keeps failing. +func TestAFailedPassIsRetriedWithBackoff(t *testing.T) { + f := newPreloadFixture(t) + retry := preloadRetryAfter + preloadRetryAfter = 10 * time.Millisecond + t.Cleanup(func() { preloadRetryAfter = retry }) + d := newFakeImageDaemon(t) + d.mu.Lock() + d.pullFails = true + d.mu.Unlock() + w := d.conn(t, f.m) + go f.m.preloadLoop(w) + time.Sleep(400 * time.Millisecond) + pulls := 0 + d.mu.Lock() + for _, r := range d.requests { + if strings.HasSuffix(r, "/images/create") { + pulls++ + } + } + d.mu.Unlock() + // Retries at 10, 20, 40, 80, 160ms make six attempts in 400ms; every + // 10ms would make forty. + if pulls < 2 || pulls > 12 { + t.Errorf("%d pull attempts in 400ms, want a retry soon that backs off", pulls) + } +} + +// A pass that could not list the worker counts as failed, and is retried. +func TestAPassThatCannotListTheWorkerFails(t *testing.T) { + f := newPreloadFixture(t) + d := newFakeImageDaemon(t) + d.mu.Lock() + d.listFails = true + d.mu.Unlock() + if !f.m.preloadPass(d.conn(t, f.m)) { + t.Error("a pass that could not list the worker reported nothing to retry") + } +} + +// preload-remove reads what it covered itself: a flag that missed a preload +// (an earlier failed read) must not turn its answer into "nothing released". +func TestRemovePreloadDoesNotTrustThePreloadingFlag(t *testing.T) { + f := newPreloadFixture(t) + f.m.preloading.Store(false) + got, err := f.m.RemovePreload("event") + if err != nil { + t.Fatalf("RemovePreload: %s", err) + } + if want := []string{f.webName}; !reflect.DeepEqual(got, want) { + t.Errorf("released %v, want %v", got, want) + } +} + +// preload-add kicks the workers, as each landing build does; other operations +// leave what they change to the sweep. +func TestOnlyAPreloadAddKicks(t *testing.T) { + f := newPreloadFixture(t) + d := newFakeImageDaemon(t) + w := d.conn(t, f.m) + f.m.lockUpdates() + f.m.unlockUpdates() + if len(w.preloadNow) != 0 { + t.Error("an operation that landed no build kicked the worker") + } + if err := f.m.AddPreload("nonesuch"); err == nil { + t.Fatal("AddPreload of an unknown schema succeeded") + } + if len(w.preloadNow) != 0 { + t.Error("a failed preload-add kicked the worker") + } + if err := f.m.AddPreload("archive"); err != nil { + t.Fatalf("AddPreload: %s", err) + } + if len(w.preloadNow) != 1 { + t.Error("preload-add did not kick the worker") + } +} + +// A version that lands while a pass runs is pulled by the next, and the old +// ones wait for it there, rather than going first and taking the layers it +// shares with them. +func TestAnOldImageWaitsForASuccessorThatLandsMidPass(t *testing.T) { + f := newPreloadFixture(t) + old := preloadTestRegistry + "/test/web:s1-aaaa-challenge" + d := newFakeImageDaemon(t, image.Summary{ID: "old-web", RepoTags: []string{old}}) + w := d.conn(t, f.m) + var once sync.Once + d.mu.Lock() + d.onPull = func() { + once.Do(func() { + build, err := f.m.lookupBuildMetadata(f.web) + if err != nil { + t.Errorf("lookupBuildMetadata: %s", err) + return + } + build.Checksum++ + if err := f.m.finalizeBuild(build); err != nil { + t.Errorf("finalizeBuild: %s", err) + } + }) + } + d.mu.Unlock() + f.m.preloadPass(w) + if _, removes := d.calls(); len(removes) > 0 { + t.Errorf("removed %v before the version that landed mid-pass was pulled", removes) + } + build, err := f.m.lookupBuildMetadata(f.web) + if err != nil { + t.Fatalf("lookupBuildMetadata: %s", err) + } + landed := f.m.instanceImageName(build.Challenge, build, Image{Host: "challenge"}) + f.m.preloadPass(w) + pulls, removes := d.calls() + if !slices.Contains(pulls, landed) { + t.Errorf("pulls = %v, want the version that landed, %s", pulls, landed) + } + if want := []string{f.webName + " force=0", old + " force=0"}; !reflect.DeepEqual(removes, want) { + t.Errorf("removes once it is held = %v, want %v", removes, want) + } +} + +// A failed pull does not hold the old version back: on a full disk, deleting +// it is what makes room for the retry. +func TestAFailedPullReleasesTheOldVersion(t *testing.T) { + f := newPreloadFixture(t) + old := preloadTestRegistry + "/test/web:s1-aaaa-challenge" + d := newFakeImageDaemon(t, image.Summary{ID: "old-web", RepoTags: []string{old}}) + d.mu.Lock() + d.pullFails = true + d.mu.Unlock() + f.m.preloadPass(d.conn(t, f.m)) + if _, removes := d.calls(); !reflect.DeepEqual(removes, []string{old + " force=0"}) { + t.Errorf("removes after the new version's pull failed = %v, want %s", removes, old) + } +} + +// A pass pulls nothing that stopped being wanted while it ran: preload-remove +// takes effect on a pass already pulling. +func TestAPassPullsNothingNoLongerWanted(t *testing.T) { + f := newPreloadFixture(t) + more := &BuildMetadata{ + Id: insertTestBuild(t, f.m, "event", "test/web", "flag{%s}", 3, 0x5555), Flag: "flag{x}", + Challenge: "test/web", Seed: 3, Checksum: 0x5555, Images: []Image{{Host: "challenge", Ports: []string{}}}, + } + if err := f.m.finalizeBuild(more); err != nil { + t.Fatalf("finalizeBuild: %s", err) + } + d := newFakeImageDaemon(t) + var once sync.Once + d.mu.Lock() + d.onPull = func() { + once.Do(func() { + if _, err := f.m.RemovePreload("event"); err != nil { + t.Errorf("RemovePreload: %s", err) + } + }) + } + d.mu.Unlock() + f.m.preloadPass(d.conn(t, f.m)) + if pulls, _ := d.calls(); len(pulls) != 1 { + t.Errorf("pulled %v, want only the pull under way when the preload was removed", pulls) + } +} + +// A timeout marking an image stale ends the worker's removals, as a delete's +// does: more calls would only queue behind a slow daemon. +func TestATimeoutMarkingStaleEndsTheRemovals(t *testing.T) { + f := newPreloadFixture(t) + timing := defaultWorkerTiming + timing.controlTimeout = 50 * time.Millisecond + f.m.workerTiming = timing + images := []image.Summary{{ID: "web", RepoTags: []string{f.webName}}} + d := newFakeImageDaemon(t) + for j := 0; j < 3; j++ { + id := fmt.Sprintf("old-%d", j) + images = append(images, image.Summary{ID: id, RepoTags: []string{fmt.Sprintf("%s/test/web:s1-%d-challenge", preloadTestRegistry, j)}}) + d.inUse[id] = true + } + d.mu.Lock() + d.images = images + d.onDelete = func() { time.Sleep(300 * time.Millisecond) } // the untag after the stale tag + d.mu.Unlock() + w := d.conn(t, f.m) + f.m.removeSuperseded(w, d.tags(), nil) + if _, removes := d.calls(); len(removes) != 2 { + t.Errorf("deletes = %v, want the refused one and the untag that timed out", removes) + } +} + +// A worker out of placement is not preloaded, and one that leaves during its +// removals stops them. +func TestAnUnreachableWorkerIsLeftAlone(t *testing.T) { + f := newPreloadFixture(t) + d := newFakeImageDaemon(t) + w := d.conn(t, f.m) + w.reachable.Store(int32(workerUnresponsive)) + f.m.preloadPass(w) + d.mu.Lock() + requests := slices.Clone(d.requests) + d.mu.Unlock() + if len(requests) > 0 { + t.Errorf("a pass on an unresponsive worker called its daemon: %v", requests) + } + + images := []image.Summary{{ID: "web", RepoTags: []string{f.webName}}} + for j := 0; j < 3; j++ { + images = append(images, image.Summary{ID: fmt.Sprintf("old-%d", j), + RepoTags: []string{fmt.Sprintf("%s/test/web:s1-%d-challenge", preloadTestRegistry, j)}}) + } + d.mu.Lock() + d.images = images + d.onDelete = func() { w.reachable.Store(int32(workerUnresponsive)) } + d.mu.Unlock() + w.reachable.Store(int32(workerReachableOk)) + f.m.removeSuperseded(w, d.tags(), nil) + if _, removes := d.calls(); len(removes) != 1 { + t.Errorf("deletes = %v, want the removals to stop once the worker left", removes) + } +} + +// A pass that succeeds after a run of failures puts the retry back to its +// shortest: the next failure is retried soon, not at the backoff reached. +func TestASucceedingPassResetsTheBackoff(t *testing.T) { + f := newPreloadFixture(t) + retry := preloadRetryAfter + preloadRetryAfter = 20 * time.Millisecond + t.Cleanup(func() { preloadRetryAfter = retry }) + d := newFakeImageDaemon(t) + pullsTried := func() int { + d.mu.Lock() + defer d.mu.Unlock() + n := 0 + for _, r := range d.requests { + if strings.HasSuffix(r, "/images/create") { + n++ + } + } + return n + } + d.mu.Lock() + d.pullFails = true + d.mu.Unlock() + w := d.conn(t, f.m) + go f.m.preloadLoop(w) + time.Sleep(700 * time.Millisecond) // retries at 20, 40, 80, 160, 320ms: the next is 640ms off + d.mu.Lock() + d.pullFails = false + d.mu.Unlock() + w.preloadNow <- struct{}{} + eventually(t, 5*time.Second, "the pass after the failures to pull", func() bool { + pulls, _ := d.calls() + return len(pulls) == 1 + }) + time.Sleep(50 * time.Millisecond) // into the wait after the pass + d.mu.Lock() + d.images, d.pullFails = nil, true + d.mu.Unlock() + before := pullsTried() + w.preloadNow <- struct{}{} + time.Sleep(250 * time.Millisecond) + if n := pullsTried() - before; n < 2 { + t.Errorf("%d pull attempt(s) in the 250ms after a new failure, want the retry back at 20ms", n) + } +} + +// A pass's failures are its own: the next pass starts with none. +func TestAPassStartsWithNoFailures(t *testing.T) { + var s preloadStatus + s.begin(map[string]bool{}) + s.pulled("r/web:1", errors.New("manifest unknown")) + s.begin(map[string]bool{}) + if got := s.report("w", workerReachableOk, []string{"r/web:1"}); len(got.Failed) > 0 { + t.Errorf("a new pass reports %v, an earlier pass's failure", got.Failed) + } +} + +// A worker admitted to placement gets its preloader. +func TestAnAdmittedWorkerPreloads(t *testing.T) { + f := newPreloadFixture(t) + f.m.workerTiming = probeTiming() + d := newFakeImageDaemon(t) + w := d.conn(t, f.m) + w.reachable.Store(int32(workerUnresponsive)) // as newWorkerConn leaves it + w.reconciled.Store(false) + go f.m.runWorker(w) + eventually(t, 5*time.Second, "the admitted worker to pull what is preloaded", func() bool { + pulls, _ := d.calls() + return slices.Contains(pulls, f.webName) + }) +} + +// A pass that cannot read the database pulls nothing on a guess: a read that +// fails before a pull, after the wait for its slot, ends the pass as failed. +func TestAPassPullsNothingItCouldNotRead(t *testing.T) { + f := newPreloadFixture(t) + f.m.preloadPulls = make(chan struct{}, 1) + f.m.preloadPulls <- struct{}{} // another worker's pull + d := newFakeImageDaemon(t) + w := d.conn(t, f.m) + failed := make(chan bool) + go func() { failed <- f.m.preloadPass(w) }() + eventually(t, time.Second, "the pass to list the worker", func() bool { + d.mu.Lock() + defer d.mu.Unlock() + return len(d.requests) > 0 + }) + time.Sleep(20 * time.Millisecond) // into the slot wait + f.m.db.MustExec("DROP TABLE preloads;") + <-f.m.preloadPulls + if !<-failed { + t.Error("a pass that could not read what to pull reported nothing to retry") + } + if pulls, _ := d.calls(); len(pulls) > 0 { + t.Errorf("pulled %v without reading whether it was still wanted", pulls) + } +} + +// A pass whose removals cannot read the database is retried, as one whose +// pulls could not is. +func TestAPassThatCannotReadWhatItSupersedesFails(t *testing.T) { + f := newPreloadFixture(t) + d := newFakeImageDaemon(t, image.Summary{ID: "web", RepoTags: []string{f.webName}}) + d.mu.Lock() + d.onList = func() { f.m.db.MustExec("DROP TABLE preloads;") } + d.mu.Unlock() + if !f.m.preloadPass(d.conn(t, f.m)) { + t.Error("a pass whose removals could not read the database reported nothing to retry") + } +} diff --git a/cork/types.go b/cork/types.go index ba2ab76..2c8432c 100644 --- a/cork/types.go +++ b/cork/types.go @@ -27,9 +27,11 @@ const ( // where it is read; it is a constant so the list of what a build plane // ignores can name it (orchestratorOnlySettings). CONCURRENT_LAUNCHES_ENV string = "CORK_CONCURRENT_LAUNCHES" - DISK_QUOTA_ENV string = "CORK_ENABLE_DISK_QUOTAS" - PRUNE_AGE_ENV string = "CORK_PRUNE_AGE" - DB_WAL_ENV string = "CORK_DB_WAL" + // PRELOAD_PULLS_ENV caps preload pulls across the fleet (see preload.go). + PRELOAD_PULLS_ENV string = "CORK_PRELOAD_PULLS" + DISK_QUOTA_ENV string = "CORK_ENABLE_DISK_QUOTAS" + PRUNE_AGE_ENV string = "CORK_PRUNE_AGE" + DB_WAL_ENV string = "CORK_DB_WAL" // Worker tunables (see workerTiming in workers.go). WORKER_POLL_INTERVAL_ENV string = "CORK_WORKER_POLL_INTERVAL" @@ -77,8 +79,18 @@ type Manager struct { // updateMu serializes rebuilds (UpdateWithOptions) and schema operations // (CreateSchema, UpdateSchema, DeleteSchema) against each other: any two // of them would tear down and relaunch the same instances against each - // other. Taken at the API boundary only; internal helpers assume it held. + // other. Taken at the API boundary only, through lockUpdates (preload.go); + // internal helpers assume it held. updateMu sync.Mutex + // preloadPulls holds the fleet's preload pull slots (acquirePreloadPull), and + // updateBefore what preloading covered when the running operation began + // (lockUpdates), for what it releases; the latter is the updateMu holder's. + preloadPulls chan struct{} + // preloading is whether any schema is preloaded, kept by unlockUpdates and + // read at startup, so that on a fleet that preloads nothing a landing build + // kicks no one and an operation takes no preload snapshot. + preloading atomic.Bool + updateBefore *preloadSets // imageMu serializes the "is this content still referenced? if not, untag" // critical sections (executeBuild cleanup, pruneReplacedImages, // destroyImages) so a concurrent remover cannot delete a tag between diff --git a/cork/workers.go b/cork/workers.go index f5f684e..0f2d0c4 100644 --- a/cork/workers.go +++ b/cork/workers.go @@ -369,6 +369,10 @@ type workerConn struct { // control call has failed. Buffered and sent to without blocking, so a // burst of failing launches costs one extra probe, not one each. probeNow chan struct{} + // preloadNow asks the preloader for a pass (kickPreloads), the same way; + // preload is what its passes found (see preload.go). + preloadNow chan struct{} + preload preloadStatus } type WorkerInfo struct { @@ -420,6 +424,10 @@ func (m *Manager) EnableWorkerPlacement() { // per worker. Unreachable workers are marked down by their pollers and skipped // by placement — never a startup failure. func (m *Manager) initWorkers() error { + // Under the lock, as in AddWorker: each conn's goroutine starts inside + // newWorkerConn and may read the map before the loop below is through. + m.workersMu.Lock() + defer m.workersMu.Unlock() m.workers = make(map[string]*workerConn) var rows []struct { @@ -503,13 +511,14 @@ func (m *Manager) newWorkerConn(ip, public string, fromRecovery bool) (*workerCo } w := &workerConn{ - ip: ip, - public: public, - cli: cli, - queue: newDaemonQueue(slots(m.launchConcurrency)), - unreachCh: make(chan struct{}), - done: make(chan struct{}), - probeNow: make(chan struct{}, 1), + ip: ip, + public: public, + cli: cli, + queue: newDaemonQueue(slots(m.launchConcurrency)), + unreachCh: make(chan struct{}), + done: make(chan struct{}), + probeNow: make(chan struct{}, 1), + preloadNow: make(chan struct{}, 1), } // Fail closed for placement until the box has proved itself: unresponsive // with its load unknown, which is what is actually true of a worker @@ -545,6 +554,9 @@ func (m *Manager) runWorker(w *workerConn) { // Its load stays unknown until telemetry answers, which does not hold // it back: unknown places. m.setReachable(w, workerReachableOk, "") + // Alongside placement, not before it: a worker added under load is + // needed now, and its launches pull as they always have meanwhile. + go m.preloadLoop(w) m.pollWorker(w) case reconcileEjected: // The daemon never answered. Poll anyway: the worker is out of From 3ffd2dd6cef46c72bce984db690c5544eedbbe8e Mon Sep 17 00:00:00 2001 From: Ivan Liang Date: Fri, 16 Oct 2026 15:15:53 -0400 Subject: [PATCH 3/4] feat(cork): preload-add, preload-remove and preload-list POST /preloads {"schema": ""}, DELETE /preloads/ and GET /preloads put the preloading of the previous commit in corkd's API, and cork preload-add, preload-remove and preload-list in front of it. An unknown schema is a 404, a daemon without a challenge registry a 409. preload-add answers whether any worker is registered, and the CLI says when none is: the preload then waits for workers to be added, which is otherwise indistinguishable from it working. preload-list shows, per schema and worker, how many of its images the worker was last seen to hold, the pull in progress for that schema, failed pulls, when the worker was last checked, and "ready" once it holds them all and has been checked ("no images" for a schema left with no builds). preload-remove answers with the images corkd no longer cleans up ({"released": [...]}), and the CLI lists them as images that may remain on the workers, since they come from the database, not from asking every worker. The README and both usage texts describe the two modes by what selects them -- whether a schema's images fit on a worker -- rather than by any one use, and what each leaves to whom: on demand, corkd removes no images and docker-reaper evicts; preloaded, cleanup is corkd's while the preload lasts, an old version deleted after any successor that landed meanwhile is pulled, never forced, with an image still in use kept under a stale tag, and what leaves preloading may stay for an operator. They also say what preloaded workers need (room for the set plus a second version of what one update redeploys, docker-reaper's image eviction off, and back on after preload-remove), that a worker serves without pulling only once it shows ready, how to list stale tags (a reference filter's * stops at a slash, so it is a grep), and CORK_PRELOAD_PULLS. Tests pin the handler's status codes and when preload-list says ready. --- README.md | 81 ++++++++++++++++++++++- cmd/cork/main.go | 31 +++++++++ cmd/cork/preloads.go | 116 +++++++++++++++++++++++++++++++++ cmd/cork/preloads_test.go | 58 +++++++++++++++++ cmd/corkd/main.go | 39 +++++++++++ cmd/corkd/preloads_api.go | 86 ++++++++++++++++++++++++ cmd/corkd/preloads_api_test.go | 29 +++++++++ 7 files changed, 439 insertions(+), 1 deletion(-) create mode 100644 cmd/cork/preloads.go create mode 100644 cmd/cork/preloads_test.go create mode 100644 cmd/corkd/preloads_api.go create mode 100644 cmd/corkd/preloads_api_test.go diff --git a/README.md b/README.md index 900022e..cc5e41f 100644 --- a/README.md +++ b/README.md @@ -300,6 +300,31 @@ Workers: show every worker, its two health states, how long it has been in the reachable one and why, and how many instances it holds +Preloads (when a schema's images fit on every worker, pull them ahead of its launches): + preload-add + pull every image the schema launches onto every worker, including + workers added later, without holding any worker out of placement: a + worker serves without pulling once preload-list shows it ready. + Cleanup is then corkd's, not docker-reaper's, for as long as the + schema is preloaded: after its pulls, each worker deletes the images + of its challenges that no current build launches, though not before + pulling a version that landed meanwhile. An image a container still + uses (an instance the update has yet to stop, a launch it overtook) + is tagged :stale- instead, for a later pass to delete + once nothing uses it. The workers' image store + has to hold the whole set plus a second version of what one update + redeploys, with docker-reaper's image eviction off. What the preload + stops covering -- the schema removed with remove-schema, a challenge + dropped from it -- may stay on the workers unless another preloaded + schema shares the challenge, and corkd's log lists those images + preload-remove + go back to pulling on demand: the workers pull no more of the + schema's images, the ones already pulled stay, and the ones that may + remain uncleaned are listed: turn docker-reaper's image eviction back + on, or remove them by hand + preload-list + each preloaded schema, and how many of its images each worker holds + Base image pins: pin-list show the pinned base images and how many challenges use each @@ -387,6 +412,59 @@ changes. Either way, your challenges are then built and ready to run. Send HTTP requests for start/stop with the appropriate build IDs and watch them run. +Workers hold challenge images in one of two modes, chosen per schema by +whether its images fit on a worker: + +- **On demand**, the default. A worker pulls an image the first time it + launches it, so the set being served never has to fit on one box. This is the + mode for a large or growing catalogue; its cost is a pull on each worker's + first launch of each image. Cleanup is out of band: corkd removes no images + from a worker, and docker-reaper evicts what has gone unused. +- **Preloaded**. When a schema's images fit on every worker, + `cork preload-add ` has every worker pull all of them ahead of time, + and `preload-list` shows when each worker holds them. A worker that shows + `ready` serves the schema's launches without pulling. Until then launches on + it pull as they would on demand: a worker just added takes launches while it + catches up, and the new version of a redeployed challenge is pulled by each + worker once its build lands, after any pulls the worker already has under + way. + + Cleanup is corkd's own, not docker-reaper's or an operator's, for as long as + a challenge is preloaded. After its pulls, each worker deletes the images of + it that no current build launches, such as the version a redeploy replaced + or a seed the schema dropped; a version that lands while the worker pulls is + pulled before its predecessor goes. The delete is never forced. A redeploy stops the old version's instances, or restarts them + on the new one; an image a container still uses -- an instance the redeploy + has yet to reach, a launch it overtook -- is tagged `:stale-` + instead, and a later pass deletes it once nothing uses it. A redeployed + challenge mostly shares its layers with the version it replaces, so these + deletes are small; while one runs, the worker's container creates and + removes, and its image checks and lists, wait for it. + + A preload ends with `preload-remove` or with its schema's deletion; an update + that leaves the schema empty, such as a converge that failed, keeps it for + the schema's return. What leaves preloading is not tracked down: deleting a + preloaded schema, or dropping a challenge from it, may leave its images on + the workers, unless another preloaded schema shares the challenge. corkd logs + the images that may remain, from its database, without asking the workers + what they hold. Stale tags left behind that way are listed by + `docker images --format '{{.Repository}}:{{.Tag}}' | grep ':stale-'`, and the + rare race of a delete with a launch can leave an untagged image, which + `docker image prune` removes. + + Size the workers' image store for the whole set, plus a second version of + whatever one update redeploys (the new version is pulled before the old one + goes), and turn docker-reaper's image eviction off there, since corkd decides + what stays. + +The modes are per schema but eviction is per worker: a worker with eviction off +also keeps every image an on-demand schema has it pull, and an on-demand schema +sharing a challenge with a preloaded one has its old versions of that challenge +removed while the preload lasts. `preload-remove` leaves the schema's images in +place and lists the ones that may remain uncleaned; turn docker-reaper's image +eviction back on for those workers, or remove them by hand, since nothing else +will. + To build or run challenges by hand, use `cork`'s `build`, `start` and `stop` — they act on one orchestrator directly. Note that a manual build belongs to no schema, so it has no destination and no build plane will route @@ -410,6 +488,7 @@ Every setting also answers to its pre-rename `CMGR_` name (`CMGR_DB` for `CORK_D | CORK_DB_WAL | Enable WAL journaling for SQLite | on | | CORK_LOGGING | Log verbosity for `corkd` and `cork-build`: `debug`, `info`, `warn`, `error` or `disabled`. There is no flag for it on `corkd`, so this is the only way to raise an orchestrator's logging; `cork-build --verbose` is the flag form and wins over it. A value that does not parse is warned about and not fatal | info | | CORK_CONCURRENT_LAUNCHES | Launch slots per daemon (1-16), and as many teardown slots: network creation plus container starts, which dockerd serializes internally on either firewall backend; no gain measured past 2 | 2 | +| CORK_PRELOAD_PULLS | How many preload pulls run at once across the fleet (each worker pulls one at a time); see `cork preload-add` | 8 | | CORK_WORKER_POLL_INTERVAL | How often each worker is probed (telemetry and docker on the same tick) | 5s | | CORK_WORKER_POLL_TIMEOUT | Telemetry poll timeout; must be under the probe interval (clamped to half of it otherwise) | 250ms | | CORK_WORKER_PING_TIMEOUT | Docker `/_ping` timeout; clamped the same way. Looser than the telemetry one — dockerd under load answers more slowly than a cached verdict, and there is TLS in the path | 1s | @@ -441,7 +520,7 @@ Every setting also answers to its pre-rename `CMGR_` name (`CMGR_DB` for `CORK_D Everything above is read by the shared `cork` library, so `corkd` and `cork-build` accept the same surface — but each ignores the other's half, and says so at startup rather than silently. A build plane runs nothing, so -`CORK_CONCURRENT_LAUNCHES`, `CORK_PORTS`, `CORK_INTERFACE`, +`CORK_CONCURRENT_LAUNCHES`, `CORK_PRELOAD_PULLS`, `CORK_PORTS`, `CORK_INTERFACE`, `CORK_ENABLE_DISK_QUOTAS`, `CORK_PRUNE_AGE` and the `CORK_WORKER_*` tunables are inert on `cork-build`; an orchestrator on an external build plane builds nothing, so `CORK_DIR`, `CORK_BASE_PINS`, `DOCKER_HOST` and diff --git a/cmd/cork/main.go b/cmd/cork/main.go index 5068aa5..0008953 100644 --- a/cmd/cork/main.go +++ b/cmd/cork/main.go @@ -88,6 +88,12 @@ func main() { exitCode = workerDownCommand(c, cmdArgs) case "worker-list": exitCode = workerListCommand(c, cmdArgs) + case "preload-add": + exitCode = preloadAddCommand(c, cmdArgs) + case "preload-remove": + exitCode = preloadRemoveCommand(c, cmdArgs) + case "preload-list": + exitCode = preloadListCommand(c, cmdArgs) case "pin-list": exitCode = pinListCommand(c, cmdArgs) case "pin-refresh": @@ -182,6 +188,31 @@ Workers: daemon; load, from its telemetry agent), how long it has been in that state, why, and how many instances it holds +Preloads (when a schema's images fit on every worker, pull them ahead of its launches): + preload-add + pull every image the schema launches onto every worker, including + workers added later, without holding any worker out of placement: a + worker serves without pulling once preload-list shows it ready. + Cleanup is then corkd's, not docker-reaper's, for as long as the + schema is preloaded: after its pulls, each worker deletes the images + of its challenges that no current build launches, though not before + pulling a version that landed meanwhile. An image a container still + uses (an instance the update has yet to stop, a launch it overtook) + is tagged :stale- instead, for a later pass to delete + once nothing uses it. The workers' image store + has to hold the whole set plus a second version of what one update + redeploys, with docker-reaper's image eviction off. What the preload + stops covering -- the schema removed with remove-schema, a challenge + dropped from it -- may stay on the workers unless another preloaded + schema shares the challenge, and corkd's log lists those images + preload-remove + go back to pulling on demand: the workers pull no more of the + schema's images, the ones already pulled stay, and the ones that may + remain uncleaned are listed: turn docker-reaper's image eviction back + on, or remove them by hand + preload-list + each preloaded schema, and how many of its images each worker holds + Base image pins: pin-list show the pinned base images and how many challenges use each diff --git a/cmd/cork/preloads.go b/cmd/cork/preloads.go new file mode 100644 index 0000000..467d0a7 --- /dev/null +++ b/cmd/cork/preloads.go @@ -0,0 +1,116 @@ +package main + +import ( + "fmt" + "io" + "net/url" + "os" + "time" +) + +// PreloadInfo mirrors cork.PreloadInfo; cork/preload.go is the source of truth. +type PreloadInfo struct { + Schema string `json:"schema"` + Images int `json:"images"` + Workers []PreloadWorker `json:"workers"` +} + +type PreloadWorker struct { + IP string `json:"ip"` + Reachable string `json:"reachable"` + Held int `json:"held"` + Pulling string `json:"pulling"` + Failed []PreloadFailure `json:"failed"` + LastPass *time.Time `json:"last_pass"` +} + +type PreloadFailure struct { + Image string `json:"image"` + Error string `json:"error"` +} + +func preloadAddCommand(c *client, args []string) int { + if len(args) != 1 { + fmt.Fprintln(os.Stderr, "usage: preload-add ") + return USAGE_ERROR + } + var resp struct { + WorkersConfigured *bool `json:"workers_configured"` + } + if err := c.doJSON("POST", "/preloads", map[string]string{"schema": args[0]}, &resp); err != nil { + return runtimeError(err) + } + if resp.WorkersConfigured != nil && !*resp.WorkersConfigured { + fmt.Println("Preloaded, but no worker is registered: workers pull the set as they are added.") + return NO_ERROR + } + fmt.Println("Preloading on every worker; preload-list shows how far each has got.") + return NO_ERROR +} + +func preloadRemoveCommand(c *client, args []string) int { + if len(args) != 1 { + fmt.Fprintln(os.Stderr, "usage: preload-remove ") + return USAGE_ERROR + } + var resp struct { + Released []string `json:"released"` + } + if err := c.doJSON("DELETE", "/preloads/"+url.PathEscape(args[0]), nil, &resp); err != nil { + return runtimeError(err) + } + printReleased(os.Stdout, resp.Released) + return NO_ERROR +} + +// printReleased lists what corkd no longer cleans up, as it knows from its +// database: no worker was asked what it holds. +func printReleased(out io.Writer, released []string) { + if len(released) == 0 { + fmt.Fprintln(out, "Pulling on demand again; other preloaded schemas still want every image it preloaded.") + return + } + fmt.Fprintf(out, "Pulling on demand again. corkd no longer cleans up these %d image(s), which may remain on the workers, as may older ones of the same repositories, until eviction or an operator removes them:\n", len(released)) + for _, name := range released { + fmt.Fprintf(out, " %s\n", name) + } +} + +func preloadListCommand(c *client, args []string) int { + var infos []PreloadInfo + if err := c.doJSON("GET", "/preloads", nil, &infos); err != nil { + return runtimeError(err) + } + printPreloads(os.Stdout, infos, time.Now()) + return NO_ERROR +} + +func printPreloads(out io.Writer, infos []PreloadInfo, now time.Time) { + if len(infos) == 0 { + fmt.Fprintln(out, "no schema is preloaded") + return + } + for _, info := range infos { + fmt.Fprintf(out, "%s: %d image(s)\n", info.Schema, info.Images) + for _, w := range info.Workers { + progress := fmt.Sprintf("%d/%d", w.Held, info.Images) + state := "" + switch { + case w.Pulling != "": + state = "pulling " + w.Pulling + case info.Images == 0: // a schema left with no builds + state = "no images" + case w.Held == info.Images && w.LastPass != nil: + state = "ready" + } + checked := "not checked yet" + if w.LastPass != nil { + checked = fmt.Sprintf("checked %s ago", now.Sub(*w.LastPass).Round(time.Second)) + } + fmt.Fprintf(out, " %-15s %-12s %9s %-16s %s\n", w.IP, w.Reachable, progress, checked, state) + for _, f := range w.Failed { + fmt.Fprintf(out, " %-15s failed %s: %s\n", "", f.Image, f.Error) + } + } + } +} diff --git a/cmd/cork/preloads_test.go b/cmd/cork/preloads_test.go new file mode 100644 index 0000000..6f5813e --- /dev/null +++ b/cmd/cork/preloads_test.go @@ -0,0 +1,58 @@ +package main + +import ( + "regexp" + "strings" + "testing" + "time" +) + +// preload-list is what an operator watches before launches start: "ready" has +// to mean every image held, on a worker that has actually been checked. +func TestPrintPreloads(t *testing.T) { + now := time.Date(2026, 10, 14, 12, 0, 0, 0, time.UTC) + checked := now.Add(-90 * time.Second) + infos := []PreloadInfo{{ + Schema: "event", + Images: 3, + Workers: []PreloadWorker{ + {IP: "10.0.0.1", Reachable: "ok", Held: 3, LastPass: &checked}, + {IP: "10.0.0.2", Reachable: "ok", Held: 1, Pulling: "r/web:s1", LastPass: &checked, + Failed: []PreloadFailure{{Image: "r/pwn:s1", Error: "manifest unknown"}}}, + {IP: "10.0.0.3", Reachable: "unresponsive"}, + {IP: "10.0.0.4", Reachable: "ok", Held: 3}, // a count from no pass of its own + }, + }} + + var out strings.Builder + printPreloads(&out, infos, now) + lines := strings.Split(strings.TrimRight(out.String(), "\n"), "\n") + if len(lines) != 6 { + t.Fatalf("got %d lines, want 6:\n%s", len(lines), out.String()) + } + for i, want := range []string{ + `^event: 3 image\(s\)$`, + "3/3 .* checked 1m30s ago .* ready$", + "1/3 .* pulling r/web:s1$", + "failed r/pwn:s1: manifest unknown$", + "unresponsive .* 0/3 .* not checked yet", + `10\.0\.0\.4 .* 3/3 .* not checked yet\s*$`, + } { + if !regexp.MustCompile(want).MatchString(lines[i]) { + t.Errorf("line %d = %q, want it to match %q", i, lines[i], want) + } + } + + // A schema left with no builds keeps its preload, and is not "ready". + out.Reset() + printPreloads(&out, []PreloadInfo{{Schema: "emptied", Workers: []PreloadWorker{{IP: "10.0.0.1", Reachable: "ok", LastPass: &checked}}}}, now) + if got := out.String(); strings.Contains(got, "ready") || !strings.Contains(got, "no images") { + t.Errorf("an emptied schema shows:\n%s", got) + } + + out.Reset() + printPreloads(&out, nil, now) + if got := out.String(); got != "no schema is preloaded\n" { + t.Errorf("with nothing preloaded: %q", got) + } +} diff --git a/cmd/corkd/main.go b/cmd/corkd/main.go index 81181a7..a7bd93e 100644 --- a/cmd/corkd/main.go +++ b/cmd/corkd/main.go @@ -120,6 +120,8 @@ func main() { http.HandleFunc("/instances/", s.instanceHandler) http.HandleFunc("/workers", s.workersHandler) http.HandleFunc("/workers/", s.workerHandler) + http.HandleFunc("/preloads", s.preloadsHandler) + http.HandleFunc("/preloads/", s.preloadHandler) http.HandleFunc("/pins", s.pinsHandler) http.HandleFunc("/schemas", s.schemaHandler) http.HandleFunc("/schemas/", s.existingSchemaHandler) @@ -437,6 +439,43 @@ Workers: variables. See https://docs.docker.com/engine/reference/commandline/cli/ for specific details. +Preloads: + Workers hold challenge images in one of two modes, chosen per schema by + whether its images fit on a worker. On demand, the default, a worker pulls + an image the first time it launches it, so the set being served never has + to fit on one box, and cleanup is out of band: corkd removes no images, and + docker-reaper evicts what goes unused. A preloaded schema (POST /preloads + {"schema": ""}, or cork preload-add) instead has every image it + launches pulled onto every worker ahead of its launches, including workers + added later. Each worker's preload pulls one image at a time, in its own + order, and at most CORK_PRELOAD_PULLS (default 8) run at once across the + fleet, without holding back placement; launches still pull their own. A + worker checks again when a schema is preloaded, as each build lands, and + every five minutes; once GET /preloads shows it holding every image, its + launches of the schema do not wait on the registry. Until then they pull + as on demand: a worker just added catches up while it serves, and a + redeployed challenge's new version is pulled by each worker once its build + lands, after any pulls already under way there. Its workers need an image + store that holds the whole set plus a second version of what one update + redeploys, with docker-reaper's image eviction off. + + A preloaded schema's cleanup is corkd's own. While a challenge is preloaded, + each worker deletes, after its pulls, the images of it that no current + build launches, pulling first any version that landed meanwhile. The + delete is never forced: an image a container still uses, such as an + instance the redeploy has yet to stop or a launch it overtook, is tagged + :stale- instead and deleted by a later pass once nothing uses + it. A worker's removals stop at the first that times + out, and a preload call that times out counts against nothing. A preload + ends with DELETE /preloads/ or with its schema's deletion. What an + operation takes out of preloading -- the schema deleted, a challenge + dropped from it -- may stay on the workers unless another preloaded schema + shares the challenge, and corkd logs the images that may remain, read from + its database rather than from the workers. DELETE /preloads/ returns + the schema to pulling on demand, so the workers pull no more of it, leaves + its images in place, and answers with the ones that may remain uncleaned: + {"released": [, ...]}. + `, os.Args[0]) } diff --git a/cmd/corkd/preloads_api.go b/cmd/corkd/preloads_api.go new file mode 100644 index 0000000..97998fe --- /dev/null +++ b/cmd/corkd/preloads_api.go @@ -0,0 +1,86 @@ +package main + +import ( + "encoding/json" + "errors" + "net/http" + "strings" + + "github.com/CyLabAcademy/challenge-orchestrator/cork" +) + +type PreloadAddRequest struct { + Schema string `json:"schema"` +} + +// preloadsHandler serves the preloaded schemas: GET lists them with each +// worker's progress, POST {"schema": ""} preloads one on every worker. +func (s state) preloadsHandler(w http.ResponseWriter, r *http.Request) { + switch r.Method { + case "GET": + infos, err := s.mgr.ListPreloads() + if err != nil { + preloadError(w, err) + return + } + body, err := json.Marshal(infos) + if err != nil { + preloadError(w, err) + return + } + w.Write(body) + case "POST": + var req PreloadAddRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil || req.Schema == "" { + w.WriteHeader(http.StatusBadRequest) + w.Write([]byte(`invalid request body: want {"schema": ""}`)) + return + } + if err := s.mgr.AddPreload(req.Schema); err != nil { + preloadError(w, err) + return + } + // Said, since with no worker registered there is nothing to preload yet. + body, _ := json.Marshal(map[string]bool{"workers_configured": s.mgr.WorkersConfigured()}) + w.WriteHeader(http.StatusCreated) + w.Write(body) + default: + w.WriteHeader(http.StatusMethodNotAllowed) + } +} + +// preloadHandler serves DELETE /preloads/, which returns the schema to +// pulling on demand, leaves its images where they are, and answers with those +// that leave corkd's cleanup and may remain: {"released": [, ...]}. +func (s state) preloadHandler(w http.ResponseWriter, r *http.Request) { + if r.Method != "DELETE" { + w.WriteHeader(http.StatusMethodNotAllowed) + return + } + schema := strings.TrimPrefix(r.URL.Path, "/preloads/") + if schema == "" || strings.Contains(schema, "/") { + w.WriteHeader(http.StatusNotFound) + return + } + released, err := s.mgr.RemovePreload(schema) + if err != nil { + preloadError(w, err) + return + } + if released == nil { + released = []string{} + } + body, _ := json.Marshal(map[string][]string{"released": released}) + w.Write(body) +} + +func preloadError(w http.ResponseWriter, err error) { + code := http.StatusInternalServerError + if _, ok := err.(*cork.UnknownIdentifierError); ok { + code = http.StatusNotFound + } else if errors.Is(err, cork.ErrPreloadNeedsRegistry) { + code = http.StatusConflict + } + w.WriteHeader(code) + w.Write([]byte(err.Error())) +} diff --git a/cmd/corkd/preloads_api_test.go b/cmd/corkd/preloads_api_test.go new file mode 100644 index 0000000..cac3593 --- /dev/null +++ b/cmd/corkd/preloads_api_test.go @@ -0,0 +1,29 @@ +package main + +import ( + "errors" + "net/http" + "net/http/httptest" + "testing" + + "github.com/CyLabAcademy/challenge-orchestrator/cork" +) + +// cork preload-add and preload-remove tell an operator's typo (404) from a +// corkd that cannot preload at all (409) by the code. +func TestPreloadError(t *testing.T) { + for _, tc := range []struct { + err error + want int + }{ + {&cork.UnknownIdentifierError{Type: "preload", Name: "event"}, http.StatusNotFound}, + {cork.ErrPreloadNeedsRegistry, http.StatusConflict}, + {errors.New("database is locked"), http.StatusInternalServerError}, + } { + rec := httptest.NewRecorder() + preloadError(rec, tc.err) + if rec.Code != tc.want { + t.Errorf("%v answered %d, want %d", tc.err, rec.Code, tc.want) + } + } +} From e54e8f3181ed8b146dcb7695f3724f76f3b0391a Mon Sep 17 00:00:00 2001 From: Ivan Liang Date: Fri, 16 Oct 2026 15:15:53 -0400 Subject: [PATCH 4/4] test(e2e): preload the schema onto both workers and follow a redeploy A full-mode step after generation 5, against real dockerd and zot, which is where the image names a worker lists and the unforced removals have to match what cork computes, and where dockerd's refusal to delete an image in use is real rather than the unit tests' fake. It takes binex101's image off both workers first, so the wait that follows cannot pass on a fleet that already held everything; then preload-add the e2e schema and waits for GET /preloads to report both workers holding all three images the schema launches (the flag-only challenge launches none), checking each daemon holds them too and that preload-list shows both ready. A container corkd does not manage then holds generation 5 of the on-demand challenge on one worker, and the step redeploys it: both workers must come to hold the new tag and drop generation 5's, the one with the container keeping the image as :stale- (the other may too, when its pass beats the rebuild's stop of its own instance), while zot keeps generation 5 as the rollback target and the other challenges' images stay. Once the container is removed, preload-add again (idempotent; it asks for a pass) must see the stale tag deleted from both. preload-remove hands the schema back and must list every image it leaves, and the step leaves tracked instances on both workers as generation 5 did. The removal of the second schema, also full mode, now preloads it first and waits for both workers to hold its two runtime images. After remove-schema they must still be on both, since what leaves preloading is an operator's to vet, and the step removes them by hand. The redeploy uses a generation number of its own (50): the base-pins step builds generation 6 next and fails if its source change is already in place, which is how the first run of this step found out. --- e2e/README.md | 6 +- e2e/run.sh | 3 +- e2e/scenario.sh | 184 +++++++++++++++++++++++++++++++++++++++++++++++- 3 files changed, 186 insertions(+), 7 deletions(-) diff --git a/e2e/README.md b/e2e/README.md index 100cdf3..e1e707c 100644 --- a/e2e/README.md +++ b/e2e/README.md @@ -63,9 +63,9 @@ docker compose run --rm e2e # the scenario `./run.sh` runs the regular mode; `./run.sh --full` (or `E2E_FULL=1`) adds the steps that are only worth their seconds when nobody is waiting: the registry's failure paths, artifact delivery, multi-container challenges, cold pulls, -docker-reaper and container resource limits. A regular run names the full-mode -steps it did not take, and its verdict says which mode it was, so the two are -never confused for one another. +preloading, docker-reaper and container resource limits. A regular run names +the full-mode steps it did not take, and its verdict says which mode it was, so +the two are never confused for one another. Full mode also converges a **second schema**, `schema-full.yaml`, alongside the first. It holds one challenge, the multi-container `aptitude-and-privileges`, diff --git a/e2e/run.sh b/e2e/run.sh index a2d1f92..59bdfb7 100755 --- a/e2e/run.sh +++ b/e2e/run.sh @@ -3,7 +3,8 @@ # # --full adds the steps that are only worth their seconds when nobody is # waiting for the answer: the registry's failure paths, artifact delivery, -# multi-container challenges, cold pulls, docker-reaper and resource limits. +# multi-container challenges, cold pulls, preloading, docker-reaper and +# resource limits. set -euo pipefail cd "$(dirname "$0")" if [[ "${1:-}" == "--full" ]]; then diff --git a/e2e/scenario.sh b/e2e/scenario.sh index 4975327..0361c8e 100755 --- a/e2e/scenario.sh +++ b/e2e/scenario.sh @@ -227,6 +227,11 @@ declare -A FAKES=() # fake telemetry container id -> the sidecar it replaced BLOCKER_NAME=e2e-reconcile-blocker BLOCKER_NET=cmgr-997 BLOCKER_WORKER="" # worker ip currently holding that blocker, if any +# The preload step holds a superseded generation on a worker the same way, so +# that corkd's delete of it is refused and the image is tagged stale. +HOLDER_NAME=e2e-preload-holder +HOLDER_WORKER="" # worker ip currently holding it, if any +HOLDER_STALE="" # the stale tag it leaves there # Full-mode state. A step plants an orphan corkd is expected to reclaim later # in the same step; until it does, the run owes the worker its removal, since a @@ -259,7 +264,7 @@ MULTI_INST="" # its multi-container instance, if the run dies between la DB_LOCK_PID="" # a sqlite3 holding corkd's write lock open on purpose DB_LOCK_FD="" # the write end of the fifo feeding it, kept open by this shell cleanup() { - local status=$? fake + local status=$? fake ip # Before anything else: while the stand-in writer holds corkd's write # lock, every API call below would be refused as a busy database. if [[ -n "$DB_LOCK_FD" ]]; then exec {DB_LOCK_FD}>&- 2>/dev/null || true; DB_LOCK_FD=""; fi @@ -294,6 +299,12 @@ cleanup() { worker_status "$BLOCKER_WORKER" "/containers/$BLOCKER_NAME?force=true" -X DELETE >/dev/null 2>&1 || true worker_status "$BLOCKER_WORKER" "/networks/$BLOCKER_NET" -X DELETE >/dev/null 2>&1 || true fi + if [[ -n "$HOLDER_WORKER" ]]; then + worker_status "$HOLDER_WORKER" "/containers/$HOLDER_NAME?force=true" -X DELETE >/dev/null 2>&1 || true + for ip in "${WORKER_IPS[@]}"; do + worker_status "$ip" "/images/$HOLDER_STALE" -X DELETE >/dev/null 2>&1 || true + done + fi if [[ -n "$ORPHAN_WORKER" ]]; then # Container first, as for the blocker above: while it runs, its network # refuses removal. @@ -4109,6 +4120,146 @@ else deselect "update, generation 5 without --prune-old" fi +if (( FULL )); then +step "preload: every worker holds every image the schema launches before any launch asks for one; a redeploy is pulled onto both and the generation it replaced is removed from both, kept as stale where a container still uses it until nothing does; preload-remove lists what it leaves" +# What the schema launches: each service build's non-builder images. +preload_want() { + local challenge seed checksum host + while read -r challenge seed checksum host; do + printf '%s\n' "$E2E_REGISTRY/$challenge:$(printf 's%d-%x-%s' "$seed" "$checksum" "$host")" + done < <(api GET "/schemas/$SCHEMA_NAME" | jq -r '.[] | select(.delivery_type == "service") | .id as $c + | .builds[] | . as $b | .images[] | select(.host != "builder") | "\($c) \($b.seed) \($b.checksum) \(.host)"') +} +mapfile -t PRELOAD_WANT < <(preload_want) +(( ${#PRELOAD_WANT[@]} > 0 )) || fail "schema $SCHEMA_NAME launches no images, so there is nothing to preload" +# The negative control: without it a fleet that already held everything would +# pass the wait below without the preload pulling a thing. Nothing runs +# $CH_MAKE at this point, so its image can go from both boxes. +PRELOAD_COLD="" +for ref in "${PRELOAD_WANT[@]}"; do + if [[ "$ref" == "$E2E_REGISTRY/$CH_MAKE:"* ]]; then PRELOAD_COLD=$ref; fi +done +[[ -n "$PRELOAD_COLD" ]] || fail "schema $SCHEMA_NAME launches no image of $CH_MAKE to take off the workers" +for ip in "${WORKER_IPS[@]}"; do + code=$(worker_status "$ip" "/images/$PRELOAD_COLD" -X DELETE) + case "$code" in + 200|404) ;; + 409) fail "worker $ip refuses to delete $PRELOAD_COLD because a container still uses it: this step expects nothing to run $CH_MAKE" ;; + *) fail "deleting $PRELOAD_COLD from worker $ip answered HTTP $code" ;; + esac + if has_line "$PRELOAD_COLD" worker_tags "$ip"; then fail "worker $ip still holds $PRELOAD_COLD after the delete"; fi +done +note "took $PRELOAD_COLD off both workers" +out=$(cork preload-add "$SCHEMA_NAME") || fail "cork preload-add $SCHEMA_NAME failed: $out" +sed 's/^/ /' <<<"$out" +preloaded_everywhere() { + api GET /preloads | jq -e --arg s "$SCHEMA_NAME" --argjson n "${#PRELOAD_WANT[@]}" \ + '[.[] | select(.schema == $s)] | length == 1 and (.[0].images == $n) + and (.[0].workers | length == 2 and all(.held == $n and .last_pass != null))' >/dev/null +} +retry 300 "both workers to hold all ${#PRELOAD_WANT[@]} image(s) of $SCHEMA_NAME" preloaded_everywhere +# Checked against each daemon too: preload-list reports what corkd last saw. +for ip in "${WORKER_IPS[@]}"; do + for ref in "${PRELOAD_WANT[@]}"; do + has_line "$ref" worker_tags "$ip" || fail "preload-list reports worker $ip complete, but its daemon does not hold $ref" + done +done +out=$(cork preload-list) +sed 's/^/ /' <<<"$out" +(( $(grep -c ' ready$' <<<"$out") == 2 )) || fail "preload-list does not show both workers ready" +note "both workers hold all ${#PRELOAD_WANT[@]} image(s) the schema launches" + +# A container corkd does not manage holds generation 5 on $WA through the +# redeploy, as an instance the rebuild has yet to stop would. Created, not +# started: dockerd refuses the unforced delete for any container. Its +# cmgr.managed is not "true", for the reason the blocker's is not. +holder=$(worker_api "$WA" "/containers/create?name=$HOLDER_NAME" -X POST -H 'Content-Type: application/json' \ + -d "$(jq -cn --arg img "$E2E_REGISTRY/$CH_ONDEMAND:$OD_TAG_G5" '{Image: $img, Labels: {"cmgr.managed": "e2e-holder"}}')" | jq -r .Id) +[[ -n "$holder" && "$holder" != null ]] || fail "could not create the holder container $HOLDER_NAME on $WA" +HOLDER_WORKER=$WA +HOLDER_STALE="$E2E_REGISTRY/$CH_ONDEMAND:stale-$OD_TAG_G5" + +# The redeploy. --prune-old so the generation this displaces from retention +# leaves the builder and registry here rather than as an orphan to clean up. +PRELOAD_GEN=50 # a generation no other step builds, so each still finds its own change +set_generation $PRELOAD_GEN ondemand +out=$(cork update --prune-old) || fail "cork update exited non-zero building generation $PRELOAD_GEN of $CH_ONDEMAND: $out" +sed 's/^/ /' <<<"$out" +grep -q " $CH_ONDEMAND$" <<<"$out" || fail "$CH_ONDEMAND was not rebuilt" +OD_TAG_PRELOAD=$(image_tag "$(api GET "/builds/$OD_BUILD")" challenge) +[[ "$OD_TAG_PRELOAD" != "$OD_TAG_G5" ]] || fail "generation $PRELOAD_GEN has generation 5's tag: nothing was redeployed" +for id in "${OD_IDS[@]}"; do + assert_torn_down "$id" +done +OD_IDS=() +redeployed_everywhere() { + local ip tags + for ip in "${WORKER_IPS[@]}"; do + tags=$(worker_tags "$ip") + grep -qx "$E2E_REGISTRY/$CH_ONDEMAND:$OD_TAG_PRELOAD" <<<"$tags" || return 1 + if grep -qx "$E2E_REGISTRY/$CH_ONDEMAND:$OD_TAG_G5" <<<"$tags"; then return 1; fi + done +} +retry 120 "both workers to pull generation $PRELOAD_GEN of $CH_ONDEMAND and drop generation 5" redeployed_everywhere +# Only the workers were told to let it go: generation 5 is the rollback target. +has_line "$OD_TAG_G5" registry_tags "$CH_ONDEMAND" || + fail "generation 5 ($OD_TAG_G5) left the registry: preloading removes from workers only, and it is the rollback target" +for ip in "${WORKER_IPS[@]}"; do + for ref in "${PRELOAD_WANT[@]}"; do + [[ "$ref" == "$E2E_REGISTRY/$CH_ONDEMAND:"* ]] && continue + has_line "$ref" worker_tags "$ip" || fail "the redeploy of $CH_ONDEMAND cost worker $ip the image $ref of a challenge it did not touch" + done +done +note "both workers pulled $OD_TAG_PRELOAD and dropped $OD_TAG_G5, which zot still holds as the rollback target" + +# On $WA the holder kept generation 5, under its stale tag. $WB may have too: +# the pass a landing build kicks can reach the delete before the rebuild has +# stopped that worker's own generation 5 instance. +has_line "$HOLDER_STALE" worker_tags "$WA" || + fail "generation 5 left $WA while $HOLDER_NAME still used it: the delete is not forced, so the image should be kept as $HOLDER_STALE" +code=$(worker_status "$WA" "/containers/$HOLDER_NAME?force=true" -X DELETE) +[[ "$code" == 204 ]] || fail "could not remove the holder container from $WA: HTTP $code" +# A later pass deletes it once nothing uses it; preload-add again (it is +# idempotent) asks for one now rather than at the sweep. +out=$(cork preload-add "$SCHEMA_NAME") || fail "cork preload-add $SCHEMA_NAME a second time failed: $out" +stale_gone() { + local ip + for ip in "${WORKER_IPS[@]}"; do + if has_line "$HOLDER_STALE" worker_tags "$ip"; then return 1; fi + done +} +retry 60 "both workers to delete $HOLDER_STALE once nothing uses it" stale_gone +HOLDER_WORKER="" +note "$WA kept generation 5 as $HOLDER_STALE while a container used it, and both workers deleted it once none did" + +out=$(cork preload-remove "$SCHEMA_NAME") || fail "cork preload-remove $SCHEMA_NAME failed: $out" +sed 's/^/ /' <<<"$out" +# Nothing else is preloaded, so all the schema launches now leaves corkd's +# cleanup, and preload-remove names each image. +mapfile -t PRELOAD_RELEASED < <(preload_want) +grep -q "no longer cleans up these ${#PRELOAD_RELEASED[@]} image(s)" <<<"$out" || + fail "preload-remove did not report the ${#PRELOAD_RELEASED[@]} image(s) it leaves on the workers" +for ref in "${PRELOAD_RELEASED[@]}"; do + grep -qx " $ref" <<<"$out" || fail "preload-remove did not list $ref among the images corkd no longer cleans up" +done +[[ "$(cork preload-list)" == "no schema is preloaded" ]] || fail "preload-list still shows a preload after preload-remove" +IMAGE_TAG[$CH_ONDEMAND]=$OD_TAG_PRELOAD +# Back to the shape the later steps expect: tracked instances on both workers. +retry 30 "both workers to report ok after the rebuild" all_workers_ok +for i in 70 71; do + inst=$(launch "$OD_BUILD" "e2e-user-$i" "e2e-value-$i") + track "$inst" "$i" +done +ensure_instance_on "$WB" 72 +ensure_instance_on "$WA" 73 +for id in "${OD_IDS[@]}"; do + check_ondemand "${INST_META[$id]}" "${INST_USER[$id]}" "E2E_GENERATION=$PRELOAD_GEN" +done +ok "preloaded ${#PRELOAD_WANT[@]} image(s) onto both workers ahead of any launch, followed a redeploy onto both and removed the generation it replaced from both, keeping it as stale on $WA until a container there let it go, then handed the schema back to on-demand pulls, listing the ${#PRELOAD_RELEASED[@]} image(s) it leaves on the workers, with ${#OD_IDS[@]} instances serving generation $PRELOAD_GEN" +else + deselect "preload" +fi + # ========================================================================= @@ -6372,8 +6523,35 @@ step "seccomp: a challenge's own policy reaches the container, through the runti MULTI_INST="" ok "$CH_SECCOMP declared the profile '$sec_profile', instance $SEC_INST ran on $SEC_WORKER under oci-interceptor carrying a seccomp policy, and its own solver got build $SEC_BUILD's flag -- which it only serves when that profile was applied" -step "removing the second schema takes its builds and its images and leaves the first schema whole" +step "removing the second schema takes its builds and its images and leaves the first schema whole; preloaded, its images stay on the workers for an operator" + # Preloaded first, so the removal also says what a preloaded schema leaves: + # corkd's cleanup lasts as long as the preload, so the images stay on the + # workers, and an operator takes them. Its instance is stopped by now, so + # nothing holds them. + MULTI_REFS=() + for host in "$MULTI_FRONT" "$MULTI_BACK"; do + MULTI_REFS+=("$E2E_REGISTRY/$CH_MULTI:$(image_tag "$MULTI_META" "$host")") + done + multi_on_workers() { # multi_on_workers all|none + local ip ref + for ip in "${WORKER_IPS[@]}"; do + for ref in "${MULTI_REFS[@]}"; do + if has_line "$ref" worker_tags "$ip"; then [[ "$1" == all ]] || return 1; else [[ "$1" == none ]] || return 1; fi + done + done + } + out=$(cork preload-add "$MULTI_SCHEMA") || fail "cork preload-add $MULTI_SCHEMA failed: $out" + retry 300 "both workers to hold $CH_MULTI's two runtime images" multi_on_workers all cork remove-schema "$MULTI_SCHEMA" || fail "remove-schema $MULTI_SCHEMA failed" + [[ "$(cork preload-list)" == "no schema is preloaded" ]] || fail "the preload of $MULTI_SCHEMA outlived the schema" + multi_on_workers all || fail "removing preloaded schema $MULTI_SCHEMA took its images off a worker; they are an operator's to vet" + for ip in "${WORKER_IPS[@]}"; do + for ref in "${MULTI_REFS[@]}"; do + code=$(worker_status "$ip" "/images/$ref" -X DELETE) + [[ "$code" == 200 ]] || fail "deleting $ref from worker $ip by hand answered HTTP $code" + done + done + multi_on_workers none || fail "worker still holds an image of $CH_MULTI after deleting it by hand" MULTI_SCHEMA_ADDED="" if has_line "$MULTI_SCHEMA" cork list-schemas; then fail "schema $MULTI_SCHEMA is still listed after remove-schema"; fi has_line "$SCHEMA_NAME" cork list-schemas || @@ -6394,7 +6572,7 @@ step "removing the second schema takes its builds and its images and leaves the fail "the $host image of build $MULTI_BUILD is still on the builder after its schema was removed; destroyImages left it behind$([[ "$host" == "$MULTI_PRIVATE" ]] && printf ' -- and that stage holds the flag and the ssh key in plain text')" fi done - ok "schema $MULTI_SCHEMA converged alongside $SCHEMA_NAME without touching its builds or its persistent instance; $CH_MULTI shipped two runtime images and kept '$MULTI_PRIVATE' out of the registry; it handed over whole to a daemon that builds nothing, all three stages recorded and a payload missing one refused; instance $multi_done ran $MULTI_FRONT (0.5 cpu, published at $mpub:$mport) and $MULTI_BACK (0.25 cpu, unpublished) on one cmgr network and returned its flag over ssh from $MULTI_BACK through $MULTI_FRONT; the stop removed both containers and the network, and remove-schema left $SCHEMA_NAME whole" + ok "schema $MULTI_SCHEMA converged alongside $SCHEMA_NAME without touching its builds or its persistent instance; $CH_MULTI shipped two runtime images and kept '$MULTI_PRIVATE' out of the registry; it handed over whole to a daemon that builds nothing, all three stages recorded and a payload missing one refused; instance $multi_done ran $MULTI_FRONT (0.5 cpu, published at $mpub:$mport) and $MULTI_BACK (0.25 cpu, unpublished) on one cmgr network and returned its flag over ssh from $MULTI_BACK through $MULTI_FRONT; the stop removed both containers and the network, and remove-schema left $SCHEMA_NAME whole and, the schema being preloaded, left both runtime images on both workers for the operator's hand" else deselect "a second schema and the multi-container delivery shape" fi