diff --git a/cmd/xtcp2/xtcp2.go b/cmd/xtcp2/xtcp2.go index 3461181..9dece9f 100644 --- a/cmd/xtcp2/xtcp2.go +++ b/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) } diff --git a/docs/observability.md b/docs/observability.md index 6d9b546..32fd870 100644 --- a/docs/observability.md +++ b/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`. diff --git a/nix/versions.nix b/nix/versions.nix index e5e2d8e..98a2b0b 100644 --- a/nix/versions.nix +++ b/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="; } diff --git a/pkg/health/health.go b/pkg/health/health.go new file mode 100644 index 0000000..4a20343 --- /dev/null +++ b/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")) +} diff --git a/pkg/health/health_test.go b/pkg/health/health_test.go new file mode 100644 index 0000000..bed9e57 --- /dev/null +++ b/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) + } +} diff --git a/pkg/xtcp/grpc_server.go b/pkg/xtcp/grpc_server.go index 85b1c66..9deb107 100644 --- a/pkg/xtcp/grpc_server.go +++ b/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) + } +} diff --git a/pkg/xtcp/poller.go b/pkg/xtcp/poller.go index 72142ca..45c7aab 100644 --- a/pkg/xtcp/poller.go +++ b/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()) diff --git a/pkg/xtcp/xtcp.go b/pkg/xtcp/xtcp.go index 730305a..f2327e0 100644 --- a/pkg/xtcp/xtcp.go +++ b/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