Skip to content

fix(gateway): proxy sandbox-routed envd metrics requests #159

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
feimingxliu wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix(gateway): proxy sandbox-routed envd metrics requests #159

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 5 additions & 5 deletions services/gateway/internal/metrics.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -114,12 +114,12 @@ func setGatewayRouteSource(w http.ResponseWriter, source routeSource) {

func (s *Server) instrumentGatewayHTTP(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path == "/metrics" || s.isLocalGatewayHealthRequest(r) {
if s.isLocalGatewayEndpointRequest(r) {
next.ServeHTTP(w, r)
return
}
// A sandbox-routed /health request is user traffic, so keep it
// instrumented like any other proxy call.
// Sandbox-routed /health and /metrics requests are user traffic, so
// keep them instrumented like any other proxy call.

method := gatewayMethodLabel(r.Method)
route := gatewayRouteLabel(r.URL.Path)
Expand All @@ -132,8 +132,8 @@ func (s *Server) instrumentGatewayHTTP(next http.Handler) http.Handler {
})
}

func (s *Server) isLocalGatewayHealthRequest(r *http.Request) bool {
if r.URL.Path != "/health" || hasProxyRoutingHeaders(r.Header) {
func (s *Server) isLocalGatewayEndpointRequest(r *http.Request) bool {
if (r.URL.Path != "/health" && r.URL.Path != "/metrics") || hasProxyRoutingHeaders(r.Header) {
return false
}
hostRoute, hostRouteErr := parseHostRoute(r.Host, s.sandboxProxyDomains)
Expand Down
62 changes: 62 additions & 0 deletions services/gateway/internal/metrics_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,10 @@ package gateway

import (
"context"
"net/http"
"net/http/httptest"
"testing"
"time"
)

func TestGatewayRouteLabelIncludesV2Sandboxes(t *testing.T) {
Expand Down Expand Up @@ -38,6 +40,66 @@ func TestStatusRecorderRouteSourceLabel(t *testing.T) {
}
}

func TestInstrumentGatewayHTTPSkipsOnlyLocalEndpoints(t *testing.T) {
server := newTestServer(
t,
stubSchedulerClient{},
time.Second,
1024,
withSandboxProxyDomains("sandbox-proxy.example.invalid"),
)

tests := []struct {
name string
path string
host string
sandboxID string
targetPort string
wantWrapped bool
}{
{name: "local health", path: "/health"},
{name: "local metrics", path: "/metrics"},
{
name: "header-routed metrics",
path: "/metrics",
sandboxID: "sbx-metrics",
targetPort: "49983",
wantWrapped: true,
},
{
name: "host-routed metrics",
path: "/metrics",
host: "49983-sbx-metrics.sandbox-proxy.example.invalid",
wantWrapped: true,
},
{name: "ordinary gateway route", path: "/sandboxes", wantWrapped: true},
}

for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
wrapped := false
handler := server.instrumentGatewayHTTP(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
_, wrapped = w.(*statusRecorder)
w.WriteHeader(http.StatusNoContent)
}))
req := httptest.NewRequest(http.MethodGet, tt.path, nil)
if tt.host != "" {
req.Host = tt.host
}
if tt.sandboxID != "" {
req.Header.Set(headerSandboxID, tt.sandboxID)
}
if tt.targetPort != "" {
req.Header.Set(headerTargetPort, tt.targetPort)
}
handler.ServeHTTP(httptest.NewRecorder(), req)
if wrapped != tt.wantWrapped {
t.Fatalf("instrumentation wrapper present = %t, want %t", wrapped, tt.wantWrapped)
}
})
}
}

