llmgw/internal/cli/serverinit_patroni.go

package cli

import (
	"context"
	"fmt"
	"net"
	"os"
	"os/exec"
	"strings"

	"gitea.mynodes.xyz/ilker/llmgw/internal/config"
	"gitea.mynodes.xyz/ilker/llmgw/internal/onboard"
	"gitea.mynodes.xyz/ilker/llmgw/internal/pgha"
	"gitea.mynodes.xyz/ilker/llmgw/internal/store"
)

// ensurePatroni puts the store in PostgreSQL under Patroni on the machines
// given with --patroni, and moves whatever store server.yaml names into it:
// SQLite, or a PostgreSQL on one machine. This machine is the third etcd vote
// when they are two.
func (in *serverInit) ensurePatroni(ctx context.Context) (err error) {
	setup := pgha.Setup{Emit: func(step, detail string, changed bool) {
		if changed {
			in.done("%s: %s", step, detail)
		} else {
			in.kept("%s: %s", step, detail)
		}
	}}
	setup.DB = in.patroniNodes
	for _, t := range in.patroni {
		if len(in.patroniNodes) > 0 {
			break
		}
		user, host, ok := strings.Cut(t, "@")
		if !ok {
			user, host = "", t
		}
		remote := &onboard.SSH{Addr: host, User: user}
		out, err := remote.Run(ctx, "hostname && sudo -n true", nil)
		if err != nil {
			return fmt.Errorf("%s: cannot log in and use sudo without a password: %v: %s",
				t, err, strings.TrimSpace(string(out)))
		}
		ip, err := ipv4Of(host)
		if err != nil {
			return err
		}
		setup.DB = append(setup.DB, pgha.Node{
			Member: pgha.Member{Name: strings.TrimSpace(strings.SplitN(string(out), "\n", 2)[0]), IP: ip},
			Remote: remote,
		})
	}
	self, err := sourceIPToward(setup.DB[0].IP)
	if err != nil {
		return err
	}
	name, _ := os.Hostname()
	setup.Witness = &pgha.Node{Member: pgha.Member{Name: name, IP: self}, Remote: pgha.Local{}}
	setup.Clients = []string{self}

	dsn, err := setup.Run(ctx)
	if err != nil {
		return err
	}

	cfg, err := config.Load(in.configPath())
	if err != nil {
		return err
	}
	current := cfg.Auth.Store
	if current == "" {
		current = in.storeAddress()
	}
	resolved, _ := store.ResolveAddress(current)
	if strings.TrimSpace(resolved) == dsn {
		in.kept("store in the Patroni cluster (%s)", in.dsnPath())
		return nil
	}

	// Still while it moves, and started again on the old store if the move
	// fails: a gateway down over a database that did not work out is worse
	// than one left where it was.
	stopped, err := in.stopService(ctx)
	if err != nil {
		return err
	}
	defer func() {
		if err != nil && stopped {
			_ = exec.Command("systemctl", "start", "llmgw-server").Run()
			fmt.Fprintf(in.out, "  ..    llmgw-server started again on its old store\n")
		}
	}()
	if err := in.moveInto(ctx, current, dsn); err != nil {
		return err
	}

	// The old address is kept beside the new one: the way back, if the
	// cluster is not what was wanted.
	if b, err := os.ReadFile(in.dsnPath()); err == nil && strings.TrimSpace(string(b)) != dsn {
		if err := os.WriteFile(in.dsnPath()+".pre-patroni", b, 0o600); err != nil {
			return err
		}
		_ = in.chown(in.dsnPath()+".pre-patroni", true)
	}
	if err := in.writeDSN(dsn); err != nil {
		return err
	}
	if current != "@"+in.dsnPath() {
		if err := setAuthStore(in.configPath(), "@"+in.dsnPath()); err != nil {
			return err
		}
	}
	// Started again by the service step that follows, on the new address.
	in.done("server.yaml now keeps the store in the Patroni cluster")
	return nil
}

// moveInto copies a store into the cluster, unless the cluster already holds
// one — from a run that copied and stopped before changing the address.
func (in *serverInit) moveInto(ctx context.Context, from, to string) error {
	src, err := store.Open(ctx, from)
	if err != nil {
		// Nothing there yet: a first install, with nothing to move.
		in.kept("no store at %s to move", from)
		return nil
	}
	defer src.Close()
	dst, err := store.Open(ctx, to)
	if err != nil {
		return fmt.Errorf("opening the cluster: %w", err)
	}
	defer dst.Close()
	empty, err := dst.Empty(ctx)
	if err != nil {
		return err
	}
	if !empty {
		in.kept("the cluster already holds a store; %s is left as it is", from)
		return nil
	}
	res, err := src.CopyTo(ctx, dst)
	if err != nil {
		return fmt.Errorf("copying the store into the cluster: %w", err)
	}
	var rows int64
	for _, r := range res {
		rows += r.Rows
	}
	in.done("copied %d rows into the cluster (the old store is kept as it was)", rows)
	return nil
}

// sourceIPToward is this machine's address on the route to another: the one
// the other machine sees it connect from, which is what etcd advertises and
// what the database lets in. Not the first --host, which names the gateway
// for its clients and may not resolve yet, or resolve to a floating address.
func sourceIPToward(ip string) (string, error) {
	c, err := net.Dial("udp", net.JoinHostPort(ip, "5432"))
	if err != nil {
		return "", fmt.Errorf("no route from this machine to %s: %w", ip, err)
	}
	defer c.Close()
	return c.LocalAddr().(*net.UDPAddr).IP.String(), nil
}

// ipv4Of is the address a host name or address stands for, as the other
// machines will be told to reach it.
func ipv4Of(host string) (string, error) {
	if ip := net.ParseIP(host); ip != nil {
		return ip.String(), nil
	}
	ips, err := net.LookupIP(host)
	if err != nil {
		return "", fmt.Errorf("%s does not resolve: %w", host, err)
	}
	for _, ip := range ips {
		if v4 := ip.To4(); v4 != nil {
			return v4.String(), nil
		}
	}
	return "", fmt.Errorf("%s has no IPv4 address", host)
}