monitoring/internal/kube/kube.go

// Package kube is the smallest in-cluster apiserver client that answers the
// three questions fleet health actually asks: what is this pod doing, what has
// it been saying, and please change this one label.
//
// It is deliberately not client-go. client-go brings a scheme, a codec, a
// informer machinery and ~40MB of transitive dependency to marshal four fields
// out of a Pod. The whole surface here is one Do method plus three callers.
package kube

import (
	"bytes"
	"context"
	"crypto/tls"
	"crypto/x509"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"os"
	"time"
)

const (
	apiServer = "https://kubernetes.default.svc"
	saDir     = "/var/run/secrets/kubernetes.io/serviceaccount"
)

// Client talks to the apiserver with the pod's own service-account identity.
type Client struct {
	http *http.Client
	base string
	dir  string
}

// New builds a client from the projected service-account volume. A missing CA
// is not fatal here — it surfaces on the first call, inside the caller's retry
// loop, rather than killing a process before its loop exists.
func New(timeout time.Duration) *Client {
	tr := &http.Transport{}
	if pem, err := os.ReadFile(saDir + "/ca.crt"); err == nil {
		pool := x509.NewCertPool()
		if pool.AppendCertsFromPEM(pem) {
			tr.TLSClientConfig = &tls.Config{RootCAs: pool, MinVersion: tls.VersionTLS12}
		}
	}
	return &Client{
		http: &http.Client{Timeout: timeout, Transport: tr},
		base: apiServer,
		dir:  saDir,
	}
}

// NewAt points a client at an arbitrary base with no credentials. It exists so
// the differential tests can drive the real code against a fake apiserver.
func NewAt(base string, timeout time.Duration) *Client {
	return &Client{http: &http.Client{Timeout: timeout}, base: base}
}

// token is re-read on every call: projected service-account tokens are rotated
// underneath a running pod, so caching one is a slow-motion outage.
func (c *Client) token() string {
	if c.dir == "" {
		return ""
	}
	b, err := os.ReadFile(c.dir + "/token")
	if err != nil {
		return ""
	}
	return string(bytes.TrimSpace(b))
}

// Do performs one apiserver call. body non-nil sends a merge-patch.
func (c *Client) Do(ctx context.Context, method, path string, body []byte) ([]byte, error) {
	var r io.Reader
	if body != nil {
		r = bytes.NewReader(body)
	}
	req, err := http.NewRequestWithContext(ctx, method, c.base+path, r)
	if err != nil {
		return nil, err
	}
	if tok := c.token(); tok != "" {
		req.Header.Set("Authorization", "Bearer "+tok)
	}
	if body != nil {
		req.Header.Set("Content-Type", "application/merge-patch+json")
	}
	resp, err := c.http.Do(req)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	out, err := io.ReadAll(resp.Body)
	if err != nil {
		return nil, err
	}
	if resp.StatusCode/100 != 2 {
		return nil, fmt.Errorf("%s %s: %s", method, path, resp.Status)
	}
	return out, nil
}

// Pod is the slice of a Pod fleet health reads.
type Pod struct {
	Metadata struct {
		Name   string            `json:"name"`
		Labels map[string]string `json:"labels"`
	} `json:"metadata"`
	Status struct {
		Phase             string `json:"phase"`
		PodIP             string `json:"podIP"`
		ContainerStatuses []struct {
			RestartCount int `json:"restartCount"`
		} `json:"containerStatuses"`
		Conditions []struct {
			Type   string `json:"type"`
			Status string `json:"status"`
		} `json:"conditions"`
	} `json:"status"`
}

// Restarts is the sum across containers, matching kubectl's own accounting.
func (p *Pod) Restarts() int {
	n := 0
	for _, cs := range p.Status.ContainerStatuses {
		n += cs.RestartCount
	}
	return n
}

// Ready reports the Ready condition. A pod with no Ready condition is not ready.
func (p *Pod) Ready() bool {
	for _, c := range p.Status.Conditions {
		if c.Type == "Ready" {
			return c.Status == "True"
		}
	}
	return false
}

// GetPod fetches one pod.
func (c *Client) GetPod(ctx context.Context, ns, name string) (*Pod, error) {
	b, err := c.Do(ctx, http.MethodGet, "/api/v1/namespaces/"+ns+"/pods/"+name, nil)
	if err != nil {
		return nil, err
	}
	var p Pod
	if err := json.Unmarshal(b, &p); err != nil {
		return nil, err
	}
	return &p, nil
}

// ListPods fetches every pod matching a label selector.
func (c *Client) ListPods(ctx context.Context, ns, selector string) ([]Pod, error) {
	p := "/api/v1/namespaces/" + ns + "/pods?labelSelector=" + url.QueryEscape(selector)
	b, err := c.Do(ctx, http.MethodGet, p, nil)
	if err != nil {
		return nil, err
	}
	var out struct {
		Items []Pod `json:"items"`
	}
	if err := json.Unmarshal(b, &out); err != nil {
		return nil, err
	}
	return out.Items, nil
}

// Logs reads a container's log with the given query string.
func (c *Client) Logs(ctx context.Context, ns, name, query string) (string, error) {
	b, err := c.Do(ctx, http.MethodGet, "/api/v1/namespaces/"+ns+"/pods/"+name+"/log?"+query, nil)
	return string(b), err
}

// SetLabel adds or (with value "") removes one label. A JSON null value is how
// the apiserver is told to delete a key in a merge-patch — the whole reason
// this takes a *string rather than a string.
func (c *Client) SetLabel(ctx context.Context, ns, name, key string, value *string) error {
	patch, err := json.Marshal(map[string]any{
		"metadata": map[string]any{"labels": map[string]any{key: value}},
	})
	if err != nil {
		return err
	}
	_, err = c.Do(ctx, http.MethodPatch, "/api/v1/namespaces/"+ns+"/pods/"+name, patch)
	return err
}