func TestHTTPStatusLabel(t *testing.T) {
cancelled, cancel := context.WithCancel(context.Background())
cancel()
Expand Down
17 changes: 10 additions & 7 deletions services/gateway/internal/server.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,11 +94,7 @@ func (s *Server) Handler() http.Handler {
// decoding %2F → / and issuing 301 redirects), which breaks proxy
// forwarding of percent-encoded path segments such as /files/%2F.
core := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path == "/metrics" {
http.NotFound(w, r)
return
}
if r.URL.Path == "/health" {
if r.URL.Path == "/health" || r.URL.Path == "/metrics" {
hostRoute, hostRouteErr := parseHostRoute(r.Host, s.sandboxProxyDomains)
if hostRoute != nil || hostRouteErr != nil {
s.handleProxy(w, r)
Expand All @@ -113,8 +109,15 @@ func (s *Server) Handler() http.Handler {
s.handleProxy(w, r)
return
}
// Keep load balancer health checks local when they are not sandbox-routed.
w.WriteHeader(http.StatusNoContent)
if r.URL.Path == "/health" {
// Keep load balancer health checks local when they are not sandbox-routed.
w.WriteHeader(http.StatusNoContent)
} else {
// Gateway Prometheus metrics use the separate metrics listener. Keep
// this path unavailable on the public HTTP listener unless it is
// explicitly routed to a sandbox.
http.NotFound(w, r)
}
return
}
s.handleProxy(w, r)
Expand Down
156 changes: 92 additions & 64 deletions services/gateway/internal/server_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1361,6 +1361,22 @@ func TestFlushInterval(t *testing.T) {
}
}

func TestMetricsEndpointReturnsNotFoundWithoutProxyRouting(t *testing.T) {
server := newTestServer(t, stubSchedulerClient{}, time.Second, 1024)
gatewayServer := httptest.NewServer(server.Handler())
defer gatewayServer.Close()

resp, err := http.Get(gatewayServer.URL + "/metrics")
if err != nil {
t.Fatalf("gateway metrics request failed: %v", err)
}
defer resp.Body.Close()

if resp.StatusCode != http.StatusNotFound {
t.Fatalf("gateway metrics status = %d, want %d", resp.StatusCode, http.StatusNotFound)
}
}

func TestHealthEndpointReturnsGatewayHealthWithoutProxyHeaders(t *testing.T) {
server := newTestServer(t, stubSchedulerClient{}, time.Second, 1024)
gatewayServer := httptest.NewServer(server.Handler())
Expand All @@ -1382,14 +1398,14 @@ func TestHealthEndpointReturnsGatewayHealthWithoutProxyHeaders(t *testing.T) {
}
}

func TestHealthEndpointWithSandboxHeadersProxiesToSandbox(t *testing.T) {
func TestHealthAndMetricsEndpointsWithSandboxHeadersProxyToSandbox(t *testing.T) {
type upstreamRequestSnapshot struct {
path string
targetPort string
forwardedURI string
}

requests := make(chan upstreamRequestSnapshot, 1)
requests := make(chan upstreamRequestSnapshot, 2)
upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
requests <- upstreamRequestSnapshot{
path: r.URL.Path,
Expand All @@ -1402,7 +1418,7 @@ func TestHealthEndpointWithSandboxHeadersProxiesToSandbox(t *testing.T) {

server := newTestServer(t, stubSchedulerClient{
lookupNodeFunc: func(_ context.Context, req *schedulerv1.LookupNodeRequest, _ ...grpc.CallOption) (*schedulerv1.LookupNodeResponse, error) {
if req.GetSandboxId() != "sbx-health" {
if req.GetSandboxId() != "sbx-service" {
return nil, fmt.Errorf("unexpected sandbox id lookup: %q", req.GetSandboxId())
}
return &schedulerv1.LookupNodeResponse{
Expand All @@ -1417,65 +1433,73 @@ func TestHealthEndpointWithSandboxHeadersProxiesToSandbox(t *testing.T) {
gatewayServer := httptest.NewServer(server.Handler())
defer gatewayServer.Close()

req, err := http.NewRequest(http.MethodGet, gatewayServer.URL+"/health", nil)
if err != nil {
t.Fatalf("build health proxy request failed: %v", err)
}
req.Header.Set(headerSandboxID, "sbx-health")
req.Header.Set(headerTargetPort, "49983")
for _, path := range []string{"/health", "/metrics"} {
t.Run(path, func(t *testing.T) {
req, err := http.NewRequest(http.MethodGet, gatewayServer.URL+path, nil)
if err != nil {
t.Fatalf("build proxy request failed: %v", err)
}
req.Header.Set(headerSandboxID, "sbx-service")
req.Header.Set(headerTargetPort, "49983")

resp, err := http.DefaultClient.Do(req)
if err != nil {
t.Fatalf("health proxy request failed: %v", err)
}
defer resp.Body.Close()
resp, err := http.DefaultClient.Do(req)
if err != nil {
t.Fatalf("proxy request failed: %v", err)
}
defer resp.Body.Close()

if resp.StatusCode != http.StatusNoContent {
t.Fatalf("health proxy status = %d, want %d", resp.StatusCode, http.StatusNoContent)
}
if resp.StatusCode != http.StatusNoContent {
t.Fatalf("proxy status = %d, want %d", resp.StatusCode, http.StatusNoContent)
}

upstreamReq := <-requests
if upstreamReq.path != "/proxy/health" {
t.Fatalf("upstream path = %q, want %q", upstreamReq.path, "/proxy/health")
}
if upstreamReq.targetPort != "49983" {
t.Fatalf("target port header = %q, want %q", upstreamReq.targetPort, "49983")
}
if upstreamReq.forwardedURI != "/health" {
t.Fatalf("X-Forwarded-URI = %q, want %q", upstreamReq.forwardedURI, "/health")
upstreamReq := <-requests
if upstreamReq.path != "/proxy"+path {
t.Fatalf("upstream path = %q, want %q", upstreamReq.path, "/proxy"+path)
}
if upstreamReq.targetPort != "49983" {
t.Fatalf("target port header = %q, want %q", upstreamReq.targetPort, "49983")
}
if upstreamReq.forwardedURI != path {
t.Fatalf("X-Forwarded-URI = %q, want %q", upstreamReq.forwardedURI, path)
}
})
}
}

func TestHealthEndpointWithProxyHeadersMissingSandboxIDReturnsBadRequest(t *testing.T) {
func TestHealthAndMetricsEndpointsWithProxyHeadersMissingSandboxIDReturnBadRequest(t *testing.T) {
server := newTestServer(t, stubSchedulerClient{}, time.Second, 1024)
gatewayServer := httptest.NewServer(server.Handler())
defer gatewayServer.Close()

req, err := http.NewRequest(http.MethodGet, gatewayServer.URL+"/health", nil)
if err != nil {
t.Fatalf("build malformed health proxy request failed: %v", err)
}
req.Header.Set(headerTargetPort, "49983")
for _, path := range []string{"/health", "/metrics"} {
t.Run(path, func(t *testing.T) {
req, err := http.NewRequest(http.MethodGet, gatewayServer.URL+path, nil)
if err != nil {
t.Fatalf("build malformed proxy request failed: %v", err)
}
req.Header.Set(headerTargetPort, "49983")

resp, err := http.DefaultClient.Do(req)
if err != nil {
t.Fatalf("malformed health proxy request failed: %v", err)
}
defer resp.Body.Close()
resp, err := http.DefaultClient.Do(req)
if err != nil {
t.Fatalf("malformed proxy request failed: %v", err)
}
defer resp.Body.Close()

if resp.StatusCode != http.StatusBadRequest {
t.Fatalf("malformed health proxy status = %d, want %d", resp.StatusCode, http.StatusBadRequest)
if resp.StatusCode != http.StatusBadRequest {
t.Fatalf("malformed proxy status = %d, want %d", resp.StatusCode, http.StatusBadRequest)
}
})
}
}

func TestHealthEndpointWithHostRoutingProxiesToSandbox(t *testing.T) {
func TestHealthAndMetricsEndpointsWithHostRoutingProxyToSandbox(t *testing.T) {
type upstreamRequestSnapshot struct {
path string
sandboxID string
targetPort string
}

requests := make(chan upstreamRequestSnapshot, 1)
requests := make(chan upstreamRequestSnapshot, 2)
upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
requests <- upstreamRequestSnapshot{
path: r.URL.Path,
Expand All @@ -1488,7 +1512,7 @@ func TestHealthEndpointWithHostRoutingProxiesToSandbox(t *testing.T) {

server := newTestServer(t, stubSchedulerClient{
lookupNodeFunc: func(_ context.Context, req *schedulerv1.LookupNodeRequest, _ ...grpc.CallOption) (*schedulerv1.LookupNodeResponse, error) {
if req.GetSandboxId() != "sbx-health" {
if req.GetSandboxId() != "sbx-service" {
return nil, fmt.Errorf("unexpected sandbox id lookup: %q", req.GetSandboxId())
}
return &schedulerv1.LookupNodeResponse{
Expand All @@ -1502,31 +1526,35 @@ func TestHealthEndpointWithHostRoutingProxiesToSandbox(t *testing.T) {
gatewayServer := httptest.NewServer(server.Handler())
defer gatewayServer.Close()

req, err := http.NewRequest(http.MethodGet, gatewayServer.URL+"/health", nil)
if err != nil {
t.Fatalf("build health request failed: %v", err)
}
req.Host = "40988-sbx-health.sandbox-proxy.example.invalid"
for _, path := range []string{"/health", "/metrics"} {
t.Run(path, func(t *testing.T) {
req, err := http.NewRequest(http.MethodGet, gatewayServer.URL+path, nil)
if err != nil {
t.Fatalf("build host-routed request failed: %v", err)
}
req.Host = "40988-sbx-service.sandbox-proxy.example.invalid"

resp, err := http.DefaultClient.Do(req)
if err != nil {
t.Fatalf("host-routed health request failed: %v", err)
}
defer resp.Body.Close()
resp, err := http.DefaultClient.Do(req)
if err != nil {
t.Fatalf("host-routed request failed: %v", err)
}
defer resp.Body.Close()

if resp.StatusCode != http.StatusAccepted {
t.Fatalf("health status = %d, want %d", resp.StatusCode, http.StatusAccepted)
}
if resp.StatusCode != http.StatusAccepted {
t.Fatalf("proxy status = %d, want %d", resp.StatusCode, http.StatusAccepted)
}

upstreamReq := <-requests
if upstreamReq.path != "/proxy/health" {
t.Fatalf("upstream path = %q, want %q", upstreamReq.path, "/proxy/health")
}
if upstreamReq.sandboxID != "sbx-health" {
t.Fatalf("sandbox routing header = %q, want %q", upstreamReq.sandboxID, "sbx-health")
}
if upstreamReq.targetPort != "40988" {
t.Fatalf("target port header = %q, want 40988", upstreamReq.targetPort)
upstreamReq := <-requests
if upstreamReq.path != "/proxy"+path {
t.Fatalf("upstream path = %q, want %q", upstreamReq.path, "/proxy"+path)
}
if upstreamReq.sandboxID != "sbx-service" {
t.Fatalf("sandbox routing header = %q, want %q", upstreamReq.sandboxID, "sbx-service")
}
if upstreamReq.targetPort != "40988" {
t.Fatalf("target port header = %q, want 40988", upstreamReq.targetPort)
}
})
}
}

Expand Down
Loading