feat(observe): export compute pressure metrics

cmd/gha-pressure-observer/main.go

@@ -0,0 +1,53 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"flag"
+	"fmt"
+	"log/slog"
+	"net/http"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
+	"github.com/NDDev-OpenNetwork/github-actions/internal/pressureobserve"
+)
+
+var version = "dev"
+var commit = "unknown"
+
+func main() {
+	statePath := flag.String("state-path", "/var/lib/gha-fleet/pressure-gate.json", "pressure state JSON path")
+	listen := flag.String("listen", "127.0.0.1:9471", "loopback listen address")
+	maxStaleness := flag.Duration("max-staleness", pressureobserve.DefaultMaxStaleness, "maximum pressure state age")
+	showVersion := flag.Bool("version", false, "print version")
+	flag.Parse()
+	if flag.NArg() != 0 {
+		fmt.Fprintln(os.Stderr, "positional arguments are not accepted")
+		os.Exit(2)
+	}
+	if *showVersion {
+		fmt.Printf("gha-pressure-observer %s (%s)\n", version, commit)
+		return
+	}
+	if *listen == "" || *maxStaleness <= 0 {
+		slog.Error("invalid pressure observer configuration")
+		os.Exit(1)
+	}
+	root, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer stop()
+	server := &http.Server{Addr: *listen, Handler: pressureobserve.Handler{StatePath: *statePath, MaxStaleness: *maxStaleness},
+		ReadHeaderTimeout: 5 * time.Second, ReadTimeout: 10 * time.Second, WriteTimeout: 10 * time.Second, IdleTimeout: 30 * time.Second}
+	go func() {
+		<-root.Done()
+		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+		defer cancel()
+		_ = server.Shutdown(ctx)
+	}()
+	if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
+		slog.Error("pressure observer stopped", "error", err)
+		os.Exit(1)
+	}
+}

internal/pressureobserve/observe.go

@@ -0,0 +1,114 @@
+package pressureobserve
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"os"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/NDDev-OpenNetwork/github-actions/internal/pressuregate"
+)
+
+const DefaultMaxStaleness = 90 * time.Second
+
+type Handler struct {
+	StatePath    string
+	MaxStaleness time.Duration
+	Now          func() time.Time
+}
+
+func (h Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodGet {
+		w.WriteHeader(http.StatusMethodNotAllowed)
+		return
+	}
+	if r.URL.RawQuery != "" {
+		w.WriteHeader(http.StatusBadRequest)
+		return
+	}
+	now := time.Now().UTC()
+	if h.Now != nil {
+		now = h.Now().UTC()
+	}
+	state, err := Load(h.StatePath)
+	fresh := err == nil && Fresh(state, now, h.maxStaleness())
+	switch r.URL.Path {
+	case "/healthz":
+		w.Header().Set("Content-Type", "application/json")
+		if !fresh {
+			w.WriteHeader(http.StatusServiceUnavailable)
+		}
+		_, _ = fmt.Fprintf(w, "{\"healthy\":%t,\"fresh\":%t}\n", fresh, fresh)
+	case "/metrics":
+		w.Header().Set("Content-Type", "text/plain; version=0.0.4")
+		_, _ = w.Write([]byte(Render(state, now, h.maxStaleness(), err)))
+	default:
+		w.WriteHeader(http.StatusNotFound)
+	}
+}
+
+func (h Handler) maxStaleness() time.Duration {
+	if h.MaxStaleness <= 0 {
+		return DefaultMaxStaleness
+	}
+	return h.MaxStaleness
+}
+
+func Load(path string) (pressuregate.State, error) {
+	info, err := os.Stat(path)
+	if err != nil {
+		return pressuregate.State{}, err
+	}
+	if !info.Mode().IsRegular() {
+		return pressuregate.State{}, fmt.Errorf("pressure state must be a regular file")
+	}
+	file, err := os.Open(path)
+	if err != nil {
+		return pressuregate.State{}, err
+	}
+	defer file.Close()
+	decoder := json.NewDecoder(file)
+	decoder.DisallowUnknownFields()
+	var state pressuregate.State
+	if err := decoder.Decode(&state); err != nil {
+		return pressuregate.State{}, fmt.Errorf("decode pressure state: %w", err)
+	}
+	if err := state.Validate(); err != nil {
+		return pressuregate.State{}, err
+	}
+	return state, nil
+}
+
+func Fresh(state pressuregate.State, now time.Time, maxStaleness time.Duration) bool {
+	return maxStaleness > 0 && !state.ObservedAt.After(now.Add(time.Second)) && now.Sub(state.ObservedAt) <= maxStaleness
+}
+
+func Render(state pressuregate.State, now time.Time, maxStaleness time.Duration, loadErr error) string {
+	var output strings.Builder
+	gauge := func(name, help string, value float64) {
+		fmt.Fprintf(&output, "# HELP %s %s\n# TYPE %s gauge\n%s %s\n", name, help, name, name, strconv.FormatFloat(value, 'f', -1, 64))
+	}
+	counter := func(name, help string, value uint64) {
+		fmt.Fprintf(&output, "# HELP %s %s\n# TYPE %s counter\n%s %d\n", name, help, name, name, value)
+	}
+	fresh := loadErr == nil && Fresh(state, now, maxStaleness)
+	age := float64(-1)
+	if !state.ObservedAt.IsZero() && !state.ObservedAt.After(now) {
+		age = now.Sub(state.ObservedAt).Seconds()
+	}
+	gauge("gha_fleet_pressure_observer_up", "Whether the local pressure state is valid and fresh.", boolFloat(fresh))
+	gauge("gha_fleet_pressure_sample_age_seconds", "Age of the local pressure publication, or -1 when unavailable.", age)
+	counter("gha_fleet_host_oom_kills_total", "Kernel OOM kills observed since host boot.", state.OOMKillsTotal)
+	gauge("gha_fleet_host_pressure_open", "Whether local pressure admission is open and fresh.", boolFloat(fresh && state.State == pressuregate.StateOpen))
+	return output.String()
+}
+
+func boolFloat(value bool) float64 {
+	if value {
+		return 1
+	}
+	return 0
+}

