liken

1// Package api is the grammar of the liken.sh API group. It holds the2// shapes and the vocabulary that every liken document shares, and3// nothing more.4//5// liken's documents include the Machine, the Cluster, and the6// smaller records that use the same machinery. Kubernetes resources7// shape all of them, and this package defines that shape. It8// declares the group and version, and the metadata each document9// contains. It declares the conditions and phases each status uses,10// the role vocabulary both documents use, and the version grammar11// that names releases. The machine package and the cluster package12// each own one document, and both build on this package. Neither13// package imports the other. Kubernetes uses the same layering:14// every typed API embeds metav1, and no typed API imports another.15//16// This package covers a narrow scope, chosen on purpose: only facts17// that every document shares belong here. Nothing behavioral belongs18// here. Nothing about storage or staging belongs here. A type that19// names one document is in the wrong package.20//21// k8s.io/apimachinery already defines these shapes, as22// metav1.ObjectMeta and metav1.Condition. liken redeclares them on23// purpose. The kubernetes package speaks to the API server with only24// net/http, for the same reason: liken's only Kubernetes dependency25// is the YAML converter. Importing apimachinery for two structs26// would add its whole dependency tree and release cadence. The27// narrow copies also do real work. metav1.ObjectMeta has about28// twenty-five fields, and liken uses four of them. Every liken29// document parses strictly, so liken refuses metadata it would30// otherwise ignore silently: labels and owner references in a31// hand-written manifest. The wire format stays compatible with the32// Kubernetes convention. The Go types represent only the fields33// liken uses.34package api3536// APIVersion is the full group/version string every liken document37// declares. It is also the URL segment operators use to speak to the38// API server: /apis/liken.sh/v1alpha1/machines. CRD groups are DNS39// names, and liken owns liken.sh.40const APIVersion = "liken.sh/v1alpha1"4142// ObjectMeta is the slice of Kubernetes object metadata that liken43// uses. Name identifies the object. ResourceVersion is the cluster's44// optimistic-concurrency counter. The operator sends it back when it45// watches, and the server resumes the stream from that point.46// Generation counts changes to the spec. The API server increases47// generation on spec writes and leaves it unchanged on status48// writes. This lets a condition record which version of the spec it49// judged.50//51// Annotations carry the signals that are not desired state. A spec52// declares what should hold, and a controller works until it holds.53// A request such as "poll the channel now" does not fit that54// contract: it asks for one action, not a standing state. Kubernetes55// puts these signals in annotations (kubectl writes restartedAt to56// request a Deployment rollout the same way), and liken follows.57// The operators that render canonical documents build their metadata58// fresh, from the name alone, so an annotation can never change a59// staged document's hash, and an annotation edit can never reboot a60// machine.61//62// UID names one instance of an object, so an owner reference to it63// does not also name an object that is deleted and created again under64// the same name. The machine operator reads its Machine's UID to name65// the Machine as the owner of its heartbeat lease.66type ObjectMeta struct {67	Name            string            `json:"name"`68	UID             string            `json:"uid,omitempty"`69	ResourceVersion string            `json:"resourceVersion,omitempty"`70	Generation      int64             `json:"generation,omitempty"`71	Annotations     map[string]string `json:"annotations,omitempty"`72}7374// Meta is the part of an object's metadata that a watch's store and an75// operator's memo of its own writes read: the shared module's76// memo.Meta. Both are aliases of the same interface literal, so the77// Machine and the Cluster answer memo.Meta with no import of that78// module. init links this package, and the module's client would bring79// net/http into init.80type Meta = interface {81	GetNamespace() string82	GetName() string83	GetResourceVersion() string84}8586// GetName answers the object's name.87func (m ObjectMeta) GetName() string { return m.Name }8889// GetNamespace answers the empty namespace, because every liken kind90// is cluster-scoped.91func (m ObjectMeta) GetNamespace() string { return "" }9293// GetResourceVersion answers the version of the copy.94func (m ObjectMeta) GetResourceVersion() string { return m.ResourceVersion }9596// Phase summarizes a document's state in one word. Go models a97// closed vocabulary with a named string type. The constants below98// are the only legal values. The compiler catches an error when code99// passes a Phase where a Role belongs. The wire format stays100// unchanged, because a named string marshals exactly like a bare101// string. Kubernetes' own API types use the same method: v1.PodPhase102// and metav1.ConditionStatus.103type Phase string104105// These are the phases a machine can report, listed from most severe106// to least severe. Each phase summarizes the conditions; it is not a107// fact on its own. The operator derives the phase from the108// conditions on every pass. The table that does this work,109// machine-operator/phase.go, maps each condition to the phase it110// produces. Lost is the exception to this rule. A machine that is111// Lost cannot write its own status, so the cluster operator writes112// Lost on the machine's behalf when the machine's heartbeat stops.113// The Cluster's status reuses this same vocabulary (Ready, Updating,114// Degraded), so a fleet and its machines use the same words to115// describe their state.116const (117	PhaseUnknown       Phase = "Unknown"       // the facts are unreadable, so the operator cannot judge the machine's state118	PhaseBooting       Phase = "Booting"       // init has not yet finished publishing this boot's record119	PhaseLost          Phase = "Lost"          // the heartbeat stopped, and the cluster operator wrote this, not the machine120	PhaseBlocked       Phase = "Blocked"       // drift exists but the system cannot stage it; it needs a different edit, not more time121	PhaseUpdating      Phase = "Updating"      // a reboot is under way to apply a staged change122	PhaseUpdatePending Phase = "UpdatePending" // a change is staged and waits for a Manual reboot123	PhaseDegraded      Phase = "Degraded"      // something is wrong that does not match one of the specific states above124	PhaseDownloading   Phase = "Downloading"   // a release downloads to the inactive slot while the machine serves its workloads125	PhaseReady         Phase = "Ready"         // every condition is True126)127128// Role is what a machine is in its cluster. There are exactly two129// roles. Leaders run a control plane: an API server, a scheduler,130// and the datastore. Followers run workloads and take direction from131// the leaders. k3s calls these roles "server" and "agent". liken132// translates between the two names in exactly one place, the moment133// it execs k3s (init/supervisor.go), and uses leader and follower134// everywhere else. Both documents share this vocabulary directly:135// the Cluster's spec declares the leaders, and each Machine's status136// reports the role it derived.137type Role string138139const (140	RoleLeader   Role = "leader"141	RoleFollower Role = "follower"142)

1package api23// Conditions are how a liken document's status carries observations:4// a set of typed, timestamped verdicts ("Ready", "SysctlsApplied")5// that controllers maintain and humans and tooling read. The shape6// and the rules here mirror metav1.Condition, the method Kubernetes7// uses everywhere (Pods, Nodes, and Deployments all carry these), so8// anyone who reads `kubectl describe` output already knows how to9// read a liken document.1011import (12	"slices"13	"time"1415	"github.com/liken-sh/liken/kubernetes/conditions"16)1718// The condition type is the one every liken component declares: the19// shared module's conditions.Condition, which has the JSON shape of20// metav1.Condition. The aliases keep the names that the Machine and21// Cluster packages, the operators, and the CLI already use. The22// shared package imports only time, so init, which reads these types23// and must link no HTTP client, links nothing more.24//25// ConditionStatus is a condition's verdict. It is a string rather26// than a bool because there is a third state: a controller must be27// able to say when it currently cannot tell.28type ConditionStatus = conditions.Status2930const (31	ConditionTrue    = conditions.True32	ConditionFalse   = conditions.False33	ConditionUnknown = conditions.Unknown34)3536// Condition mirrors metav1.Condition. ObservedGeneration records37// which metadata.generation the condition judged. Generation counts38// spec edits, so a reader can tell "Ready, for the spec as it stands"39// apart from "Ready, but for a spec two edits ago". That difference40// matters in liken, where edits wait for a reboot to take effect. (The41// convergence conditions make the stronger, content-hashed version of42// this claim. The generation is for tooling that reads the43// convention.)44//45// The CRDs require a reason of at least one character, and accept an46// empty message, so a condition encodes the same way with or without47// omitempty on those two fields.48type Condition = conditions.Condition4950// SetCondition adds or updates a condition by type, through51// conditions.Set, at the time now. It preserves the Kubernetes rule52// that makes lastTransitionTime meaningful: the time moves only when53// Status flips, not on every write. This is what lets `kubectl get`54// answer "how long has this machine been Ready?" instead of only55// "when did the operator last say so?".56func SetCondition(list []Condition, c Condition, now time.Time) []Condition {57	c.LastTransitionTime = now58	conditions.Set(&list, c)59	return list60}6162// Transitions answers each condition in after that transitioned from63// before, by the rule of conditions.Set: the condition is new, or its64// status or its reason changed. An operator composes a whole status,65// writes it, and then posts one Event for each transition. Comparing66// the written list with the stored one, after the write lands, means67// a refused write posts nothing, and the next pass finds the same68// transition again.69func Transitions(before, after []Condition) []Condition {70	var out []Condition71	for _, c := range after {72		held := FindCondition(before, c.Type)73		if held == nil || held.Status != c.Status || held.Reason != c.Reason {74			out = append(out, c)75		}76	}77	return out78}7980// FindCondition returns the condition of the named type, or nil when81// the list holds none.82func FindCondition(list []Condition, conditionType string) *Condition {83	for i := range list {84		if list[i].Type == conditionType {85			return &list[i]86		}87	}88	return nil89}9091// RemoveCondition drops the condition of the named type. Most92// conditions are observations: they stay in the list and flip93// between True and False. Removal exists for the conditions that are94// grants. A grant is present while it is extended and gone when it95// is revoked. Its absence carries the meaning, so there is no False96// state for other machinery to misread as trouble.97func RemoveCondition(list []Condition, conditionType string) []Condition {98	for i := range list {99		if list[i].Type == conditionType {100			return slices.Delete(list, i, i+1)101		}102	}103	return list104}

1package api23// This file defines liken's version grammar and the ordering over it.4//5// A liken version is a calendar date and a serial: yyyy.mm.dd-nnn,6// with every field zero-padded, and the serial starts at 001 and7// counts up within the day. The number says when a release was cut8// and nothing else. liken does not use semantic versioning on9// purpose: an OS release carries a kernel version and a Kubernetes10// version of its own, and a semver major on the outside would only11// invite readings like "liken 2.0 must mean kubernetes 2". There is12// also no compatibility boundary for a major version to mark. Every13// release is expected to take over from the release before it, and14// to carry the machine's layer and on-disk state forward, so the15// code keeps compatibility, and the version number does not report16// it. What shipped inside a release is recorded where it belongs, in17// the release document's components.18//19// Versions are immutable: a new serial supersedes a bad release; the20// system never rebuilds a release under the same name. The catalog21// on the Cluster names releases by these versions, and two questions22// need an ordering over them: which catalog entry is the newest (the23// NEWEST printer column), and whether a machine is behind its24// target.2526import (27	"fmt"28	"regexp"29	"strings"30	"time"31)3233// versionShape is the grammar, shared verbatim with the Cluster34// CRD's pattern on spec.version and the catalog: a four-digit year,35// a two-digit month and day, a dash, and a three-digit serial.36var versionShape = regexp.MustCompile(`^\d{4}\.\d{2}\.\d{2}-\d{3}$`)3738// ValidVersion checks a version against the grammar at authoring39// time. This lets the system refuse a malformed version when it40// bundles a release, rather than let it be discovered when a machine41// fails to fetch it. Beyond the shape, the date must be a real42// calendar date: the CRD's pattern cannot check that 2026.02.30 never43// happened, but this function can.44func ValidVersion(v string) error {45	if !versionShape.MatchString(v) {46		return fmt.Errorf("version %q is not yyyy.mm.dd-nnn (for example 2026.07.11-001)", v)47	}48	date, _, _ := strings.Cut(v, "-")49	if _, err := time.Parse("2006.01.02", date); err != nil {50		return fmt.Errorf("version %q does not name a real date", v)51	}52	return nil53}5455// CompareVersions orders two version strings. It returns a negative56// number when a is older, zero when they name the same version, and57// a positive number when a is newer. Every field of the grammar is58// zero-padded to a fixed width, so plain string comparison gives the59// correct order. The padding exists exactly so that no field ever60// needs numeric parsing to sort: as bare numbers, "010" and "9"61// would sort backwards.62func CompareVersions(a, b string) int {63	return strings.Compare(a, b)64}

1package main23// approve-reboot is the one cluster command that carries liken's own4// meaning instead of handing the terminal to another tool. A machine5// whose rebootPolicy is Manual stages a change and waits for a6// person. This command is how the person answers: it reads7// status.pending, reports what is waiting, and writes the8// approve-disruption annotation with the staged document's hash. The9// hash makes the grant one-shot (machine.ApproveDisruptionAnnotation10// explains why), and writing the same annotation twice is the same11// grant, so the command is repeatable.1213import (14	"flag"15	"fmt"16	"io"17	"strings"1819	"github.com/liken-sh/liken/liken/api"20	"github.com/liken-sh/liken/liken/kubernetes"21	"github.com/liken-sh/liken/liken/machine"22)2324// chooseApproval picks which pending entry to approve. A reboot25// applies every staged document at once, so when a reboot-class26// change is pending, approving it covers the rest. Otherwise the27// first entry stands for the list: with several restart-class28// changes pending, each applies on its own grant, and running the29// command again approves the next.30func chooseApproval(pending []machine.PendingDisruption) *machine.PendingDisruption {31	for i := range pending {32		if pending[i].Kind == machine.DisruptionReboot {33			return &pending[i]34		}35	}36	if len(pending) > 0 {37		return &pending[0]38	}39	return nil40}4142// renderPending reports what the machine waits for, in the shape a43// person reads before deciding to grant: each change's condition and44// reason, its summary with the short hash a grant names, and what45// kind of disruption applies it.46func renderPending(m *machine.Machine) string {47	name := m.Metadata.Name48	pending := m.Status.Pending49	if len(pending) == 0 {50		return fmt.Sprintf("%s is converged; nothing is waiting\n", name)51	}52	count := fmt.Sprintf("%d changes", len(pending))53	if len(pending) == 1 {54		count = "one change"55	}56	var b strings.Builder57	fmt.Fprintf(&b, "%s is waiting on %s:\n", name, count)58	for _, p := range pending {59		condition := p.Condition60		if c := api.FindCondition(m.Status.Conditions, p.Condition); c != nil {61			condition = fmt.Sprintf("%s  %s", p.Condition, c.Reason)62		}63		applies := "a reboot applies this, and every other staged change with it"64		if p.Kind == machine.DisruptionRestart {65			applies = "a k3s restart applies this; the machine does not reboot"66		}67		fmt.Fprintf(&b, "  %s\n  %s (%.12s)\n  %s\n", condition, p.Summary, p.Hash, applies)68	}69	return b.String()70}7172// approveReboot is the command: resolve the credential, read the73// machine, report, and grant.74func approveReboot(args []string, out io.Writer) error {75	fs := flag.NewFlagSet("approve-reboot", flag.ContinueOnError)76	server := fs.String("server", "", "the API server address, when cluster.yaml's endpoint is not reachable from here")77	if err := fs.Parse(args); err != nil {78		return err79	}80	if fs.NArg() != 2 {81		return fmt.Errorf("usage: liken approve-reboot [-server URL] <deployment-dir> <machine>")82	}83	kubeconfigPath, err := writeKubeconfig(fs.Arg(0), *server, io.Discard)84	if err != nil {85		return err86	}87	c, err := kubernetes.KubeconfigClient(kubeconfigPath)88	if err != nil {89		return err90	}91	m, err := kubernetes.GetMachine(c, fs.Arg(1))92	if err != nil {93		return err94	}95	fmt.Fprint(out, renderPending(m))96	choice := chooseApproval(m.Status.Pending)97	if choice == nil {98		return nil99	}100	short := fmt.Sprintf("%.12s", choice.Hash)101	patch := fmt.Sprintf(`{"metadata":{"annotations":{%q:%q}}}`, machine.ApproveDisruptionAnnotation, short)102	if err := kubernetes.PatchJSON(c, kubernetes.MachinesPath+"/"+m.Metadata.Name, []byte(patch)); err != nil {103		return err104	}105	fmt.Fprintf(out, "\napproved: %s=%s\n", machine.ApproveDisruptionAnnotation, short)106	return nil107}

