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 }