Use CloseNotifier to supress spurious HTTP 400 errors on early disconnect

When a client disconnects without completing a HTTP request, we were
attempting to process the partial request, which usually leads to a 400
error. These errors can pollute the logs and make it more difficult to
track down real bugs.

This change uses CloseNotifier to detect disconnects. In combination
with checking Content-Length, we can detect a disconnect before sending
the full payload, and avoid logging a 400 error.

This logic is only applied to PUT, POST, and PATCH endpoints, as these
are the places where disconnects during a request are most likely to
happen.

Signed-off-by: Aaron Lehmann <aaron.lehmann@docker.com>

context/http.go

@@ -103,8 +103,13 @@ func GetRequestID(ctx Context) string {
 // WithResponseWriter returns a new context and response writer that makes
 // interesting response statistics available within the context.
 func WithResponseWriter(ctx Context, w http.ResponseWriter) (Context, http.ResponseWriter) {
+	closeNotifier, ok := w.(http.CloseNotifier)
+	if !ok {
+		panic("the ResponseWriter does not implement CloseNotifier")
+	}
 	irw := &instrumentedResponseWriter{
 		ResponseWriter: w,
+		CloseNotifier:  closeNotifier,
 		Context:        ctx,
 	}
 
@@ -262,6 +267,7 @@ func (ctx *muxVarsContext) Value(key interface{}) interface{} {
 // context.
 type instrumentedResponseWriter struct {
 	http.ResponseWriter
+	http.CloseNotifier
 	Context
 
 	mu      sync.Mutex

context/http_test.go

@@ -110,6 +110,13 @@ func (trw *testResponseWriter) Header() http.Header {
 	return trw.header
 }
 
+// CloseNotify is only here to make the testResponseWriter implement the
+// http.CloseNotifier interface, which WithResponseWriter expects to be
+// implemented.
+func (trw *testResponseWriter) CloseNotify() <-chan bool {
+	return nil
+}
+
 func (trw *testResponseWriter) Write(p []byte) (n int, err error) {
 	if trw.status == 0 {
 		trw.status = http.StatusOK

registry/handlers/blobupload.go

@@ -170,8 +170,28 @@ func (buh *blobUploadHandler) PatchBlobData(w http.ResponseWriter, r *http.Reque
 
 	// TODO(dmcgowan): support Content-Range header to seek and write range
 
+	// Get a channel that tells us if the client disconnects
+	var clientClosed <-chan bool
+	if notifier, ok := w.(http.CloseNotifier); ok {
+		clientClosed = notifier.CloseNotify()
+	} else {
+		panic("the ResponseWriter does not implement CloseNotifier")
+	}
+
 	// Copy the data
-	if _, err := io.Copy(buh.Upload, r.Body); err != nil {
+	copied, err := io.Copy(buh.Upload, r.Body)
+	if clientClosed != nil && (err != nil || (r.ContentLength > 0 && copied < r.ContentLength)) {
+		// Didn't recieve as much content as expected. Did the client
+		// disconnect during the request? If so, avoid returning a 400
+		// error to keep the logs cleaner.
+		select {
+		case <-clientClosed:
+			ctxu.GetLogger(buh).Error("client disconnected during blob PATCH")
+			return
+		default:
+		}
+	}
+	if err != nil {
 		ctxu.GetLogger(buh).Errorf("unknown error copying into upload: %v", err)
 		buh.Errors = append(buh.Errors, errcode.ErrorCodeUnknown.WithDetail(err))
 		return
@@ -211,8 +231,28 @@ func (buh *blobUploadHandler) PutBlobUploadComplete(w http.ResponseWriter, r *ht
 		return
 	}
 
+	// Get a channel that tells us if the client disconnects
+	var clientClosed <-chan bool
+	if notifier, ok := w.(http.CloseNotifier); ok {
+		clientClosed = notifier.CloseNotify()
+	} else {
+		panic("the ResponseWriter does not implement CloseNotifier")
+	}
+
 	// Read in the data, if any.
-	if _, err := io.Copy(buh.Upload, r.Body); err != nil {
+	copied, err := io.Copy(buh.Upload, r.Body)
+	if clientClosed != nil && (err != nil || (r.ContentLength > 0 && copied < r.ContentLength)) {
+		// Didn't recieve as much content as expected. Did the client
+		// disconnect during the request? If so, avoid returning a 400
+		// error to keep the logs cleaner.
+		select {
+		case <-clientClosed:
+			ctxu.GetLogger(buh).Error("client disconnected during blob PUT")
+			return
+		default:
+		}
+	}
+	if err != nil {
 		ctxu.GetLogger(buh).Errorf("unknown error copying into upload: %v", err)
 		buh.Errors = append(buh.Errors, errcode.ErrorCodeUnknown.WithDetail(err))
 		return

registry/handlers/images.go

@@ -3,6 +3,7 @@ package handlers
 import (
 	"encoding/json"
 	"fmt"
+	"io/ioutil"
 	"net/http"
 	"strings"
 
@@ -112,10 +113,35 @@ func (imh *imageManifestHandler) PutImageManifest(w http.ResponseWriter, r *http
 		return
 	}
 
-	dec := json.NewDecoder(r.Body)
+	// Get a channel that tells us if the client disconnects
+	var clientClosed <-chan bool
+	if notifier, ok := w.(http.CloseNotifier); ok {
+		clientClosed = notifier.CloseNotify()
+	} else {
+		panic("the ResponseWriter does not implement CloseNotifier")
+	}
+
+	// Copy the data
+	jsonBytes, err := ioutil.ReadAll(r.Body)
+	if clientClosed != nil && (err != nil || (r.ContentLength > 0 && int64(len(jsonBytes)) < r.ContentLength)) {
+		// Didn't recieve as much content as expected. Did the client
+		// disconnect during the request? If so, avoid returning a 400
+		// error to keep the logs cleaner.
+		select {
+		case <-clientClosed:
+			ctxu.GetLogger(imh).Error("client disconnected during image manifest PUT")
+			return
+		default:
+		}
+	}
+	if err != nil {
+		ctxu.GetLogger(imh).Errorf("unknown error reading payload: %v", err)
+		imh.Errors = append(imh.Errors, errcode.ErrorCodeUnknown.WithDetail(err))
+		return
+	}
 
 	var manifest manifest.SignedManifest
-	if err := dec.Decode(&manifest); err != nil {
+	if err := json.Unmarshal(jsonBytes, &manifest); err != nil {
 		imh.Errors = append(imh.Errors, v2.ErrorCodeManifestInvalid.WithDetail(err))
 		return
 	}