monitoring/internal/fleet/gate.go

package fleet

import (
	"context"
	"fmt"
	"net/http"
	"time"

	"github.com/luxfi/monitoring/internal/kube"
	"github.com/luxfi/monitoring/internal/luxd"
)

// GateConfig is what the tip gate needs to decide who serves public traffic.
type GateConfig struct {
	Namespace  string
	Selector   string
	RPCPort    string
	RPCPath    string
	LabelKey   string
	LabelValue string
	MaxLag     uint64
	Interval   time.Duration
	Timeout    time.Duration

	// KubeBase and NodeBase override in-cluster addressing, as in Config.
	KubeBase string
	NodeBase func(podIP string) string
}

// Gate labels the luxd pods that are keeping up and unlabels the ones that are
// not. The Service that fronts public RPC selects on that label, so this is the
// thing standing between a lagging node and a user reading stale chain state.
type Gate struct {
	cfg  GateConfig
	kube *kube.Client
	http *http.Client
	// Log is where each decision goes. Injectable so tests can read it.
	Log func(string)
}

func NewGate(cfg GateConfig) *Gate {
	k := kube.New(cfg.Timeout)
	if cfg.KubeBase != "" {
		k = kube.NewAt(cfg.KubeBase, cfg.Timeout)
	}
	return &Gate{
		cfg:  cfg,
		kube: k,
		http: &http.Client{Timeout: cfg.Timeout},
		Log:  func(s string) { fmt.Println(s) },
	}
}

// Run reconciles until ctx is done. It never exits the loop on error: a
// crash-looping gate stops reconciling, and the labels it leaves behind freeze
// whatever it last decided.
func (g *Gate) Run(ctx context.Context) {
	for {
		if err := g.Reconcile(ctx); err != nil {
			g.Log(fmt.Sprintf("reconcile failed: %v", err))
		}
		select {
		case <-ctx.Done():
			return
		case <-time.After(g.cfg.Interval):
		}
	}
}

// Reconcile performs one pass.
func (g *Gate) Reconcile(ctx context.Context) error {
	pods, err := g.kube.ListPods(ctx, g.cfg.Namespace, g.cfg.Selector)
	if err != nil {
		return err
	}

	tips := make(map[string]*uint64, len(pods))
	for i := range pods {
		p := &pods[i]
		tips[p.Metadata.Name] = g.tip(ctx, p)
	}

	var top uint64
	answered := false
	for _, t := range tips {
		if t != nil && (!answered || *t > top) {
			top, answered = *t, true
		}
	}
	if !answered {
		// Fail static. Losing sight of the fleet is not evidence that the fleet
		// is unfit, and emptying the pool would take the endpoint down outright.
		g.Log("no node answered; labels unchanged")
		return nil
	}

	for i := range pods {
		p := &pods[i]
		name := p.Metadata.Name
		t := tips[name]
		want := t != nil && top-*t <= g.cfg.MaxLag
		has := p.Metadata.Labels[g.cfg.LabelKey] == g.cfg.LabelValue

		changed := ""
		if want != has {
			var v *string
			if want {
				v = &g.cfg.LabelValue
			}
			if err := g.kube.SetLabel(ctx, g.cfg.Namespace, name, g.cfg.LabelKey, v); err != nil {
				return err
			}
			changed = " (changed)"
		}
		lag := "-"
		if t != nil {
			lag = fmt.Sprintf("%d", top-*t)
		}
		g.Log(fmt.Sprintf("%s tip=%s lag=%s serving=%v%s", name, tipStr(t), lag, want, changed))
	}
	return nil
}

// tip asks one pod for its accepted head. A pod that is not Running or has no
// IP is not asked — there is nothing to ask.
func (g *Gate) tip(ctx context.Context, p *kube.Pod) *uint64 {
	ip := p.Status.PodIP
	if ip == "" || p.Status.Phase != "Running" {
		return nil
	}
	base := fmt.Sprintf("http://%s:%s", ip, g.cfg.RPCPort)
	if g.cfg.NodeBase != nil {
		base = g.cfg.NodeBase(ip)
	}
	// eth_blockNumber, not a whole block: the gate runs every few seconds
	// against every validator and only ever needed the height. A node that
	// will not answer returns nil, which is the honest answer for every
	// failure mode that matters — refused connection, a bootstrapping node
	// that 404s the chain handler, a body that is not JSON — and they all
	// mean the same thing: do not send it public traffic.
	c := &luxd.Client{Base: base, HTTP: g.http, CPath: g.cfg.RPCPath}
	n, err := c.BlockNumber(ctx)
	if err != nil {
		return nil
	}
	return &n
}

func tipStr(t *uint64) string {
	if t == nil {
		return "None"
	}
	return fmt.Sprintf("%d", *t)
}