llmgw/internal/cli/controlplanes.go

package cli

import (
	"context"
	"net"
	"net/http"
	"net/url"
	"os"
	"strings"
	"sync"
	"time"

	"gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi"
	"gitea.mynodes.xyz/ilker/llmgw/internal/store"
)

// floatingPath is where `server add --vip` records the floating address.
const floatingPath = "/etc/llmgw/floating"

// controlPlaneProbe answers the summary's question of which control plane
// leads: it asks each one's /leader, on the client port every one of them
// listens on, since they share a server.yaml.
//
// Asked, rather than read from the lease, because the lease names whoever
// held it last and says nothing about a standby that has stopped — and a
// standby that is not there is the thing worth seeing before it is needed.
type controlPlaneProbe struct {
	st   *store.DB
	self string
	// floating is the file the floating address is recorded in.
	floating string
	client   *http.Client
	// leaderURL is where a host is asked; tests point it elsewhere.
	leaderURL func(host string) string

	mu   sync.Mutex
	at   time.Time
	last adminapi.ControlPlanes
}

func newControlPlaneProbe(st *store.DB, listen string) *controlPlaneProbe {
	_, port, _ := net.SplitHostPort(listen)
	if port == "" {
		port = "8080"
	}
	self, _ := os.Hostname()
	return &controlPlaneProbe{st: st, self: self,
		leaderURL: func(host string) string { return "http://" + net.JoinHostPort(host, port) + "/leader" },
		floating:  floatingPath,
		client:    &http.Client{Timeout: time.Second}}
}

// Status is the answer, kept for a few seconds: the console asks every three,
// and each asking is a request to every control plane.
func (p *controlPlaneProbe) Status() adminapi.ControlPlanes {
	p.mu.Lock()
	defer p.mu.Unlock()
	if time.Since(p.at) < 2*time.Second {
		return p.last
	}
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	cps, err := p.st.ControlPlanes(ctx)
	if err != nil || len(cps) < 2 {
		// One control plane is not a set of them; nothing to say.
		p.at, p.last = time.Now(), adminapi.ControlPlanes{}
		return p.last
	}
	// Read each time: server add writes it after this process may have
	// started.
	out := adminapi.ControlPlanes{Members: make([]adminapi.ControlPlane, len(cps)), Floating: floatingAddress(p.floating)}
	var wg sync.WaitGroup
	for i, cp := range cps {
		host := cp.URL
		if u, err := url.Parse(cp.URL); err == nil && u.Hostname() != "" {
			host = u.Hostname()
		}
		out.Members[i] = adminapi.ControlPlane{Name: cp.Name, Address: host, This: cp.Name == p.self}
		wg.Go(func() { out.Members[i].State = p.ask(ctx, host) })
	}
	if out.Floating != "" {
		wg.Go(func() { out.FloatingUp = p.ask(ctx, out.Floating) == "leading" })
	}
	wg.Wait()
	p.at, p.last = time.Now(), out
	return out
}

// ask is what one address's /leader says.
func (p *controlPlaneProbe) ask(ctx context.Context, host string) string {
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, p.leaderURL(host), nil)
	if err != nil {
		return "not answering"
	}
	resp, err := p.client.Do(req)
	if err != nil {
		return "not answering"
	}
	_ = resp.Body.Close()
	switch resp.StatusCode {
	case http.StatusOK:
		return "leading"
	case http.StatusServiceUnavailable:
		return "standing by"
	}
	return "not answering"
}

// floatingAddress is the address in the file server add keeps beside the
// keepalived configuration, without its prefix; "" when nothing floats.
// keepalived's own file holds its password and is root's alone, and this
// process is not root.
func floatingAddress(path string) string {
	b, err := os.ReadFile(path) //nolint:gosec // a fixed path of our own
	if err != nil {
		return ""
	}
	addr, _, _ := strings.Cut(strings.TrimSpace(string(b)), "/")
	if net.ParseIP(addr) == nil {
		return ""
	}
	return addr
}