-
Notifications
You must be signed in to change notification settings - Fork 2.7k
[WIP] Implement proper payload chunked encoding in HTTP api_v3 #6479
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
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -4,12 +4,14 @@ | |
| package apiv3 | ||
|
|
||
| import ( | ||
| "compress/gzip" | ||
| "encoding/json" | ||
| "errors" | ||
| "fmt" | ||
| "net/http" | ||
| "net/url" | ||
| "strconv" | ||
| "strings" | ||
| "time" | ||
|
|
||
| "github.com/gogo/protobuf/jsonpb" | ||
|
|
@@ -51,6 +53,12 @@ | |
| Tracer trace.TracerProvider | ||
| } | ||
|
|
||
| type chunkedGzipWriter struct { | ||
| http.ResponseWriter | ||
| gzipWriter *gzip.Writer | ||
| flusher http.Flusher | ||
| } | ||
|
|
||
| // RegisterRoutes registers HTTP endpoints for APIv3 into provided mux. | ||
| // The called can create a subrouter if it needs to prepend a base path. | ||
| func (h *HTTPGateway) RegisterRoutes(router *mux.Router) { | ||
|
|
@@ -60,15 +68,58 @@ | |
| h.addRoute(router, h.getOperations, routeGetOperations).Methods(http.MethodGet) | ||
| } | ||
|
|
||
| // addRoute adds a new endpoint to the router with given path and handler function. | ||
| // This code is mostly copied from ../http_handler. | ||
| func (w *chunkedGzipWriter) Write(b []byte) (int, error) { | ||
| n, err := w.gzipWriter.Write(b) | ||
| if err != nil { | ||
| return n, err | ||
| } | ||
|
|
||
| if err := w.gzipWriter.Flush(); err != nil { | ||
| return n, err | ||
| } | ||
|
|
||
| w.flusher.Flush() | ||
| return n, nil | ||
| } | ||
|
|
||
| func gzipMiddleware(next http.Handler) http.Handler { | ||
| return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { | ||
| if !strings.Contains(r.Header.Get("Accept-Encoding"), "gzip") { | ||
| next.ServeHTTP(w, r) | ||
| return | ||
| } | ||
|
|
||
| flusher, ok := w.(http.Flusher) | ||
| if !ok { | ||
| http.Error(w, "Streaming unsupported!", http.StatusInternalServerError) | ||
| return | ||
| } | ||
|
|
||
| // Set chunked transfer encoding before any writes | ||
| w.Header().Set("Transfer-Encoding", "chunked") | ||
| w.Header().Set("Content-Encoding", "gzip") | ||
| w.Header().Set("Vary", "Accept-Encoding") | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. what is this? |
||
|
|
||
| gz := gzip.NewWriter(w) | ||
| defer gz.Close() | ||
|
Comment on lines
+103
to
+104
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. The Consider either:
This would ensure the gzip writer remains open until all data has been properly flushed and written to the response. Spotted by Diamond |
||
|
|
||
| writer := &chunkedGzipWriter{ | ||
| ResponseWriter: w, | ||
| gzipWriter: gz, | ||
| flusher: flusher, | ||
| } | ||
|
|
||
| next.ServeHTTP(writer, r) | ||
| }) | ||
| } | ||
|
|
||
| func (*HTTPGateway) addRoute( | ||
| router *mux.Router, | ||
| f func(http.ResponseWriter, *http.Request), | ||
| route string, | ||
| _ ...any, /* args */ | ||
| ) *mux.Route { | ||
| var handler http.Handler = http.HandlerFunc(f) | ||
| handler = gzipMiddleware(handler) | ||
| handler = otelhttp.WithRouteTag(route, handler) | ||
| handler = spanNameHandler(route, handler) | ||
| return router.HandleFunc(route, handler.ServeHTTP) | ||
|
|
||
| Original file line number | Diff line number | Diff line change | ||||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
|
|
@@ -7,6 +7,7 @@ import ( | |||||||||||||
| "bytes" | ||||||||||||||
| "context" | ||||||||||||||
| "fmt" | ||||||||||||||
| "io" | ||||||||||||||
| "net" | ||||||||||||||
| "net/http" | ||||||||||||||
| "testing" | ||||||||||||||
|
|
@@ -915,6 +916,21 @@ func TestServerHTTP_TracesRequest(t *testing.T) { | |||||||||||||
| resp, err := client.Do(req) | ||||||||||||||
| require.NoError(t, err) | ||||||||||||||
|
|
||||||||||||||
| var fullResponse bytes.Buffer | ||||||||||||||
|
|
||||||||||||||
| buf := make([]byte, 1024) | ||||||||||||||
| for { | ||||||||||||||
| n, err := resp.Body.Read(buf) | ||||||||||||||
|
|
||||||||||||||
|
Comment on lines
+923
to
+924
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. The n, err := resp.Body.Read(buf)
if err != nil && err != io.EOF {
require.NoError(t, err)
break
}This ensures the test fails appropriately if there's a network issue or other error during response reading, rather than silently continuing.
Suggested change
Spotted by Diamond |
||||||||||||||
| if n > 0 { | ||||||||||||||
| fullResponse.Write(buf[:n]) | ||||||||||||||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. if you concatenate the chunks you will end up with invalid JSON |
||||||||||||||
| } | ||||||||||||||
|
|
||||||||||||||
| if err == io.EOF { | ||||||||||||||
| break | ||||||||||||||
| } | ||||||||||||||
| } | ||||||||||||||
|
|
||||||||||||||
| if test.expectedTrace != "" { | ||||||||||||||
| assert.Len(t, exporter.GetSpans(), 1, "HTTP request was traced and span reported") | ||||||||||||||
| assert.Equal(t, test.expectedTrace, exporter.GetSpans()[0].Name) | ||||||||||||||
|
|
||||||||||||||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
chunked encoding requires each chunk to be prefixed with its size