johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package bootstrap

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"net"
	"os"
	"path/filepath"
	"time"

	"strings"

	"github.com/johnandersen777/socialweb-computer-kcp/internal/apiserverfront"
	"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/versions"
)

type InstalledRoll struct {
	Kcp           string
	Kine          string
	PlannedOutage bool
	Retention     kcpdatastore.BackupRetention
	Compat        []versions.KcpCompat
	ReadyTimeout  time.Duration
	OnListen      func(name string, addr net.Addr)
	Log           *slog.Logger
}

type InstalledRollReport struct {
	Pair      kcpserver.PairVerdict `json:"pair"`
	KcpFrom   kcpserver.Asset       `json:"kcpFrom"`
	KcpTo     kcpserver.Asset       `json:"kcpTo"`
	KineFrom  string                `json:"kineFrom"`
	KineTo    string                `json:"kineTo"`
	Roll      kcpserver.RollReport  `json:"roll"`
	Slots     kcpserver.Slots       `json:"slots"`
	Boot      []string              `json:"boot,omitempty"`
	BootError string                `json:"bootError,omitempty"`
}

func resolveTarget(link, target string) (string, error) {
	cur, err := filepath.EvalSymlinks(link)
	if err != nil {
		return "", fmt.Errorf("bootstrap: the asset link %s: %w", link, err)
	}
	if target == "" {
		return cur, nil
	}
	if !filepath.IsAbs(target) {
		return "", fmt.Errorf("bootstrap: asset %s must be an absolute path", target)
	}
	return filepath.EvalSymlinks(target)
}