1package main23// This file is the cluster half of the toolkit: how a command on an4// operator's workstation finds the cluster and holds a credential5// for it.6//7// The one positional argument these commands share is the8// deployment directory, the layout GETTING-STARTED.md documents:9// identity sits at <dir>/identity, and the endpoint comes from10// <dir>/cluster.yaml. The endpoint needs a caveat, and -server11// exists because of it. spec.endpoint is the address a follower12// joins through from inside the cluster's own network segment. The13// operator's workstation often cannot reach that address: a dev lab14// reaches its guests through a forwarded localhost port instead.15// So cluster.yaml is the default and -server overrides it.16//17// The kubeconfig goes to a predictable path that these commands18// rewrite and reuse: <dir>/identity/kubeconfig, with 060019// permissions, beside the identity it is derived from, because that20// is already the directory an operator guards. A fresh file per run21// would leak, because the passthrough commands replace this process22// with the tool (exec), so no deferred cleanup ever runs.2324import (25	"flag"26	"fmt"27	"io"28	"os"29	"os/exec"30	"path/filepath"31	"strings"3233	"github.com/liken-sh/liken/liken/cluster"34	"github.com/liken-sh/liken/liken/identity"35	"golang.org/x/sys/unix"36)3738// resolveCluster decides where the cluster is and what it is named.39// The endpoint is the -server flag when given, otherwise the40// endpoint in the deployment's cluster.yaml. A deployment with no41// cluster.yaml at all is a single machine on its own (see42// cluster.LoadCluster), which never declares an endpoint, so that43// case also asks for -server. The name comes from the same44// document's metadata, so that the kubeconfig for each cluster an45// operator holds labels its entries with that cluster's own name. A46// deployment with no cluster.yaml has no name, and falls back to47// liken.48func resolveCluster(dir, server string) (name, endpoint string, err error) {49	c, err := cluster.LoadCluster(filepath.Join(dir, "cluster.yaml"))50	if err != nil {51		return "", "", fmt.Errorf("reading the deployment's cluster.yaml: %w", err)52	}53	name = "liken"54	if c != nil && c.Metadata.Name != "" {55		name = c.Metadata.Name56	}57	if server != "" {58		return name, server, nil59	}60	if c == nil || c.Spec.Endpoint == "" {61		return "", "", fmt.Errorf("%s/cluster.yaml declares no endpoint; pass -server", dir)62	}63	return name, c.Spec.Endpoint, nil64}6566// writeKubeconfig resolves the cluster's name and endpoint, mints67// the admin credential into the deployment's identity directory, and68// returns the kubeconfig's path.69func writeKubeconfig(dir, server string, out io.Writer) (string, error) {70	name, endpoint, err := resolveCluster(dir, server)71	if err != nil {72		return "", err73	}74	identityDir := filepath.Join(dir, "identity")75	if err := identity.Kubeconfig(identityDir, name, endpoint, out); err != nil {76		return "", err77	}78	return filepath.Join(identityDir, "kubeconfig"), nil79}8081// execTool is exec(2): it replaces this process with the tool, so82// the tool owns the terminal and its exit code is liken's. It is a83// variable so tests can observe the call instead of vanishing into84// the exec.85var execTool = unix.Exec8687// envWith returns the environment with one variable bound to one88// value, whether or not it was set before. Appending a duplicate89// would not do: when a variable appears twice in an environment,90// which copy a program reads is the program's own accident.91func envWith(environ []string, key, value string) []string {92	kept := make([]string, 0, len(environ)+1)93	for _, entry := range environ {94		if !strings.HasPrefix(entry, key+"=") {95			kept = append(kept, entry)96		}97	}98	return append(kept, key+"="+value)99}100101// passthrough runs one of the tools an operator already uses,102// against the right cluster: it resolves the credential, sets103// KUBECONFIG, and replaces this process with the tool from PATH.104// liken does not reimplement kubectl, and it does not vendor these105// binaries either: every command here is a credential and an106// endpoint handed to a program the operator already chose to107// install.108func passthrough(tool string, args []string) error {109	fs := flag.NewFlagSet(tool, flag.ContinueOnError)110	server := fs.String("server", "", "the API server address, when cluster.yaml's endpoint is not reachable from here")111	if err := fs.Parse(args); err != nil {112		return err113	}114	if fs.NArg() < 1 {115		return fmt.Errorf("usage: liken %s [-server URL] <deployment-dir> [args...]", tool)116	}117	kubeconfigPath, err := writeKubeconfig(fs.Arg(0), *server, io.Discard)118	if err != nil {119		return err120	}121	path, err := exec.LookPath(tool)122	if err != nil {123		return fmt.Errorf("%s is not on PATH; install it and run this again", tool)124	}125	return execTool(path,126		append([]string{tool}, fs.Args()[1:]...),127		envWith(os.Environ(), "KUBECONFIG", kubeconfigPath))128}

1package main23// How the base CLI completes a command line for the shell, and4// the contract the per-operator CLIs copy. kubectl completes a plugin5// by running an executable named kubectl_complete-<name> on PATH, and6// that shim runs this binary's hidden __complete verb. The verb answers7// in cobra's completion protocol: one candidate per line, then a final8// ":N" line that carries the ShellCompDirective bitmask. The base CLI9// answers from static lists, because its verbs, its plugins10// subcommands, and its plugin domains are all known without a cluster.1112import (13	"fmt"14	"io"15	"sort"16	"strings"17)1819// The ShellCompDirective bits this CLI emits, cobra's own20// numbering. The default is 0, which leaves the shell's own file21// completion in place. NoFileComp stops the shell from offering file22// names when the CLI has a closed set of its own, and Error marks a23// request the CLI could not answer.24const (25	compDirectiveError      = 126	compDirectiveNoFileComp = 427	compDirectiveDefault    = 028)2930// The plugins group, the one verb with subcommands of its own.31const pluginsVerb = "plugins"3233// Every top-level verb the dispatcher answers, minus completion34// and __complete, which serve the shell and never a person. A new verb35// in run() joins this list so the shell offers it.36var topLevelVerbs = []string{37	"new", "mint", "adopt", "kubeconfig", "kubectl", "stern", "flux",38	"approve-reboot", "request-reboot", "plugins", "layer", "fetch",39	"media", "stick", "bundle", "index", "serve", "version",40}4142// The plugin domains the walk dispatches to a43// kubectl-liken-<domain> binary. The list is static because the base44// CLI ships with a release and completes before any cluster is reached,45// so the shell offers a domain the same whether its CLI is installed or46// not.47var pluginDomains = []string{"audio", "bluetooth", "display", "library", "media"}4849// The plugins group's own subcommands.50var pluginsSubcommands = []string{"list", "remove", "sync"}5152// topLevelCandidates is the verbs and the plugin domains as one53// sorted set with no repeat, so "liken <TAB>" offers a command and a54// domain side by side. media names both a verb and a domain, and it55// appears once.56func topLevelCandidates() []string {57	seen := map[string]bool{}58	var all []string59	for _, item := range topLevelVerbs {60		if !seen[item] {61			seen[item] = true62			all = append(all, item)63		}64	}65	for _, item := range pluginDomains {66		if !seen[item] {67			seen[item] = true68			all = append(all, item)69		}70	}71	sort.Strings(all)72	return all73}7475// One flag the CLI accepts, and whether it reads a value from the76// next word. Completion walks the words with this table so it can tell a77// flag's value apart from a positional argument.78type completionFlag struct {79	name       string80	takesValue bool81}8283// Every flag the subcommands bind, in the completion table's own84// form. The base CLI parses each subcommand with the standard flag85// package, which reads one and two dashes alike, so the table names the86// one-dash spelling the usage prints. Each of these reads a value.87var completionFlags = []completionFlag{88	{name: "-server", takesValue: true},89	{name: "-digest", takesValue: true},90	{name: "-source", takesValue: true},91	{name: "-slot-size", takesValue: true},92	{name: "-console", takesValue: true},93}9495// What the words before the cursor amount to: the positional96// arguments already given, and the one flag whose value the cursor is97// about to type.98type completionParse struct {99	positionals      []string100	pendingValueFlag string101}102103// runComplete answers a __complete request and writes the protocol to104// stdout. It stays silent on every failure, because a completer that105// prints an error corrupts the command line the shell is drawing.106func runComplete(args []string, stdout io.Writer) error {107	candidates, directive := completeArgs(args)108	for _, candidate := range candidates {109		fmt.Fprintln(stdout, candidate)110	}111	fmt.Fprintf(stdout, ":%d\n", directive)112	return nil113}114115// completeArgs reads the request words and returns the candidates. The last116// word is the one under the cursor, and the words before it are already117// typed. It is the whole of the completion logic, so a test drives it118// with words alone.119func completeArgs(args []string) ([]string, int) {120	if len(args) == 0 {121		return topLevelCandidates(), compDirectiveNoFileComp122	}123	toComplete := args[len(args)-1]124	parse := parseCompletionArgs(args[:len(args)-1])125126	// The word under the cursor is the value of a flag that reads127	// one, so complete that flag's values, not a positional.128	if parse.pendingValueFlag != "" {129		return completeFlagValue(parse.pendingValueFlag, toComplete)130	}131132	// The word under the cursor starts a flag, so complete flag133	// names.134	if strings.HasPrefix(toComplete, "-") {135		return filterByPrefix(flagNames(), toComplete), compDirectiveNoFileComp136	}137138	// The word is a positional. With none given yet it names a139	// verb or a plugin domain; after the plugins verb it names a plugins140	// subcommand; a verb's own argument is a path, so the shell's file141	// completion answers it.142	switch {143	case len(parse.positionals) == 0:144		return filterByPrefix(topLevelCandidates(), toComplete), compDirectiveNoFileComp145	case len(parse.positionals) == 1 && parse.positionals[0] == pluginsVerb:146		return filterByPrefix(pluginsSubcommands, toComplete), compDirectiveNoFileComp147	default:148		return nil, compDirectiveDefault149	}150}151152// parseCompletionArgs walks the words before the cursor into their153// positionals, and marks a flag left waiting for its value.154func parseCompletionArgs(prior []string) completionParse {155	var parse completionParse156	for i := 0; i < len(prior); i++ {157		token := prior[i]158		if len(token) > 1 && token[0] == '-' {159			name, _, hasEquals := splitFlag(token)160			flag, ok := lookupCompletionFlag(name)161			if !ok || !flag.takesValue {162				continue163			}164			if hasEquals {165				continue166			}167			if i+1 < len(prior) {168				i++169				continue170			}171			parse.pendingValueFlag = name172			continue173		}174		parse.positionals = append(parse.positionals, token)175	}176	return parse177}178179// splitFlag splits a "-name=value" word into its name and value, and180// reports a bare "-name" as a name with no value. The standard flag181// package reads one and two dashes alike, so a "--name" spelling splits182// the same way.183func splitFlag(token string) (name, value string, hasEquals bool) {184	if equals := strings.IndexByte(token, '='); equals >= 0 {185		return token[:equals], token[equals+1:], true186	}187	return token, "", false188}189190// lookupCompletionFlag finds a flag by its exact name, and reads a191// "--name" spelling as the "-name" the table holds.192func lookupCompletionFlag(name string) (completionFlag, bool) {193	name = "-" + strings.TrimLeft(name, "-")194	for _, flag := range completionFlags {195		if flag.name == name {196			return flag, true197		}198	}199	return completionFlag{}, false200}201202// flagNames lists every flag name completion can offer.203func flagNames() []string {204	names := make([]string, 0, len(completionFlags))205	for _, flag := range completionFlags {206		names = append(names, flag.name)207	}208	return names209}210211// completeFlagValue offers the values a flag accepts. Every base212// flag names an address, a digest, a size, or a console, none of which213// this CLI can enumerate, so it offers nothing and leaves the file214// fallback off.215func completeFlagValue(flag, toComplete string) ([]string, int) {216	return nil, compDirectiveNoFileComp217}218219// filterByPrefix keeps the candidates the cursor's word begins.220func filterByPrefix(items []string, prefix string) []string {221	var kept []string222	for _, item := range items {223		if strings.HasPrefix(item, prefix) {224			kept = append(kept, item)225		}226	}227	return kept228}229230// completionScript prints the bash script that completes a direct231// invocation of the binary, under the liken name and the kubectl-liken232// name both. The shell sources it, and from then on it runs the hidden233// __complete verb and reads the protocol back. kubectl needs none of234// this: it runs the kubectl_complete-liken shim instead, which runs the235// same verb.236func completionScript(shell string, stdout io.Writer) error {237	if shell != "bash" {238		return fmt.Errorf("completion supports bash, not %q", shell)239	}240	fmt.Fprint(stdout, bashCompletionScript)241	return nil242}243244// The bash completion for a direct liken or kubectl-liken call. It245// mirrors the small part of cobra's generated script this CLI needs: run246// __complete with the words before the cursor and the word under it, read247// the candidates, and read the final ":N" directive. A per-operator CLI248// copies this text and changes only the binary name and the two function249// names.250const bashCompletionScript = `# bash completion for liken and kubectl-liken251__kubectl_liken_complete()252{253    local cur args out comp directive254    COMPREPLY=()255    cur="${COMP_WORDS[COMP_CWORD]}"256257    # The words after the binary name and before the cursor, then the258    # word under it, go to the hidden __complete verb.259    args=("${COMP_WORDS[@]:1:COMP_CWORD-1}")260    out="$("${COMP_WORDS[0]}" __complete "${args[@]}" "$cur" 2>/dev/null)"261262    # The last line is ":N", the ShellCompDirective bitmask.263    directive="${out##*$'\n'}"264    directive="${directive#:}"265    [[ "$directive" =~ ^[0-9]+$ ]] || directive=0266267    # Every other line is a candidate; a tab and a description may follow268    # the value, so keep only the value.269    while IFS= read -r comp; do270        [[ -z "$comp" || "$comp" == :* ]] && continue271        COMPREPLY+=("${comp%%$'\t'*}")272    done <<< "${out%$'\n'*}"273274    # Bit 2 keeps the cursor against the completion; bit 4 stops the file275    # fallback the "complete -o default" registration would otherwise add.276    (( (directive & 2) != 0 )) && compopt -o nospace 2>/dev/null277    (( (directive & 4) != 0 )) && compopt +o default 2>/dev/null278}279complete -o default -F __kubectl_liken_complete liken kubectl-liken280`

1package main23// This file is the base binary's own prefix walk. kubectl dispatches4// kubectl liken audio capture by finding kubectl-liken-audio on PATH,5// so the two-layer command falls out of kubectl's own rule with no6// code here. The walk below gives the bare liken and kubectl-liken7// names the same behavior: liken audio capture finds the same binary8// and runs it with capture. The walk reads the domain from the9// arguments, never from this program's own name, so it behaves the10// same under either name.1112import (13	"os"14	"os/exec"15	"path/filepath"1617	"github.com/liken-sh/liken/liken/plugins"18)1920// findPlugin locates a domain's CLI. It reads PATH first, because a21// person who added the plugin directory to PATH expects that copy,22// and then the plugin directory itself, which is not always on PATH.23// It returns the path, or the empty string when neither holds the24// CLI.25func findPlugin(binDir, domain string) string {26	name := plugins.Name(domain)27	if path, err := exec.LookPath(name); err == nil {28		return path29	}30	candidate := filepath.Join(binDir, name)31	if info, err := os.Stat(candidate); err == nil && !info.IsDir() {32		return candidate33	}34	return ""35}3637// dispatchPlugin runs a domain's CLI with the remaining arguments, or38// reports that no plugin answers the domain. It replaces this process39// with the plugin (execTool), so the plugin owns the terminal and its40// exit code is the base binary's.41func dispatchPlugin(binDir, domain string, rest []string) (bool, error) {42	path := findPlugin(binDir, domain)43	if path == "" {44		return false, nil45	}46	return true, execTool(path, append([]string{plugins.Name(domain)}, rest...), os.Environ())47}

