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) + } + } +} 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/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. 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 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