pkg/k8score/request_deadlines_test.go
Rama main · réplica de lectura del repositorio original.
package k8score
import (
"context"
"errors"
"io"
"net/http"
"net/http/httptest"
"strings"
"sync/atomic"
"testing"
"time"
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/watch"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/rest"
)
// read marks a request bounded by the idle, header and ceiling timeouts.
const read = time.Duration(-1)
func TestRequestDeadlineClassify(t *testing.T) {
cases := []struct {
name string
method string
url string
header http.Header
want time.Duration
server string // path prefix of the configured server URL
}{
{"cluster list", "GET", "/api/v1/secrets?resourceVersion=5", nil, read, ""},
{"namespaced list page", "GET", "/apis/apps/v1/namespaces/a/deployments?limit=500&continue=x", nil, read, ""},
{"single get", "GET", "/api/v1/namespaces/a/secrets/db", nil, read, ""},
{"namespace object", "GET", "/api/v1/namespaces/log", nil, read, ""},
{"namespace status", "GET", "/api/v1/namespaces/a/status", nil, read, ""},
{"object named log", "GET", "/api/v1/namespaces/a/configmaps/log", nil, read, ""},
{"crd resource named proxy", "GET", "/apis/example.com/v1/namespaces/a/proxy", nil, read, ""},
{"proxied server prefix", "GET", "/k8s/clusters/c-1/api/v1/namespaces/a/pods/p/log", nil, 0, "/k8s/clusters/c-1"},
{"discovery", "GET", "/version", nil, read, ""},
{"API-looking server prefix, log", "GET", "/api/v1/clusters/c-1/api/v1/namespaces/a/pods/p/log", nil, 0, "/api/v1/clusters/c-1"},
{"API-looking server prefix, proxy tail", "GET", "/api/v1/clusters/c-1/api/v1/namespaces/a/services/prom:9090/proxy/api/v1/query", nil, 0, "/api/v1/clusters/c-1"},
{"server prefix ending in proxy, list", "GET", "/api/v1/clusters/c-1/proxy/api/v1/secrets", nil, read, "/api/v1/clusters/c-1/proxy/"},
{"prefix-less list under apis", "GET", "/apis/apps/v1beta2/deployments", nil, read, ""},
{"api path inside a proxied tail", "GET", "/api/v1/namespaces/m/services/prom:9090/proxy/api/v1/query", nil, 0, ""},
{"watch with timeout", "GET", "/api/v1/secrets?watch=true&timeoutSeconds=300", nil, 6 * time.Minute, ""},
{"watch=1", "GET", "/api/v1/secrets?watch=1&timeoutSeconds=10", nil, 70 * time.Second, ""},
{"watchlist", "GET", "/api/v1/secrets?watch=true&sendInitialEvents=true&timeoutSeconds=540", nil, 10 * time.Minute, ""},
{"legacy watch path", "GET", "/api/v1/watch/namespaces/a/pods?timeoutSeconds=60", nil, 2 * time.Minute, ""},
{"watch without timeout", "GET", "/api/v1/secrets?watch=true", nil, 0, ""},
{"pod log", "GET", "/api/v1/namespaces/a/pods/p/log?container=c", nil, 0, ""},
{"follow", "GET", "/api/v1/namespaces/a/pods/p/log?follow=true", nil, 0, ""},
{"service proxy", "GET", "/api/v1/namespaces/m/services/prom:9090/proxy/api/v1/query", nil, 0, ""},
{"node proxy", "GET", "/api/v1/nodes/n1/proxy/metrics", nil, 0, ""},
{"exec upgrade", "GET", "/api/v1/namespaces/a/pods/p/exec?command=sh", http.Header{"Upgrade": {"websocket"}}, 0, ""},
{"portforward", "GET", "/api/v1/namespaces/a/pods/p/portforward", nil, 0, ""},
{"post", "POST", "/api/v1/namespaces/a/secrets", nil, 0, ""},
{"patch", "PATCH", "/api/v1/namespaces/a/secrets/db", nil, 0, ""},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
req := httptest.NewRequest(tc.method, "https://apiserver"+tc.url, nil)
for k, v := range tc.header {
req.Header[k] = v
}
rt := WrapRequestDeadlines(tc.server)(http.DefaultTransport).(*requestDeadlineRoundTripper)
bounded, watch := rt.classify(req)
got := watch
if bounded && watch == 0 {
got = read
}
if got != tc.want {
t.Fatalf("classify(%s %s) = %v, want %v", tc.method, tc.url, got, tc.want)
}
})
}
}
// A client built from a Host with an API-looking path prefix classifies its
// requests by what follows the prefix: the pod log stream is not cut, and a
// stalled LIST still is.
func TestRequestDeadlines_PrefixedServerURL(t *testing.T) {
const prefix = "/api/v1/clusters/c-1/proxy"
release := make(chan struct{})
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch strings.TrimPrefix(r.URL.Path, prefix) {
case "/api/v1/namespaces/a/pods/p/log":
w.WriteHeader(http.StatusOK)
w.(http.Flusher).Flush()
time.Sleep(300 * time.Millisecond)
_, _ = io.WriteString(w, "done")
case "/api/v1/namespaces/a/secrets":
select {
case <-r.Context().Done():
case <-release:
}
default:
http.NotFound(w, r)
}
}))
defer srv.Close()
defer close(release)
cfg := &rest.Config{Host: srv.URL + prefix}
serverURL, _, err := rest.DefaultServerUrlFor(cfg)
if err != nil {
t.Fatalf("DefaultServerUrlFor: %v", err)
}
rt := WrapRequestDeadlines(serverURL.Path)(http.DefaultTransport).(*requestDeadlineRoundTripper)
rt.headerTimeout = 100 * time.Millisecond
cfg.Transport = rt
client := kubernetes.NewForConfigOrDie(cfg)
ctx := context.Background()
body, err := client.CoreV1().Pods("a").GetLogs("p", &corev1.PodLogOptions{}).DoRaw(ctx)
if err != nil || string(body) != "done" {
t.Fatalf("log stream behind a prefixed server was cut: body=%q err=%v", body, err)
}
start := time.Now()
if _, err := client.CoreV1().Secrets("a").List(ctx, metav1.ListOptions{}); err == nil {
t.Fatal("stalled LIST behind a prefixed server should have hit the deadline")
}
if time.Since(start) > 5*time.Second {
t.Fatal("stalled LIST behind a prefixed server was not ended by the deadline")
}
}
// testRoundTripper is the production wrapper with short bounds.
func testRoundTripper(idle, header, ceiling time.Duration) *requestDeadlineRoundTripper {
rt := WrapRequestDeadlines("")(http.DefaultTransport).(*requestDeadlineRoundTripper)
rt.idle = idle
rt.headerTimeout = header
rt.ceiling = ceiling
return rt
}
func TestRequestDeadlines_StallsEndSlowTransfersFinish(t *testing.T) {
release := make(chan struct{})
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
hold := func() {
select {
case <-r.Context().Done():
case <-release:
}
}
switch r.URL.Path {
case "/api/v1/secrets": // never sends headers
hold()
case "/api/v1/configmaps": // headers and a first chunk, then nothing
w.WriteHeader(http.StatusOK)
_, _ = io.WriteString(w, `{"kind":`)
w.(http.Flusher).Flush()
hold()
case "/api/v1/pods": // a slow transfer that keeps delivering bytes
w.WriteHeader(http.StatusOK)
for i := 0; i < 12; i++ {
_, _ = io.WriteString(w, "x")
w.(http.Flusher).Flush()
select {
case <-r.Context().Done():
return
case <-time.After(50 * time.Millisecond):
}
}
case "/api/v1/events": // trickles bytes forever
w.WriteHeader(http.StatusOK)
for {
_, _ = io.WriteString(w, "x")
w.(http.Flusher).Flush()
select {
case <-r.Context().Done():
return
case <-time.After(20 * time.Millisecond):
}
}
}
}))
defer srv.Close()
defer close(release)
client := &http.Client{Transport: testRoundTripper(200*time.Millisecond, 300*time.Millisecond, 2*time.Second)}
t.Run("no headers ends at the header timeout", func(t *testing.T) {
start := time.Now()
_, err := client.Get(srv.URL + "/api/v1/secrets")
if !errors.Is(err, errRequestEnded) {
t.Fatalf("err = %v, want a stall", err)
}
if d := time.Since(start); d < 250*time.Millisecond || d > 2*time.Second {
t.Fatalf("ended after %v, want about the 300ms header timeout", d)
}
})
t.Run("a body that stops ends at the idle timeout", func(t *testing.T) {
resp, err := client.Get(srv.URL + "/api/v1/configmaps")
if err != nil {
t.Fatalf("headers should arrive: %v", err)
}
start := time.Now()
_, err = io.ReadAll(resp.Body)
_ = resp.Body.Close()
if !errors.Is(err, errRequestEnded) {
t.Fatalf("err = %v, want a stall", err)
}
if n := strings.Count(err.Error(), errRequestEnded.Error()); n != 1 {
t.Fatalf("the cause should appear once, got %q", err)
}
if d := time.Since(start); d < 150*time.Millisecond || d > 2*time.Second {
t.Fatalf("ended after %v, want about the 200ms idle timeout", d)
}
})
t.Run("a slow transfer longer than the idle timeout is not cut", func(t *testing.T) {
start := time.Now()
resp, err := client.Get(srv.URL + "/api/v1/pods")
if err != nil {
t.Fatalf("get: %v", err)
}
body, err := io.ReadAll(resp.Body)
_ = resp.Body.Close()
if err != nil || len(body) != 12 {
t.Fatalf("slow transfer was cut: %d bytes, err=%v", len(body), err)
}
if d := time.Since(start); d < 400*time.Millisecond {
t.Fatalf("transfer took %v; the test needs it to outlast the idle and header timeouts", d)
}
})
t.Run("a transfer that never ends stops at the ceiling", func(t *testing.T) {
start := time.Now()
resp, err := client.Get(srv.URL + "/api/v1/events")
if err != nil {
t.Fatalf("get: %v", err)
}
_, err = io.ReadAll(resp.Body)
_ = resp.Body.Close()
if !errors.Is(err, errRequestEnded) || !strings.Contains(err.Error(), "still running") {
t.Fatalf("err = %v, want the ceiling", err)
}
if d := time.Since(start); d < 1500*time.Millisecond || d > 4*time.Second {
t.Fatalf("ended after %v, want about the 2s ceiling", d)
}
})
t.Run("a caller's shorter deadline still wins", func(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
defer cancel()
req, _ := http.NewRequestWithContext(ctx, http.MethodGet, srv.URL+"/api/v1/secrets", nil)
start := time.Now()
if _, err := client.Do(req); !errors.Is(err, context.DeadlineExceeded) {
t.Fatalf("err = %v", err)
}
if time.Since(start) > 200*time.Millisecond {
t.Fatal("caller's shorter deadline was not honored")
}
})
}
// A quiet watch is healthy: it is bounded only by its timeoutSeconds plus
// grace, never by the idle timeout.
func TestRequestDeadlines_QuietWatchIsNotIdleTimed(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusOK)
w.(http.Flusher).Flush()
select {
case <-r.Context().Done():
case <-time.After(500 * time.Millisecond):
_, _ = io.WriteString(w, "event")
}
}))
defer srv.Close()
client := &http.Client{Transport: testRoundTripper(100*time.Millisecond, 100*time.Millisecond, time.Second)}
resp, err := client.Get(srv.URL + "/api/v1/secrets?watch=true&timeoutSeconds=60")
if err != nil {
t.Fatalf("watch: %v", err)
}
body, err := io.ReadAll(resp.Body)
_ = resp.Body.Close()
if err != nil || string(body) != "event" {
t.Fatalf("quiet watch was cut: body=%q err=%v", body, err)
}
}
// A watch that outlives its timeoutSeconds plus grace is ended with an error
// client-go passes on instead of retrying, so the reflector lists again rather
// than silently re-watching a dead path.
func TestRequestDeadlines_DeadWatchSurfacesAsError(t *testing.T) {
var attempts atomic.Int32
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Query().Get("watch") != "true" {
http.NotFound(w, r)
return
}
if attempts.Add(1) == 1 || r.URL.Query().Get("resourceVersion") == "hang-headers" {
<-r.Context().Done() // no headers
return
}
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusOK)
w.(http.Flusher).Flush()
<-r.Context().Done() // headers, then nothing
}))
defer srv.Close()
rt := WrapRequestDeadlines("")(http.DefaultTransport).(*requestDeadlineRoundTripper)
rt.watchGrace = 200 * time.Millisecond
cfg := &rest.Config{Host: srv.URL, Transport: rt}
client := kubernetes.NewForConfigOrDie(cfg)
timeout := int64(1) // the deadline is timeoutSeconds plus grace: 1.2s
ctx := context.Background()
_, err := client.CoreV1().Secrets("").Watch(ctx, metav1.ListOptions{TimeoutSeconds: &timeout, ResourceVersion: "hang-headers"})
if !errors.Is(err, errRequestEnded) {
t.Fatalf("a watch with no headers: err = %v, want it ended by Radar", err)
}
if n := attempts.Load(); n != 1 {
t.Fatalf("client-go retried the ended watch %d times; it must surface the error", n-1)
}
w, err := client.CoreV1().Secrets("").Watch(ctx, metav1.ListOptions{TimeoutSeconds: &timeout})
if err != nil {
t.Fatalf("watch: %v", err)
}
defer w.Stop()
select {
case ev, ok := <-w.ResultChan():
if !ok || ev.Type != watch.Error {
t.Fatalf("a dead established watch must end with an error event, got ok=%v %v", ok, ev.Type)
}
case <-time.After(5 * time.Second):
t.Fatal("a dead established watch was not ended")
}
}