1// The liken CLI is the toolkit for producing and operating a2// deployment of liken.3//4// This program does everything an operator needs to do to a5// deployment that is not running a machine: minting or adopting a6// cluster identity, computing credentials from that identity,7// packing a deployment layer, and assembling install media from a8// public release. The other Go programs in this repo run inside the9// machine: init as PID 1, and the operators as pods. This program10// runs on the operator's workstation, and it ships with public11// releases, so producing a cluster never requires this repo or a12// build.13//14// The command is a thin dispatcher. The logic lives with the domain15// that owns it (the identity package, and later the image and16// releases packages). Because of this, the CLI stays a table of17// names, while each capability keeps its own file, its own tests,18// and its own documentation.19package main2021import (22	"bufio"23	"flag"24	"fmt"25	"io"26	"os"27	"strings"2829	"github.com/liken-sh/liken/liken/identity"30	"github.com/liken-sh/liken/liken/image"31	"github.com/liken-sh/liken/liken/machine"32	"github.com/liken-sh/liken/liken/plugins"33	"github.com/liken-sh/liken/liken/releases"34	"github.com/liken-sh/liken/liken/scaffold"35)3637// A consoleList collects repeated -console flags in order.38type consoleList []string3940func (c *consoleList) String() string { return fmt.Sprint([]string(*c)) }4142func (c *consoleList) Set(v string) error {43	*c = append(*c, v)44	return nil45}4647const usage = `liken: the toolkit for setting up and running a liken cluster4849usage:5051  liken new <directory>52      Start a deployment: answer a few questions and get a directory53      of manifests: cluster.yaml and one file per machine, with54      comments that teach every field. The other commands build on55      this directory.5657  liken mint <identity-dir>58      Create a new cluster identity: the certificates and join token59      that every machine in one cluster shares.6061  liken adopt <harvest-dir> <identity-dir>62      Take identity files copied off an existing cluster's disk and63      arrange them as an identity directory. The cluster does not64      have to be a liken cluster: any k3s cluster's identity can be65      adopted, so liken machines can join a cluster you already run.6667  liken kubeconfig [-server URL] <deployment-dir>68      Write an admin kubeconfig: the credential kubectl uses to69      administer the cluster. The endpoint comes from the70      deployment's cluster.yaml; -server overrides it, for when71      that address is not reachable from this machine.7273  liken approve-reboot [-server URL] <deployment-dir> <machine>74      Report what a machine waits for, and grant it the one75      disruption its rebootPolicy withholds. The grant is an76      annotation naming the staged change's hash, so it spends77      itself when the change applies, and running this twice is78      the same grant. When nothing waits, this writes nothing.7980  liken request-reboot [-server URL] <deployment-dir> <machine>81      Ask a machine to reboot when no change asks it to: a driver82      bound to the wrong device, or a machine you are experimenting83      on. The machine still takes its turn from the cluster,84      cordons, and drains. On rebootPolicy: Auto that is the whole85      procedure. On Manual the request waits for approve-reboot.86      The request names the running boot, so it spends itself when87      the machine comes back.8889  liken kubectl [-server URL] <deployment-dir> [args...]90  liken stern   [-server URL] <deployment-dir> [args...]91  liken flux    [-server URL] <deployment-dir> [args...]92      Run the named tool from your PATH against the deployment's93      cluster: resolve the credential, set KUBECONFIG, and hand the94      terminal to the tool. Give liken's -server before the95      directory; everything after the directory goes to the tool.9697  liken layer <manifests-dir> <identity-dir> <output.cpio>98      Pack your cluster's half of the operating system into one small99      archive: your cluster and machine manifests, and your identity.100      Kernel modules are not part of it: the system image carries the101      kernel's whole module tree, and spec.modules loads from that102      tree at boot.103104  liken fetch [-digest sha256:<hex>] <source-url> <version|latest> <channel-dir>105      Download a published release from a channel into a local106      channel directory, verifying every artifact against the107      release's document. Pass "latest" to take whatever the channel108      currently names newest. -digest pins the document itself to a109      catalog entry's digest, closing the trust chain end to end.110111  liken media <release-dir> <deployment.cpio> <output.cpio>112      Build a bootable install image from a downloaded release and113      your deployment layer. Machines install themselves from it.114115  liken stick [-console ttyS0] <release-dir> <deployment.cpio> <output.img>116      Build the USB install stick's disk image from a downloaded117      release and your deployment layer: one stick for the whole118      deployment, with a boot menu listing every machine by name.119      Boot it, pick the machine you're standing at, and follow the120      console: the install holds until you remove the stick and press121      Enter, then powers the machine off. -console (repeatable) adds a122      console= argument to every entry; the machines keep it123      permanently.124125  liken bundle [-slot-size 1Gi] <vmlinuz> <liken.sqfs> <boot.cpio> <microcode.cpio> <liken-cli> <systemd-boot.efi> <grub-boot.img> <grub-core.img> <licenses.md> <channel-dir> <version> [component=version ...]126      Lay out a release: copy the nine files into the channel and127      write the release.yaml that names each one by its digest. The128      version is a calendar date and serial (2026.07.11-001); the129      component=version pairs record which upstreams shipped inside,130      since the date deliberately doesn't say.131132  liken serve <channel-dir> [address]133      Share a release channel over plain HTTP so machines can134      download from it. The address defaults to :8017.135136  liken index -source <url> <output-dir> < keys137      Render a channel's index: a front page listing every release, a138      page per release giving its catalog entry and its artifacts, a139      page for the source mirror, and the versions.yaml document that140      lists every release for a script. Read the channel's object keys141      from standard input, one per line, and read each release's142      document from the channel itself. The index is derived, so143      rendering it again over the same channel repairs whatever is144      stale.145146  liken plugins sync|list|remove [-server URL] <deployment-dir>147      The plugins group. sync reads the operators the cluster148      runs, pulls each one's CLI from the registry into149      ~/.liken/plugins/bin, and prints the line to add that directory150      to PATH. list reports each installed CLI's version and the151      operator version it faces, and marks drift. remove <domain>152      deletes one installed CLI. A domain that names no command, like153      "liken audio capture", walks to kubectl-liken-<domain>.154155  liken completion bash156      Print the bash completion script for this toolkit.157158  liken version159      Print this toolkit's version.160161An identity directory holds the certificates and join token that162make a cluster one cluster. Some of the files are private keys, so163keep the directory out of version control.164165A deployment layer is a small archive holding everything about the166operating system that is yours and not liken's. A machine boots the167generic liken image and your layer together, and the kernel joins168them into one system.169170A release channel is a directory any web server can share: one171subdirectory per version, each holding the release's files and a172release.yaml that names every file by its sha256 digest, so a173machine can check that what it downloaded is what was published.174`175176func main() {177	if err := run(os.Args[1:]); err != nil {178		fmt.Fprintf(os.Stderr, "liken: %v\n", err)179		os.Exit(1)180	}181}182183func run(args []string) error {184	// The __complete verb answers the shell's completion request. It185	// runs before the command check, because a request for the first186	// word carries no command and a person waits on the answer.187	if len(args) > 0 && args[0] == "__complete" {188		return runComplete(args[1:], os.Stdout)189	}190	if len(args) == 0 {191		fmt.Fprint(os.Stderr, usage)192		return fmt.Errorf("a command is required")193	}194	switch args[0] {195	case "new":196		if len(args) != 2 {197			return fmt.Errorf("usage: liken new <directory>")198		}199		return scaffold.New(args[1], os.Stdin, os.Stdout)200	case "mint":201		if len(args) != 2 {202			return fmt.Errorf("usage: liken mint <identity-dir>")203		}204		return identity.Mint(args[1], os.Stdout)205	case "adopt":206		if len(args) != 3 {207			return fmt.Errorf("usage: liken adopt <harvest-dir> <identity-dir>")208		}209		return identity.Adopt(args[1], args[2], os.Stdout)210	case "kubeconfig":211		fs := flag.NewFlagSet("kubeconfig", flag.ContinueOnError)212		server := fs.String("server", "", "the API server address, when cluster.yaml's endpoint is not reachable from here")213		if err := fs.Parse(args[1:]); err != nil {214			return err215		}216		if fs.NArg() != 1 {217			return fmt.Errorf("usage: liken kubeconfig [-server URL] <deployment-dir>")218		}219		_, err := writeKubeconfig(fs.Arg(0), *server, os.Stdout)220		return err221	case "kubectl", "stern", "flux":222		return passthrough(args[0], args[1:])223	case "approve-reboot":224		return approveReboot(args[1:], os.Stdout)225	case "request-reboot":226		return requestReboot(args[1:], os.Stdout)227	case "plugins":228		return pluginsCommand(args[1:], os.Stdout)229	case "layer":230		if len(args) != 4 {231			return fmt.Errorf("usage: liken layer <manifests-dir> <identity-dir> <output.cpio>")232		}233		return image.Layer(args[1], args[2], args[3])234	case "fetch":235		fs := flag.NewFlagSet("fetch", flag.ContinueOnError)236		digest := fs.String("digest", "", "pin the release document to a catalog entry's sha256:<hex> digest")237		if err := fs.Parse(args[1:]); err != nil {238			return err239		}240		if fs.NArg() != 3 {241			return fmt.Errorf("usage: liken fetch [-digest sha256:<hex>] <source-url> <version|latest> <channel-dir>")242		}243		return releases.Fetch(fs.Arg(0), fs.Arg(1), *digest, fs.Arg(2), os.Stdout)244	case "media":245		if len(args) != 4 {246			return fmt.Errorf("usage: liken media <release-dir> <deployment.cpio> <output.cpio>")247		}248		return image.Media(args[1], args[2], args[3], os.Stdout)249	case "stick":250		// The CLI's first flags. This uses the standard library's251		// flag package over a subcommand's own FlagSet, so the252		// positional arguments stay positional.253		fs := flag.NewFlagSet("stick", flag.ContinueOnError)254		var consoles consoleList255		fs.Var(&consoles, "console", "add console=<value> to every menu entry (repeatable)")256		if err := fs.Parse(args[1:]); err != nil {257			return err258		}259		if fs.NArg() != 3 {260			return fmt.Errorf("usage: liken stick [-console ttyS0] <release-dir> <deployment.cpio> <output.img>")261		}262		return image.Stick(fs.Arg(0), fs.Arg(1), fs.Arg(2), consoles, os.Stdout)263	case "bundle":264		fs := flag.NewFlagSet("bundle", flag.ContinueOnError)265		slotSize := fs.String("slot-size", "1Gi", "the boot slot size that every artifact, plus headroom, must fit")266		if err := fs.Parse(args[1:]); err != nil {267			return err268		}269		if fs.NArg() < 11 {270			return fmt.Errorf("usage: liken bundle [-slot-size 1Gi] <vmlinuz> <liken.sqfs> <boot.cpio> <microcode.cpio> <liken-cli> <systemd-boot.efi> <grub-boot.img> <grub-core.img> <licenses.md> <channel-dir> <version> [component=version ...]")271		}272		var components []machine.ReleaseComponent273		for _, arg := range fs.Args()[11:] {274			name, version, ok := strings.Cut(arg, "=")275			if !ok || name == "" || version == "" {276				return fmt.Errorf("component %q must be name=version", arg)277			}278			components = append(components, machine.ReleaseComponent{Name: name, Version: version})279		}280		return releases.Bundle(fs.Arg(0), fs.Arg(1), fs.Arg(2), fs.Arg(3), fs.Arg(4), fs.Arg(5), fs.Arg(6), fs.Arg(7), fs.Arg(8), fs.Arg(9), fs.Arg(10), *slotSize, components, os.Stdout)281	case "index":282		fs := flag.NewFlagSet("index", flag.ContinueOnError)283		source := fs.String("source", "", "the channel's base URL, where the pages will be served")284		if err := fs.Parse(args[1:]); err != nil {285			return err286		}287		if fs.NArg() != 1 || *source == "" {288			return fmt.Errorf("usage: liken index -source <url> <output-dir> < keys")289		}290		keys, err := readLines(os.Stdin)291		if err != nil {292			return err293		}294		return releases.Index(*source, keys, fs.Arg(0), os.Stdout)295	case "serve":296		addr := ":8017"297		switch len(args) {298		case 2:299		case 3:300			addr = args[2]301		default:302			return fmt.Errorf("usage: liken serve <channel-dir> [address]")303		}304		return releases.Serve(args[1], addr)305	case "version":306		fmt.Println(machine.Version)307		return nil308	case "completion":309		shell := ""310		if len(args) > 1 {311			shell = args[1]312		}313		return completionScript(shell, os.Stdout)314	default:315		// A first argument that names no command is read as a plugin316		// domain: liken audio capture walks the prefix to317		// kubectl-liken-audio and runs it, the same walk kubectl runs318		// for kubectl liken audio capture.319		if binDir, err := plugins.BinDir(); err == nil {320			if dispatched, err := dispatchPlugin(binDir, args[0], args[1:]); dispatched {321				return err322			}323		}324		fmt.Fprint(os.Stderr, usage)325		return fmt.Errorf("unknown command %q", args[0])326	}327}328329// readLines collects a list given on standard input, one entry per330// line, and drops blank lines. The index command takes a channel's331// object keys this way, because the command that lists a bucket holds332// a credential and this program holds none.333func readLines(r io.Reader) ([]string, error) {334	var lines []string335	scanner := bufio.NewScanner(r)336	for scanner.Scan() {337		if line := strings.TrimSpace(scanner.Text()); line != "" {338			lines = append(lines, line)339		}340	}341	return lines, scanner.Err()342}

1package main23// The plugins command group installs and manages the per-operator4// CLIs that give liken its two-layer commands. Each operator ships a5// binary named kubectl-liken-<domain> beside its operator image at6// the same version. sync reads the operators the cluster runs, pulls7// each CLI out of the registry, and writes it into the plugin8// directory. list reports what is installed against what the cluster9// runs. remove deletes one. The registry pull, the filesystem layout,10// and the comparison live in the plugins package; this file is the11// cluster read and the dispatch.1213import (14	"flag"15	"fmt"16	"io"17	"os"18	"os/exec"19	"path/filepath"20	"runtime"21	"strings"2223	"github.com/liken-sh/liken/kubernetes/apiclient"24	"github.com/liken-sh/liken/liken/kubernetes"25	"github.com/liken-sh/liken/liken/plugins"26)2728// clusterClient resolves the deployment's credential and builds a29// client for its cluster, the same resolution the other cluster30// commands use (cluster.go).31func clusterClient(dir, server string) (*apiclient.Client, error) {32	kubeconfigPath, err := writeKubeconfig(dir, server, io.Discard)33	if err != nil {34		return nil, err35	}36	return kubernetes.KubeconfigClient(kubeconfigPath)37}3839// operatorsFromWorkloads maps the plugin workloads to the domains and40// images the sync pulls, dropping any that name no domain or no image.41func operatorsFromWorkloads(workloads []kubernetes.Workload) []plugins.Operator {42	var operators []plugins.Operator43	for _, w := range workloads {44		domain, image := w.PluginDomain(), w.OperatorImage()45		if domain == "" || image == "" {46			continue47		}48		operators = append(operators, plugins.Operator{Domain: domain, Image: image})49	}50	return operators51}5253// operatorVersions maps each domain to the version its operator image54// carries, the version every CLI compares against.55func operatorVersions(workloads []kubernetes.Workload) map[string]string {56	versions := map[string]string{}57	for _, w := range workloads {58		domain, image := w.PluginDomain(), w.OperatorImage()59		if domain == "" || image == "" {60			continue61		}62		if version, err := plugins.ImageVersion(image); err == nil {63			versions[domain] = version64		}65	}66	return versions67}6869// pluginVersion asks an installed CLI for its stamped version. It is70// a variable so tests observe the call instead of running a real71// binary.72var pluginVersion = execPluginVersion7374func execPluginVersion(binDir, domain string) (string, error) {75	out, err := exec.Command(filepath.Join(binDir, plugins.Name(domain)), "--version").Output()76	if err != nil {77		return "", err78	}79	return strings.TrimSpace(string(out)), nil80}8182// installedVersions reads the stamped version of each installed CLI.83// A CLI that does not answer --version reports an empty version, so84// the list still names it rather than dropping it.85func installedVersions(binDir string, domains []string) map[string]string {86	versions := map[string]string{}87	for _, domain := range domains {88		version, err := pluginVersion(binDir, domain)89		if err != nil {90			version = ""91		}92		versions[domain] = version93	}94	return versions95}9697func pluginsCommand(args []string, out io.Writer) error {98	if len(args) == 0 {99		return fmt.Errorf("usage: liken plugins sync|list|remove")100	}101	switch args[0] {102	case "sync":103		return pluginsSync(args[1:], out)104	case "list":105		return pluginsList(args[1:], out)106	case "remove":107		return pluginsRemove(args[1:], out)108	default:109		return fmt.Errorf("usage: liken plugins sync|list|remove")110	}111}112113func pluginsSync(args []string, out io.Writer) error {114	fs := flag.NewFlagSet("plugins sync", flag.ContinueOnError)115	server := fs.String("server", "", "the API server address, when cluster.yaml's endpoint is not reachable from here")116	if err := fs.Parse(args); err != nil {117		return err118	}119	if fs.NArg() != 1 {120		return fmt.Errorf("usage: liken plugins sync [-server URL] <deployment-dir>")121	}122	c, err := clusterClient(fs.Arg(0), *server)123	if err != nil {124		return err125	}126	workloads, err := kubernetes.ListPluginWorkloads(c)127	if err != nil {128		return err129	}130	binDir, err := plugins.BinDir()131	if err != nil {132		return err133	}134	return plugins.Sync(operatorsFromWorkloads(workloads), runtime.GOARCH, binDir, os.Getenv("PATH"), out)135}136137func pluginsList(args []string, out io.Writer) error {138	fs := flag.NewFlagSet("plugins list", flag.ContinueOnError)139	server := fs.String("server", "", "the API server address, when cluster.yaml's endpoint is not reachable from here")140	if err := fs.Parse(args); err != nil {141		return err142	}143	if fs.NArg() != 1 {144		return fmt.Errorf("usage: liken plugins list [-server URL] <deployment-dir>")145	}146	c, err := clusterClient(fs.Arg(0), *server)147	if err != nil {148		return err149	}150	workloads, err := kubernetes.ListPluginWorkloads(c)151	if err != nil {152		return err153	}154	binDir, err := plugins.BinDir()155	if err != nil {156		return err157	}158	domains, err := plugins.Installed(binDir)159	if err != nil {160		return err161	}162	entries := plugins.Merge(installedVersions(binDir, domains), operatorVersions(workloads))163	plugins.Render(entries, out)164	return nil165}166167func pluginsRemove(args []string, out io.Writer) error {168	if len(args) != 1 {169		return fmt.Errorf("usage: liken plugins remove <domain>")170	}171	binDir, err := plugins.BinDir()172	if err != nil {173		return err174	}175	if err := plugins.Remove(binDir, args[0]); err != nil {176		return err177	}178	fmt.Fprintf(out, "removed %s\n", plugins.Name(args[0]))179	return nil180}

