Skip to content
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.

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

Automated cherry pick of #107452: Fix header mutation race in timeout filter #107458

Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
19 changes: 16 additions & 3 deletions staging/src/k8s.io/apiserver/pkg/server/filters/timeout.go
Original file line number Diff line number Diff line change
Expand Up @@ -149,7 +149,7 @@ type timeoutWriter interface {
}

func newTimeoutWriter(w http.ResponseWriter) (timeoutWriter, http.ResponseWriter) {
base := &baseTimeoutWriter{w: w}
base := &baseTimeoutWriter{w: w, handlerHeaders: w.Header().Clone()}
wrapped := responsewriter.WrapForHTTP1Or2(base)

return base, wrapped
Expand All @@ -161,6 +161,9 @@ var _ responsewriter.UserProvidedDecorator = &baseTimeoutWriter{}
type baseTimeoutWriter struct {
w http.ResponseWriter

// headers written by the normal handler
handlerHeaders http.Header

mu sync.Mutex
// if the timeout handler has timeout
timedOut bool
Expand All @@ -182,7 +185,7 @@ func (tw *baseTimeoutWriter) Header() http.Header {
return http.Header{}
}

return tw.w.Header()
return tw.handlerHeaders
}

func (tw *baseTimeoutWriter) Write(p []byte) (int, error) {
Expand All @@ -196,7 +199,10 @@ func (tw *baseTimeoutWriter) Write(p []byte) (int, error) {
return 0, http.ErrHijacked
}

tw.wroteHeader = true
if !tw.wroteHeader {
copyHeaders(tw.w.Header(), tw.handlerHeaders)
tw.wroteHeader = true
}
return tw.w.Write(p)
}

Expand All @@ -221,10 +227,17 @@ func (tw *baseTimeoutWriter) WriteHeader(code int) {
return
}

copyHeaders(tw.w.Header(), tw.handlerHeaders)
tw.wroteHeader = true
tw.w.WriteHeader(code)
}

func copyHeaders(dst, src http.Header) {
for k, v := range src {
dst[k] = v
}
}

func (tw *baseTimeoutWriter) timeout(err *apierrors.StatusError) {
tw.mu.Lock()
defer tw.mu.Unlock()
Expand Down
46 changes: 46 additions & 0 deletions staging/src/k8s.io/apiserver/pkg/server/filters/timeout_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -203,6 +203,52 @@ func TestTimeout(t *testing.T) {
}
}

func TestTimeoutHeaders(t *testing.T) {
origReallyCrash := runtime.ReallyCrash
runtime.ReallyCrash = false
defer func() {
runtime.ReallyCrash = origReallyCrash
}()

ctx, cancel := context.WithCancel(context.Background())
defer cancel()

withDeadline := func(handler http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
handler.ServeHTTP(w, req.WithContext(ctx))
})
}

ts := httptest.NewServer(
withDeadline(
WithTimeout(
http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
h := w.Header()
// trigger the timeout
cancel()
// mutate response Headers
for j := 0; j < 1000; j++ {
h.Set("Test", "post")
}
}),
func(req *http.Request) (*http.Request, bool, func(), *apierrors.StatusError) {
return req, false, func() {}, apierrors.NewServerTimeout(schema.GroupResource{Group: "foo", Resource: "bar"}, "get", 0)
},
),
),
)
defer ts.Close()

res, err := http.Get(ts.URL)
if err != nil {
t.Fatal(err)
}
if res.StatusCode != http.StatusGatewayTimeout {
t.Errorf("got res.StatusCode %d; expected %d", res.StatusCode, http.StatusServiceUnavailable)
}
res.Body.Close()
}

func captureStdErr() (func() string, func(), error) {
var buf bytes.Buffer
reader, writer, err := os.Pipe()
Expand Down