github/containerd
4
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 2 Wiki Activity

retry request on writer reset

Browse Source 
when a put request is retried due to the response from registry,
the body of the request should be seekable. A dynamic pipe is added
to the body so that the content of the body can be read again.
Currently a maximum of 5 resets are allowed, above which will fail the
request. A new error ErrReset is introduced which informs that a
reset has occured and request needs to be retried.

also added tests for Copy() and push() to test the new functionality

Signed-off-by: Akhil Mohan <makhil@vmware.com>
This commit is contained in:
Akhil Mohan
2022-05-13 12:04:24 +05:30
parent de509c0682
commit 8f4c23b69f
4 changed files with 499 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files

160
remotes/docker/pusher.go
View File

@@ -264,27 +264,20 @@ func (p dockerPusher) push(ctx context.Context, desc ocispec.Descriptor, ref str
// TODO: Support chunked upload
pr, pw := io.Pipe()
respC := make(chan response, 1)
body := io.NopCloser(pr)
pushw := newPushWriter(p.dockerBase, ref, desc.Digest, p.tracker, isManifest)
req.body = func() (io.ReadCloser, error) {
if body == nil {
return nil, errors.New("cannot reuse body, request must be retried")
}
// Only use the body once since pipe cannot be seeked
ob := body
body = nil
return ob, nil
pr, pw := io.Pipe()
pushw.setPipe(pw)
return io.NopCloser(pr), nil
}
req.size = desc.Size
go func() {
defer close(respC)
resp, err := req.doWithRetries(ctx, nil)
if err != nil {
respC <- response{err: err}
pr.CloseWithError(err)
pushw.setError(err)
pushw.Close()
return
}
@@ -293,20 +286,13 @@ func (p dockerPusher) push(ctx context.Context, desc ocispec.Descriptor, ref str
default:
err := remoteserrors.NewUnexpectedStatusErr(resp)
log.G(ctx).WithField("resp", resp).WithField("body", string(err.(remoteserrors.ErrUnexpectedStatus).Body)).Debug("unexpected response")
pr.CloseWithError(err)
pushw.setError(err)
pushw.Close()
}
respC <- response{Response: resp}
pushw.setResponse(resp)
}()
return &pushWriter{
base: p.dockerBase,
ref: ref,
pipe: pw,
responseC: respC,
isManifest: isManifest,
expected: desc.Digest,
tracker: p.tracker,
}, nil
return pushw, nil
}
func getManifestPath(object string, dgst digest.Digest) []string {
@@ -328,28 +314,80 @@ func getManifestPath(object string, dgst digest.Digest) []string {
return []string{"manifests", object}
}
type response struct {
*http.Response
err error
}
type pushWriter struct {
base *dockerBase
ref string
pipe *io.PipeWriter
responseC <-chan response
pipe *io.PipeWriter
pipeC chan *io.PipeWriter
respC chan *http.Response
errC chan error
isManifest bool
expected digest.Digest
tracker StatusTracker
}
func newPushWriter(db *dockerBase, ref string, expected digest.Digest, tracker StatusTracker, isManifest bool) *pushWriter {
// Initialize and create response
return &pushWriter{
base: db,
ref: ref,
expected: expected,
tracker: tracker,
pipeC: make(chan *io.PipeWriter, 1),
respC: make(chan *http.Response, 1),
errC: make(chan error, 1),
isManifest: isManifest,
}
}
func (pw *pushWriter) setPipe(p *io.PipeWriter) {
pw.pipeC <- p
}
func (pw *pushWriter) setError(err error) {
pw.errC <- err
}
func (pw *pushWriter) setResponse(resp *http.Response) {
pw.respC <- resp
}
func (pw *pushWriter) Write(p []byte) (n int, err error) {
status, err := pw.tracker.GetStatus(pw.ref)
if err != nil {
return n, err
}
if pw.pipe == nil {
p, ok := <-pw.pipeC
if !ok {
return 0, io.ErrClosedPipe
}
pw.pipe = p
} else {
select {
case p, ok := <-pw.pipeC:
if !ok {
return 0, io.ErrClosedPipe
}
pw.pipe.CloseWithError(content.ErrReset)
pw.pipe = p
// If content has already been written, the bytes
// cannot be written and the caller must reset
if status.Offset > 0 {
status.Offset = 0
status.UpdatedAt = time.Now()
pw.tracker.SetStatus(pw.ref, status)
return 0, content.ErrReset
}
default:
}
}
n, err = pw.pipe.Write(p)
status.Offset += int64(n)
status.UpdatedAt = time.Now()
@@ -358,13 +396,26 @@ func (pw *pushWriter) Write(p []byte) (n int, err error) {
}
func (pw *pushWriter) Close() error {
status, err := pw.tracker.GetStatus(pw.ref)
if err == nil && !status.Committed {
// Closing an incomplete writer. Record this as an error so that following write can retry it.
status.ErrClosed = errors.New("closed incomplete writer")
pw.tracker.SetStatus(pw.ref, status)
// Ensure pipeC is closed but handle `Close()` being
// called multiple times without panicking
select {
case _, ok := <-pw.pipeC:
if ok {
close(pw.pipeC)
}
default:
close(pw.pipeC)
}
return pw.pipe.Close()
if pw.pipe != nil {
status, err := pw.tracker.GetStatus(pw.ref)
if err == nil && !status.Committed {
// Closing an incomplete writer. Record this as an error so that following write can retry it.
status.ErrClosed = errors.New("closed incomplete writer")
pw.tracker.SetStatus(pw.ref, status)
}
return pw.pipe.Close()
}
return nil
}
func (pw *pushWriter) Status() (content.Status, error) {
@@ -391,18 +442,43 @@ func (pw *pushWriter) Commit(ctx context.Context, size int64, expected digest.Di
return err
}
// TODO: timeout waiting for response
resp := <-pw.responseC
if resp.err != nil {
return resp.err
var resp *http.Response
select {
case err := <-pw.errC:
if err != nil {
return err
}
case resp = <-pw.respC:
defer resp.Body.Close()
case p, ok := <-pw.pipeC:
// check whether the pipe has changed in the commit, because sometimes Write
// can complete successfully, but the pipe may have changed. In that case, the
// content needs to be reset.
if !ok {
return io.ErrClosedPipe
}
pw.pipe.CloseWithError(content.ErrReset)
pw.pipe = p
status, err := pw.tracker.GetStatus(pw.ref)
if err != nil {
return err
}
// If content has already been written, the bytes
// cannot be written again and the caller must reset
if status.Offset > 0 {
status.Offset = 0
status.UpdatedAt = time.Now()
pw.tracker.SetStatus(pw.ref, status)
return content.ErrReset
}
}
defer resp.Response.Body.Close()
// 201 is specified return status, some registries return
// 200, 202 or 204.
switch resp.StatusCode {
case http.StatusOK, http.StatusCreated, http.StatusNoContent, http.StatusAccepted:
default:
return remoteserrors.NewUnexpectedStatusErr(resp.Response)
return remoteserrors.NewUnexpectedStatusErr(resp)
}
status, err := pw.tracker.GetStatus(pw.ref)