1package main23// request-reboot is approve-reboot's counterpart. Every other reboot4// liken performs applies a staged change, so a machine that agrees5// with every document it was given has no way to reboot, and a6// machine with no shell has no other way to be told. This command is7// how a person asks: it reads status.boot.time, derives the running8// boot's identity, and writes the request-reboot annotation naming9// it. Naming the running boot makes the request one-shot10// (machine.RequestRebootAnnotation explains why), so a later request11// works with no cleanup in between.12//13// The request skips no coordination. The machine still takes its14// turn from the cluster's disruption budget, cordons, and drains. On15// rebootPolicy: Manual it parks where a staged change parks, and16// liken approve-reboot releases it.1718import (19	"flag"20	"fmt"21	"io"22	"strings"2324	"github.com/liken-sh/liken/liken/kubernetes"25	"github.com/liken-sh/liken/liken/machine"26)2728// renderRequest reports what the request set in motion. What happens29// next depends on the machine's rebootPolicy, so the report says30// which of the two paths this machine is on, and names the command31// that finishes the second one.32func renderRequest(m *machine.Machine, dir, short string) string {33	var b strings.Builder34	fmt.Fprintf(&b, "requested: %s=%s\n", machine.RequestRebootAnnotation, short)35	if m.Spec.RebootPolicyOrDefault() == machine.RebootAuto {36		fmt.Fprintf(&b, "%s takes a reboot turn from the cluster, drains its workloads, and reboots.\n", m.Metadata.Name)37	} else {38		fmt.Fprintf(&b, "%s has rebootPolicy: Manual, so the request waits for you. Release it with:\n", m.Metadata.Name)39		fmt.Fprintf(&b, "  liken approve-reboot %s %s\n", dir, m.Metadata.Name)40	}41	fmt.Fprint(&b, "Nothing is staged, so the machine comes back on the documents it runs now.\n")42	return b.String()43}4445// requestReboot is the command: resolve the credential, read the46// machine, name its boot, and ask.47func requestReboot(args []string, out io.Writer) error {48	fs := flag.NewFlagSet("request-reboot", flag.ContinueOnError)49	server := fs.String("server", "", "the API server address, when cluster.yaml's endpoint is not reachable from here")50	if err := fs.Parse(args); err != nil {51		return err52	}53	if fs.NArg() != 2 {54		return fmt.Errorf("usage: liken request-reboot [-server URL] <deployment-dir> <machine>")55	}56	kubeconfigPath, err := writeKubeconfig(fs.Arg(0), *server, io.Discard)57	if err != nil {58		return err59	}60	c, err := kubernetes.KubeconfigClient(kubeconfigPath)61	if err != nil {62		return err63	}64	m, err := kubernetes.GetMachine(c, fs.Arg(1))65	if err != nil {66		return err67	}68	// A machine that reports no boot time has published nothing for69	// the request to name. That is a machine still starting, or one70	// that has never reported at all, and either way there is no71	// boot to ask about yet.72	bootID := machine.BootID(m.Status.Boot)73	if bootID == "" {74		return fmt.Errorf("%s reports no boot time, so there is no boot to request a reboot of; the machine is still starting, or it has never reported", m.Metadata.Name)75	}76	short := fmt.Sprintf("%.12s", bootID)77	patch := fmt.Sprintf(`{"metadata":{"annotations":{%q:%q}}}`, machine.RequestRebootAnnotation, short)78	if err := kubernetes.PatchJSON(c, kubernetes.MachinesPath+"/"+m.Metadata.Name, []byte(patch)); err != nil {79		return err80	}81	fmt.Fprint(out, renderRequest(m, fs.Arg(0), short))82	return nil83}

1package main23// This poller checks the release channel. It shows how a cluster4// detects that a new version exists.5//6// The channel's root document, channel.yaml, names the latest7// published version. This poller fetches that document so the sweep8// can show the answer in the Cluster's status.9//10// Two facts shape this design. First, the answer is advisory. The11// machine package's channel.go explains why the design keeps the12// answer outside the trust chain. Because the answer is advisory, a13// stale or missing answer costs nothing. The poller keeps the last14// answer it received and tries again later. Second, the sweep runs15// every ten seconds, but the channel changes about once a week. So16// the poller fetches the document on a long interval instead of every17// sweep. Two things make the poller fetch immediately: a new source,18// or a new value in the Cluster's liken.sh/check-releases annotation.19// The annotation is a declared request to poll immediately (the20// cluster package's CheckReleasesAnnotation explains why the request21// is an annotation and not a spec field).22//23// The fetch runs on its own goroutine. The machine operator's release24// fetcher uses the same method. The sweep must keep judging the fleet25// at its own speed, and a slow or dead release server must never26// delay a Lost verdict. The Observe function determines what to do27// and returns immediately; the goroutine reports its result back28// through the mutex.2930import (31	"fmt"32	"io"33	"net/http"34	"strings"35	"sync"36	"time"3738	"github.com/liken-sh/liken/liken/cluster"39	"github.com/liken-sh/liken/liken/machine"40)4142// channelPollInterval sets how long an answer stays fresh, so the43// poller does not ask again during this time. People publish releases44// on a human timescale, and the poll answers the question "should45// someone think about upgrading?", not a real-time telemetry46// question. A person who just published a release and wants the47// answer now can set the check-releases annotation to request an48// immediate poll.49const channelPollInterval = 6 * time.Hour5051type channelPoller struct {52	mu        sync.Mutex53	source    string    // the channel this poller last asked54	check     string    // the last check-releases annotation this poller handled55	polled    time.Time // when the last attempt started56	inFlight  bool57	available string // the channel's last known answer5859	// fetch performs the actual HTTP request. Tests can replace it60	// with a fake function, so they do not need a real channel61	// server.62	fetch func(url string) ([]byte, error)63}6465func newChannelPoller() *channelPoller {66	return &channelPoller{fetch: fetchChannelDocument}67}6869// Observe runs once per sweep. It receives the spec's current70// releases section and the Cluster's check-releases annotation. It71// determines whether a poll is due: the source or the annotation72// changed, or the last answer is too old. If a poll is due, Observe73// starts it in the background. Observe never blocks.74func (p *channelPoller) Observe(releases cluster.ClusterReleasesSpec, check string, now time.Time) {75	p.mu.Lock()76	defer p.mu.Unlock()7778	// If there is no source, there is nothing to poll. A cluster79	// without a channel has no available version, and it does not80	// keep an available version from a channel it had before.81	if releases.Source == "" {82		p.source, p.check, p.available = "", "", ""83		return84	}8586	if releases.Source != p.source {87		// A different channel's answer does not apply here. The88		// poller drops it and asks the new channel immediately.89		p.source, p.check, p.available = releases.Source, check, ""90		p.polled = time.Time{}91	} else if check != p.check {92		// This is the request to poll immediately. The poller honors93		// it once, by marking the last answer as stale right away.94		// The last answer stays visible until the fresh answer95		// replaces it.96		p.check = check97		p.polled = time.Time{}98	}99100	if p.inFlight || now.Sub(p.polled) < channelPollInterval {101		return102	}103	// The poller records the time of this attempt before it runs.104	// This way, a dead server is asked again on the interval, not on105	// every ten-second sweep.106	p.polled = now107	p.inFlight = true108	go p.poll(p.source)109}110111// Available returns the latest version the channel last announced.112// It is empty until a poll succeeds.113func (p *channelPoller) Available() string {114	p.mu.Lock()115	defer p.mu.Unlock()116	return p.available117}118119// poll fetches and parses the channel document. It keeps the answer120// only if the sweep is still asking about the same source.121func (p *channelPoller) poll(source string) {122	url := strings.TrimSuffix(source, "/") + "/channel.yaml"123	raw, err := p.fetch(url)124	var latest string125	if err == nil {126		var channel *machine.Channel127		if channel, err = machine.ParseChannel(raw); err == nil {128			latest = channel.Latest129		}130	}131132	p.mu.Lock()133	defer p.mu.Unlock()134	p.inFlight = false135	if source != p.source {136		return137	}138	if err != nil {139		// The last answer stands. Advisory data may be stale. The140		// interval, or an edit to the check value, will trigger141		// another poll.142		fmt.Printf("polling the release channel %s: %v\n", url, err)143		return144	}145	p.available = latest146}147148// fetchChannelDocument sends a GET request for the channel document149// through the default transport.150func fetchChannelDocument(url string) ([]byte, error) {151	return fetchChannelDocumentVia(nil, url)152}153154// fetchChannelDocumentVia sends a GET request for the channel document155// through transport, or through the default transport when transport156// is nil, using a client with a bounded size and time limit. The157// document is only a few lines of YAML, so the size limit is generous.158// The time limit stops a stuck server from keeping the poll in progress159// forever. A test passes a transport that reaches its fake server160// without a socket.161func fetchChannelDocumentVia(transport http.RoundTripper, url string) ([]byte, error) {162	client := &http.Client{Transport: transport, Timeout: 30 * time.Second}163	resp, err := client.Get(url)164	if err != nil {165		return nil, err166	}167	defer resp.Body.Close()168	if resp.StatusCode != http.StatusOK {169		return nil, fmt.Errorf("%s: %s", url, resp.Status)170	}171	return io.ReadAll(io.LimitReader(resp.Body, 1<<20))172}

1package main23// The Events the cluster operator posts about the Machines and the4// Cluster. `kubectl describe` lists them under the status, so a person5// reads the history of a rollout, or of a machine that went silent,6// with no log to open. kubernetes/events writes them, and root plan 787// gives the rule for a condition, an Event, and a log line.8//9// Both kinds are cluster-scoped, so their Events are in the namespace10// default. `kubectl describe` finds them there, and `kubectl events11// --for cluster/<name>` finds them only with -n default or -A.12//13//   - Each condition transition of the Cluster posts one Event, with14//     the condition's reason, after the status write lands: a fleet15//     that goes MachinesDegraded or AllMachinesReady, and a rollout16//     that goes RollingOut, RolloutStalled, or RolloutComplete.17//   - The verdicts this program writes onto a Machine post one Event18//     each on that Machine: MachineLost, RebootTurnGranted, and19//     RebootTurnReclaimed. Each one is an action this program takes on20//     an object that another program owns, so each Event names the21//     action and not the condition's reason.22//   - The flux feature's one-off actions post one Event each on the23//     Cluster: the mint of the deploy key and the planting of the24//     engine.25//26// Nothing that repeats on each sweep posts an Event: a poll of the27// release channel, an eviction the budget refuses, or the sweep28// itself.2930import (31	"github.com/liken-sh/liken/kubernetes/conditions"32	"github.com/liken-sh/liken/kubernetes/events"33	"github.com/liken-sh/liken/liken/api"34	"github.com/liken-sh/liken/liken/cluster"35	"github.com/liken-sh/liken/liken/machine"36)3738// The reasons of the Events that are not condition transitions. A39// transition's Event takes the condition's own reason.40const (41	reasonMachineLost         = "MachineLost"42	reasonRebootTurnGranted   = "RebootTurnGranted"43	reasonRebootTurnReclaimed = "RebootTurnReclaimed"44	reasonFluxDeployKeyMinted = "FluxDeployKeyMinted"45	reasonFluxEngineSeeded    = "FluxEngineSeeded"46)4748// machineReference answers the object reference of a Machine, for an49// Event about it.50func machineReference(m *machine.Machine) events.ObjectReference {51	return events.ObjectReference{52		APIVersion: api.APIVersion, Kind: machineKind,53		Name: m.Metadata.Name, UID: m.Metadata.UID,54	}55}5657// clusterReference answers the object reference of the Cluster, for an58// Event about it.59func clusterReference(c *cluster.Cluster) events.ObjectReference {60	return events.ObjectReference{61		APIVersion: api.APIVersion, Kind: clusterKind,62		Name: c.Metadata.Name, UID: c.Metadata.UID,63	}64}6566// postClusterTransitions posts one Event for each Cluster condition67// that transitioned from stored to written, with the condition's reason68// and message.69func postClusterTransitions(recorder *events.Recorder, c *cluster.Cluster, stored, written []api.Condition) {70	for _, t := range api.Transitions(stored, written) {71		recorder.Transition(clusterReference(c), t, clusterBadStatus(t))72	}73}7475// clusterBadStatus answers the status of a Cluster condition that76// needs a person, for events.Recorder.Transition. A degraded machine,77// a stalled rollout, and a declined flux teardown each need one. A78// fleet in the middle of a rollout is MachinesReady False too, and79// that needs no one.80func clusterBadStatus(c api.Condition) conditions.Status {81	switch c.Reason {82	case "MachinesDegraded", "RolloutStalled", "NotPlantedByLiken":83		return c.Status84	}85	return ""86}

