This is an automated email from the ASF dual-hosted git repository.
wilfred-s pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/yunikorn-core.git
The following commit(s) were added to refs/heads/master by this push:
new 075eb1dc [YUNIKORN-3341] Allow external web handler gzip encoding
(#1115)
075eb1dc is described below
commit 075eb1dc72a24ea286a0cc4384c7751d7fc3a1c4
Author: Ilia <[email protected]>
AuthorDate: Tue Aug 4 14:04:38 2026 +1000
[YUNIKORN-3341] Allow external web handler gzip encoding (#1115)
The gzip middleware added to the REST router compresses any response whose
buffered body exceeds minCompressionSize, without checking whether the
wrapped handler had already encoded the body itself.
/ws/v1/metrics is served by promhttp.Handler(), which performs its own gzip
content negotiation: when the client sends Accept-Encoding: gzip it
compresses the payload and sets Content-Encoding: gzip. The middleware then
sees the same Accept-Encoding, buffers promhttp's already-gzipped bytes,
finds them over the threshold and compresses a second time. switchToGzip
uses Header().Set, so the response still advertises a single
Content-Encoding: gzip while carrying a doubly encoded body.
Signed-off-by: Ilia <[email protected]>
Closes: #1115
Signed-off-by: Wilfred Spiegelenburg <[email protected]>
---
pkg/webservice/gzip.go | 47 +++++++--
pkg/webservice/gzip_test.go | 241 ++++++++++++++++++++++++++++++++++++++++++++
2 files changed, 279 insertions(+), 9 deletions(-)
diff --git a/pkg/webservice/gzip.go b/pkg/webservice/gzip.go
index 6fc5debd..fe416ba1 100644
--- a/pkg/webservice/gzip.go
+++ b/pkg/webservice/gzip.go
@@ -64,6 +64,14 @@ func (d *deferredGzipResponseWriter) WriteHeader(code int) {
// commits to gzip streaming. After the decision is made, writes go directly
to the
// chosen writer.
func (d *deferredGzipResponseWriter) Write(b []byte) (int, error) {
+ // A handler is allowed to encode the response body itself, in which
case it has
+ // already set Content-Encoding by the time it writes. Compressing
again would
+ // produce a doubly encoded body advertised by a single
Content-Encoding, which no
+ // client can decode. Hand the response over untouched instead.
+ if !d.decided && d.ResponseWriter.Header().Get("Content-Encoding") !=
"" {
+ d.passThrough()
+ }
+
if d.decided {
if d.useGzip {
return d.gz.Write(b)
@@ -78,6 +86,31 @@ func (d *deferredGzipResponseWriter) Write(b []byte) (int,
error) {
return n, err
}
+// passThrough commits to sending the body as the handler produced it: it
flushes any
+// buffered bytes to the underlying writer and marks the decision as final.
+//
+// Vary is added only when the response actually carries an encoding. A
handler that
+// compressed the body itself (promhttp) sets Content-Encoding but never Vary,
and a
+// compressed response without Vary lets a shared cache serve those bytes to a
client
+// that cannot decode them. A body this middleware simply chose not to
compress carries
+// neither header, which is the contract Test_GzipMinCompressionSize asserts.
+func (d *deferredGzipResponseWriter) passThrough() {
+ d.decided = true
+ if d.ResponseWriter.Header().Get("Content-Encoding") != "" {
+ d.ResponseWriter.Header().Add("Vary", "Accept-Encoding")
+ }
+ if d.statusCode != 0 {
+ d.ResponseWriter.WriteHeader(d.statusCode)
+ }
+ if d.buf.Len() > 0 {
+ if _, err := d.ResponseWriter.Write(d.buf.Bytes()); err != nil {
+ log.Log(log.REST).Error("failed to write buffered
response bytes",
+ zap.Error(err))
+ }
+ d.buf.Reset()
+ }
+}
+
// switchToGzip commits to gzip encoding: sets response headers, flushes the
buffered
// bytes through the gzip writer, and marks the decision as final.
func (d *deferredGzipResponseWriter) switchToGzip() {
@@ -100,14 +133,7 @@ func (d *deferredGzipResponseWriter) switchToGzip() {
// The gzip writer is closed if compression was used.
func (d *deferredGzipResponseWriter) finalize() {
if !d.decided {
- d.decided = true
- if d.statusCode != 0 {
- d.ResponseWriter.WriteHeader(d.statusCode)
- }
- if _, err := d.ResponseWriter.Write(d.buf.Bytes()); err != nil {
- log.Log(log.REST).Error("failed to write buffered
response bytes",
- zap.Error(err))
- }
+ d.passThrough()
}
if d.useGzip {
if closeErr := d.gz.Close(); closeErr != nil {
@@ -124,7 +150,10 @@ func (d *deferredGzipResponseWriter) finalize() {
// gzip framing overhead exceeding the size savings.
//
// Responses are always served uncompressed when the client has not requested
gzip,
-// ensuring full backwards compatibility.
+// ensuring full backwards compatibility. Handlers that encode the response
body
+// themselves (for example promhttp, which performs its own gzip content
negotiation)
+// are detected via the Content-Encoding header they set and are passed through
+// untouched, so their bodies are never compressed twice.
//
// The event-stream endpoint (/ws/v1/events/stream) is excluded because it uses
// server-sent events with http.Flusher, which requires writing directly to the
diff --git a/pkg/webservice/gzip_test.go b/pkg/webservice/gzip_test.go
new file mode 100644
index 00000000..46282f97
--- /dev/null
+++ b/pkg/webservice/gzip_test.go
@@ -0,0 +1,241 @@
+/*
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+package webservice
+
+import (
+ "bytes"
+ "compress/gzip"
+ "encoding/binary"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "strings"
+ "testing"
+
+ "gotest.tools/v3/assert"
+)
+
+// largeBody returns a compressible body comfortably above minCompressionSize
so the
+// middleware commits to compression rather than passing the buffered bytes
through.
+func largeBody() []byte {
+ return bytes.Repeat([]byte("yunikorn metrics payload "), 200)
+}
+
+// incompressibleBody returns a deterministic high-entropy body whose
*gzipped* form is
+// still well above minCompressionSize. Self-encoded handler tests need this:
a highly
+// compressible body would gzip down below the threshold, the middleware would
never
+// commit to a second compression pass, and the test would pass even against
the
+// double-encoding bug it is meant to catch.
+func incompressibleBody(t *testing.T) []byte {
+ t.Helper()
+ // Deterministic xorshift keeps the test reproducible without depending
on
+ // math/rand defaults across Go versions. The length is a multiple of 4
so each
+ // round fills a whole word.
+ out := make([]byte, 16*1024)
+ state := uint32(0x9E3779B9)
+ for i := 0; i < len(out); i += 4 {
+ state ^= state << 13
+ state ^= state >> 17
+ state ^= state << 5
+ binary.LittleEndian.PutUint32(out[i:], state)
+ }
+
+ var buf bytes.Buffer
+ zw := gzip.NewWriter(&buf)
+ _, err := zw.Write(out)
+ assert.NilError(t, err)
+ assert.NilError(t, zw.Close())
+ assert.Assert(t, buf.Len() > minCompressionSize,
+ "gzipped fixture must exceed minCompressionSize to exercise the
double-encoding path, got %d", buf.Len())
+
+ return out
+}
+
+func gunzip(t *testing.T, b []byte) []byte {
+ t.Helper()
+ zr, err := gzip.NewReader(bytes.NewReader(b))
+ assert.NilError(t, err, "body is not valid gzip")
+ defer zr.Close()
+ out, err := io.ReadAll(zr)
+ assert.NilError(t, err)
+ return out
+}
+
+func serve(handler http.Handler, acceptEncoding string)
*httptest.ResponseRecorder {
+ req := httptest.NewRequest(http.MethodGet, "/ws/v1/metrics", nil)
+ if acceptEncoding != "" {
+ req.Header.Set("Accept-Encoding", acceptEncoding)
+ }
+ rec := httptest.NewRecorder()
+ compressResponse(handler).ServeHTTP(rec, req)
+ return rec
+}
+
+// TestCompressResponseSelfEncodedHandler covers the case where the wrapped
handler
+// compresses the body itself and sets Content-Encoding, as promhttp does when
the
+// client advertises gzip support. The middleware must not compress a second
time:
+// a single gzip decode has to yield the original payload, otherwise clients
such as
+// the Prometheus scraper fail to parse the response.
+func TestCompressResponseSelfEncodedHandler(t *testing.T) {
+ payload := incompressibleBody(t)
+
+ handler := http.HandlerFunc(func(w http.ResponseWriter, r
*http.Request) {
+ w.Header().Set("Content-Encoding", "gzip")
+ zw := gzip.NewWriter(w)
+ _, err := zw.Write(payload)
+ assert.NilError(t, err)
+ assert.NilError(t, zw.Close())
+ })
+
+ rec := serve(handler, "gzip")
+
+ assert.Equal(t, http.StatusOK, rec.Code)
+ assert.Equal(t, "gzip", rec.Header().Get("Content-Encoding"))
+ // promhttp sets Content-Encoding but never Vary, so the middleware
must add it.
+ assert.Equal(t, "Accept-Encoding", rec.Header().Get("Vary"))
+ // Exactly one layer of gzip: decoding once must produce the original
bytes.
+ assert.DeepEqual(t, payload, gunzip(t, rec.Body.Bytes()))
+}
+
+// TestCompressResponseSelfEncodedSmallBody covers a self-encoded body that
stays below
+// minCompressionSize. The buffered bytes must still be flushed verbatim
rather than
+// being dropped or re-encoded when the handler returns.
+func TestCompressResponseSelfEncodedSmallBody(t *testing.T) {
+ payload := []byte("small self-encoded payload")
+
+ handler := http.HandlerFunc(func(w http.ResponseWriter, r
*http.Request) {
+ w.Header().Set("Content-Encoding", "gzip")
+ zw := gzip.NewWriter(w)
+ _, err := zw.Write(payload)
+ assert.NilError(t, err)
+ assert.NilError(t, zw.Close())
+ })
+
+ rec := serve(handler, "gzip")
+
+ assert.Equal(t, http.StatusOK, rec.Code)
+ assert.Equal(t, "gzip", rec.Header().Get("Content-Encoding"))
+ assert.Equal(t, "Accept-Encoding", rec.Header().Get("Vary"))
+ assert.DeepEqual(t, payload, gunzip(t, rec.Body.Bytes()))
+}
+
+// TestCompressResponseSelfEncodedPreservesStatus verifies that a status code
set
+// before the first write survives the pass-through path.
+func TestCompressResponseSelfEncodedPreservesStatus(t *testing.T) {
+ handler := http.HandlerFunc(func(w http.ResponseWriter, r
*http.Request) {
+ w.Header().Set("Content-Encoding", "gzip")
+ w.WriteHeader(http.StatusAccepted)
+ zw := gzip.NewWriter(w)
+ _, err := zw.Write(incompressibleBody(t))
+ assert.NilError(t, err)
+ assert.NilError(t, zw.Close())
+ })
+
+ rec := serve(handler, "gzip")
+
+ assert.Equal(t, http.StatusAccepted, rec.Code)
+ assert.Equal(t, "gzip", rec.Header().Get("Content-Encoding"))
+ assert.Equal(t, "Accept-Encoding", rec.Header().Get("Vary"))
+}
+
+// TestCompressResponseCompressesLargeBody is the happy path: a plain handler
with a
+// body over the threshold gets compressed by the middleware.
+func TestCompressResponseCompressesLargeBody(t *testing.T) {
+ payload := largeBody()
+
+ handler := http.HandlerFunc(func(w http.ResponseWriter, r
*http.Request) {
+ _, err := w.Write(payload)
+ assert.NilError(t, err)
+ })
+
+ rec := serve(handler, "gzip")
+
+ assert.Equal(t, http.StatusOK, rec.Code)
+ assert.Equal(t, "gzip", rec.Header().Get("Content-Encoding"))
+ assert.Equal(t, "Accept-Encoding", rec.Header().Get("Vary"))
+ assert.Assert(t, rec.Body.Len() < len(payload), "body was not
compressed")
+ assert.DeepEqual(t, payload, gunzip(t, rec.Body.Bytes()))
+}
+
+// TestCompressResponseSmallBodyUncompressed verifies the sub-threshold path
still
+// sends the body raw.
+func TestCompressResponseSmallBodyUncompressed(t *testing.T) {
+ payload := []byte("tiny")
+
+ handler := http.HandlerFunc(func(w http.ResponseWriter, r
*http.Request) {
+ _, err := w.Write(payload)
+ assert.NilError(t, err)
+ })
+
+ rec := serve(handler, "gzip")
+
+ assert.Equal(t, http.StatusOK, rec.Code)
+ assert.Equal(t, "", rec.Header().Get("Content-Encoding"))
+ // Nothing was encoded, so neither header is set - see
Test_GzipMinCompressionSize.
+ assert.Equal(t, "", rec.Header().Get("Vary"))
+ assert.DeepEqual(t, payload, rec.Body.Bytes())
+}
+
+// TestCompressResponseNoGzipRequested verifies a client that does not
advertise gzip
+// always receives an uncompressed body.
+func TestCompressResponseNoGzipRequested(t *testing.T) {
+ payload := largeBody()
+
+ handler := http.HandlerFunc(func(w http.ResponseWriter, r
*http.Request) {
+ _, err := w.Write(payload)
+ assert.NilError(t, err)
+ })
+
+ rec := serve(handler, "")
+
+ assert.Equal(t, http.StatusOK, rec.Code)
+ assert.Equal(t, "", rec.Header().Get("Content-Encoding"))
+ // The middleware hands off before wrapping, so it adds no headers of
its own.
+ assert.Equal(t, "", rec.Header().Get("Vary"))
+ assert.DeepEqual(t, payload, rec.Body.Bytes())
+}
+
+func TestClientAcceptsGzip(t *testing.T) {
+ tests := []struct {
+ acceptEncoding string
+ expected bool
+ }{
+ {"", false},
+ {"gzip", true},
+ {"GZIP", true},
+ {" gzip ", true},
+ {"gzip;q=1.0", true},
+ {"gzip;q=0.5", true},
+ {"gzip;q=0", false},
+ {"deflate, gzip", true},
+ {"deflate", false},
+ {"identity", false},
+ {"deflate, gzip;q=0", false},
+ }
+
+ for _, tt := range tests {
+ t.Run(strings.ReplaceAll(tt.acceptEncoding, " ", "_"), func(t
*testing.T) {
+ req := httptest.NewRequest(http.MethodGet,
"/ws/v1/metrics", nil)
+ if tt.acceptEncoding != "" {
+ req.Header.Set("Accept-Encoding",
tt.acceptEncoding)
+ }
+ assert.Equal(t, tt.expected, clientAcceptsGzip(req))
+ })
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
