diff --git a/README.md b/README.md index c06dc33..bfcf172 100644 --- a/README.md +++ b/README.md @@ -370,7 +370,7 @@ a spell of running clean. `down` is the exception, and the only state you can set by hand (`cork worker-down `). An operator taking a box out of service knows something the probes do not, so no probe will lift it: it stands until -`worker-add`. +`worker-add`, across corkd restarts too. Use it before a **termination**, not before a reboot. A reboot needs nothing: the box is ejected while it is away and rejoins by itself, which is the whole diff --git a/cork/database.go b/cork/database.go index 6cc5ba7..f813ca5 100644 --- a/cork/database.go +++ b/cork/database.go @@ -163,12 +163,16 @@ const schemaQuery string = ` -- Docker workers for multi-host instance placement. ip is the private -- orchestration address cork dials; public is the player-facing address -- (IP or hostname) surfaced in instance metadata, '' meaning the private - -- IP doubles as public. worker-remove purges the row along with the - -- worker's instance records. (DBs created before this scheme carry a - -- leftover 'removed' column; it is never read.) + -- IP doubles as public. downsince is when worker-down took the worker + -- down (unix nanos), NULL until then and again after a worker-add: down + -- is the one asserted state, so the one a restart cannot re-derive from a + -- probe. worker-remove purges the row along with the worker's instance + -- records. (DBs created before this scheme carry a leftover 'removed' + -- column; it is never read.) CREATE TABLE IF NOT EXISTS workers ( ip TEXT NOT NULL PRIMARY KEY, - public TEXT NOT NULL DEFAULT '' + public TEXT NOT NULL DEFAULT '', + downsince INTEGER ); -- Schemas every worker holds the images of (see preload.go). Not a @@ -500,6 +504,22 @@ func (m *Manager) initDatabase() error { } } + // Migrate older DBs: add workers.downsince. NULL is "not down", and a + // worker downed before the column existed came back up at that restart. + var downSinceCols int + err = db.QueryRow("SELECT COUNT(*) FROM pragma_table_info('workers') WHERE name = 'downsince';").Scan(&downSinceCols) + if err != nil { + m.log.errorf("could not check workers table schema: %s", err) + return err + } + if downSinceCols == 0 { + _, err = db.Exec("ALTER TABLE workers ADD COLUMN downsince INTEGER;") + if err != nil { + m.log.errorf("could not migrate workers.downsince column: %s", err) + return err + } + } + // Drop the images.digest column left behind by the digest-based worker // freshness check: content-addressed tags make the stored registry digest // redundant (the start path now only checks tag existence). Plain ALTER ... diff --git a/cork/database_operations_test.go b/cork/database_operations_test.go index b8152cf..fd7e5e7 100644 --- a/cork/database_operations_test.go +++ b/cork/database_operations_test.go @@ -268,6 +268,51 @@ func TestChallengesDifficultyMigration(t *testing.T) { } } +// TestWorkersDownSinceMigration verifies that a workers table from before +// worker-down was persisted gains downsince in place, that its rows read back +// as not down, and that re-running the migration is a no-op. +func TestWorkersDownSinceMigration(t *testing.T) { + dbFile, err := os.CreateTemp("", "cmgr-downsince-migrate-*.db") + if err != nil { + t.Fatalf("failed to create temp file: %s", err) + } + dbFile.Close() + t.Cleanup(func() { removeDBFiles(dbFile.Name()) }) + + dsn := dbFile.Name() + "?_fk=false&_journal_mode=WAL&_busy_timeout=50&_synchronous=NORMAL" + legacy, err := sqlx.Open("sqlite3", dsn) + if err != nil { + t.Fatalf("failed to open db: %s", err) + } + // With the leftover 'removed' column that older databases still carry. + schema := ` + CREATE TABLE workers ( + ip TEXT NOT NULL PRIMARY KEY, + public TEXT NOT NULL DEFAULT '', + removed INTEGER NOT NULL DEFAULT 0 + ); + INSERT INTO workers(ip, public) VALUES ('10.0.0.1', 'worker-a.example');` + if _, err := legacy.Exec(schema); err != nil { + t.Fatalf("failed to create legacy schema: %s", err) + } + legacy.Close() + + os.Setenv(DB_ENV, dbFile.Name()) + defer os.Unsetenv(DB_ENV) + + for _, pass := range []string{"migration", "idempotence"} { + mgr := new(Manager) + mgr.log = newLogger(DISABLED) + if err := mgr.initDatabase(); err != nil { + t.Fatalf("initDatabase failed (%s pass): %s", pass, err) + } + if storedDownSince(t, mgr, "10.0.0.1").Valid { + t.Errorf("a worker from before the migration reads back as down after the %s pass", pass) + } + mgr.db.Close() + } +} + // TestDatabaseBuildLifecycle tests the build database operations func TestDatabaseBuildLifecycle(t *testing.T) { mgr := setupTestManager(t) diff --git a/cork/worker_down_test.go b/cork/worker_down_test.go new file mode 100644 index 0000000..6ef9615 --- /dev/null +++ b/cork/worker_down_test.go @@ -0,0 +1,117 @@ +package cork + +import ( + "context" + "database/sql" + "strings" + "testing" +) + +// restarted is another corkd started on m's database, as a restart would +// leave it. +func restarted(t *testing.T, m *Manager) *Manager { + t.Helper() + r := &Manager{log: newLogger(DISABLED), ctx: context.Background(), workerTiming: m.workerTiming, db: m.db} + if err := r.initWorkers(); err != nil { + t.Fatalf("initWorkers after the restart: %s", err) + } + t.Cleanup(func() { + r.workersMu.Lock() + defer r.workersMu.Unlock() + for _, w := range r.workers { + close(w.done) + } + }) + return r +} + +func storedDownSince(t *testing.T, m *Manager, ip string) sql.NullInt64 { + t.Helper() + var downSince sql.NullInt64 + if err := m.db.Get(&downSince, "SELECT downsince FROM workers WHERE ip = ?;", ip); err != nil { + t.Fatalf("reading back downsince of %s: %s", ip, err) + } + return downSince +} + +// Down is the one state a restart cannot re-derive from a probe, so it is the +// one that survives it: the worker comes back down, as of when it was taken +// down, and out of placement. An ejection is an observation and does not. +func TestWorkerDownSurvivesARestart(t *testing.T) { + const downIP, ejectedIP = "127.0.0.1", "127.0.0.2" + m := addWorkerTestManager(t) + for _, ip := range []string{downIP, ejectedIP} { + if err := m.AddWorker(ip, ""); err != nil { + t.Fatalf("worker-add %s: %s", ip, err) + } + } + if err := m.SetWorkerDown(downIP); err != nil { + t.Fatalf("worker-down: %s", err) + } + m.workersMu.RLock() + since := m.workers[downIP].since.Load() + m.eject(m.workers[ejectedIP], "its daemon did not answer") + m.workersMu.RUnlock() + + r := restarted(t, m) + down := r.workers[downIP] + if got := reachableOf(down); got != workerDown { + t.Fatalf("reachable %s after a restart, want down: the restart returned a downed worker to placement", got) + } + if got := down.since.Load(); got != since { + t.Errorf("since %d after a restart, want %d, when the operator took it down", got, since) + } + if reason := down.reason.Load(); reason == nil || *reason != downReason { + t.Errorf("reason %v after a restart, want %q", reason, downReason) + } + if got := reachableOf(r.workers[ejectedIP]); got == workerDown { + t.Fatal("an ejected worker came back down: an observation was persisted as an assertion, and no probe will ever lift it") + } +} + +// Marked before the write: a worker-down that cannot persist still takes the +// worker out of placement for the life of this corkd, and its error says so. +func TestWorkerDownThatCannotPersistStillTakesTheWorkerDown(t *testing.T) { + m := &Manager{log: newLogger(DISABLED), ctx: context.Background()} + w := testWorkerConn(workerReachableOk) + m.workers = map[string]*workerConn{w.ip: w} + m.db = workerDB(t, w.ip) + m.db.Close() + + err := m.SetWorkerDown(w.ip) + if err == nil { + t.Fatal("worker-down reported success on a database it could not write") + } + if got := reachableOf(w); got != workerDown { + t.Fatalf("reachable %s after a worker-down whose write failed, want down", got) + } + if !strings.Contains(err.Error(), "is down") { + t.Errorf("error %q does not say the worker is down, so worker-down reads as having done nothing", err) + } +} + +// worker-add lifts the persisted assertion along with the one in memory, or +// the next restart takes the worker back out of placement. +func TestWorkerAddLiftsAPersistedDown(t *testing.T) { + const ip = "127.0.0.1" + m := addWorkerTestManager(t) + if err := m.AddWorker(ip, ""); err != nil { + t.Fatalf("worker-add: %s", err) + } + if err := m.SetWorkerDown(ip); err != nil { + t.Fatalf("worker-down: %s", err) + } + if !storedDownSince(t, m, ip).Valid { + t.Fatal("worker-down was not persisted") + } + + if err := m.AddWorker(ip, ""); err != nil { + t.Fatalf("worker-add after worker-down: %s", err) + } + if storedDownSince(t, m, ip).Valid { + t.Fatal("worker-add left worker-down persisted, so the next restart takes the worker out of placement again") + } + if got := reachableOf(restarted(t, m).workers[ip]); got == workerDown { + t.Fatal("a worker brought back with worker-add is down again after a restart") + } +} diff --git a/cork/worker_probe_test.go b/cork/worker_probe_test.go index 6b345e7..a4f33bb 100644 --- a/cork/worker_probe_test.go +++ b/cork/worker_probe_test.go @@ -262,6 +262,7 @@ func TestRecoveryDoesNotResurrectAnAssertedDownWorker(t *testing.T) { w := testWorkerConn(workerUnresponsive) w.done = make(chan struct{}) m.workers = map[string]*workerConn{w.ip: w} + m.db = workerDB(t, w.ip) if err := m.SetWorkerDown(w.ip); err != nil { t.Fatalf("worker-down: %s", err) @@ -317,9 +318,10 @@ func TestWorkerDownAndRecoveryCannotBothWin(t *testing.T) { timing.pollInterval = 5 * time.Millisecond timing.maxMisses = 1 + db := workerDB(t, testWorkerConn(workerUnresponsive).ip) var swapped, declined int for i := 0; i < 100; i++ { - m := &Manager{log: newLogger(DISABLED), ctx: context.Background(), workerTiming: timing} + m := &Manager{log: newLogger(DISABLED), ctx: context.Background(), workerTiming: timing, db: db} w := testWorkerConn(workerUnresponsive) w.done = make(chan struct{}) m.workers = map[string]*workerConn{w.ip: w} @@ -381,6 +383,7 @@ func TestWorkerDownLandsOnTheConnThatRecoveryInstalled(t *testing.T) { w := testWorkerConn(workerUnresponsive) w.done = make(chan struct{}) m.workers = map[string]*workerConn{w.ip: w} + m.db = workerDB(t, w.ip) m.recoverWorker(w) fresh := m.workers[w.ip] diff --git a/cork/worker_reconcile.go b/cork/worker_reconcile.go index cf83b43..d33c6c5 100644 --- a/cork/worker_reconcile.go +++ b/cork/worker_reconcile.go @@ -56,7 +56,7 @@ func (r reconcileResult) String() string { // DB-only stop leaves behind while the worker is down or purged // (stopInstance, RemoveWorker). They are not merely untidy. A box that // rejoins placement, through worker-add or a corkd restart that starts every -// worker afresh, would otherwise refuse any new instance that draws one of +// worker but a downed one afresh, would otherwise refuse any new instance that draws one of // their published host ports. It therefore runs before the worker's poller // starts (runWorker), so nothing is placed on the worker until they are gone. // diff --git a/cork/workers.go b/cork/workers.go index 0f2d0c4..e76afb1 100644 --- a/cork/workers.go +++ b/cork/workers.go @@ -3,6 +3,7 @@ package cork import ( "context" "crypto/tls" + "database/sql" "encoding/json" "errors" "fmt" @@ -421,8 +422,9 @@ func (m *Manager) EnableWorkerPlacement() { } // initWorkers loads the persisted worker list and starts a telemetry poller -// per worker. Unreachable workers are marked down by their pollers and skipped -// by placement — never a startup failure. +// per worker. Unreachable workers stay unresponsive, out of placement, until +// they answer — never a startup failure. A worker an operator took down comes +// back down, as of when they did it. 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. @@ -431,16 +433,23 @@ func (m *Manager) initWorkers() error { m.workers = make(map[string]*workerConn) var rows []struct { - Ip string `db:"ip"` - Public string `db:"public"` + Ip string `db:"ip"` + Public string `db:"public"` + DownSince sql.NullInt64 `db:"downsince"` } - if err := m.db.Select(&rows, "SELECT ip, public FROM workers;"); err != nil { + if err := m.db.Select(&rows, "SELECT ip, public, downsince FROM workers;"); err != nil { m.log.errorf("could not load workers: %s", err) return err } for _, row := range rows { - w, err := m.newWorkerConn(row.Ip, row.Public, false) + var downSince time.Time + if row.DownSince.Valid { + downSince = time.Unix(0, row.DownSince.Int64) + m.log.infof("worker %s: down since %s, as worker-down left it; worker-add returns it to placement", + row.Ip, downSince.UTC().Format(time.RFC3339)) + } + w, err := m.newWorkerConn(row.Ip, row.Public, false, downSince) if err != nil { return err } @@ -477,7 +486,9 @@ func workerTLSConfig() (*tls.Config, error) { return cfg, nil } -func (m *Manager) newWorkerConn(ip, public string, fromRecovery bool) (*workerConn, error) { +// newWorkerConn builds a worker's conn and starts it. downSince is non-zero +// only for a worker initWorkers restores down. +func (m *Manager) newWorkerConn(ip, public string, fromRecovery bool, downSince time.Time) (*workerConn, error) { tlsCfg, err := workerTLSConfig() if err != nil { m.log.errorf("could not load worker TLS material: %s", err) @@ -530,6 +541,14 @@ func (m *Manager) newWorkerConn(ip, public string, fromRecovery bool) (*workerCo w.reachable.Store(int32(workerUnresponsive)) w.load.Store(int32(workerLoadUnknown)) w.since.Store(time.Now().UnixNano()) + if !downSince.IsZero() { + // Set before the goroutine starts, so nothing sees the worker as + // anything but down. + reason := downReason + w.reachable.Store(int32(workerDown)) + w.since.Store(downSince.UnixNano()) + w.reason.Store(&reason) + } // Before the goroutine below exists, which is the only thing that makes // this plain field safe to read from it. w.fromRecovery = fromRecovery @@ -544,6 +563,13 @@ func (m *Manager) newWorkerConn(ip, public string, fromRecovery bool) (*workerCo // still starting when corkd starts or the worker is added therefore costs // seconds, not a worker-add. func (m *Manager) runWorker(w *workerConn) { + // Down before this ran (restored at startup, or downed the moment the + // conn was published): left as worker-down leaves it, neither reconciled + // nor probed, until a worker-add replaces the conn. + if workerReachable(w.reachable.Load()) == workerDown { + <-w.done + return + } t := m.timing() switch m.reconcileWithRetries(w, m.reconcileWorker, time.Duration(t.maxMisses)*t.pollInterval, t.pollInterval) { case reconcileReady: @@ -1038,7 +1064,7 @@ func (m *Manager) recoverWorker(w *workerConn) { return } - fresh, err := m.newWorkerConn(w.ip, w.public, true) + fresh, err := m.newWorkerConn(w.ip, w.public, true, time.Time{}) if err != nil { // Rebuilding the client failed (TLS material unreadable): leave the // worker ejected and try again on the next probe. @@ -1058,10 +1084,14 @@ func (m *Manager) recoverWorker(w *workerConn) { w.ip, time.Since(time.Unix(0, w.since.Load())).Round(time.Second)) } +// downReason is the reason a down worker reports, whether it was taken down +// in this corkd or restored down at startup. +const downReason = "taken down by an operator" + // markWorkerDown asserts down, which no probe will lift. Reached only from // SetWorkerDown; the observed paths eject instead. func (m *Manager) markWorkerDown(w *workerConn) { - m.setReachable(w, workerDown, "taken down by an operator") + m.setReachable(w, workerDown, downReason) } // noteWorkerTransportError ejects a worker after a connection-level docker @@ -1183,6 +1213,9 @@ func (w *workerConn) noteTimeout(now time.Time, callTimeout, window time.Duratio // placements. Holding the read lock across both means the two orderings are // the only ones left, and recoverWorker refuses to recover a worker it finds // down. +// +// The assertion is persisted under the same lock, so that a worker-add (which +// clears it) cannot land between the two, and initWorkers restores it. func (m *Manager) SetWorkerDown(ip string) error { m.workersMu.RLock() defer m.workersMu.RUnlock() @@ -1190,7 +1223,16 @@ func (m *Manager) SetWorkerDown(ip string) error { if !ok { return &UnknownIdentifierError{Type: "worker", Name: ip} } + // Marked before the write, so a write that fails still leaves the worker + // down for as long as this corkd runs. m.markWorkerDown(w) + if _, err := m.db.Exec("UPDATE workers SET downsince = ? WHERE ip = ?;", w.since.Load(), ip); err != nil { + // Said in the error too, which is all the PATCH answers with: the + // worker is down, and worker-down again retries the write. + err = fmt.Errorf("worker %s is down, but a restart would return it to placement: could not persist it: %w", ip, err) + m.log.errorf("%s", err) + return err + } return nil } @@ -1261,10 +1303,13 @@ func (m *Manager) AddWorker(ip, public string) error { // safe to run against a worker that is already registered. This column is // the address a player connects to, so overwriting it with "" would // silently repoint every instance cork went on to place there. + // + // It also lifts a persisted worker-down, as it lifts the one in memory. _, err := m.db.Exec( `INSERT INTO workers(ip, public) VALUES (?, ?) ON CONFLICT(ip) DO UPDATE SET - public = CASE WHEN excluded.public = '' THEN workers.public ELSE excluded.public END;`, + public = CASE WHEN excluded.public = '' THEN workers.public ELSE excluded.public END, + downsince = NULL;`, ip, public, ) if err != nil { @@ -1280,7 +1325,7 @@ func (m *Manager) AddWorker(ip, public string) error { } } - w, err := m.newWorkerConn(ip, public, false) + w, err := m.newWorkerConn(ip, public, false, time.Time{}) if err != nil { return err } diff --git a/cork/workers_test.go b/cork/workers_test.go index f31bf30..0933382 100644 --- a/cork/workers_test.go +++ b/cork/workers_test.go @@ -14,6 +14,7 @@ import ( "testing" "time" + "github.com/jmoiron/sqlx" "github.com/moby/moby/client" ) @@ -35,6 +36,18 @@ func testWorkerConn(r workerReachable) *workerConn { return w } +// workerDB is a database holding a row for each of ips, which is where +// SetWorkerDown persists its assertion. +func workerDB(t *testing.T, ips ...string) *sqlx.DB { + t.Helper() + db := setupTestManager(t).db + t.Cleanup(func() { db.Close() }) + for _, ip := range ips { + db.MustExec("INSERT INTO workers(ip, public) VALUES (?, '');", ip) + } + return db +} + func reachableOf(w *workerConn) workerReachable { return workerReachable(w.reachable.Load()) } func loadOf(w *workerConn) workerLoad { return workerLoad(w.load.Load()) }