1package main23// The fleet sweep is the cluster operator's main task.4//5// Each machine's operator reports on itself. This leaves one gap6// that nothing else covers: a dead machine cannot report that it is7// dead. Its last written status stays in the API and reads Ready8// forever, which is worse than no status at all. Kubernetes has this9// same problem with kubelets, and it solves the problem with10// heartbeats. The kubelet renews a lease every few seconds, and the11// node controller turns a silent lease into a NotReady Node. The12// Machine object gets the same treatment here. Each machine's13// operator renews its own heartbeat lease (see the kubernetes14// package), and this sweep marks a machine with the Lost phase when15// its lease goes silent. The sweep also produces the cluster's16// headcount. The same pass over the Machine list produces the17// ready-out-of-total tally that the Cluster's status carries.18//19// Writing another machine's status breaks the one-writer-per-object20// rule that the machine operators otherwise follow. So the sweep is21// careful about when it writes: it only writes a machine whose22// heartbeat is already stale. A stale heartbeat means the machine's23// own operator has stopped writing, so the two writers can never24// actually write to the same object at the same time. The moment the25// machine returns, its own operator's next pass overwrites the Lost26// verdict with fresh observations. No cleanup step is needed.2728import (29	"errors"30	"fmt"31	"slices"32	"strings"33	"time"3435	"github.com/liken-sh/liken/kubernetes/apiclient"36	"github.com/liken-sh/liken/liken/api"37	"github.com/liken-sh/liken/liken/cluster"38	"github.com/liken-sh/liken/liken/machine"39)4041// A fleetSweep holds one pass's verdict over the whole fleet: which42// machines to declare Lost, the headcount for the Cluster's status,43// the MachinesReady condition that carries the full detail, and the44// phase that summarizes the verdict in one word. Every Machine uses45// this same conditions-then-phase arrangement, applied here to the46// fleet. The function decideFleetSweep only computes the verdict; the47// function sweepFleet carries it out.48type fleetSweep struct {49	lost      []string50	tally     cluster.MachineTally51	condition api.Condition52	phase     api.Phase5354	// The two counts that the fleet's metrics publish (metrics.go).55	// They are part of the verdict because this loop is the one place56	// that judges every machine at once, and a second pass over the57	// same list could only ever repeat this one.58	phases    map[api.Phase]int59	approvals int60}6162// decideFleetSweep judges each machine by its effective phase. A63// machine counts toward ready only when its phase is Ready and its64// heartbeat is fresh. The verdict sorts each machine that is not65// Ready into one of two groups: mid-transition (rebooting into a66// change, waiting for a reboot, or booting) or unwell (Lost, Blocked,67// or otherwise degraded). Unwell outranks mid-transition. The68// MachinesReady condition names the affected machines, so nobody has69// to search for them.70func decideFleetSweep(machines []machine.Machine, heard map[string]time.Time, now time.Time) fleetSweep {71	s := fleetSweep{tally: cluster.MachineTally{Total: len(machines)}, phases: map[api.Phase]int{}}72	var transitioning, unwell []string73	for i := range machines {74		m := &machines[i]75		effective := effectivePhase(m, heard, now)76		if effective == api.PhaseLost && m.Status.Phase != api.PhaseLost {77			s.lost = append(s.lost, m.Metadata.Name)78		}79		s.phases[effective]++80		if awaitsApproval(m) {81			s.approvals++82		}8384		switch effective {85		case api.PhaseReady:86			s.tally.Ready++87		case api.PhaseUpdating, api.PhaseUpdatePending, api.PhaseDownloading, api.PhaseBooting:88			transitioning = append(transitioning, m.Metadata.Name)89		default:90			unwell = append(unwell, m.Metadata.Name)91		}92	}93	s.tally.Summary = fmt.Sprintf("%d/%d", s.tally.Ready, s.tally.Total)9495	switch {96	case len(unwell) > 0:97		s.phase = api.PhaseDegraded98		s.condition = api.Condition{99			Type: "MachinesReady", Status: api.ConditionFalse, Reason: "MachinesDegraded",100			Message: fmt.Sprintf("%s machines ready; unwell: %s", s.tally.Summary, strings.Join(unwell, ", ")),101		}102	case len(transitioning) > 0:103		s.phase = api.PhaseUpdating104		s.condition = api.Condition{105			Type: "MachinesReady", Status: api.ConditionFalse, Reason: "MachinesUpdating",106			Message: fmt.Sprintf("%s machines ready; mid-transition: %s", s.tally.Summary, strings.Join(transitioning, ", ")),107		}108	default:109		s.phase = api.PhaseReady110		s.condition = api.Condition{111			Type: "MachinesReady", Status: api.ConditionTrue, Reason: "AllMachinesReady",112			Message: fmt.Sprintf("all %d machines are ready", s.tally.Total),113		}114	}115	return s116}117118// awaitsApproval reports whether this machine holds a staged change119// that waits for a person. A machine on the Manual reboot policy120// stages its change and stops there, and the reason on the121// convergence condition says so. The cluster operator counts these,122// because the count is a fleet fact: it says how many machines a123// person has to visit before the fleet converges.124func awaitsApproval(m *machine.Machine) bool {125	for _, c := range m.Status.Conditions {126		if c.Reason == "RebootPending" || c.Reason == "RestartPending" {127			return true128		}129	}130	return false131}132133// sweepFleet carries out the sweep. It lists the fleet and its134// heartbeats, determines the verdict, marks the silent machines Lost,135// and publishes the verdict on the Cluster. The available parameter136// is the channel poller's last answer, probe holds when the flux137// engine probe last asked, and steward holds the terminating pods the138// steward has reported. The caller passes all three in, so the sweep139// itself stays a function of its arguments.140//141// A pass that cannot read the fleet returns the reason. The reading142// is the whole pass here: with no list of machines there is no143// verdict to reach, so this is the failure that the layer 2 error144// counter counts (metrics.go).145//146// The order of the writes matters. The Cluster's status goes first,147// because it carries the grant ledger, and a grant goes out only after148// the API server stores the grant's entry (rollout.go). The Machine149// writes follow it: the grants, the confirms, the reclaims, and the150// Lost verdicts.151func sweepFleet(reads *fleetReader, clusterDoc *cluster.Cluster, available string, probe *engineProbe,152	steward *podSteward, cm *clusterMetrics, now time.Time) error {153	c := reads.client154	machines, err := reads.machines()155	if err != nil {156		fmt.Printf("listing machines for the fleet sweep: %v\n", err)157		return err158	}159	// Each Machine whose ledger entry shows no grant in its copy comes160	// from the API server, because the decision to free the entry's161	// slot needs the Machine's current version (rollout.go).162	machines, err = reads.readTurns(clusterDoc.Status.RebootTurns, machines)163	if err != nil {164		fmt.Printf("reading the machines that hold reboot turns: %v\n", err)165		return err166	}167	heard, err := reads.heartbeats(now)168	if err != nil {169		fmt.Printf("listing heartbeats for the fleet sweep: %v\n", err)170		return err171	}172	s := decideFleetSweep(machines, heard, now)173174	// The rollout decision uses the same listing: which machines may175	// take their reboot turn now, and which spent grants return to176	// the budget (see rollout.go). This sequencing happens here for177	// the same reason the tally does: the sweep is the one place that178	// sees the whole fleet at once. appliedVersion is the release the179	// machine-operator DaemonSet's template actually carries180	// (daemonSetVersion, steward.go); decideRollout compares it181	// against the fleet's target so a leader can go first while the182	// applied template lags, because only a leader's boot can advance183	// it.184	appliedVersion, err := daemonSetVersion(reads, machineOperatorDaemonSet)185	r := decideRollout(machines, heard, reads.sightings.since(), clusterDoc, appliedVersion, now)186	if err != nil {187		fmt.Printf("reading the %s DaemonSet for the rollout: %v\n", machineOperatorDaemonSet, err)188		r = r.withoutGrants(err)189	}190191	// The OS's own pods, the operator's pods and the log relay pods,192	// are also fleet state. An upgraded machine keeps pods from193	// before its upgrade until the steward refreshes them (see194	// steward.go).195	stewardOSPods(reads, machines, steward, now)196197	// A retracted feature leaves its workloads behind. k3s only198	// deletes an addon when it sees the addon's manifest disappear199	// while k3s is running, but retraction removes the manifest while200	// k3s is down (see janitor.go). The flux feature's teardown is201	// always the janitor's, in a deliberate order (janitorFlux). That202	// teardown can also decline, when liken did not plant the Flux it203	// found, and it reports the refusal on the Cluster.204	janitorFeatureWorkloads(reads, clusterDoc)205	fluxTeardown := janitorFlux(c, clusterDoc)206207	// The flux feature's deploy key is fleet state too: minted once,208	// then read on every pass so the status always carries the209	// public half (see flux.go). The engine gets the same standing210	// care on a slower cadence: the probe asks once a minute, and gone211	// means re-planted on the pass that asked.212	publicKey := ensureFluxDeployKey(c, reads.recorder, clusterDoc)213	if seed, err := engineSeed(); err != nil {214		if clusterDoc.FeatureEnabled(cluster.FeatureFlux) {215			fmt.Printf("this build carries no engine seed (a plain go build outside make): %v\n", err)216		}217	} else {218		ensureFluxEngine(c, reads.recorder, clusterDoc, seed, probe, now)219	}220221	// The Cluster's status goes first, ahead of every Machine write222	// (see above).223	stored := publishClusterStatus(reads, clusterDoc, s, r, fluxTeardown, available, publicKey, now)224	carryOutRollout(reads, machines, r, stored, now)225	markLost(reads, machines, s.lost, heard, now)226227	// The fleet's metrics come from the verdict that the write above228	// carries, so the graph and the Cluster's status report one229	// observation (metrics.go).230	cm.observeSweep(s, r)231	return nil232}233234// markLost writes the Lost verdict onto each machine that the sweep235// found silent. Each verdict that lands posts MachineLost on the236// machine, with the time this program last saw its heartbeat lease237// renewed, because the Ready condition holds only that the heartbeat238// is stale, not since when.239func markLost(reads *fleetReader, machines []machine.Machine, lost []string, heard map[string]time.Time, now time.Time) {240	for _, m := range machines {241		if !slices.Contains(lost, m.Metadata.Name) {242			continue243		}244		// Everything else in the status stays as the machine last245		// wrote it. Those fields are still the machine's last known246		// facts, and only the machine can revise them. The phase and247		// the Ready condition change because they are claims about248		// the present, and the machine has stopped making those249		// claims.250		status := m.Status251		status.Phase = api.PhaseLost252		// The clone matters here. status.Conditions still shares its253		// backing array with the machines slice, and SetCondition254		// rewrites an existing entry in place.255		status.Conditions = api.SetCondition(slices.Clone(m.Status.Conditions), api.Condition{256			Type: "Ready", Status: api.ConditionUnknown, Reason: "HeartbeatStale",257			ObservedGeneration: m.Metadata.Generation,258			Message:            "the machine's operator has stopped renewing its heartbeat lease; the machine is presumed down",259		}, now)260		// A conflict here means the machine just came back and wrote261		// its own status first. That is the exact outcome this write262		// exists to allow, so the sweep skips this machine. The sweep263		// never retries a write onto another machine's status.264		if err := reads.publishStatus(&m, &status); errors.Is(err, apiclient.ErrConflict) {265			continue266		} else if err != nil {267			fmt.Printf("marking %s lost: %v\n", m.Metadata.Name, err)268		} else {269			fmt.Printf("machine %s has gone silent; marked Lost\n", m.Metadata.Name)270			reads.recorder.Warning(machineReference(&m), reasonMachineLost, lostMessage(heard, m.Metadata.Name))271		}272	}273}274275// lostMessage answers the message of a MachineLost Event: when this276// program last saw the machine renew its heartbeat lease, on this277// program's clock, or that it never did. The time is the one the278// verdict measured from, so it is HeartbeatStaleAfter or more before279// the Event.280func lostMessage(heard map[string]time.Time, name string) string {281	renewed, ok := heard[name]282	if !ok {283		return "the machine has never renewed a heartbeat lease; marked Lost"284	}285	return "the heartbeat lease was last renewed at " + renewed.UTC().Format(time.RFC3339) + "; marked Lost"286}287288// publishClusterStatus publishes the sweep's verdict on the Cluster.289// The MachinesReady condition carries the observation, stamped with290// the generation of the spec it judged. The Progressing condition291// reports the rollout, and the FluxTeardown condition appears only292// while the flux janitor has a refusal to report. The phase293// summarizes the first two conditions, and the tally is the headcount294// that the printer shows. This function also295// derives the release fields: the catalog's newest version, and the296// channel's last polled version. The sweep is the only writer of the297// Cluster's status, so deriving every field is its job. The same write298// carries the grant ledger, status.rebootTurns. The function299// writes the status only when something actually changed, so a300// settled fleet causes no write. Each condition that transitioned301// posts its Event after the write lands, so a refused write posts302// nothing and the next sweep finds the same transition again.303//304// It answers the ledger that the API server holds after the sweep: the305// one the Cluster already held when nothing changed, the one the306// write's answer holds, or none when the write failed. A 409 means307// another copy wrote the status since this sweep read it, so this308// sweep's grants must not go out.309func publishClusterStatus(reads *fleetReader, clusterDoc *cluster.Cluster, s fleetSweep, r rollout, fluxTeardown *api.Condition, available, publicKey string, now time.Time) []cluster.RebootTurn {310	newest := cluster.NewestVersion(clusterDoc.Spec.Releases.Catalog)311	s.condition.ObservedGeneration = clusterDoc.Metadata.Generation312	r.progressing.ObservedGeneration = clusterDoc.Metadata.Generation313	conditions := api.SetCondition(slices.Clone(clusterDoc.Status.Conditions), s.condition, now)314	conditions = api.SetCondition(conditions, r.progressing, now)315	// The flux janitor reports only a refusal, so this condition is316	// present exactly while there is a refusal to report. Removing it317	// when the janitor returns nothing keeps the list steady, which is318	// what the comparison below needs to leave a settled cluster319	// unwritten.320	if fluxTeardown != nil {321		fluxTeardown.ObservedGeneration = clusterDoc.Metadata.Generation322		conditions = api.SetCondition(conditions, *fluxTeardown, now)323	} else {324		conditions = api.RemoveCondition(conditions, fluxTeardownCondition)325	}326	// The deploy key's public half. A declared feature with an empty327	// answer means this pass could not read the Secret (a permission328	// still being seeded, a lost mint race); the last published329	// value is still the fleet's key, so the status keeps it. A330	// retracted feature clears the section: the Secret keeps the key331	// for a re-enable, but a status should not advertise a feature332	// the spec no longer declares.333	flux := clusterDoc.Status.Flux334	if !clusterDoc.FeatureEnabled(cluster.FeatureFlux) {335		flux = cluster.ClusterFluxStatus{}336	} else if publicKey != "" {337		flux = cluster.ClusterFluxStatus{PublicKey: publicKey}338	}339	if clusterDoc.Status.Machines != s.tally || clusterDoc.Status.Phase != s.phase ||340		clusterDoc.Status.Releases.Newest != newest ||341		clusterDoc.Status.Releases.Available != available ||342		clusterDoc.Status.Flux != flux ||343		clusterDoc.Status.ObservedGeneration != clusterDoc.Metadata.Generation ||344		!sameTurns(r.turns, clusterDoc.Status.RebootTurns) ||345		!slices.Equal(conditions, clusterDoc.Status.Conditions) {346		updated := *clusterDoc347		updated.Status.Machines = s.tally348		updated.Status.Phase = s.phase349		updated.Status.Releases.Newest = newest350		updated.Status.Releases.Available = available351		updated.Status.Flux = flux352		updated.Status.ObservedGeneration = clusterDoc.Metadata.Generation353		updated.Status.Conditions = conditions354		updated.Status.RebootTurns = r.turns355		stored, err := reads.publishClusterStatus(&updated)356		if err != nil {357			fmt.Printf("publishing cluster status: %v\n", err)358			return nil359		}360		postClusterTransitions(reads.recorder, clusterDoc, clusterDoc.Status.Conditions, conditions)361		if stored == nil {362			return nil363		}364		if !sameTurns(r.turns, stored.Status.RebootTurns) {365			return schemaPrunedTurns(reads, stored, now)366		}367		return stored.Status.RebootTurns368	}369	return clusterDoc.Status.RebootTurns370}371372// schemaPrunedTurns reports a Cluster write whose answer lacks the373// ledger that it sent. The served CRD then comes from a release that374// does not declare status.rebootTurns, and the API server pruned the375// field. That happens while a newer cluster-operator runs against an376// older CRD: in the seconds after the first leader boots a new release,377// or when a leader on an older release starts k3s again and applies378// its own copy of the CRD. The sweep grants nothing until the newer379// schema returns, and the Progressing message says why. The reason380// stays RollingOut, so the change posts no Event. It answers the381// ledger the API server stored.382func schemaPrunedTurns(reads *fleetReader, stored *cluster.Cluster, now time.Time) []cluster.RebootTurn {383	fmt.Println("the served Cluster schema does not store status.rebootTurns; granting no reboot turn")384	updated := *stored385	updated.Status.Conditions = api.SetCondition(slices.Clone(stored.Status.Conditions), api.Condition{386		Type: "Progressing", Status: api.ConditionTrue, Reason: "RollingOut",387		ObservedGeneration: stored.Metadata.Generation,388		Message: "reboot turns are waiting for a Cluster schema that stores status.rebootTurns; " +389			"the served CustomResourceDefinition comes from an older release",390	}, now)391	if _, err := reads.publishClusterStatus(&updated); err != nil {392		fmt.Printf("publishing cluster status: %v\n", err)393	}394	return stored.Status.RebootTurns395}396397// sameTurns reports whether two ledgers hold the same entries in the398// same order. It compares times with Equal, because a time decoded399// from the API server carries another location than one the sweep400// made.401func sameTurns(a, b []cluster.RebootTurn) bool {402	return slices.EqualFunc(a, b, func(x, y cluster.RebootTurn) bool {403		return x.Machine == y.Machine && x.MachineResourceVersion == y.MachineResourceVersion && x.Since.Equal(y.Since)404	})405}

