package bootstrap import ( "bufio" "bytes" "context" "errors" "fmt" "log/slog" "net" "os" "path/filepath" "slices" "strconv" "strings" "time" "github.com/johnandersen777/socialweb-computer-kcp/internal/apiserverfront" "github.com/johnandersen777/socialweb-computer-kcp/internal/bootstrap/table" "github.com/johnandersen777/socialweb-computer-kcp/internal/devinstance" "github.com/johnandersen777/socialweb-computer-kcp/internal/hostsvc" "github.com/johnandersen777/socialweb-computer-kcp/internal/kcpdatastore" "github.com/johnandersen777/socialweb-computer-kcp/internal/kcpserver" "github.com/johnandersen777/socialweb-computer-kcp/internal/kit/opt" "github.com/johnandersen777/socialweb-computer-kcp/internal/pgunit" ) const ( PortPoll = 20 * time.Millisecond ReadyPoll = 200 * time.Millisecond FrontAddress = "127.0.0.1:0" LeaseStaleAfter = 6 * time.Second LeaseWatchWindow = 12 * time.Second RestartSettle = 3 * time.Second DefaultReadyTimeout = time.Minute ) func CheckProfile(p string) error { if p == ProfileQuick || p == ProfileProdlike || p == ProfileMarketNetwork { return nil } return fmt.Errorf("bootstrap: unknown profile %q: want quick, prodlike or market-network", p) } func NeedsPostgres(profile string) bool { return profile == ProfileProdlike } type Config struct { Layout devinstance.Layout KcpPath string KinePath string PostgresRoot string Exe string FrontAddress string ReadyTimeout time.Duration OnListen func(name string, addr net.Addr) OnStep func(step string) OnTick func() Application *table.App Retention kcpdatastore.BackupRetention fdStore bool Log *slog.Logger silent map[string]bool portDeadline time.Time } func SlotRole(slot string) string { return table.SlotRole(slot) } func (c Config) App(rec *devinstance.Record) (table.App, error) { if c.Application != nil { return *c.Application, nil } apps, err := DevApps(rec.Profile, c.Layout) if err != nil { return table.App{}, err } return apps[0], nil } func (c Config) log() *slog.Logger { if c.Log != nil { return c.Log } return slog.Default() } func (c Config) frontAddress() string { if c.FrontAddress != "" { return c.FrontAddress } return FrontAddress } func (c Config) onListen(name string, port int) { if c.OnListen != nil { c.OnListen(name, &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1), Port: port}) } } func (c Config) onStep(step string) { c.onTick() if c.OnStep != nil { c.OnStep(step) } } func (c Config) onTick() { if c.OnTick != nil { c.OnTick() } } func (p Params) slotSpec(slot, frontURL string) (hostsvc.UnitSpec, string, error) { spec, err := p.Spec(p.App.SlotEntry(slot), frontURL) return spec, p.SlotRoot(slot), err } func writeSlotEnv(p Params, spec hostsvc.UnitSpec) error { if _, err := os.Stat(p.EnvFile(p.App.Names().Front())); err != nil { return nil } f, err := hostsvc.RenderInstalled(spec, hostsvc.InstallOptions{EnvFile: p.EnvFile(spec.Name)}) if err != nil { return err } if cur, err := os.ReadFile(f.EnvPath); err == nil && bytes.Equal(cur, f.Env) { return nil } return writePrivate(f.EnvPath, f.Env) } func waitUnixSocket(ctx context.Context, path string, progress ...func()) error { tick := func() {} if len(progress) > 0 && progress[0] != nil { tick = progress[0] } for { tick() c, err := net.Dial("unix", path) if err == nil { c.Close() return nil } select { case <-ctx.Done(): return fmt.Errorf("bootstrap: %s never accepted a connection: %w (last: %v)", path, ctx.Err(), err) case <-time.After(PortPoll): } } } func initdb(ctx context.Context, p Params) error { in, err := p.Postgres() if err != nil { return err } _, err = in.Initdb(ctx, p.storeConfig("", in).Server()) return err } func (p Params) storeBackup(r kcpdatastore.BackupRetention, log *slog.Logger) func(ctx context.Context) (string, error) { prune := func(dir, prefix, suffix, newest string) { removed, err := kcpdatastore.PruneBackups(dir, prefix, suffix, newest, r, time.Now()) if err != nil { log.Warn("store backup prune incomplete", "dir", dir, "err", err) } if len(removed) > 0 { log.Info("store backups pruned", "dir", dir, "removed", removed, "keep", r.Keep, "maxAge", r.MaxAge.String()) } } if p.App.Store != table.StorePostgres { return func(context.Context) (string, error) { kineDir := p.KineDir(p.App.KineOf(p.App.Slots[0])) out, err := kcpdatastore.BackupSqlite(kineDir, time.Now()) if err != nil { return "", err } prune(filepath.Join(kineDir, kcpdatastore.BackupsDir), "", "", out) return out, nil } } return func(ctx context.Context) (string, error) { in, err := p.Postgres() if err != nil { return "", err } out, err := in.Dump(ctx, p.PgSocketDir(), kcpdatastore.PostgresUser, kcpdatastore.KineDatabase, p.BackupsDir(), time.Now()) if err != nil { return "", err } prune(p.BackupsDir(), kcpdatastore.KineDatabase+"-", ".sql", out) return out, nil } } func waitPostgres(ctx context.Context, p Params, progress ...func()) error { if err := pgunit.WaitReady(ctx, p.PgSocketDir(), kcpdatastore.PostgresUser, progress...); err != nil { return fmt.Errorf("bootstrap: %w", err) } if _, err := pgunit.CreateDatabase(ctx, p.PgSocketDir(), kcpdatastore.PostgresUser, kcpdatastore.KineDatabase); err != nil { return fmt.Errorf("bootstrap: creating the kine database before any kine starts: %w", err) } return nil } func RSSKiB(pid int) int64 { f, err := os.Open("/proc/" + strconv.Itoa(pid) + "/status") if err != nil { return 0 } defer f.Close() sc := bufio.NewScanner(f) for sc.Scan() { if v, ok := strings.CutPrefix(sc.Text(), "VmRSS:"); ok { n, _ := strconv.ParseInt(strings.TrimSpace(strings.TrimSuffix(strings.TrimSpace(v), "kB")), 10, 64) return n } } return 0 } func Observe(ctx context.Context, m hostsvc.Manager, want hostsvc.Identity) (hostsvc.Observation, bool, error) { st, err := m.Inspect(ctx, want.Unit) if err == nil && st.State == hostsvc.Activating { sctx, cancel := context.WithTimeout(ctx, RestartSettle) st, err = hostsvc.Await(sctx, m, want.Unit, func(st hostsvc.UnitStatus) bool { return st.State != hostsvc.Activating }) cancel() if ctx.Err() != nil { return hostsvc.Observation{}, false, ctx.Err() } if errors.Is(err, context.DeadlineExceeded) { err = nil } } if err == nil && st.State == hostsvc.Activating { return hostsvc.Observation{Verdict: hostsvc.VerdictUnknown, Status: st, Detail: fmt.Sprintf("%s is still activating after %s", want.Unit, RestartSettle)}, false, nil } if err == nil && st.State == hostsvc.Active && st.InvocationID != "" && want.InvocationID != "" && st.InvocationID != want.InvocationID && st.MainPID > 0 { return hostsvc.Observation{Verdict: hostsvc.VerdictRunning, Status: st, Detail: fmt.Sprintf("%s was restarted in place by its manager (invocation %s, recorded %s)", want.Unit, st.InvocationID, want.InvocationID)}, true, nil } obs, err := m.Check(ctx, want) return obs, false, err } func identifyUnit(ctx context.Context, m hostsvc.Manager, cfg Config, role, name string, adopted bool, placedAt time.Time) (devinstance.Unit, error) { id, err := m.Identify(ctx, name, 0) if err != nil { return devinstance.Unit{}, err } if placedAt.IsZero() { placedAt = time.Now().UTC() } if cfg.silent[role] { return devinstance.Unit{Role: role, Unit: name, Identity: id, Adopted: adopted, PlacedAt: placedAt, PortAt: time.Now().UTC()}, nil } pctx, cancel := context.WithTimeout(ctx, cfg.ReadyTimeout) if !cfg.portDeadline.IsZero() { cancel() pctx, cancel = context.WithDeadline(ctx, cfg.portDeadline) } defer cancel() port, err := hostsvc.WaitListenPort(pctx, id.PID, PortPoll, cfg.onTick) if err != nil { return devinstance.Unit{}, fmt.Errorf("bootstrap: reading the bound port of %s (pid %d): %w", name, id.PID, err) } return devinstance.Unit{Role: role, Unit: name, Identity: id, Port: port, Endpoint: endpoint(role, port), Adopted: adopted, PlacedAt: placedAt, PortAt: time.Now().UTC()}, nil } func ensureUnit(ctx context.Context, m hostsvc.Manager, cfg Config, rec *devinstance.Record, role string, spec hostsvc.UnitSpec, beforeStart func() error) (devinstance.Unit, bool, error) { log := cfg.log().With("role", role, "unit", spec.Name) if prev := rec.Unit(role); prev != nil && prev.Identity.Unit == spec.Name { obs, restarted, err := Observe(ctx, m, prev.Identity) if err != nil { return devinstance.Unit{}, false, fmt.Errorf("bootstrap: checking %s: %w", spec.Name, err) } if restarted { u, err := identifyUnit(ctx, m, cfg, role, spec.Name, true, obs.Status.ActiveSince) if err != nil { return devinstance.Unit{}, false, err } log.Info("restarted in place by its manager; adopting the new invocation", "pid", u.Identity.PID, "invocation", u.Identity.InvocationID, "port", u.Port, "recordedPort", prev.Port) return u, false, nil } switch obs.Verdict { case hostsvc.VerdictRunning: u := *prev u.Adopted = true if port, err := hostsvc.ListenPort(u.Identity.PID); err == nil && port != u.Port && !cfg.silent[role] { log.Warn("an adopted unit listens on another port than recorded", "recorded", u.Port, "now", port) u.Port = port u.Endpoint = endpoint(role, port) } log.Info("adopted", "pid", u.Identity.PID, "invocation", u.Identity.InvocationID, "port", u.Port) return u, false, nil case hostsvc.VerdictUnknown: return devinstance.Unit{}, false, fmt.Errorf("bootstrap: %s is Unknown (%s); never re-created while the manager is unreachable", spec.Name, obs.Detail) } log.Info("recorded unit lost; placing it again", "reason", string(obs.Reason), "detail", obs.Detail) } if beforeStart != nil { if st, err := m.Inspect(ctx, spec.Name); err != nil || st.State != hostsvc.Active { if err := beforeStart(); err != nil { return devinstance.Unit{}, false, err } } } st, err := m.Start(ctx, spec) if err != nil { return devinstance.Unit{}, false, err } if st.Adopted && st.State != hostsvc.Active { log.Info("a loaded unit of this name is not active; resetting it", "state", string(st.State), "result", st.Result) if err := m.Stop(ctx, spec.Name); err != nil { return devinstance.Unit{}, false, err } if err := m.ResetFailed(ctx, spec.Name); err != nil { return devinstance.Unit{}, false, err } if beforeStart != nil { if err := beforeStart(); err != nil { return devinstance.Unit{}, false, err } } if st, err = m.Start(ctx, spec); err != nil { return devinstance.Unit{}, false, err } if st.Adopted && st.State != hostsvc.Active { log.Info("an installed unit of this name is loaded; starting it by name") if st, err = hostsvc.StartInstalled(ctx, m, spec.Name); err != nil { return devinstance.Unit{}, false, err } } } u, err := identifyUnit(ctx, m, cfg, role, spec.Name, st.Adopted, st.ActiveSince) if err != nil { return devinstance.Unit{}, false, err } log.Info("placed", "pid", u.Identity.PID, "invocation", u.Identity.InvocationID, "port", u.Port, "adopted", st.Adopted, "portAfter", u.PortAt.Sub(u.PlacedAt).String()) return u, !st.Adopted, nil } func endpoint(role string, port int) string { return "https://127.0.0.1:" + strconv.Itoa(port) } func readSlots(l devinstance.Layout) kcpserver.Slots { s, err := kcpserver.ReadSlots(l.SlotsFile()) if err != nil || s.Active == "" { return kcpserver.Slots{Shape: kcpserver.ShapeBlueGreen, Active: kcpserver.SlotA, Slots: map[string]kcpserver.SlotRecord{}} } return s } func stopRecordedSlots(ctx context.Context, m hostsvc.Manager, cfg Config, rec *devinstance.Record, why string) error { for _, slot := range kcpserver.SlotNames(kcpserver.ShapeBlueGreen) { prev := rec.Unit(SlotRole(slot)) if prev == nil || prev.Identity.Unit == "" { continue } cfg.log().Info(why+"; the slot that dialed the old one is restarted", "slot", slot) if err := m.Stop(ctx, prev.Unit); err != nil { return err } m.ResetFailed(ctx, prev.Unit) } return nil } func HandoverStaleLease(ctx context.Context, admin *kcpserver.HTTPAdmin, log *slog.Logger) (string, error) { first, err := admin.Lease(ctx) if err != nil || !first.Found { return "absent", err } live := func(l kcpserver.Lease) bool { return !l.Found || l.Holder != first.Holder || l.RenewTime.After(first.RenewTime) } staleIn := time.Until(first.RenewTime.Add(LeaseStaleAfter)) if staleIn > 0 { _, err := admin.AwaitLease(ctx, min(staleIn, LeaseWatchWindow), live) switch { case err == nil: return "live", nil case !errors.Is(err, kcpserver.ErrAwaitTimeout): return "", err case staleIn > LeaseWatchWindow: return "unchanged", nil } } else if l, err := admin.Lease(ctx); err != nil { return "", err } else if live(l) { return "live", nil } log.Info("the controller lease is held by a gone process; deleting it", "holder", first.Holder, "renewed", first.RenewTime.UTC().Format(time.RFC3339Nano)) return "deleted", admin.DeleteLease(ctx) } func Ensure(ctx context.Context, m hostsvc.Manager, cfg Config, rec *devinstance.Record) error { cfg.ReadyTimeout = opt.Positive(cfg.ReadyTimeout, DefaultReadyTimeout) cfg.fdStore = m.Capabilities().FDStore if len(rec.StandIns) > 0 { return ensureStandIns(ctx, m, cfg, rec) } if rec.SharedKcp != "" { rec.Ready, rec.Updated = true, time.Now().UTC() return nil } l := cfg.Layout log := cfg.log() app, err := cfg.App(rec) if err != nil { return err } p := cfg.Params(app) tbl, err := table.Table(app) if err != nil { return err } cfg.silent = map[string]bool{table.RolePostgres: true} for _, s := range app.KineSlots() { cfg.silent[table.KineRole(s)] = true if err := os.MkdirAll(p.KineDir(s), 0o700); err != nil { return err } } for _, d := range []string{p.Dir, p.Logs, p.Run} { if err := os.MkdirAll(d, 0o700); err != nil { return err } } slots := readSlots(l) active := slots.Active if !slices.Contains(app.Slots, active) { active = app.Slots[0] } rctx, cancel := context.WithTimeout(ctx, cfg.ReadyTimeout) defer cancel() if d, ok := rctx.Deadline(); ok { cfg.portDeadline = d } var kine, front devinstance.Unit frontPlaced := false for _, e := range tbl { if e.Kind == table.KindSlot || e.Kind == table.KindCore { continue } spec, err := p.Spec(e, "") if err != nil { return err } var before func() error if e.Kind == table.KindStore { if err := os.MkdirAll(p.PgSocketDir(), 0o700); err != nil { return err } before = func() error { return initdb(ctx, p) } } cfg.onStep("placing " + e.Role) u, placed, err := ensureUnit(ctx, m, cfg, rec, e.Role, spec, before) if err != nil { return err } switch e.Kind { case table.KindStore: cfg.onStep("waiting for postgres") if err := waitPostgres(rctx, p, cfg.onTick); err != nil { return err } u.Endpoint = "unix://" + p.PgSocketDir() case table.KindKine: cfg.onStep("waiting for kine's unix socket") if err := waitUnixSocket(rctx, p.KineSocket(e.Slot), cfg.onTick); err != nil { return err } u.Endpoint = p.KineURL(e.Slot) if e.Slot == app.KineOf(active) { kine = u rec.KineURL = u.Endpoint } case table.KindFront: cfg.onListen(e.Role, u.Port) front, frontPlaced = u, placed } u.RSSKiB = RSSKiB(u.Identity.PID) rec.SetUnit(u) } frontURL := kcpserver.FrontURL(front.Port) if frontPlaced { if err := stopRecordedSlots(ctx, m, cfg, rec, "the front was placed again on a new port"); err != nil { return err } } role := SlotRole(active) slotSpec, root, err := p.slotSpec(active, frontURL) if err != nil { return err } if err := kcpserver.WriteSlotFiles(root); err != nil { return err } if err := writeSlotEnv(p, slotSpec); err != nil { return err } seeded := false seed := func() error { if !kcpserver.HasShared(l.SharedDir()) { return nil } seeded = true return kcpserver.SeedSlot(l.SharedDir(), root) } prevInvocation := "" if prev := rec.Unit(role); prev != nil { prevInvocation = prev.Identity.InvocationID } cfg.onStep("placing the kcp slot") kcp, kcpPlaced, err := ensureUnit(ctx, m, cfg, rec, role, slotSpec, seed) if err != nil { return err } cfg.onListen(role, kcp.Port) cfg.onStep("waiting for kcp /readyz") var lastReady string reportReady := func(attempt int, last error) { cfg.onTick() msg := "" if last != nil { msg = last.Error() } if msg == lastReady { return } lastReady = msg log.Info("waiting for kcp /readyz", "attempt", attempt, "detail", msg) } var readyAt time.Time if kcpserver.HasShared(l.SharedDir()) { a, err := kcpserver.ReadSharedAdmin(l.SharedDir()) if err != nil { return err } admin, err := kcpserver.NewHTTPAdmin(a, frontURL) if err != nil { return err } if err := kcpserver.WaitOwnPortReady(rctx, admin, kcp.Port, ReadyPoll, reportReady); err != nil { return fmt.Errorf("bootstrap: %s on 127.0.0.1:%d: %w", kcp.Unit, kcp.Port, err) } readyAt = time.Now().UTC() } else { notBefore := time.Time{} if kcpPlaced { notBefore = kcp.PlacedAt } readyAt, err = kcpserver.WaitReady(rctx, kcp.Port, filepath.Join(root, "admin.kubeconfig"), notBefore, ReadyPoll, reportReady) if err != nil { return fmt.Errorf("bootstrap: %s on 127.0.0.1:%d: %w", kcp.Unit, kcp.Port, err) } if err := kcpserver.SaveShared(root, l.SharedDir()); err != nil { return err } log.Info("shared root material saved", "dir", l.SharedDir(), "from", root) } if kcpPlaced || kcp.ReadyAt.IsZero() { kcp.ReadyAt = readyAt kcp.StartedTo = readyAt.Sub(kcp.PlacedAt).Round(time.Millisecond).String() } a, err := kcpserver.ReadSharedAdmin(l.SharedDir()) if err != nil { return err } if err := kcpserver.WriteFrontKubeconfig(l.FrontKubeconfig(), frontURL, a); err != nil { return err } slots.Shape = p.shape() slots.Active = active slots.KineURL = kine.Endpoint slots.Front = kcpserver.FrontRecord{Unit: front.Unit, Address: "127.0.0.1:" + strconv.Itoa(front.Port), Port: front.Port, Control: l.FrontControl(), PID: front.Identity.PID} slots.Slots[active] = kcpserver.SlotRecord{Unit: kcp.Unit, Root: root, Port: kcp.Port, PID: kcp.Identity.PID, InvocationID: kcp.Identity.InvocationID, Exec: cfg.KcpPath, Kine: kine.Endpoint, StartedAt: kcp.PlacedAt, ReadyAt: kcp.ReadyAt} for _, other := range kcpserver.SlotNames(kcpserver.ShapeBlueGreen) { if other != active { delete(slots.Slots, other) rec.RemoveUnit(SlotRole(other)) } } if err := kcpserver.WriteSlots(l.SlotsFile(), slots); err != nil { return err } fc := apiserverfront.Client{Path: l.FrontControl()} backend := "127.0.0.1:" + strconv.Itoa(kcp.Port) fst, err := fc.Set(ctx, backend) if err != nil { return fmt.Errorf("bootstrap: pointing the front at %s: %w", backend, err) } if admin, err := kcpserver.NewHTTPAdmin(a, frontURL); err == nil { if moved, err := admin.SyncRootShard(ctx); err != nil { log.Warn("pointing the root shard at the front", "front", frontURL, "err", err.Error()) } else if moved { log.Info("the root shard advertises the front it runs behind now", "front", frontURL) } if n, err := admin.SyncEndpointSlices(ctx); err != nil { log.Warn("pointing the APIExport endpoint slices at the front", "front", frontURL, "err", err.Error()) } else if n > 0 { log.Info("the APIExport endpoint slices name the front it runs behind now", "front", frontURL, "slices", n) } } if kcpPlaced || prevInvocation != kcp.Identity.InvocationID { admin, err := kcpserver.NewHTTPAdmin(a, frontURL) if err != nil { return err } verdict, err := HandoverStaleLease(ctx, admin, log) if err != nil { log.Warn("checking the controller lease after a slot start", "err", err.Error()) } log.Info("controller lease after the slot start", "lease", verdict, "seeded", seeded) } kcp.RSSKiB = RSSKiB(kcp.Identity.PID) kine.RSSKiB = RSSKiB(kine.Identity.PID) front.RSSKiB = RSSKiB(front.Identity.PID) rec.SetUnit(kine) rec.SetUnit(front) rec.SetUnit(kcp) rec.KcpURL = frontURL rec.AdminKubeconfig = l.FrontKubeconfig() rec.Ready = true rec.Updated = time.Now().UTC() log.Info("kcp ready behind its front", "slot", active, "port", kcp.Port, "front", fst.Address, "backend", fst.Backend, "startToReady", kcp.StartedTo, "kcpRSSKiB", kcp.RSSKiB, "kineRSSKiB", kine.RSSKiB, "frontRSSKiB", front.RSSKiB) return nil } func Roll(ctx context.Context, m hostsvc.Manager, cfg Config, rec *devinstance.Record) (kcpserver.RollReport, error) { cfg.fdStore = m.Capabilities().FDStore l := cfg.Layout slots, err := kcpserver.ReadSlots(l.SlotsFile()) if err != nil { return kcpserver.RollReport{}, fmt.Errorf("bootstrap: no kcp-slots.json; dev up first: %w", err) } app, err := cfg.App(rec) if err != nil { return kcpserver.RollReport{}, err } p := cfg.Params(app) a, err := kcpserver.ReadSharedAdmin(l.SharedDir()) if err != nil { return kcpserver.RollReport{}, err } frontURL := slots.Front.URL() admin, err := kcpserver.NewHTTPAdmin(a, frontURL) if err != nil { return kcpserver.RollReport{}, err } for _, slot := range kcpserver.SlotNames(kcpserver.ShapeBlueGreen) { if err := kcpserver.WriteSlotFiles(l.SlotRoot(slot)); err != nil { return kcpserver.RollReport{}, err } } rep, after, err := kcpserver.Roll(ctx, m, apiserverfront.Client{Path: l.FrontControl()}, admin, kcpserver.RollConfig{ SlotsFile: l.SlotsFile(), Shared: l.SharedDir(), MemberToken: kcpserver.MemberTokenPath(p.Dir), SpecFor: func(slot string) (hostsvc.UnitSpec, string, error) { return p.slotSpec(slot, frontURL) }, KineFor: func(slot string) string { return p.KineURL(app.KineOf(slot)) }, Backup: p.storeBackup(cfg.Retention, cfg.log()), ReadyTimeout: cfg.ReadyTimeout, OnListen: cfg.OnListen, Log: cfg.log(), }) if after.Active == rep.To && rep.NewUnit != "" { if id, ierr := m.Identify(ctx, rep.NewUnit, 0); ierr == nil { rec.SetUnit(devinstance.Unit{Role: SlotRole(rep.To), Unit: rep.NewUnit, Identity: id, Port: rep.NewPort, Endpoint: endpoint(SlotRole(rep.To), rep.NewPort), PlacedAt: rep.StartedAt, PortAt: rep.PortAt, ReadyAt: rep.ReadyAt, StartedTo: rep.ReadyAt.Sub(rep.StartedAt).Round(time.Millisecond).String(), RSSKiB: RSSKiB(id.PID)}) } if !rep.OldExitedAt.IsZero() { rec.RemoveUnit(SlotRole(rep.From)) } rec.Updated = time.Now().UTC() } return rep, err }