func RollInstalled(ctx context.Context, m hostsvc.Manager, p Params, o InstalledRoll) (InstalledRollReport, error) {
	log := o.Log
	if log == nil {
		log = slog.Default()
	}
	var out InstalledRollReport
	slots, err := kcpserver.ReadSlots(p.SlotsFile())
	if err != nil {
		return out, fmt.Errorf("bootstrap: no %s; this installed kcp has not booted: %w", p.SlotsFile(), err)
	}
	from := slots.Active
	if out.KcpFrom, err = kcpserver.ReadAsset(ctx, p.KcpPath); err != nil {
		return out, err
	}
	out.KcpTo = out.KcpFrom
	if o.Kcp != "" {
		if out.KcpTo, err = kcpserver.ReadAsset(ctx, o.Kcp); err != nil {
			return out, err
		}
	}
	if out.KineFrom, err = resolveTarget(p.KinePath, ""); err != nil {
		return out, err
	}
	if out.KineTo, err = resolveTarget(p.KinePath, o.Kine); err != nil {
		return out, err
	}
	out.Pair, err = kcpserver.CheckPair(out.KcpFrom, out.KcpTo, o.Compat, o.PlannedOutage)
	if err != nil {
		return out, err
	}
	app := p.App
	kcpChanged := out.KcpTo.Path != out.KcpFrom.Path
	kineChanged := out.KineTo != out.KineFrom
	outage := out.Pair.Mode == kcpserver.ModePlannedOutage || slots.Shape != kcpserver.ShapeBlueGreen
	to := kcpserver.Other(from)
	if slots.Shape != kcpserver.ShapeBlueGreen {
		to = from
		out.Pair.Mode = kcpserver.ModePlannedOutage
		out.Pair.Reason += "; a single-slot kcp rolls as a short restart of its slot"
	}
	if kineChanged && !outage && app.KineOf(to) == app.KineOf(from) {
		return out, fmt.Errorf("bootstrap: slots %s and %s share kine %s; a kine change needs --planned-outage", from, to, app.KineOf(from))
	}
	a, err := kcpserver.ReadSharedAdmin(p.SharedDir())
	if err != nil {
		return out, err
	}
	frontURL := slots.Front.URL()
	admin, err := kcpserver.NewHTTPAdmin(a, frontURL)
	if err != nil {
		return out, err
	}
	for _, slot := range app.Slots {
		if err := kcpserver.WriteSlotFiles(p.SlotRoot(slot)); err != nil {
			return out, err
		}
	}
	names := app.Names()
	retargeted, aborted := false, false
	point := func(kcp, kine string) error {
		if err := retarget(p.KcpPath, kcp); err != nil {
			return err
		}
		return retarget(p.KinePath, kine)
	}
	timeout := o.ReadyTimeout
	timeout = opt.Positive(timeout, DefaultReadyTimeout)
	start := func(ctx context.Context, slot string, spec hostsvc.UnitSpec) (hostsvc.UnitStatus, error) {
		if (kcpChanged || kineChanged) && !retargeted && !aborted {
			if err := point(out.KcpTo.Path, out.KineTo); err != nil {
				return hostsvc.UnitStatus{}, err
			}
			retargeted = true
			log.Info("asset links retargeted", "kcp", out.KcpTo.Path, "kine", out.KineTo)
		}
		ks := app.KineOf(slot)
		kine := names.Kine(ks)
		if ks != app.KineOf(from) || kineChanged {
			m.Stop(ctx, kine)
			m.ResetFailed(ctx, kine)
		}
		if err := os.MkdirAll(p.KineDir(ks), 0o700); err != nil {
			return hostsvc.UnitStatus{}, err
		}
		if _, err := hostsvc.StartInstalled(ctx, m, kine); err != nil {
			return hostsvc.UnitStatus{}, err
		}
		kctx, cancel := context.WithTimeout(ctx, timeout)
		defer cancel()
		if err := waitUnixSocket(kctx, p.KineSocket(ks)); err != nil {
			return hostsvc.UnitStatus{}, fmt.Errorf("bootstrap: %s: %w", kine, err)
		}
		f, err := hostsvc.RenderInstalled(spec, hostsvc.InstallOptions{EnvFile: p.EnvFile(spec.Name)})
		if err != nil {
			return hostsvc.UnitStatus{}, err
		}
		if err := writePrivate(f.EnvPath, f.Env); err != nil {
			return hostsvc.UnitStatus{}, err
		}
		return hostsvc.StartInstalled(ctx, m, spec.Name)
	}
	abort := func(context.Context) error {
		if aborted {
			return nil
		}
		aborted = true
		if !retargeted {
			return nil
		}
		log.Warn("roll aborted; the asset links point back at the previous assets", "kcp", out.KcpFrom.Path, "kine", out.KineFrom)
		return point(out.KcpFrom.Path, out.KineFrom)
	}
	retire := func(ctx context.Context, old, current string) error {
		if app.KineOf(old) == app.KineOf(current) {
			return nil
		}
		kine := names.Kine(app.KineOf(old))
		if err := m.Stop(ctx, kine); err != nil {
			return err
		}
		m.ResetFailed(ctx, kine)
		return nil
	}
	rep, after, rerr := kcpserver.Roll(ctx, m, apiserverfront.Client{Path: p.FrontControl()}, admin, kcpserver.RollConfig{
		SlotsFile: p.SlotsFile(), Shared: p.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(o.Retention, log),
		Start:    start,
		Retire:   retire,
		Abort:    abort,
		Describe: func(string) (string, string) { return out.KcpTo.SHA256, out.KcpTo.Version },
		Outage:   outage, ReadyTimeout: o.ReadyTimeout, OnListen: o.OnListen, Log: log,
	})
	out.Roll, out.Slots = rep, after
	if rerr != nil && after.Active != rep.To {
		if err := abort(ctx); err != nil {
			rerr = errors.Join(rerr, err)
		}
	}
	if after.Active == rep.To && rep.To != rep.From {
		enable := []string{names.Slot(rep.To), names.Kine(app.KineOf(rep.To))}
		disable := []string{names.Slot(rep.From)}
		if app.KineOf(rep.From) != app.KineOf(rep.To) {
			disable = append(disable, names.Kine(app.KineOf(rep.From)))
		}
		if err := hostsvc.SetBoot(ctx, m, enable, disable); err != nil {
			out.BootError = err.Error()
			if !errors.Is(err, hostsvc.ErrUnsupported) {
				rerr = errors.Join(rerr, err)
			}
		} else {
			out.Boot = enable
		}
		pruneInactiveAuditLogs(p.SlotRoot(rep.From), log)
	}
	return out, rerr
}

func pruneInactiveAuditLogs(root string, log *slog.Logger) {
	entries, err := os.ReadDir(root)
	if err != nil {
		return
	}
	removed := 0
	for _, e := range entries {
		if e.IsDir() || !strings.HasPrefix(e.Name(), "audit-") || !strings.HasSuffix(e.Name(), ".log") {
			continue
		}
		if err := os.Remove(filepath.Join(root, e.Name())); err == nil {
			removed++
		}
	}
	if removed > 0 {
		log.Info("removed the audit logs of the retired kcp slot", "slotRoot", root, "files", removed)
	}
}

func writePrivate(path string, b []byte) error {
	if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
		return err
	}
	tmp := path + ".tmp"
	if err := os.WriteFile(tmp, b, 0o600); err != nil {
		return err
	}
	return os.Rename(tmp, path)
}