1package main23// The deploy key minter: the cluster operator's part of the flux4// feature.5//6// The flux feature syncs the fleet's declared state from a git7// repository, and that repository is expected to be private, so the8// sync engine needs a credential. liken mints that credential inside9// the cluster, instead of asking a person to generate a key and10// carry it in. The person's whole job is to copy the public half11// from the Cluster's status and register it at the forge as a deploy12// key. Private material never travels, never sits on a USB stick,13// and never depends on someone getting key formats right by hand.14//15// The key is one per cluster, not one per machine, because a16// narrower key would not narrow anything: the private half lives in17// a Secret, every Secret lives in the cluster datastore, and every18// leader carries that datastore. The datastore is the unit of19// exposure, so one key per datastore states the truth. Rotation is a20// person's decision, made by deleting the Secret; the next sweep21// mints a fresh pair and publishes the new half to register.22//23// This job belongs to the cluster operator because the credential is24// cluster-scoped: the sweep is the one writer of Cluster status, and25// the flux-system Secret is a cluster-level object that any leader's26// sweep may create. Init cannot do it, because init runs before k3s27// exists. The permission to touch the Secret arrives with the28// feature itself: the flux feature's manifests carry a Role in29// flux-system (flux/manifests/flux-system.yaml), seeded only while30// the feature is declared, so this operator holds no standing Secret31// access on fleets that never asked for GitOps.3233import (34	"crypto/ed25519"35	"embed"36	"encoding/json"37	"encoding/pem"38	"errors"39	"fmt"40	"net/http"41	"strings"42	"time"4344	"github.com/liken-sh/liken/kubernetes/apiclient"45	"github.com/liken-sh/liken/kubernetes/events"46	"github.com/liken-sh/liken/liken/cluster"47	"github.com/liken-sh/liken/liken/kubernetes"48	"golang.org/x/crypto/ssh"49	"sigs.k8s.io/yaml"50)5152// The Secret's home, by Flux's conventions: the flux-system53// namespace, a Secret named flux-system, and the identity /54// identity.pub keys that Flux's git client reads. Following the55// conventions means the sync engine finds the credential with no56// configuration pointing at it.57const (58	fluxSecretPath  = "/api/v1/namespaces/flux-system/secrets/flux-system"59	fluxSecretsPath = "/api/v1/namespaces/flux-system/secrets"60)6162// ensureFluxDeployKey makes sure the fleet's deploy key exists when63// the flux feature is declared, and returns the public half for the64// status to carry. It returns "" when the feature is off, and also65// on any API failure, because the sweep runs again and a missing66// status one pass long is cheaper than a wrong one. The first passes67// after the feature turns on can fail here legitimately: the Role68// that grants this operator its Secret access is itself seeded by69// the feature, and k3s may not have applied it yet. The sweep's70// level-triggered loop absorbs that window; nothing here needs to71// wait or retry.72//73// A mint posts FluxDeployKeyMinted on the Cluster, because the person74// must register the new public half at the forge before the first75// sync.76func ensureFluxDeployKey(c *apiclient.Client, recorder *events.Recorder, clusterDoc *cluster.Cluster) string {77	if !clusterDoc.FeatureEnabled(cluster.FeatureFlux) {78		return ""79	}80	// The declared configuration supplies known_hosts. A81	// malformed declaration does not block the mint: the key is82	// per-cluster and parameter-independent, and init's feature pass83	// already reports the configuration problem.84	cfg, _ := clusterDoc.FluxConfig()8586	secret := &kubernetes.Secret{}87	err := c.RequestJSON(http.MethodGet, fluxSecretPath, nil, secret)88	if err == nil {89		// The key exists and never changes here, but the Secret's90		// known_hosts entry follows the declaration: a forge that91		// rotates its host key must not strand the sync on a stale92		// pin.93		if cfg != nil && string(secret.Data["known_hosts"]) != cfg.KnownHosts {94			patch, _ := json.Marshal(map[string]any{95				"stringData": map[string]string{"known_hosts": cfg.KnownHosts},96			})97			if err := kubernetes.PatchJSON(c, fluxSecretPath, patch); err != nil {98				fmt.Printf("refreshing the flux known_hosts: %v\n", err)99			}100		}101		return string(secret.Data["identity.pub"])102	}103	if !errors.Is(err, apiclient.ErrNotFound) {104		fmt.Printf("reading the flux deploy key: %v\n", err)105		return ""106	}107108	pub, priv, err := mintDeployKey(clusterDoc.Metadata.Name)109	if err != nil {110		fmt.Printf("minting the flux deploy key: %v\n", err)111		return ""112	}113	knownHosts := ""114	if cfg != nil {115		knownHosts = cfg.KnownHosts116	}117	body, err := json.Marshal(map[string]any{118		"apiVersion": "v1",119		"kind":       "Secret",120		"metadata": map[string]string{121			"name":      "flux-system",122			"namespace": "flux-system",123		},124		"type": "Opaque",125		// stringData is the write-side convenience: the API server126		// encodes it into data, so this code never handles base64.127		"stringData": map[string]string{128			"identity":     priv,129			"identity.pub": pub,130			"known_hosts":  knownHosts,131		},132	})133	if err != nil {134		fmt.Printf("encoding the flux deploy key secret: %v\n", err)135		return ""136	}137	if err := c.RequestJSON(http.MethodPost, fluxSecretsPath, body, nil); err != nil {138		// A conflict means another leader's sweep minted first. That139		// copy's key is the fleet's key; the next pass reads it.140		if !errors.Is(err, apiclient.ErrConflict) {141			fmt.Printf("creating the flux deploy key secret: %v\n", err)142		}143		return ""144	}145	fmt.Printf("minted the flux deploy key; register the public half from the Cluster's status at the forge\n")146	recorder.Normal(clusterReference(clusterDoc), reasonFluxDeployKeyMinted,147		"minted the flux deploy key; register the public half from status.flux.publicKey at the forge")148	return pub149}150151// The engine seed: the pinned gotk-components.yaml that the flux152// domain renders at build time and the Makefile copies here for153// the embed below, which cannot reach outside the package. The154// directory always holds a README, so a plain `go build` works155// before the flux domain has fetched anything; a binary built that156// way carries no seed, and the planter reports it instead of157// planting nothing silently.158//159//go:embed seed160var seedFS embed.FS161162func engineSeed() ([]byte, error) {163	return seedFS.ReadFile("seed/gotk-components.yaml")164}165166// engineProbePath names the one object whose absence means the167// engine is gone: the kustomize-controller Deployment. That168// controller is the applier that heals everything else from git, so169// while it exists, git owns the engine, whatever its state; when it170// does not, nothing can heal, and liken re-plants the seed.171const engineProbePath = "/apis/apps/v1/namespaces/flux-system/deployments/kustomize-controller"172173// seedObject is one document of the seed: its addressing fields,174// and its whole body as JSON, ready to send.175type seedObject struct {176	APIVersion string177	Kind       string178	Name       string179	Namespace  string180	body       []byte181}182183// parseSeed splits the seed into its documents. The seed is184// flux-rendered YAML with clean document separators, so the split is185// by separator lines; each document converts to JSON once, here,186// and never again on the apply path.187func parseSeed(raw []byte) ([]seedObject, error) {188	var objects []seedObject189	for _, doc := range strings.Split("\n"+string(raw), "\n---\n") {190		if strings.TrimSpace(stripYAMLComments(doc)) == "" {191			continue192		}193		body, err := yaml.YAMLToJSON([]byte(doc))194		if err != nil {195			return nil, fmt.Errorf("parsing a seed document: %w", err)196		}197		var meta struct {198			APIVersion string `json:"apiVersion"`199			Kind       string `json:"kind"`200			Metadata   struct {201				Name      string `json:"name"`202				Namespace string `json:"namespace"`203			} `json:"metadata"`204		}205		if err := json.Unmarshal(body, &meta); err != nil {206			return nil, err207		}208		objects = append(objects, seedObject{209			APIVersion: meta.APIVersion,210			Kind:       meta.Kind,211			Name:       meta.Metadata.Name,212			Namespace:  meta.Metadata.Namespace,213			body:       body,214		})215	}216	return objects, nil217}218219// stripYAMLComments drops comment and blank lines, so a document220// that is only commentary (the seed's generated-file header) reads221// as empty.222func stripYAMLComments(doc string) string {223	var kept []string224	for _, line := range strings.Split(doc, "\n") {225		trimmed := strings.TrimSpace(line)226		if trimmed == "" || strings.HasPrefix(trimmed, "#") {227			continue228		}229		kept = append(kept, line)230	}231	return strings.Join(kept, "\n")232}233234// seedKinds maps each kind the seed may contain to its REST235// resource. liken's API client has no discovery machinery, and it236// does not need any: the seed's vocabulary is small and known, and237// an unknown kind is a loud error naming what to add, which happens238// exactly when a new Flux version grows a new kind.239var seedKinds = map[string]struct {240	resource   string241	namespaced bool242}{243	"Namespace":                {"namespaces", false},244	"ResourceQuota":            {"resourcequotas", true},245	"NetworkPolicy":            {"networkpolicies", true},246	"CustomResourceDefinition": {"customresourcedefinitions", false},247	"ServiceAccount":           {"serviceaccounts", true},248	"ClusterRole":              {"clusterroles", false},249	"ClusterRoleBinding":       {"clusterrolebindings", false},250	"Role":                     {"roles", true},251	"RoleBinding":              {"rolebindings", true},252	"Service":                  {"services", true},253	"Deployment":               {"deployments", true},254}255256// collectionPath builds the create URL for one seed object: the core257// group lives under /api/v1, every other group under /apis/<group>/258// <version>, with the namespace segment for namespaced kinds.259func collectionPath(o seedObject) (string, error) {260	kind, known := seedKinds[o.Kind]261	if !known {262		return "", fmt.Errorf("the seed contains a %s, a kind the planter's table does not map; add it to seedKinds", o.Kind)263	}264	base := "/apis/" + o.APIVersion265	if !strings.Contains(o.APIVersion, "/") {266		base = "/api/" + o.APIVersion267	}268	if kind.namespaced {269		return base + "/namespaces/" + o.Namespace + "/" + kind.resource, nil270	}271	return base + "/" + kind.resource, nil272}273274// engineProbeInterval sets how long the sweep waits between engine275// probes. The probe is a backstop for one failure: someone deletes276// the kustomize-controller `Deployment` while the flux feature is277// on. This program watches only Machines, so no event announces that278// deletion. The sweep runs at least every ten seconds and again after279// each Machine change, and a probe on every sweep would make this GET280// the most frequent request `liken` sends, for an answer that almost281// never changes. One GET a minute costs less than a second watch and282// its restart handling. An engine that heals within a minute is283// still far faster than the only other repair, which is this284// program's next start.285const engineProbeInterval = 60 * time.Second286287// engineProbe holds the one fact the engine's care must remember288// across passes: when the sweep last asked whether the engine is289// there. The channel poller (channel.go) keeps its state the same290// way, and for the same reason. The sweep stays level-triggered and291// keeps nothing itself, so what must outlive a pass lives in a value292// the caller threads through. This one needs no mutex, because the293// probe runs inline on the sweep's goroutine and one pass runs at a294// time, while the channel poller's fetch runs on its own goroutine.295type engineProbe struct {296	asked time.Time297}298299// TryAsk reports whether this sweep may probe, and records the ask300// when it may. The zero value probes on the first sweep. A feature301// that is off does not use up an interval, because the caller returns302// before it calls this. Planting the seed counts as an ask, because303// the probe records the ask before it reads the answer. That keeps304// the state at one timestamp, and the interval then holds with no305// exception. A plant that only partly landed waits for the next306// probe, inside the same minute this interval promises.307func (p *engineProbe) TryAsk(now time.Time) bool {308	if now.Sub(p.asked) < engineProbeInterval {309		return false310	}311	p.asked = now312	return true313}314315// ensureFluxEngine plants the engine seed when the engine is gone.316// The probe asks once an interval, so a deleted engine heals within317// about a minute, not at the next boot. Each object is a plain318// create, and a conflict means the object already exists, which the319// planter leaves exactly as it found it: the seed only ever fills320// absence. Present but broken stays the repository's problem on321// purpose; liken answers only for gone. A planting posts322// FluxEngineSeeded on the Cluster, with how many objects it created.323func ensureFluxEngine(c *apiclient.Client, recorder *events.Recorder, clusterDoc *cluster.Cluster, seed []byte, probe *engineProbe, now time.Time) {324	if !clusterDoc.FeatureEnabled(cluster.FeatureFlux) {325		return326	}327	if !probe.TryAsk(now) {328		return329	}330	err := c.RequestJSON(http.MethodGet, engineProbePath, nil, nil)331	if err == nil {332		return333	}334	if !errors.Is(err, apiclient.ErrNotFound) {335		fmt.Printf("probing for the flux engine: %v\n", err)336		return337	}338	objects, err := parseSeed(seed)339	if err != nil {340		fmt.Printf("reading the engine seed: %v\n", err)341		return342	}343	fmt.Printf("the flux engine is absent; planting the seed (%d objects)\n", len(objects))344	created, present, failed := 0, 0, 0345	for _, o := range objects {346		path, err := collectionPath(o)347		if err != nil {348			fmt.Printf("planting the engine seed: %v\n", err)349			failed++350			continue351		}352		err = c.RequestJSON(http.MethodPost, path, o.body, nil)353		switch {354		case errors.Is(err, apiclient.ErrConflict):355			present++ // it already exists; whatever is there stays356		case err != nil:357			fmt.Printf("planting %s %s: %v\n", o.Kind, o.Name, err)358			failed++359		default:360			created++361		}362	}363	recorder.Normal(clusterReference(clusterDoc), reasonFluxEngineSeeded, fmt.Sprintf(364		"the flux engine was absent; planted its seed: %d objects created, %d present already, %d failed",365		created, present, failed))366}367368// mintDeployKey generates the pair: an ed25519 key, the modern SSH369// default, small enough to read aloud and accepted by every current370// forge. The public half is one authorized_keys line, commented with371// the cluster's name so a forge's key list stays legible. The372// private half is OpenSSH PEM, the format Flux's git client parses.373func mintDeployKey(clusterName string) (publicKey, privateKey string, err error) {374	pub, priv, err := ed25519.GenerateKey(nil)375	if err != nil {376		return "", "", err377	}378	comment := "liken:" + clusterName379	pemBlock, err := ssh.MarshalPrivateKey(priv, comment)380	if err != nil {381		return "", "", err382	}383	sshPub, err := ssh.NewPublicKey(pub)384	if err != nil {385		return "", "", err386	}387	// MarshalAuthorizedKey ends the line with a newline and no388	// comment; the status reads better with the comment and without389	// the newline.390	line := string(ssh.MarshalAuthorizedKey(sshPub))391	line = line[:len(line)-1] + " " + comment392	return line, string(pem.EncodeToMemory(pemBlock)), nil393}