internal/pressureobserve/observe_test.go

@@ -0,0 +1,53 @@
+package pressureobserve
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestHandlerExportsOOMCounterAndFreshPressureState(t *testing.T) {
+	now := time.Date(2026, 8, 27, 0, 0, 0, 0, time.UTC)
+	path := filepath.Join(t.TempDir(), "pressure.json")
+	content := `{"schema_version":1,"state":"open","reason":"healthy","state_since":"2026-08-26T23:59:00Z","observed_at":"2026-08-26T23:59:30Z","oom_kills_total":23}`
+	if err := os.WriteFile(path, []byte(content), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	handler := Handler{StatePath: path, MaxStaleness: time.Minute, Now: func() time.Time { return now }}
+	response := httptest.NewRecorder()
+	handler.ServeHTTP(response, httptest.NewRequest(http.MethodGet, "/metrics", nil))
+	if response.Code != http.StatusOK {
+		t.Fatalf("status=%d", response.Code)
+	}
+	for _, wanted := range []string{"gha_fleet_pressure_observer_up 1\n", "gha_fleet_host_oom_kills_total 23\n", "gha_fleet_host_pressure_open 1\n"} {
+		if !strings.Contains(response.Body.String(), wanted) {
+			t.Fatalf("metrics missing %q\n%s", wanted, response.Body.String())
+		}
+	}
+}
+
+func TestHandlerFailsClosedForStaleOrInvalidState(t *testing.T) {
+	now := time.Date(2026, 8, 27, 0, 2, 0, 0, time.UTC)
+	path := filepath.Join(t.TempDir(), "pressure.json")
+	if err := os.WriteFile(path, []byte(`{"schema_version":1,"state":"closed","reason":"cpu-pressure","state_since":"2026-08-26T23:58:00Z","observed_at":"2026-08-26T23:59:00Z","oom_kills_total":8}`), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	handler := Handler{StatePath: path, MaxStaleness: time.Minute, Now: func() time.Time { return now }}
+	response := httptest.NewRecorder()
+	handler.ServeHTTP(response, httptest.NewRequest(http.MethodGet, "/healthz", nil))
+	if response.Code != http.StatusServiceUnavailable {
+		t.Fatalf("stale health status=%d", response.Code)
+	}
+	if err := os.WriteFile(path, []byte(`{"schema_version":1,"state":"invented"}`), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	response = httptest.NewRecorder()
+	handler.ServeHTTP(response, httptest.NewRequest(http.MethodGet, "/metrics", nil))
+	if !strings.Contains(response.Body.String(), "gha_fleet_pressure_observer_up 0\n") || !strings.Contains(response.Body.String(), "gha_fleet_host_pressure_open 0\n") {
+		t.Fatalf("invalid state did not fail closed\n%s", response.Body.String())
+	}
+}