Merge pull request #2707 from davidswu/go-1.11

remove dependencies on resumable

.travis.yml

@@ -7,8 +7,7 @@ services:
 language: go
 
 go:
-  - "1.9.x"
-  - "1.10.x"
+  - "1.11.x"
 
 go_import_path: github.com/docker/distribution
 

context/http.go

@@ -104,18 +104,6 @@ func GetRequestID(ctx context.Context) string {
 // WithResponseWriter returns a new context and response writer that makes
 // interesting response statistics available within the context.
 func WithResponseWriter(ctx context.Context, w http.ResponseWriter) (context.Context, http.ResponseWriter) {
-	if closeNotifier, ok := w.(http.CloseNotifier); ok {
-		irwCN := &instrumentedResponseWriterCN{
-			instrumentedResponseWriter: instrumentedResponseWriter{
-				ResponseWriter: w,
-				Context:        ctx,
-			},
-			CloseNotifier: closeNotifier,
-		}
-
-		return irwCN, irwCN
-	}
-
 	irw := instrumentedResponseWriter{
 		ResponseWriter: w,
 		Context:        ctx,
@@ -270,14 +258,6 @@ func (ctx *muxVarsContext) Value(key interface{}) interface{} {
 	return ctx.Context.Value(key)
 }
 
-// instrumentedResponseWriterCN provides response writer information in a
-// context. It implements http.CloseNotifier so that users can detect
-// early disconnects.
-type instrumentedResponseWriterCN struct {
-	instrumentedResponseWriter
-	http.CloseNotifier
-}
-
 // instrumentedResponseWriter provides response writer information in a
 // context. This variant is only used in the case where CloseNotifier is not
 // implemented by the parent ResponseWriter.
@@ -355,13 +335,3 @@ func (irw *instrumentedResponseWriter) Value(key interface{}) interface{} {
 fallback:
 	return irw.Context.Value(key)
 }
-
-func (irw *instrumentedResponseWriterCN) Value(key interface{}) interface{} {
-	if keyStr, ok := key.(string); ok {
-		if keyStr == "http.response" {
-			return irw
-		}
-	}
-
-	return irw.instrumentedResponseWriter.Value(key)
-}

registry/handlers/api_test.go

@@ -2529,35 +2529,6 @@ func TestRegistryAsCacheMutationAPIs(t *testing.T) {
 
 }
 
-// TestCheckContextNotifier makes sure the API endpoints get a ResponseWriter
-// that implements http.ContextNotifier.
-func TestCheckContextNotifier(t *testing.T) {
-	env := newTestEnv(t, false)
-	defer env.Shutdown()
-
-	// Register a new endpoint for testing
-	env.app.router.Handle("/unittest/{name}/", env.app.dispatcher(func(ctx *Context, r *http.Request) http.Handler {
-		return handlers.MethodHandler{
-			"GET": http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-				if _, ok := w.(http.CloseNotifier); !ok {
-					t.Fatal("could not cast ResponseWriter to CloseNotifier")
-				}
-				w.WriteHeader(200)
-			}),
-		}
-	}))
-
-	resp, err := http.Get(env.server.URL + "/unittest/reponame/")
-	if err != nil {
-		t.Fatalf("unexpected error issuing request: %v", err)
-	}
-	defer resp.Body.Close()
-
-	if resp.StatusCode != 200 {
-		t.Fatalf("wrong status code - expected 200, got %d", resp.StatusCode)
-	}
-}
-
 func TestProxyManifestGetByTag(t *testing.T) {
 	truthConfig := configuration.Configuration{
 		Storage: configuration.Storage{

registry/handlers/helpers.go

@@ -27,13 +27,7 @@ func closeResources(handler http.Handler, closers ...io.Closer) http.Handler {
 // The copy will be limited to `limit` bytes, if limit is greater than zero.
 func copyFullPayload(ctx context.Context, responseWriter http.ResponseWriter, r *http.Request, destWriter io.Writer, limit int64, action string) error {
 	// Get a channel that tells us if the client disconnects
-	var clientClosed <-chan bool
-	if notifier, ok := responseWriter.(http.CloseNotifier); ok {
-		clientClosed = notifier.CloseNotify()
-	} else {
-		dcontext.GetLogger(ctx).Warnf("the ResponseWriter does not implement CloseNotifier (type: %T)", responseWriter)
-	}
-
+	clientClosed := r.Context().Done()
 	var body = r.Body
 	if limit > 0 {
 		body = http.MaxBytesReader(responseWriter, body, limit)

registry/storage/blobwriter.go

@@ -33,7 +33,7 @@ type blobWriter struct {
 	id        string
 	startedAt time.Time
 	digester  digest.Digester
-	written   int64 // track the contiguous write
+	written   int64 // track the write to digester
 
 	fileWriter storagedriver.FileWriter
 	driver     storagedriver.StorageDriver
@@ -119,7 +119,12 @@ func (bw *blobWriter) Write(p []byte) (int, error) {
 		return 0, err
 	}
 
-	n, err := io.MultiWriter(bw.fileWriter, bw.digester.Hash()).Write(p)
+	_, err := bw.fileWriter.Write(p)
+	if err != nil {
+		return 0, err
+	}
+
+	n, err := bw.digester.Hash().Write(p)
 	bw.written += int64(n)
 
 	return n, err
@@ -133,7 +138,11 @@ func (bw *blobWriter) ReadFrom(r io.Reader) (n int64, err error) {
 		return 0, err
 	}
 
-	nn, err := io.Copy(io.MultiWriter(bw.fileWriter, bw.digester.Hash()), r)
+	// Using a TeeReader instead of MultiWriter ensures Copy returns
+	// the amount written to the digester as well as ensuring that we
+	// write to the fileWriter first
+	tee := io.TeeReader(r, bw.fileWriter)
+	nn, err := io.Copy(bw.digester.Hash(), tee)
 	bw.written += nn
 
 	return nn, err

registry/storage/blobwriter_resumable.go

@@ -4,17 +4,14 @@ package storage
 
 import (
 	"context"
+	"encoding"
 	"fmt"
+	"hash"
 	"path"
 	"strconv"
 
 	storagedriver "github.com/docker/distribution/registry/storage/driver"
 	"github.com/sirupsen/logrus"
-	"github.com/stevvooe/resumable"
-
-	// register resumable hashes with import
-	_ "github.com/stevvooe/resumable/sha256"
-	_ "github.com/stevvooe/resumable/sha512"
 )
 
 // resumeDigest attempts to restore the state of the internal hash function
@@ -24,12 +21,13 @@ func (bw *blobWriter) resumeDigest(ctx context.Context) error {
 		return errResumableDigestNotAvailable
 	}
 
-	h, ok := bw.digester.Hash().(resumable.Hash)
+	h, ok := bw.digester.Hash().(encoding.BinaryUnmarshaler)
 	if !ok {
 		return errResumableDigestNotAvailable
 	}
+
 	offset := bw.fileWriter.Size()
-	if offset == h.Len() {
+	if offset == bw.written {
 		// State of digester is already at the requested offset.
 		return nil
 	}
@@ -52,20 +50,21 @@ func (bw *blobWriter) resumeDigest(ctx context.Context) error {
 
 	if hashStateMatch.offset == 0 {
 		// No need to load any state, just reset the hasher.
-		h.Reset()
+		h.(hash.Hash).Reset()
 	} else {
 		storedState, err := bw.driver.GetContent(ctx, hashStateMatch.path)
 		if err != nil {
 			return err
 		}
 
-		if err = h.Restore(storedState); err != nil {
+		if err = h.UnmarshalBinary(storedState); err != nil {
 			return err
 		}
+		bw.written = hashStateMatch.offset
 	}
 
 	// Mind the gap.
-	if gapLen := offset - h.Len(); gapLen > 0 {
+	if gapLen := offset - bw.written; gapLen > 0 {
 		return errResumableDigestNotAvailable
 	}
 
@@ -120,26 +119,26 @@ func (bw *blobWriter) storeHashState(ctx context.Context) error {
 		return errResumableDigestNotAvailable
 	}
 
-	h, ok := bw.digester.Hash().(resumable.Hash)
+	h, ok := bw.digester.Hash().(encoding.BinaryMarshaler)
 	if !ok {
 		return errResumableDigestNotAvailable
 	}
 
+	state, err := h.MarshalBinary()
+	if err != nil {
+		return err
+	}
+
 	uploadHashStatePath, err := pathFor(uploadHashStatePathSpec{
 		name:   bw.blobStore.repository.Named().String(),
 		id:     bw.id,
 		alg:    bw.digester.Digest().Algorithm(),
-		offset: h.Len(),
+		offset: bw.written,
 	})
 
 	if err != nil {
 		return err
 	}
 
-	hashState, err := h.State()
-	if err != nil {
-		return err
-	}
-
-	return bw.driver.PutContent(ctx, uploadHashStatePath, hashState)
+	return bw.driver.PutContent(ctx, uploadHashStatePath, state)
 }

registry/storage/driver/s3-aws/s3.go

@@ -476,11 +476,11 @@ func New(params DriverParameters) (*Driver, error) {
 	// }
 
 	d := &driver{
-		S3:        s3obj,
-		Bucket:    params.Bucket,
-		ChunkSize: params.ChunkSize,
-		Encrypt:   params.Encrypt,
-		KeyID:     params.KeyID,
+		S3:                          s3obj,
+		Bucket:                      params.Bucket,
+		ChunkSize:                   params.ChunkSize,
+		Encrypt:                     params.Encrypt,
+		KeyID:                       params.KeyID,
 		MultipartCopyChunkSize:      params.MultipartCopyChunkSize,
 		MultipartCopyMaxConcurrency: params.MultipartCopyMaxConcurrency,
 		MultipartCopyThresholdSize:  params.MultipartCopyThresholdSize,

registry/storage/linkedblobstore.go

@@ -312,14 +312,14 @@ func (lbs *linkedBlobStore) newBlobUpload(ctx context.Context, uuid, path string
 	}
 
 	bw := &blobWriter{
-		ctx:        ctx,
-		blobStore:  lbs,
-		id:         uuid,
-		startedAt:  startedAt,
-		digester:   digest.Canonical.Digester(),
-		fileWriter: fw,
-		driver:     lbs.driver,
-		path:       path,
+		ctx:                    ctx,
+		blobStore:              lbs,
+		id:                     uuid,
+		startedAt:              startedAt,
+		digester:               digest.Canonical.Digester(),
+		fileWriter:             fw,
+		driver:                 lbs.driver,
+		path:                   path,
 		resumableDigestEnabled: lbs.resumableDigestEnabled,
 	}
 

registry/storage/linkedblobstore_test.go

@@ -162,8 +162,8 @@ type mockBlobDescriptorServiceFactory struct {
 func (f *mockBlobDescriptorServiceFactory) BlobAccessController(svc distribution.BlobDescriptorService) distribution.BlobDescriptorService {
 	return &mockBlobDescriptorService{
 		BlobDescriptorService: svc,
-		t:     f.t,
-		stats: f.stats,
+		t:                     f.t,
+		stats:                 f.stats,
 	}
 }
 

vendor.conf

@@ -30,7 +30,6 @@ github.com/prometheus/common 89604d197083d4781071d3c65855d24ecfb0a563
 github.com/prometheus/procfs cb4147076ac75738c9a7d279075a253c0cc5acbd
 github.com/spf13/cobra 312092086bed4968099259622145a0c9ae280064
 github.com/spf13/pflag 5644820622454e71517561946e3d94b9f9db6842
-github.com/stevvooe/resumable 2aaf90b2ceea5072cb503ef2a620b08ff3119870
 github.com/xenolf/lego a9d8cec0e6563575e5868a005359ac97911b5985
 github.com/yvasiyarov/go-metrics 57bccd1ccd43f94bb17fdd8bf3007059b802f85e
 github.com/yvasiyarov/gorelic a9bba5b9ab508a086f9a12b8c51fab68478e2128