1package main23// This file records when this program saw each machine's heartbeat4// change, so a sweep judges the heartbeat's age on one clock.5//6// A machine's operator writes the renewTime of its heartbeat Lease from7// the machine's own clock. A sweep that subtracted that time from this8// program's clock would add the difference between the two clocks to9// every heartbeat's age. A machine whose clock runs 45 seconds behind10// would read Lost while it renews every 8 seconds, and a machine whose11// clock runs ahead would read Lost late after it stopped. Clocks12// differ most at boot, before the first time synchronization, and on a13// machine with a wrong hardware clock and no time server.14//15// So this program reads renewTime only to learn that it changed. The16// Leases' watch calls the handler below for each write, and the handler17// records the moment on this program's clock. The node lifecycle18// controller in Kubernetes judges a kubelet's lease the same way19// (pkg/controller/nodelifecycle/node_lifecycle_controller.go).20//21// The record is a cache of what the watch delivered, so a program that22// starts records each Lease as seen at that moment. This program opens23// its watches only after it takes the lead, and exits when it loses24// the lead, so a new leader is always a new process with an empty25// record. A machine that is already gone then reads Lost26// HeartbeatStaleAfter after that, and until then it reads its last27// written status, which is usually Ready. So the record also keeps the28// time of the first sweep's read of the Leases, and the rollout grants29// no reboot turn until the record is HeartbeatStaleAfter older than30// that read (decideRollout). Without the hold, a new leader that31// started a minute after the old leader's machine lost power could32// count that dead machine as up and grant a second leader its turn.3334import (35	"sync"36	"time"3738	"github.com/liken-sh/liken/kubernetes/informer"39	"github.com/liken-sh/liken/liken/kubernetes"40	"k8s.io/client-go/tools/cache"41)4243// heartbeatSightings maps each Lease's name to the last renewTime this44// program saw, and when it first saw it. The watch's handler writes it45// from the informer's goroutine and a sweep reads it from the loop's46// goroutine, so a lock guards it. Its zero value is an empty record.47type heartbeatSightings struct {48	mu   sync.Mutex49	seen map[string]sighting5051	// began is when the first sweep read the Leases, on this program's52	// clock. That read holds every Lease, so each Lease that existed53	// then has a sighting no later than began, and each Lease that54	// stayed unchanged since reads stale after began plus55	// HeartbeatStaleAfter. A Lease that appears later was written by a56	// machine that was up to write it.57	began time.Time58}5960type sighting struct {61	// renewed is the Lease's renewTime, from the machine's clock. It is62	// compared only for equality.63	renewed time.Time64	// at is when this program first saw that renewTime, from its own65	// clock.66	at time.Time67}6869// observe records that the Lease held renewed at now, and answers when70// this program first saw that renewTime. A renewTime that differs from71// the recorded one, earlier or later, is a renewal: a machine's clock72// can step back.73func (s *heartbeatSightings) observe(name string, renewed, now time.Time) time.Time {74	s.mu.Lock()75	defer s.mu.Unlock()76	if last, ok := s.seen[name]; ok && last.renewed.Equal(renewed) {77		return last.at78	}79	if s.seen == nil {80		s.seen = map[string]sighting{}81	}82	s.seen[name] = sighting{renewed: renewed, at: now}83	return now84}8586func (s *heartbeatSightings) forget(name string) {87	s.mu.Lock()88	defer s.mu.Unlock()89	delete(s.seen, name)90}9192// heard observes each renewal of one read of the Leases, and maps each93// Lease's name to when this program saw its renewTime change. A sweep94// reads the Leases from the watch's copy or from the API server, and95// both reads pass through here. The handler has already recorded each96// change the watch delivered, at the moment it arrived, so this records97// only a change that a direct read finds first.98func (s *heartbeatSightings) heard(renewals map[string]time.Time, now time.Time) map[string]time.Time {99	s.mu.Lock()100	if s.began.IsZero() {101		s.began = now102	}103	s.mu.Unlock()104	heard := make(map[string]time.Time, len(renewals))105	for name, renewed := range renewals {106		heard[name] = s.observe(name, renewed, now)107	}108	return heard109}110111// since answers when the first sweep read the Leases, or the zero time112// before any sweep has read them.113func (s *heartbeatSightings) since() time.Time {114	s.mu.Lock()115	defer s.mu.Unlock()116	return s.began117}118119// handler records each Lease the watch delivers. A Lease whose120// renewTime does not parse carries no liveness claim121// (kubernetes.Renewals), so it records nothing. A deleted Lease122// leaves the record, so a Lease created again later counts as seen123// when it arrives.124func (s *heartbeatSightings) handler(source informer.Source) cache.ResourceEventHandler {125	seen := func(object any) {126		lease, err := informer.Convert[kubernetes.Lease](object)127		if err != nil {128			informer.Report(source.String(), err)129			return130		}131		now := time.Now()132		for name, renewed := range kubernetes.Renewals([]kubernetes.Lease{lease}) {133			s.observe(name, renewed, now)134		}135	}136	return cache.ResourceEventHandlerFuncs{137		AddFunc:    func(object any) { seen(object) },138		UpdateFunc: func(_, object any) { seen(object) },139		DeleteFunc: func(object any) {140			// A tombstone can hold no copy of the Lease, but its key141			// always names it.142			key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(object)143			if err != nil {144				informer.Report(source.String(), err)145				return146			}147			if _, name, err := cache.SplitMetaNamespaceKey(key); err == nil {148				s.forget(name)149			}150		},151	}152}

1package main23// The feature janitor deletes the workloads that belong to retracted4// features.5//6// A feature's workload manifests ship in the image. init seeds them7// into k3s's auto-deploy directory only while the cluster document8// declares the feature (see init/features.go). Retraction removes the9// file, and that removal deletes nothing on its own.10//11// State that plainly, because the opposite is easy to assume: k3s12// never deletes an addon's objects because its manifest file went13// away. The deploy controller walks the files that exist, and applies14// them.15// Nothing reconciles an Addon against a source file that is no longer16// there, so a deleted manifest leaves both the Addon and everything17// it created in place. k3s does delete an addon's objects when the18// component is on its disable list, because then it still has the19// file to read and can act on what the file declares. A liken feature20// is not a k3s component and is never on that list.21//22// So the janitor is not a fallback for one boot path. It is the only23// thing that removes a retracted feature's workloads, on every path.24// Every feature-seeded workload carries a liken.sh/feature annotation25// that names the feature it belongs to. Each sweep deletes any26// liken-system workload whose annotation names a feature that the27// cluster document no longer declares. Whether the feature is28// declared is the only question the janitor asks. It does not wait29// for the retraction to roll through the fleet: once the document no30// longer claims a workload's feature, the fleet does not need that31// workload running anywhere. The janitor does not act on objects that32// carry no annotation. The operator and log-relay DaemonSets live in33// the same namespace, and no feature owns them.3435import (36	"errors"37	"fmt"38	"net/http"39	"net/url"4041	"github.com/liken-sh/liken/kubernetes/apiclient"42	"github.com/liken-sh/liken/liken/api"43	"github.com/liken-sh/liken/liken/cluster"44	"github.com/liken-sh/liken/liken/kubernetes"45)4647// featureAnnotation names the feature a workload belongs to. It is48// the janitor's whole contract with the manifests: a feature's49// manifests carry this annotation (for example,50// open-iscsi/manifests/iscsid.yaml), and everything else omits it.51const featureAnnotation = "liken.sh/feature"5253// featureWorkloadKinds lists the workload kinds the janitor sweeps.54// Each kind is a list endpoint in liken-system. Features seed55// DaemonSets today. When a feature ships its first Deployment or56// Service, add that kind here, in the same change.57var featureWorkloadKinds = []struct {58	kind     string59	listPath string60}{61	{"daemonset", daemonSetsPath},62}6364// featureWorkload holds the small amount of information the janitor65// needs about an object: its name, and which feature, if any, claims66// it.67type featureWorkload struct {68	Metadata struct {69		Name        string            `json:"name"`70		Annotations map[string]string `json:"annotations"`71	} `json:"metadata"`72}7374// decideRetractions computes which workloads belong to features that75// the document no longer declares. Presence in spec.features is the76// same opt-in test that init uses, so the two gates always agree.77func decideRetractions(features map[string]*cluster.FeatureConfig, workloads []featureWorkload) []featureWorkload {78	var retracted []featureWorkload79	for _, w := range workloads {80		slug := w.Metadata.Annotations[featureAnnotation]81		if slug == "" {82			continue83		}84		if _, declared := features[slug]; !declared {85			retracted = append(retracted, w)86		}87	}88	return retracted89}9091// janitorFeatureWorkloads carries out the janitor's work, once per92// sweep: it lists each swept kind, selects which workloads to93// retract, and deletes them. Deletion is by name with background94// propagation, so the workload's pods are deleted along with it.95func janitorFeatureWorkloads(r *fleetReader, clusterDoc *cluster.Cluster) {96	for _, k := range featureWorkloadKinds {97		workloads, err := r.workloads(k.listPath)98		if err != nil {99			fmt.Printf("listing %ss for the feature janitor: %v\n", k.kind, err)100			continue101		}102		for _, w := range decideRetractions(clusterDoc.Spec.Features, workloads) {103			name := w.Metadata.Name104			slug := w.Metadata.Annotations[featureAnnotation]105			path := k.listPath + "/" + name + "?propagationPolicy=Background"106			if err := r.client.RequestJSON(http.MethodDelete, path, nil, nil); err != nil {107				fmt.Printf("deleting %s %s for the retracted %s feature: %v\n", k.kind, name, slug, err)108			} else {109				fmt.Printf("the cluster no longer declares the %s feature; deleted %s %s\n", slug, k.kind, name)110			}111		}112	}113}114115// The flux janitor: the teardown half of the seed-once engine.116//117// The generic janitor above deletes only a workload that carries the118// feature annotation, and the flux janitor makes that same ownership119// test first, because here the answer is not always yes. An adopted120// cluster may already run a Flux that somebody else installed, long121// before liken arrived. The document an operator writes during122// adoption describes machines, but spec.features is an opt-in list,123// so the same document is also a claim over everything else in the124// cluster, and a slug left out of it is a retraction. So the janitor125// reads the ownership mark first, and tears down only the126// installation liken planted.127//128// The rest of the teardown cannot use the generic janitor's shape,129// because these objects are not inert. Deleting the Kustomization130// while its controller still runs triggers the engine's own deletion131// finalizer, and with prune on, that finalizer garbage-collects132// everything the repository ever applied: the workloads, the Machine133// documents, and the Cluster document itself. So flux's retraction is134// ordered, and the order is the whole design: kill the controllers135// first, so the finalizer can never fire, then remove the finalizers136// by hand, and only then delete the objects.137//138// Each sweep advances one stage and returns, so the stages are139// separated by real observations, never by in-process waits:140//141//  0. The flux-system Namespace carries no liken.sh/feature mark:142//     delete nothing, and report the refusal on the Cluster.143//  1. Engine Deployments still exist: delete them.144//  2. Controller pods still exist: wait. A Deployment's deletion is145//     asynchronous, and a controller that is still terminating could146//     still process a finalizer.147//  3. Controllers provably gone: strip the sync objects' finalizers,148//     delete them, and delete the engine's cluster-scoped remains:149//     the CRDs, the engine's RBAC, the planter's grant, and the150//     namespace.151//152// What survives is deliberate: nothing. Retraction burns the deploy153// key with the namespace, and a re-enabled feature mints a fresh154// key to register. Keeping the key was considered and rejected: it155// would mean k3s's addon machinery must never touch the namespace,156// and the ground would outlive the feature as unowned state. Off157// means off. The repository's own workloads also survive, as158// orphans: stopping the sync must not undeploy what the sync159// deployed.160//161// The janitor's rights are standing, in the operator's own manifest,162// unlike the planter's, which arrive with the feature. This is163// deliberate too: the janitor's job begins exactly when the164// feature's delivered grants disappear, so delivered rights could165// never clean up after the feature that delivered them. The166// standing rights are deletes on exact names, powerless to create167// or read anything.168169// The engine's namespace, by name and by path. The Namespace is also170// the ownership record: liken's own feature manifest declares it with171// the feature annotation (flux/manifests/flux-system.yaml), and this172// janitor reads that annotation before it deletes anything. k3s173// applies that manifest as an addon the moment the document declares174// the feature, so the mark lands with the namespace itself, ahead of175// anything the cluster operator does.176const (177	fluxNamespace     = "flux-system"178	fluxNamespacePath = "/api/v1/namespaces/" + fluxNamespace179)180181// fluxTeardownCondition is the type of the Cluster condition that182// reports a declined teardown. It is present only while there is a183// refusal to report, so the sweep removes it whenever the janitor184// returns nothing (see publishClusterStatus in fleet.go).185const fluxTeardownCondition = "FluxTeardown"186187// fluxTeardownPaths are the delete targets of the final stage, in188// order. The sync objects come first, finalizers already stripped;189// the namespace's own deletion then sweeps everything namespaced190// that remains, the deploy key Secret included; the cluster-scoped191// remains close it out. The CRD and RBAC names must match the192// engine seed; the parity test in flux_test.go holds them to it.193var fluxTeardownPaths = []string{194	"/apis/kustomize.toolkit.fluxcd.io/v1/namespaces/flux-system/kustomizations/flux-system",195	"/apis/source.toolkit.fluxcd.io/v1/namespaces/flux-system/gitrepositories/flux-system",196	fluxNamespacePath,197	"/apis/apiextensions.k8s.io/v1/customresourcedefinitions/buckets.source.toolkit.fluxcd.io",198	"/apis/apiextensions.k8s.io/v1/customresourcedefinitions/externalartifacts.source.toolkit.fluxcd.io",199	"/apis/apiextensions.k8s.io/v1/customresourcedefinitions/gitrepositories.source.toolkit.fluxcd.io",200	"/apis/apiextensions.k8s.io/v1/customresourcedefinitions/helmcharts.source.toolkit.fluxcd.io",201	"/apis/apiextensions.k8s.io/v1/customresourcedefinitions/helmrepositories.source.toolkit.fluxcd.io",202	"/apis/apiextensions.k8s.io/v1/customresourcedefinitions/ocirepositories.source.toolkit.fluxcd.io",203	"/apis/apiextensions.k8s.io/v1/customresourcedefinitions/kustomizations.kustomize.toolkit.fluxcd.io",204	"/apis/rbac.authorization.k8s.io/v1/clusterroles/crd-controller-flux-system",205	"/apis/rbac.authorization.k8s.io/v1/clusterroles/flux-edit-flux-system",206	"/apis/rbac.authorization.k8s.io/v1/clusterroles/flux-view-flux-system",207	"/apis/rbac.authorization.k8s.io/v1/clusterrolebindings/cluster-reconciler-flux-system",208	"/apis/rbac.authorization.k8s.io/v1/clusterrolebindings/crd-controller-flux-system",209	"/apis/rbac.authorization.k8s.io/v1/clusterroles/liken-engine-planter",210	"/apis/rbac.authorization.k8s.io/v1/clusterrolebindings/liken-engine-planter",211}212213// fluxControllerPodsPath lists the engine's controller pods, alive214// or terminating. The engine's own labels select them.215var fluxControllerPodsPath = "/api/v1/namespaces/flux-system/pods?labelSelector=" +216	url.QueryEscape("app in (source-controller,kustomize-controller)")217218// declinedFluxTeardown is the report the janitor returns when it219// refuses to tear down a Flux that liken did not plant. The message220// carries the remedy, because an operator who really does want liken221// to own this installation needs one command and should not have to222// find it.223func declinedFluxTeardown() *api.Condition {224	mark := featureAnnotation + "=" + cluster.FeatureFlux225	return &api.Condition{226		Type:   fluxTeardownCondition,227		Status: api.ConditionFalse,228		Reason: "NotPlantedByLiken",229		Message: "the cluster no longer declares flux, and the " + fluxNamespace +230			" Namespace carries no " + mark + " annotation, so liken did not plant this" +231			" installation and deleted nothing. To give liken this installation, run:" +232			" kubectl annotate namespace " + fluxNamespace + " " + mark,233	}234}235236// janitorFlux tears the flux feature down when the cluster document237// no longer declares it, and only when liken planted it. Every call238// is one stage at most; the sweep calls it again within ten seconds,239// and silence is the converged state. The returned condition reports240// a refusal, and nil means there is nothing to report.241func janitorFlux(c *apiclient.Client, clusterDoc *cluster.Cluster) *api.Condition {242	if clusterDoc.FeatureEnabled(cluster.FeatureFlux) {243		return nil244	}245246	// Stage 0: ownership, before a single delete. An installation247	// liken did not plant is not one liken removes, so the Namespace's248	// mark gates everything below. An absent Namespace means there is249	// no installation to tear down, and a read that fails outright250	// says nothing about ownership, so both answers stop the pass251	// without a verdict.252	//253	// This gate has one real cost, and it falls on a cluster that254	// liken founded before the mark existed: that cluster's255	// flux-system Namespace carries no annotation, so retracting the256	// feature now deletes nothing and raises the condition instead.257	// The operator annotates the Namespace and the teardown proceeds258	// on the next sweep. Failing closed is the correct direction here,259	// because the failure in the other direction deletes somebody260	// else's GitOps installation, and the condition names the one261	// command that clears this one.262	namespace := &featureWorkload{}263	err := c.RequestJSON(http.MethodGet, fluxNamespacePath, nil, namespace)264	if errors.Is(err, apiclient.ErrNotFound) {265		return nil266	}267	if err != nil {268		fmt.Printf("reading the %s namespace for the flux teardown: %v\n", fluxNamespace, err)269		return nil270	}271	if namespace.Metadata.Annotations[featureAnnotation] != cluster.FeatureFlux {272		return declinedFluxTeardown()273	}274275	// Stage 1: the controllers must die before anything else is276	// touched. A successful delete means this pass's work is done;277	// the pod check below needs a fresh observation anyway. A delete278	// that failed also ends the pass, because the Deployment still279	// exists: its pods can be gone for a moment, and a controller280	// that comes back would run the prune finalizers that stage 3281	// strips.282	stop := false283	for _, name := range []string{"source-controller", "kustomize-controller"} {284		path := "/apis/apps/v1/namespaces/flux-system/deployments/" + name285		if err := c.RequestJSON(http.MethodGet, path, nil, nil); errors.Is(err, apiclient.ErrNotFound) {286			continue287		}288		err := c.RequestJSON(http.MethodDelete, path+"?propagationPolicy=Background", nil, nil)289		switch {290		case err == nil:291			fmt.Printf("liken planted this flux and the cluster no longer declares it; deleted the %s Deployment\n", name)292			stop = true293		case !errors.Is(err, apiclient.ErrNotFound):294			fmt.Printf("flux teardown, deleting the %s Deployment: %v\n", name, err)295			stop = true296		}297	}298	if stop {299		return nil300	}301302	// Stage 2: wait out terminating controller pods. This gate is303	// what makes the finalizer unreachable: no controller process304	// exists past it.305	pods, err := kubernetes.List[featureWorkload](c, fluxControllerPodsPath)306	if err != nil || len(pods) > 0 {307		return nil308	}309310	// Stage 3: nothing can react anymore. Strip the sync objects'311	// finalizers so their deletion, and the namespace's, completes312	// instead of waiting forever for a controller that no longer313	// exists. Then delete what remains, and let 404s stay silent:314	// this function runs on every sweep, and the converged state is315	// nothing but 404s.316	for _, path := range fluxTeardownPaths[:2] {317		_ = kubernetes.PatchJSON(c, path, []byte(`{"metadata": {"finalizers": null}}`))318	}319	// Each line states why the delete was safe to make, because a320	// reader who meets these lines in a log is asking exactly that:321	// liken planted this installation, so it is liken's to remove, and322	// no controller survives to run a prune finalizer over it.323	for _, path := range fluxTeardownPaths {324		err := c.RequestJSON(http.MethodDelete, path, nil, nil)325		if err == nil {326			fmt.Printf("liken planted this flux and the cluster no longer declares it; the controllers are gone, so no prune can fire; deleted %s\n", path)327		} else if !errors.Is(err, apiclient.ErrNotFound) && !errors.Is(err, apiclient.ErrConflict) {328			fmt.Printf("flux teardown, deleting %s: %v\n", path, err)329		}330	}331	return nil332}

