| 1 | // Package ops is the box-side operations surface — verbs a systemd timer (or |
| 2 | // an operator's shell on the box) runs to keep replication honest. Inspired by |
| 3 | // titogo internal/ops: every verb encodes knowledge that otherwise lives only |
| 4 | // in prose, and prose drifts silently where a command that runs drifts loudly. |
| 5 | package ops |
| 6 | |
| 7 | import ( |
| 8 | "github.com/carlosframework/platform/internal/cliflag" |
| 9 | |
| 10 | "context" |
| 11 | "errors" |
| 12 | "flag" |
| 13 | "fmt" |
| 14 | "log" |
| 15 | "os" |
| 16 | "strings" |
| 17 | "time" |
| 18 | |
| 19 | "github.com/carlosframework/platform/internal/activator" |
| 20 | "github.com/carlosframework/platform/internal/dbverify" |
| 21 | "github.com/carlosframework/platform/internal/registry" |
| 22 | "github.com/carlosframework/platform/internal/release" |
| 23 | ) |
| 24 | |
| 25 | const ( |
| 26 | defaultConfigPath = "/etc/litestream.yml" |
| 27 | defaultDataDir = "/data" |
| 28 | ) |
| 29 | |
| 30 | // Run dispatches `carlos ops <verb>`. |
| 31 | func Run(args []string) { |
| 32 | verb := "" |
| 33 | if len(args) > 0 { |
| 34 | verb = args[0] |
| 35 | args = args[1:] |
| 36 | } |
| 37 | switch verb { |
| 38 | case "litestream": |
| 39 | runLitestream(args) |
| 40 | case "restore-verify": |
| 41 | runRestoreVerify(args) |
| 42 | case "hibernate-verify": |
| 43 | runHibernateVerify(args) |
| 44 | case "", "-h", "--help", "help": |
| 45 | usage() |
| 46 | if verb == "" { |
| 47 | os.Exit(2) |
| 48 | } |
| 49 | default: |
| 50 | fmt.Fprintf(os.Stderr, "ops: unknown verb %q\n\n", verb) |
| 51 | usage() |
| 52 | os.Exit(2) |
| 53 | } |
| 54 | } |
| 55 | |
| 56 | func usage() { |
| 57 | fmt.Fprintf(os.Stderr, `carlos ops — box-side operations verbs |
| 58 | |
| 59 | carlos ops litestream (re)generate the fleet litestream config from the |
| 60 | registry + data dir; prints "changed"/"unchanged" |
| 61 | so the caller knows whether to restart replication |
| 62 | carlos ops restore-verify restore every configured replica from S3 and |
| 63 | integrity-check each (run by the daily timer) |
| 64 | carlos ops hibernate-verify |
| 65 | the same drill for the replicas the config does |
| 66 | NOT name: hibernating instances, whose |
| 67 | replication the activator owns under its lease. |
| 68 | Publishes a per-host verdict to the deployment |
| 69 | bucket so the result is readable off-box. |
| 70 | |
| 71 | No verb for your task? Extend ops in a small PR — never a hand-run sequence |
| 72 | reconstructed from docs. |
| 73 | `) |
| 74 | } |
| 75 | |
| 76 | // dataDir resolves the data directory: flag wins, then CARLOS_DATA_DIR, then |
| 77 | // the fleet default. |
| 78 | func dataDir(flagVal string) string { |
| 79 | if flagVal != "" { |
| 80 | return flagVal |
| 81 | } |
| 82 | if env := os.Getenv("CARLOS_DATA_DIR"); env != "" { |
| 83 | return env |
| 84 | } |
| 85 | return defaultDataDir |
| 86 | } |
| 87 | |
| 88 | // runLitestream — regenerate the fleet replication config. Idempotent: run it |
| 89 | // after every `carlos add`/`remove` (or on a timer; a no-op re-run is cheap |
| 90 | // and prints "unchanged"). Databases of hibernating routes are excluded — see |
| 91 | // FleetDBs for Eleven's two-replicators lesson. |
| 92 | func runLitestream(args []string) { |
| 93 | fs := flag.NewFlagSet("ops litestream", flag.ExitOnError) |
| 94 | regPath := fs.String("registry", "", "registry database (default <data>/registry.db)") |
| 95 | data := fs.String("data", "", "data directory holding the fleet's *.db files (default $CARLOS_DATA_DIR or /data)") |
| 96 | bucket := fs.String("bucket", "", "S3 bucket receiving the replicas (required)") |
| 97 | prefix := fs.String("prefix", "", "key prefix inside the bucket for this fleet (required)") |
| 98 | region := fs.String("region", "", "S3 region for the replica stanza") |
| 99 | out := fs.String("out", defaultConfigPath, "config file to write") |
| 100 | lsBin := fs.String("litestream-bin", "litestream", "litestream binary to check the snapshot policy against (pinned release "+litestreamVersion+")") |
| 101 | allowShrink := fs.Bool("allow-shrink", false, "permit a config that replicates FEWER databases than the one it replaces") |
| 102 | fs.Usage = cliflag.Usage(fs, "") |
| 103 | _ = fs.Parse(args) |
| 104 | |
| 105 | if *bucket == "" || *prefix == "" { |
| 106 | log.Fatal("litestream: --bucket and --prefix are required") |
| 107 | } |
| 108 | dir := dataDir(*data) |
| 109 | rp := *regPath |
| 110 | if rp == "" { |
| 111 | rp = dir + "/registry.db" |
| 112 | } |
| 113 | reg, err := registry.Open(rp) |
| 114 | if err != nil { |
| 115 | log.Fatalf("litestream: open registry %s: %v", rp, err) |
| 116 | } |
| 117 | dbs, skipped, err := FleetDBsWithSkipped(reg, dir, LayoutJournalPresent(dir)) |
| 118 | if err != nil { |
| 119 | log.Fatalf("litestream: %v", err) |
| 120 | } |
| 121 | if len(dbs) == 0 { |
| 122 | log.Fatalf("litestream: no non-empty *.db files under %s — wrong box?", dir) |
| 123 | } |
| 124 | if len(skipped) > 0 { |
| 125 | // Named, never silent: these are files under the data dir that no route |
| 126 | // claims. Usually leftovers (a purge backup, a restore scratch); if one |
| 127 | // is a database someone still wants, the fix is a route that says so. |
| 128 | fmt.Fprintf(os.Stderr, "litestream: %d file(s) under %s are claimed by no route and will NOT be replicated:\n %s\n", |
| 129 | len(skipped), dir, strings.Join(skipped, "\n ")) |
| 130 | } |
| 131 | |
| 132 | // Never quietly replicate less than the file being replaced — see |
| 133 | // CoverageShortfall. This is what makes a cutover from a hand-maintained |
| 134 | // config to a generated one safe to do unattended. |
| 135 | if prior, err := os.ReadFile(*out); err == nil { |
| 136 | if missing := CoverageShortfall(string(prior), dbs); len(missing) > 0 && !*allowShrink { |
| 137 | log.Fatalf(`litestream: REFUSING to write %s — it would stop replicating %d database(s) that %s names today: |
| 138 | %s |
| 139 | Nothing has been changed. Either make the registry describe them (a route with |
| 140 | --db, or a unit tenant's state directory), or pass --allow-shrink if they really |
| 141 | should no longer be replicated.`, |
| 142 | *out, len(missing), *out, strings.Join(missing, "\n ")) |
| 143 | } else if len(missing) > 0 { |
| 144 | fmt.Fprintf(os.Stderr, "litestream: --allow-shrink: dropping %d database(s) from replication:\n %s\n", |
| 145 | len(missing), strings.Join(missing, "\n ")) |
| 146 | } |
| 147 | } |
| 148 | |
| 149 | cfg := LitestreamConfig(dbs, *bucket, *prefix, *region) |
| 150 | wrote, err := writeIfChanged(*out, cfg) |
| 151 | if err != nil { |
| 152 | log.Fatalf("litestream: write %s: %v", *out, err) |
| 153 | } |
| 154 | fmt.Fprintf(os.Stderr, "litestream: %d databases -> s3://%s/%s/ (%s)\n", len(dbs), *bucket, *prefix, *out) |
| 155 | |
| 156 | // The config now STATES a recovery window, so say plainly when the binary |
| 157 | // on this box will ignore it. Loud, not fatal: a box on an old litestream |
| 158 | // still wants its database SET kept current (that is what carries |
| 159 | // console.db), and refusing here would withhold the fix that does work in |
| 160 | // order to protest the one that does not. |
| 161 | if v := litestreamVersionOutput(*lsBin); !SnapshotPolicySupported(v) { |
| 162 | fmt.Fprintf(os.Stderr, `litestream: WARNING — this box's litestream (%s) predates the snapshot |
| 163 | policy (0.5+), so the "snapshot: retention: %s" this file now states is |
| 164 | IGNORED and the real recovery window stays at litestream's 24h default. |
| 165 | The database set below is still correct and still replicated. Upgrade the |
| 166 | binary to %s to make the stated window real. |
| 167 | `, strings.TrimSpace(firstLine(v)), SnapshotRetention, litestreamVersion) |
| 168 | } |
| 169 | // The one contract line on stdout: the calling unit restarts the |
| 170 | // replicator on "changed" and leaves it alone on "unchanged". |
| 171 | if wrote { |
| 172 | fmt.Println("changed") |
| 173 | } else { |
| 174 | fmt.Println("unchanged") |
| 175 | } |
| 176 | } |
| 177 | |
| 178 | // runRestoreVerify — restore every database in the generated config from S3 |
| 179 | // into a temp dir and prove each opens clean. Exits non-zero on any failure so |
| 180 | // the oneshot unit lands in 'failed' where the fleet sweeps see it. |
| 181 | func runRestoreVerify(args []string) { |
| 182 | fs := flag.NewFlagSet("ops restore-verify", flag.ExitOnError) |
| 183 | cfgPath := fs.String("config", defaultConfigPath, "generated litestream config to read the database set from") |
| 184 | lsBin := fs.String("litestream-bin", "litestream", "litestream binary (pinned release "+litestreamVersion+")") |
| 185 | fs.Usage = cliflag.Usage(fs, "") |
| 186 | _ = fs.Parse(args) |
| 187 | |
| 188 | tmp, err := os.MkdirTemp("", "carlos-restore") |
| 189 | if err != nil { |
| 190 | log.Fatalf("restore-verify: %v", err) |
| 191 | } |
| 192 | defer os.RemoveAll(tmp) |
| 193 | |
| 194 | if err := RestoreVerify(*lsBin, *cfgPath, tmp); err != nil { |
| 195 | os.RemoveAll(tmp) // log.Fatal skips the defer |
| 196 | log.Fatal(err) |
| 197 | } |
| 198 | } |
| 199 | |
| 200 | // firstLine is the version banner's first line, or a stand-in when the binary |
| 201 | // answered nothing at all. |
| 202 | func firstLine(s string) string { |
| 203 | if s = strings.TrimSpace(s); s == "" { |
| 204 | return "version unknown — binary missing or unreadable" |
| 205 | } |
| 206 | line, _, _ := strings.Cut(s, "\n") |
| 207 | return line |
| 208 | } |
| 209 | |
| 210 | // defaultHibernateMax caps one drill run. Each verification is a full restore |
| 211 | // from S3, so an uncapped nightly pass over a growing fleet is a bill that |
| 212 | // grows with it (the request-cost discipline internal/lease's LeaseTTL comment |
| 213 | // records). 25 covers today's fleets in a single run and keeps a large one |
| 214 | // fully covered within days, oldest-first. |
| 215 | const defaultHibernateMax = 25 |
| 216 | |
| 217 | // runHibernateVerify — restore-drill the hibernating instances' replicas, the |
| 218 | // set /etc/litestream.yml deliberately omits, and publish the verdicts. |
| 219 | // |
| 220 | // Exit status carries corruption only: a damaged replica fails the run (the |
| 221 | // oneshot unit lands in 'failed' where fleet sweeps see it), while replicas |
| 222 | // the pass could not decide are recorded as unverified and do NOT fail it. |
| 223 | // Conflating the two would either cry wolf on every S3 blip or — worse — |
| 224 | // train an operator to ignore a red run. |
| 225 | func runHibernateVerify(args []string) { |
| 226 | fs := flag.NewFlagSet("ops hibernate-verify", flag.ExitOnError) |
| 227 | regPath := fs.String("registry", "", "registry database (default <data>/registry.db)") |
| 228 | data := fs.String("data", "", "data directory (default $CARLOS_DATA_DIR or "+defaultDataDir+")") |
| 229 | max := fs.Int("max", defaultHibernateMax, "most replicas to restore in one run (0 = no cap)") |
| 230 | fs.Usage = cliflag.Usage(fs, "") |
| 231 | _ = fs.Parse(args) |
| 232 | |
| 233 | ctx := context.Background() |
| 234 | cfg, err := activator.ConfigFromEnv() |
| 235 | if err != nil { |
| 236 | log.Fatalf("hibernate-verify: %v", err) |
| 237 | } |
| 238 | if cfg == nil { |
| 239 | // Hibernation disabled on this box: there is no such replica set, so |
| 240 | // there is nothing to drill and nothing is wrong. |
| 241 | fmt.Println("hibernate-verify: CARLOS_HIBERNATE_BUCKET unset — no hibernating replicas on this box") |
| 242 | return |
| 243 | } |
| 244 | |
| 245 | dir := dataDir(*data) |
| 246 | rp := *regPath |
| 247 | if rp == "" { |
| 248 | rp = dir + "/registry.db" |
| 249 | } |
| 250 | reg, err := registry.Open(rp) |
| 251 | if err != nil { |
| 252 | log.Fatalf("hibernate-verify: open registry %s: %v", rp, err) |
| 253 | } |
| 254 | hosts, err := HibernatingHosts(reg) |
| 255 | if err != nil { |
| 256 | log.Fatalf("hibernate-verify: %v", err) |
| 257 | } |
| 258 | if len(hosts) == 0 { |
| 259 | fmt.Println("hibernate-verify: no hibernating instances registered on this box") |
| 260 | return |
| 261 | } |
| 262 | |
| 263 | // The prior record does two jobs: it orders this run (stalest first) and |
| 264 | // it carries the verdicts this run's cap will not revisit. |
| 265 | store, storeErr := openDeployStore(ctx) |
| 266 | var prior *dbverify.Record |
| 267 | if storeErr == nil { |
| 268 | if prior, err = dbverify.Load(ctx, store); err != nil { |
| 269 | log.Printf("hibernate-verify: reading %s: %v (continuing; this run starts the record over)", dbverify.Key, err) |
| 270 | prior = nil |
| 271 | } |
| 272 | } |
| 273 | |
| 274 | tmp, err := os.MkdirTemp("", "carlos-hibernate-verify") |
| 275 | if err != nil { |
| 276 | log.Fatalf("hibernate-verify: %v", err) |
| 277 | } |
| 278 | defer os.RemoveAll(tmp) |
| 279 | |
| 280 | now := time.Now() |
| 281 | results := VerifyHibernating(ctx, cfg, OldestFirst(hosts, prior), *max, now, tmp) |
| 282 | fleet := make([]string, 0, len(hosts)) |
| 283 | for _, h := range hosts { |
| 284 | fleet = append(fleet, h.Host) |
| 285 | } |
| 286 | rec := dbverify.Merge(prior, boxLabel(), now, fleet, results) |
| 287 | |
| 288 | // Publication is best-effort and LOUD when it fails: the drill's value is |
| 289 | // that someone can read the result without a session on this box, so a |
| 290 | // silent failure to publish would leave exactly the blind spot this verb |
| 291 | // exists to remove. |
| 292 | switch { |
| 293 | case storeErr != nil: |
| 294 | log.Printf("hibernate-verify: NOT PUBLISHED (%v) — the verdicts below exist only in this log", storeErr) |
| 295 | default: |
| 296 | if err := dbverify.Publish(ctx, store, rec); err != nil { |
| 297 | log.Printf("hibernate-verify: NOT PUBLISHED: %v — the verdicts below exist only in this log", err) |
| 298 | } else { |
| 299 | fmt.Printf("hibernate-verify: published %s (%d hosts, %d unverified)\n", dbverify.Key, len(rec.Hosts), rec.Unverified()) |
| 300 | } |
| 301 | } |
| 302 | |
| 303 | if n := rec.Failures(); n > 0 { |
| 304 | os.RemoveAll(tmp) // log.Fatal skips the defer |
| 305 | log.Fatalf("hibernate-verify: %d replica(s) did not restore clean", n) |
| 306 | } |
| 307 | // Count what was PROVEN, not what was visited: an unverified host is not |
| 308 | // coverage, and a summary that said otherwise would be the same comfortable |
| 309 | // lie this verb exists to remove. |
| 310 | proven := 0 |
| 311 | for _, v := range results { |
| 312 | if v.OK { |
| 313 | proven++ |
| 314 | } |
| 315 | } |
| 316 | fmt.Printf("hibernate-verify: %d of %d hibernating replicas proven this run (%d visited, %d in the fleet)\n", |
| 317 | proven, len(hosts), len(results), len(hosts)) |
| 318 | } |
| 319 | |
| 320 | // boxLabel names the writing box in the published record. The hostname is |
| 321 | // enough — the record is informational, unlike the cold-standby snapshot |
| 322 | // sidecar, whose Box field guards against a standby clobbering the dead box's |
| 323 | // registry. |
| 324 | func boxLabel() string { |
| 325 | h, err := os.Hostname() |
| 326 | if err != nil || h == "" { |
| 327 | return "unknown" |
| 328 | } |
| 329 | return h |
| 330 | } |
| 331 | |
| 332 | // openDeployStore resolves the deployment bucket the same way the agent's |
| 333 | // bucket/dir env vars do. A fleet box that reaches its deployment through the |
| 334 | // console door instead (openDeployStoreEnv's third case in internal/host) is |
| 335 | // NOT covered here and is told so out loud rather than publishing nothing |
| 336 | // quietly. |
| 337 | func openDeployStore(ctx context.Context) (release.Store, error) { |
| 338 | if b := os.Getenv("CARLOS_DEPLOYMENT_BUCKET"); b != "" { |
| 339 | return release.NewDeploymentS3Store(ctx, b) |
| 340 | } |
| 341 | if d := os.Getenv("CARLOS_DEPLOYMENT_DIR"); d != "" { |
| 342 | ds := release.NewDirStore(d) |
| 343 | ds.EnableChangeIndex() |
| 344 | return ds, nil |
| 345 | } |
| 346 | return nil, errors.New("neither CARLOS_DEPLOYMENT_BUCKET nor CARLOS_DEPLOYMENT_DIR is set") |
| 347 | } |
| 348 | |
amadan