Merge pull request #63 from randomizedcoder/feat/listener-ip-ttl

feat(health): /healthz + /readyz + gRPC health (land on main)

Author: randomizedcoder

cmd/xtcp2/xtcp2.go

@@ -31,6 +31,7 @@ import (
 	"github.com/pkg/profile"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"github.com/randomizedcoder/xtcp2/pkg/health"
 	"github.com/randomizedcoder/xtcp2/pkg/ipsockopt"
 	"github.com/randomizedcoder/xtcp2/pkg/misc"
 	"github.com/randomizedcoder/xtcp2/pkg/xtcp"
@@ -689,6 +690,9 @@ func initPromHandler(promPath string, promListen string, ipv4TTL, ipv6HopLimit u
 			MaxRequestsInFlight: promMaxRequestsInFlight,
 		},
 	))
+	// Liveness + readiness for container/k8s deployment, on the same listener.
+	http.HandleFunc("/healthz", health.Healthz)
+	http.HandleFunc("/readyz", health.Readyz)
 	go servePromHandler(promListen, ipv4TTL, ipv6HopLimit)
 }
 

docs/observability.md

@@ -5,6 +5,7 @@ xtcp2 is built to run as a long-lived daemon, so it ships first-class observabil
 ## Table of contents
 
 - [Prometheus metrics](#prometheus-metrics)
+- [Health & readiness](#health--readiness)
 - [pprof](#pprof)
 - [Pyroscope continuous profiling](#pyroscope-continuous-profiling)
 - [Capability checks](#capability-checks)
@@ -15,6 +16,21 @@ xtcp2 is built to run as a long-lived daemon, so it ships first-class observabil
 
 `pkg/xtcp/prometheus.go` registers the daemon's metrics and serves them over HTTP. By default they are exposed at `:9088/metrics` (`-promListen`, `-promPath`). Metrics cover the collection pipeline — netlink reads, deserialization, envelope rows flushed, destination sends, and namespace counts — which is what you scrape to alarm on a stalled collector or a destination backpressure problem. The `metrics-audit` tool/check (`nix build .#test-tools-metrics-audit`) guards metric registration.
 
+## Health & readiness
+
+For container / Kubernetes deployment the metrics HTTP server also serves two
+probe endpoints (same `-promListen` address):
+
+- **`/healthz`** — liveness. Returns `200` as soon as the HTTP server is up. Use
+  it for a Docker `HEALTHCHECK` or a k8s `livenessProbe`.
+- **`/readyz`** — readiness. Returns `200` only once the daemon has initialised
+  its destination and netlinkers and started polling; `503` until then and again
+  during shutdown. Use it for a k8s `readinessProbe` / `startupProbe` so traffic
+  and rollouts wait until xtcp2 is actually collecting.
+
+The gRPC port additionally serves the standard `grpc.health.v1` service, which
+reports `SERVING` on the same readiness condition (for native k8s gRPC probes).
+
 ## pprof
 
 The standard Go `net/http/pprof` endpoints are mounted on the metrics HTTP server, so `/debug/pprof/*` is available on the same `-promListen` address for live CPU, heap, goroutine, mutex, and block profiles. For one-shot file-based profiling, `-profile.mode` enables a profiling session of mode `cpu`, `mem`, `mutex`, or `block`.

nix/versions.nix

@@ -99,5 +99,5 @@
   # Go vendor hash. Update by running `nix build .#xtcp2` and pasting the
   # `got:` value from the hash mismatch error. Used by every Nix check that
   # needs deps in the sandbox (see nix/lib/goModules.nix).
-  goVendorHash = "sha256-pP+rYZgVijaF6sF8DLyW997ZH3+ype0iylkYtapEJnY=";
+  goVendorHash = "sha256-KpZrd1NhcLMEFrNehiGBwt7sKFZlwa+uankR1itYizw=";
 }

pkg/health/health.go

@@ -0,0 +1,40 @@
+// Package health exposes liveness/readiness for containerised deployment
+// (Docker healthcheck, Kubernetes httpGet probes). Readiness is a single
+// process-wide flag — there is one xtcp2 daemon per process — that the daemon
+// flips true once it has initialised its destination and netlinkers and started
+// polling, and false on shutdown. The gRPC health service (see grpc_server.go)
+// is driven from the same flag.
+package health
+
+import (
+	"net/http"
+	"sync/atomic"
+)
+
+var ready atomic.Bool
+
+// SetReady sets the process readiness state reported by Readyz.
+func SetReady(r bool) { ready.Store(r) }
+
+// Ready reports the current readiness state.
+func Ready() bool { return ready.Load() }
+
+// Healthz is a liveness handler: 200 as soon as the HTTP server is serving. It
+// says nothing about whether the daemon is polling yet — that is Readyz.
+func Healthz(w http.ResponseWriter, _ *http.Request) {
+	w.WriteHeader(http.StatusOK)
+	_, _ = w.Write([]byte("ok\n"))
+}
+
+// Readyz is a readiness handler: 200 once the daemon has initialised its
+// destination + netlinkers and started polling, else 503 (and again on
+// shutdown) — so an orchestrator holds traffic/rollout until xtcp2 is live.
+func Readyz(w http.ResponseWriter, _ *http.Request) {
+	if !ready.Load() {
+		w.WriteHeader(http.StatusServiceUnavailable)
+		_, _ = w.Write([]byte("not ready\n"))
+		return
+	}
+	w.WriteHeader(http.StatusOK)
+	_, _ = w.Write([]byte("ready\n"))
+}

pkg/health/health_test.go

@@ -0,0 +1,36 @@
+package health
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+)
+
+func TestHealthz_alwaysOK(t *testing.T) {
+	rr := httptest.NewRecorder()
+	Healthz(rr, httptest.NewRequest(http.MethodGet, "/healthz", nil))
+	if rr.Code != http.StatusOK {
+		t.Fatalf("Healthz = %d, want 200", rr.Code)
+	}
+}
+
+func TestReadyz_reflectsState(t *testing.T) {
+	t.Cleanup(func() { SetReady(false) })
+
+	SetReady(false)
+	rr := httptest.NewRecorder()
+	Readyz(rr, httptest.NewRequest(http.MethodGet, "/readyz", nil))
+	if rr.Code != http.StatusServiceUnavailable {
+		t.Fatalf("Readyz(not ready) = %d, want 503", rr.Code)
+	}
+
+	SetReady(true)
+	if !Ready() {
+		t.Fatal("Ready() = false after SetReady(true)")
+	}
+	rr = httptest.NewRecorder()
+	Readyz(rr, httptest.NewRequest(http.MethodGet, "/readyz", nil))
+	if rr.Code != http.StatusOK {
+		t.Fatalf("Readyz(ready) = %d, want 200", rr.Code)
+	}
+}

pkg/xtcp/grpc_server.go

@@ -8,11 +8,14 @@ import (
 	"net"
 	"time"
 
+	"github.com/randomizedcoder/xtcp2/pkg/health"
 	"github.com/randomizedcoder/xtcp2/pkg/ipsockopt"
 	"github.com/randomizedcoder/xtcp2/pkg/xtcp_config"
 	"github.com/randomizedcoder/xtcp2/pkg/xtcp_flat_record"
 	"google.golang.org/grpc"
 	_ "google.golang.org/grpc/encoding/gzip"
+	grpchealth "google.golang.org/grpc/health"
+	healthpb "google.golang.org/grpc/health/grpc_health_v1"
 	"google.golang.org/grpc/keepalive"
 	"google.golang.org/grpc/reflection"
 )
@@ -82,6 +85,12 @@ func (x *XTCP) startGRPCflatRecordService(ctx context.Context) {
 	x.configService = NewXtcpConfigService(ctx, x.registry, x.config, &x.changePollFrequencyCh, x.debugLevel)
 	xtcp_config.RegisterConfigServiceServer(grpcServer, x.configService)
 
+	// Standard gRPC health service (grpc.health.v1) so k8s gRPC probes work.
+	// Starts NOT_SERVING; setReady flips it to SERVING once the daemon polls.
+	x.grpcHealth = grpchealth.NewServer()
+	x.grpcHealth.SetServingStatus("", healthpb.HealthCheckResponse_NOT_SERVING)
+	healthpb.RegisterHealthServer(grpcServer, x.grpcHealth)
+
 	// Stop the gRPC server when ctx fires. grpcServer.Serve blocks
 	// indefinitely on lis.Accept and is NOT ctx-aware on its own —
 	// without this goroutine the gRPC server outlives Run() and would
@@ -101,3 +110,17 @@ func (x *XTCP) startGRPCflatRecordService(ctx context.Context) {
 		log.Printf("startGRPCflatRecordService grpcServer.Serve err:%v", serveErr)
 	}
 }
+
+// setReady flips the process readiness in one place: the HTTP /readyz flag and
+// the gRPC health status move together. The daemon calls setReady(true) once it
+// starts polling and setReady(false) on shutdown.
+func (x *XTCP) setReady(r bool) {
+	health.SetReady(r)
+	if x.grpcHealth != nil {
+		status := healthpb.HealthCheckResponse_NOT_SERVING
+		if r {
+			status = healthpb.HealthCheckResponse_SERVING
+		}
+		x.grpcHealth.SetServingStatus("", status)
+	}
+}

pkg/xtcp/poller.go

@@ -31,6 +31,12 @@ func (x *XTCP) Poller(ctx context.Context, wg *sync.WaitGroup) {
 		log.Printf("Poller DestinationReady")
 	}
 
+	// Destination is up and netlinkers have started — the daemon is now
+	// operational, so flip liveness/readiness (HTTP /readyz + gRPC health) to
+	// ready, and back to not-ready when the poller returns (shutdown).
+	x.setReady(true)
+	defer x.setReady(false)
+
 	ticker := time.NewTicker(x.config.PollFrequency.AsDuration())
 	defer ticker.Stop()
 	x.pollTimeoutTimer = time.NewTimer(x.config.PollTimeout.AsDuration())

pkg/xtcp/xtcp.go

@@ -14,6 +14,8 @@ import (
 	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
+	grpchealth "google.golang.org/grpc/health"
+
 	"github.com/randomizedcoder/xtcp2/pkg/cgroupid"
 	"github.com/randomizedcoder/xtcp2/pkg/xsync"
 	"github.com/randomizedcoder/xtcp2/pkg/xtcp_config"
@@ -118,6 +120,7 @@ type XTCP struct {
 
 	flatRecordService *xtcpFlatRecordService
 	configService     *xtcpConfigService
+	grpcHealth        *grpchealth.Server
 
 	pC *prometheus.CounterVec
 	pH *prometheus.SummaryVec