1package main23// Leader election: only one copy of this program acts on the fleet.4//5// The sweep writes Lost verdicts, the Cluster's status, and reboot6// grants, and it evicts and deletes. Two copies that sweep at the same7// time can each grant a reboot turn from a different view of the8// fleet, and together exceed the disruption budget. A rolling update, a9// node partition, and a replica count above one each run a second10// copy, so this program elects one acting copy through11// kubernetes/election, on a coordination.k8s.io Lease named12// liken-cluster-operator in liken-system. That package explains the13// election, its timings, and its limits.14//15// The election is not a fence. A former leader that paused can still16// land a grant it sent before the pause, after another copy took the17// Lease. The grant ledger in the Cluster's status, not the election,18// keeps the disruption budget against that grant (rollout.go).19//20// Only the copy that holds the Lease opens its watches and sweeps. A21// copy that the API server refuses on the Lease waits, as for any other22// failure of the election. The OS images carry this binary, so a23// rollback runs it under the RBAC that the target release's manifests24// grant. The Cluster CRD refuses a target older than 2026.09.28-001, the25// first release whose RBAC grants this program the create and update of26// its Lease, so every release that a rollback can reach lets a copy27// take the Lease.28//29// The election links client-go's typed clientset. That costs this30// binary about 13 MB, and it is accepted here because the cluster31// operator runs as one pod in the fleet. The machine operator runs on32// every machine and never imports the package.3334import (35	"context"36	"errors"37	"fmt"38	"os"3940	"k8s.io/client-go/rest"4142	"github.com/liken-sh/liken/kubernetes/election"43)4445// leaseName names the Lease the copies compete for, and leaseNamespace46// is the namespace the whole OS uses. The sweep also reads this Lease47// from its watch, to judge how fresh its copies are (watches.go).48const (49	leaseName      = "liken-cluster-operator"50	leaseNamespace = "liken-system"51)5253// electionOptions names this program's Lease and pod for the election.54// report receives each line with this program's prefix.55func electionOptions(pod string, report func(string)) election.Options {56	return election.Options{57		Name:      leaseName,58		Namespace: leaseNamespace,59		Pod:       pod,60		Report:    func(line string) { report(component + ": " + line) },61	}62}6364// lead blocks until this process may act. It exits the process when65// stop ends first, and whenever the Lease is lost after that. The pod's66// hostname is its name, which makes the identity readable in `kubectl67// get lease`.68func lead(stop context.Context) *election.Lead {69	pod, err := os.Hostname()70	if err != nil {71		fatal("reading the pod's name for the leader election: %v", err)72	}73	config, err := rest.InClusterConfig()74	if err != nil {75		fatal("in-cluster config for the leader election: %v", err)76	}77	options := electionOptions(pod, func(line string) { fmt.Println(line) })78	l, err := election.Start(stop, config, options)79	if errors.Is(err, election.ErrStopped) {80		os.Exit(0)81	}82	if err != nil {83		fatal("leader election: %v", err)84	}85	return l86}

1// liken-cluster-operator is the program that watches the fleet.2//3// Each machine's operator reports on itself. This leaves verdicts4// that no single machine can write: a dead machine cannot report5// that it is dead, no machine can total a headcount that includes6// itself without conflicting with the other machines' operators, and7// someone who can see every machine's request must hand out reboot8// turns. This program writes exactly those verdicts: the Lost phase9// on silent machines, the Cluster's status, and the rollout's10// grants. It also refreshes the OS's own DaemonSet pods after11// upgrades (see steward.go).12//13// The machine operator is privileged and node-local: a DaemonSet14// with hostPath mounts, running on the machine it manages. This15// program is different. It is an ordinary workload: a single-replica16// Deployment with no mounts, no host network, and no privilege. The17// Kubernetes API is its only input and its only output. This is what18// lets its RBAC role match exactly what a fleet observer needs: read19// machines and heartbeats, write statuses, evict stale OS pods.20//21// Only one copy of this program acts at a time. The Deployment rolls a22// new pod in beside the old one, and each copy competes for a leader23// election Lease; only the copy that holds it watches the fleet and24// writes. leader.go names the Lease and the reason for the election.25package main2627import (28	"context"29	"errors"30	"flag"31	"fmt"32	"os"33	"os/signal"34	"syscall"35	"time"3637	"k8s.io/client-go/dynamic"3839	"github.com/liken-sh/liken/kubernetes/apiclient"40	"github.com/liken-sh/liken/kubernetes/election"41	"github.com/liken-sh/liken/kubernetes/events"42	"github.com/liken-sh/liken/kubernetes/informer"43	"github.com/liken-sh/liken/liken/cluster"44	"github.com/liken-sh/liken/liken/kubernetes"45	"github.com/liken-sh/liken/liken/kubernetes/watch"46	"github.com/liken-sh/liken/liken/machine"47	"github.com/liken-sh/liken/liken/metrics"48)4950// component is this program's name in every metric that carries one.51const component = "liken-cluster-operator"5253// metricsAddress is where this operator answers a Prometheus scrape.54// The default is the port that55// `plans/completed/65-prometheus-metrics.md` at the top of the56// repository gives every process on the cluster network, so the binary57// carries the contract and the pod template only has to name the port58// it exposes. An empty value turns the listener off, for an owner who59// runs no Prometheus.60var metricsAddress = flag.String("metrics-address", ":9200",61	"the address to serve /metrics on; empty serves no metrics")6263func main() {64	flag.Parse()65	fmt.Println(component, machine.Version)6667	// A SIGTERM from the kubelet ends the loop after the sweep in68	// flight, and the process then releases the Lease, so the copy that69	// waits beside it takes over at once.70	stop, _ := signal.NotifyContext(context.Background(), syscall.SIGTERM, os.Interrupt)7172	// The metrics registry outlives every pass, because a counter's73	// whole value is that it accumulates (metrics.go). A copy that74	// waits for the Lease serves its runtime metrics too.75	operatorMetrics, clusterLayer := serveMetrics(*metricsAddress)7677	// A failure during setup ends the process deliberately. This is78	// the same crash-only method the machine operator uses: kubelet79	// restarts the pod with backoff, and the failure shows in80	// `kubectl get pods`.81	client, err := kubernetes.InClusterClient("")82	if err != nil {83		fatal("in-cluster config: %v", err)84	}85	watcher, err := informer.InCluster()86	if err != nil {87		fatal("in-cluster config for the watches: %v", err)88	}8990	// Nothing below runs until this copy holds the Lease.91	leader := lead(stop)9293	// The ticker is a clock. A heartbeat ages past the staleness limit94	// with no event, because an aging Lease is not a write. A granted95	// reboot turn passes rolloutStallAfter the same way, and the96	// channel poller and the engine probe keep their own intervals.97	// Ten seconds keeps those verdicts within one sweep of the moment98	// they come due. The objects a sweep judges come from99	// the watches' copies, so a tick costs the API server one read: the100	// flux deploy key Secret while the feature is declared, or the101	// flux-system Namespace while it is not (watches.go says why those102	// two stay direct).103	ticker := time.NewTicker(10 * time.Second)104	run(stop, context.Background(), leader, client, watcher, ticker.C, operatorMetrics, clusterLayer)105}106107// run is the work of a copy that may act: it opens the watches, sweeps108// until stop ends, and then steps down. The watches live until watches109// ends, which in main is the end of the process, so the sweep in110// flight at a shutdown still reads from them.111func run(stop, watches context.Context, leader *election.Lead, client *apiclient.Client,112	watcher dynamic.Interface, ticks <-chan time.Time, om *metrics.Operator, cm *clusterMetrics) {113	// Every write asks the election first (election.Lead.MayWrite).114	client = client.WithWriteGuard(leader.MayWrite)115116	// The watches start with the lead, so a copy that has never acted117	// holds no copy of the fleet and opens no stream (watches.go names each watch and118	// what wakes the loop). The wake channel has one slot, so a burst119	// of Machine changes makes one wake, and one sweep over the newest120	// state answers the whole burst.121	wakes := make(chan struct{}, 1)122	fleet := watchFleet(watches, watcher, client, watch.Signal(wakes), om.WatchRestarted)123	// The recorder posts the Events about the Machines and the Cluster124	// (events.go). It writes through the guarded client, so a replica125	// that lost the election posts nothing, and from its own goroutine,126	// so a sweep never waits on an Event. It outlives stop, so the127	// Events of the last sweep are written.128	fleet.recorder = events.New(context.Background(), client, component, events.Options{})129130	operate(stop, fleet, wakes, ticks, om, cm)131	// The sweep has returned, so no write of the sweep is in flight.132	// The recorder writes the Events the sweep posted, and Shutdown133	// returns after its last request has returned. So the Lease is134	// released after the last write, and a waiting copy takes it on its135	// next try. A recorder whose last write did not return in time136	// leaves the Lease to expire, after that write can land.137	flush, cancel := context.WithTimeout(context.Background(), events.ShutdownTimeout)138	defer cancel()139	if err := fleet.recorder.Shutdown(flush); err != nil {140		fmt.Printf("%v; leaving the Lease to expire\n", err)141		return142	}143	leader.StepDown()144}145146// operate finds the Cluster and sweeps the fleet until stop ends. A147// wake or a tick starts the next sweep, and stop ends the loop only148// between two sweeps, so the sweep in flight finishes its writes.149func operate(stop context.Context, fleet *fleetReader,150	wakes <-chan struct{}, ticks <-chan time.Time, om *metrics.Operator, cm *clusterMetrics) {151	// This program takes no configuration at all. It finds the152	// Cluster it operates in the Clusters' copy, because a fleet has153	// exactly one Cluster, and the machine operators seed it from the154	// image. Until the CRD is served and some machine's seed lands, in155	// the first minutes of a brand-new cluster, there is nothing to156	// operate, so this program waits.157	clusterDoc := awaitCluster(fleet, wakes, stop)158	if clusterDoc == nil {159		return160	}161	name := clusterDoc.Metadata.Name162	fmt.Printf("operating cluster %s\n", name)163164	// Three values outlive a pass, the same way the machine operator's165	// release fetcher does. The sweep stays level-triggered and166	// stateless, and each of these remembers only what it needs to167	// avoid asking or saying the same thing too often. The channel168	// poller keeps when it last asked and the channel's last answer.169	// The flux engine probe keeps when it last asked whether the engine170	// is still there. The pod steward keeps the terminating pods it has171	// already reported.172	poller := newChannelPoller()173	probe := &engineProbe{}174	steward := &podSteward{}175176	for stop.Err() == nil {177		started := time.Now()178		err := sweep(fleet, name, poller, probe, steward, cm)179		om.ObserveReconcile(clusterKind, time.Since(started), err)180		select {181		case <-wakes:182		case <-ticks:183		case <-stop.Done():184		}185	}186}187188// sweep runs one pass of the cluster operator's whole job, always189// starting from the cluster's current state. It reads the Cluster190// fresh, because its spec drives the rollout. It gives the channel191// poller its look at the spec. Then it lets the fleet sweep list the192// fleet, judge it, and write the result, carrying the engine probe193// along so the flux engine's care keeps its own cadence.194func sweep(reads *fleetReader, name string, poller *channelPoller, probe *engineProbe, steward *podSteward, cm *clusterMetrics) error {195	clusterDoc, err := reads.cluster(name)196	if err != nil {197		fmt.Printf("reading cluster %s: %v\n", name, err)198		return err199	}200	poller.Observe(clusterDoc.Spec.Releases,201		clusterDoc.Metadata.Annotations[cluster.CheckReleasesAnnotation], time.Now())202	return sweepFleet(reads, clusterDoc, poller.Available(), probe, steward, cm, time.Now())203}204205// awaitCluster waits until a Cluster exists, and answers it. A 404206// response only means the CRD is not served yet. An empty list means207// no machine has seeded the object yet. Both conditions resolve208// themselves as the fleet boots. The Clusters' copy wakes the loop209// when the first Cluster arrives. While the copy cannot answer, each210// look reads the API server, and the ticker's pace bounds those reads.211// It answers nil when stop ends first.212func awaitCluster(reads *fleetReader, wakes <-chan struct{}, stop context.Context) *cluster.Cluster {213	retry := time.NewTicker(10 * time.Second)214	defer retry.Stop()215	for {216		clusters, err := reads.clusters()217		if err != nil && !errors.Is(err, apiclient.ErrNotFound) {218			fmt.Printf("listing clusters: %v\n", err)219		}220		if len(clusters) > 0 {221			return &clusters[0]222		}223		select {224		case <-wakes:225		case <-retry.C:226		case <-stop.Done():227			return nil228		}229	}230}231232func fatal(format string, args ...any) {233	fmt.Fprintf(os.Stderr, format+"\n", args...)234	os.Exit(1)235}