bootnode/fleet/cluster.go

package fleet

import (
	"context"
	"crypto/tls"
	"crypto/x509"
	"encoding/json"
	"fmt"
	"net/http"
	"os"
	"strconv"
	"strings"
	"time"
)

// Cluster is the read-only Kubernetes view the collector needs: which luxd
// pods exist and what the operator declared for the network.
type Cluster interface {
	Pods(ctx context.Context, namespace string) ([]Pod, error)
	Declared(ctx context.Context, namespace string) (Declared, error)
}

// Pod is one luxd StatefulSet member.
type Pod struct {
	Name  string
	Index int
	Phase string
	Ready bool
	IP    string
}

// Declared is the shape the LuxNetwork operator CR asks for. It is an
// intention, not an observation — the CR routinely lags the running fleet.
type Declared struct {
	Phase      string
	Validators int
	Image      string
	PluginSet  string
	VMPlugins  []string
}

const (
	saDir     = "/var/run/secrets/kubernetes.io/serviceaccount"
	crdPath   = "/apis/lux.cloud/v1/namespaces/%s/luxnetworks/luxd"
	podsPath  = "/api/v1/namespaces/%s/pods"
	podPrefix = "luxd-"
)

// InCluster reads the API server with the pod's own ServiceAccount. It is the
// whole Kubernetes dependency: two GETs, no client-go.
type InCluster struct {
	base string
	http *http.Client
}

// Available reports whether a ServiceAccount is projected into this pod.
func Available() bool {
	_, err := os.Stat(saDir + "/token")
	return err == nil
}

// NewInCluster builds the API-server reader from the projected ServiceAccount.
func NewInCluster() (*InCluster, error) {
	host, port := os.Getenv("KUBERNETES_SERVICE_HOST"), os.Getenv("KUBERNETES_SERVICE_PORT")
	if host == "" || port == "" {
		return nil, fmt.Errorf("not running in a cluster: KUBERNETES_SERVICE_HOST unset")
	}
	ca, err := os.ReadFile(saDir + "/ca.crt")
	if err != nil {
		return nil, err
	}
	pool := x509.NewCertPool()
	if !pool.AppendCertsFromPEM(ca) {
		return nil, fmt.Errorf("serviceaccount ca.crt is not a PEM bundle")
	}
	return &InCluster{
		base: "https://" + host + ":" + port,
		http: &http.Client{
			Timeout:   8 * time.Second,
			Transport: &http.Transport{TLSClientConfig: &tls.Config{RootCAs: pool, MinVersion: tls.VersionTLS12}},
		},
	}, nil
}

// get decodes one API-server object into out. The token is re-read per call
// because projected ServiceAccount tokens rotate while the process runs.
func (k *InCluster) get(ctx context.Context, path string, out any) error {
	token, err := os.ReadFile(saDir + "/token")
	if err != nil {
		return err
	}
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, k.base+path, nil)
	if err != nil {
		return err
	}
	req.Header.Set("Authorization", "Bearer "+strings.TrimSpace(string(token)))
	req.Header.Set("Accept", "application/json")
	resp, err := k.http.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return fmt.Errorf("GET %s: HTTP %d", path, resp.StatusCode)
	}
	return json.NewDecoder(resp.Body).Decode(out)
}

// Pods lists the luxd-N pods in a namespace with the facts needed to dial and
// classify them.
func (k *InCluster) Pods(ctx context.Context, namespace string) ([]Pod, error) {
	var list struct {
		Items []struct {
			Metadata struct {
				Name string `json:"name"`
			} `json:"metadata"`
			Status struct {
				Phase      string `json:"phase"`
				PodIP      string `json:"podIP"`
				Conditions []struct {
					Type   string `json:"type"`
					Status string `json:"status"`
				} `json:"conditions"`
			} `json:"status"`
		} `json:"items"`
	}
	if err := k.get(ctx, fmt.Sprintf(podsPath, namespace), &list); err != nil {
		return nil, err
	}
	var out []Pod
	for _, it := range list.Items {
		name := it.Metadata.Name
		if !strings.HasPrefix(name, podPrefix) {
			continue
		}
		idx, err := strconv.Atoi(name[strings.LastIndex(name, "-")+1:])
		if err != nil {
			continue
		}
		p := Pod{Name: name, Index: idx, Phase: it.Status.Phase, IP: it.Status.PodIP}
		if p.Phase == "" {
			p.Phase = "Unknown"
		}
		for _, c := range it.Status.Conditions {
			if c.Type == "Ready" {
				p.Ready = c.Status == "True"
			}
		}
		out = append(out, p)
	}
	return out, nil
}

// Declared reads the LuxNetwork CR for a namespace.
func (k *InCluster) Declared(ctx context.Context, namespace string) (Declared, error) {
	var cr struct {
		Spec struct {
			Validators int `json:"validators"`
			Image      struct {
				Repository string `json:"repository"`
				Tag        string `json:"tag"`
			} `json:"image"`
			PluginSource struct {
				Bucket    string `json:"bucket"`
				VMPlugins []struct {
					VMID string `json:"vmID"`
				} `json:"vmPlugins"`
			} `json:"pluginSource"`
		} `json:"spec"`
		Status struct {
			Phase string `json:"phase"`
		} `json:"status"`
	}
	if err := k.get(ctx, fmt.Sprintf(crdPath, namespace), &cr); err != nil {
		return Declared{}, err
	}
	d := Declared{
		Phase:      cr.Status.Phase,
		Validators: cr.Spec.Validators,
		PluginSet:  cr.Spec.PluginSource.Bucket,
		VMPlugins:  []string{},
	}
	if d.Phase == "" {
		d.Phase = "Unknown"
	}
	switch repo, tag := cr.Spec.Image.Repository, cr.Spec.Image.Tag; {
	case repo != "" && tag != "":
		d.Image = repo + ":" + tag
	case tag != "":
		d.Image = tag
	default:
		d.Image = repo
	}
	for _, p := range cr.Spec.PluginSource.VMPlugins {
		if p.VMID != "" {
			d.VMPlugins = append(d.VMPlugins, p.VMID)
		}
	}
	return d, nil
}