Merge pull request #15086 from n1hility/increase-buffer

Use 8k buffer to help clients w/ broken parsing

pkg/api/server/handler_api.go

@@ -1,7 +1,10 @@
 package server
 
 import (
+	"bufio"
+	"errors"
 	"fmt"
+	"net"
 	"net/http"
 	"runtime"
 
@@ -9,11 +12,30 @@ import (
 	"github.com/sirupsen/logrus"
 )
 
+type BufferedResponseWriter struct {
+	b *bufio.Writer
+	w http.ResponseWriter
+}
+
 // APIHandler is a wrapper to enhance HandlerFunc's and remove redundant code
 func (s *APIServer) APIHandler(h http.HandlerFunc) http.HandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request) {
 		// Wrapper to hide some boilerplate
-		fn := func(w http.ResponseWriter, r *http.Request) {
+		s.apiWrapper(h, w, r, false)
+	}
+}
+
+// An API Handler to help historical clients with broken parsing that expect
+// streaming JSON payloads to be reliably messaged framed (full JSON record
+// always fits in each read())
+func (s *APIServer) StreamBufferedAPIHandler(h http.HandlerFunc) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		// Wrapper to hide some boilerplate
+		s.apiWrapper(h, w, r, true)
+	}
+}
+
+func (s *APIServer) apiWrapper(h http.HandlerFunc, w http.ResponseWriter, r *http.Request, buffer bool) {
 	if err := r.ParseForm(); err != nil {
 		logrus.WithFields(logrus.Fields{
 			"X-Reference-Id": r.Header.Get("X-Reference-Id"),
@@ -33,10 +55,11 @@ func (s *APIServer) APIHandler(h http.HandlerFunc) http.HandlerFunc {
 		w.Header().Set("Access-Control-Allow-Methods", "HEAD, GET, POST, DELETE, PUT, OPTIONS")
 	}
 
+	if buffer {
+		w = newBufferedResponseWriter(w)
+	}
+
 	h(w, r)
-		}
-		fn(w, r)
-	}
 }
 
 // VersionedPath prepends the version parsing code
@@ -44,3 +67,37 @@ func (s *APIServer) APIHandler(h http.HandlerFunc) http.HandlerFunc {
 func VersionedPath(p string) string {
 	return "/v{version:[0-9][0-9A-Za-z.-]*}" + p
 }
+
+func (w *BufferedResponseWriter) Header() http.Header {
+	return w.w.Header()
+}
+
+func (w *BufferedResponseWriter) Hijack() (net.Conn, *bufio.ReadWriter, error) {
+	_ = w.b.Flush()
+	if wrapped, ok := w.w.(http.Hijacker); ok {
+		return wrapped.Hijack()
+	}
+
+	return nil, nil, errors.New("ResponseWriter does not support hijacking")
+}
+
+func (w *BufferedResponseWriter) Write(b []byte) (int, error) {
+	return w.b.Write(b)
+}
+
+func (w *BufferedResponseWriter) WriteHeader(statusCode int) {
+	w.w.WriteHeader(statusCode)
+}
+
+func (w *BufferedResponseWriter) Flush() {
+	_ = w.b.Flush()
+	if wrapped, ok := w.w.(http.Flusher); ok {
+		wrapped.Flush()
+	}
+}
+func newBufferedResponseWriter(rw http.ResponseWriter) *BufferedResponseWriter {
+	return &BufferedResponseWriter{
+		bufio.NewWriterSize(rw, 8192),
+		rw,
+	}
+}

pkg/api/server/register_containers.go

@@ -397,9 +397,9 @@ func (s *APIServer) registerContainersHandlers(r *mux.Router) error {
 	//     $ref: "#/responses/containerNotFound"
 	//   500:
 	//     $ref: "#/responses/internalError"
-	r.HandleFunc(VersionedPath("/containers/{name}/stats"), s.APIHandler(compat.StatsContainer)).Methods(http.MethodGet)
+	r.HandleFunc(VersionedPath("/containers/{name}/stats"), s.StreamBufferedAPIHandler(compat.StatsContainer)).Methods(http.MethodGet)
 	// Added non version path to URI to support docker non versioned paths
-	r.HandleFunc("/containers/{name}/stats", s.APIHandler(compat.StatsContainer)).Methods(http.MethodGet)
+	r.HandleFunc("/containers/{name}/stats", s.StreamBufferedAPIHandler(compat.StatsContainer)).Methods(http.MethodGet)
 	// swagger:operation POST /containers/{name}/stop compat ContainerStop
 	// ---
 	// tags:
@@ -455,9 +455,9 @@ func (s *APIServer) registerContainersHandlers(r *mux.Router) error {
 	//     $ref: "#/responses/containerNotFound"
 	//   500:
 	//     $ref: "#/responses/internalError"
-	r.HandleFunc(VersionedPath("/containers/{name}/top"), s.APIHandler(compat.TopContainer)).Methods(http.MethodGet)
+	r.HandleFunc(VersionedPath("/containers/{name}/top"), s.StreamBufferedAPIHandler(compat.TopContainer)).Methods(http.MethodGet)
 	// Added non version path to URI to support docker non versioned paths
-	r.HandleFunc("/containers/{name}/top", s.APIHandler(compat.TopContainer)).Methods(http.MethodGet)
+	r.HandleFunc("/containers/{name}/top", s.StreamBufferedAPIHandler(compat.TopContainer)).Methods(http.MethodGet)
 	// swagger:operation POST /containers/{name}/unpause compat ContainerUnpause
 	// ---
 	// tags:

pkg/api/server/register_events.go

@@ -34,9 +34,9 @@ func (s *APIServer) registerEventsHandlers(r *mux.Router) error {
 	//     description: returns a string of json data describing an event
 	//   500:
 	//     "$ref": "#/responses/internalError"
-	r.Handle(VersionedPath("/events"), s.APIHandler(compat.GetEvents)).Methods(http.MethodGet)
+	r.Handle(VersionedPath("/events"), s.StreamBufferedAPIHandler(compat.GetEvents)).Methods(http.MethodGet)
 	// Added non version path to URI to support docker non versioned paths
-	r.Handle("/events", s.APIHandler(compat.GetEvents)).Methods(http.MethodGet)
+	r.Handle("/events", s.StreamBufferedAPIHandler(compat.GetEvents)).Methods(http.MethodGet)
 	// swagger:operation GET /libpod/events system SystemEventsLibpod
 	// ---
 	// tags:

pkg/api/server/register_images.go

@@ -702,9 +702,9 @@ func (s *APIServer) registerImagesHandlers(r *mux.Router) error {
 	//     $ref: "#/responses/badParamError"
 	//   500:
 	//     $ref: "#/responses/internalError"
-	r.Handle(VersionedPath("/build"), s.APIHandler(compat.BuildImage)).Methods(http.MethodPost)
+	r.Handle(VersionedPath("/build"), s.StreamBufferedAPIHandler(compat.BuildImage)).Methods(http.MethodPost)
 	// Added non version path to URI to support docker non versioned paths
-	r.Handle("/build", s.APIHandler(compat.BuildImage)).Methods(http.MethodPost)
+	r.Handle("/build", s.StreamBufferedAPIHandler(compat.BuildImage)).Methods(http.MethodPost)
 	/*
 		libpod endpoints
 	*/