forked from containerd/containerd
-
Notifications
You must be signed in to change notification settings - Fork 0
/
fetcher.go
122 lines (99 loc) · 3.19 KB
/
fetcher.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
package docker
import (
"context"
"fmt"
"io"
"net/http"
"path"
"strings"
"github.com/containerd/containerd/errdefs"
"github.com/containerd/containerd/images"
"github.com/containerd/containerd/log"
ocispec "github.com/opencontainers/image-spec/specs-go/v1"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
)
type dockerFetcher struct {
*dockerBase
}
func (r dockerFetcher) Fetch(ctx context.Context, desc ocispec.Descriptor) (io.ReadCloser, error) {
ctx = log.WithLogger(ctx, log.G(ctx).WithFields(
logrus.Fields{
"base": r.base.String(),
"digest": desc.Digest,
},
))
urls, err := r.getV2URLPaths(ctx, desc)
if err != nil {
return nil, err
}
ctx, err = contextWithRepositoryScope(ctx, r.refspec, false)
if err != nil {
return nil, err
}
return newHTTPReadSeeker(desc.Size, func(offset int64) (io.ReadCloser, error) {
for _, u := range urls {
rc, err := r.open(ctx, u, desc.MediaType, offset)
if err != nil {
if errdefs.IsNotFound(err) {
continue // try one of the other urls.
}
return nil, err
}
return rc, nil
}
return nil, errors.Wrapf(errdefs.ErrNotFound,
"could not fetch content descriptor %v (%v) from remote",
desc.Digest, desc.MediaType)
})
}
func (r dockerFetcher) open(ctx context.Context, u, mediatype string, offset int64) (io.ReadCloser, error) {
req, err := http.NewRequest(http.MethodGet, u, nil)
if err != nil {
return nil, err
}
req.Header.Set("Accept", strings.Join([]string{mediatype, `*`}, ", "))
if offset > 0 {
// TODO(stevvooe): Only set this header in response to the
// "Accept-Ranges: bytes" header.
req.Header.Set("Range", fmt.Sprintf("bytes=%d-", offset))
}
resp, err := r.doRequestWithRetries(ctx, req, nil)
if err != nil {
return nil, err
}
if resp.StatusCode > 299 {
// TODO(stevvooe): When doing a offset specific request, we should
// really distinguish between a 206 and a 200. In the case of 200, we
// can discard the bytes, hiding the seek behavior from the
// implementation.
resp.Body.Close()
if resp.StatusCode == http.StatusNotFound {
return nil, errors.Wrapf(errdefs.ErrNotFound, "content at %v not found", u)
}
return nil, errors.Errorf("unexpected status code %v: %v", u, resp.Status)
}
return resp.Body, nil
}
// getV2URLPaths generates the candidate urls paths for the object based on the
// set of hints and the provided object id. URLs are returned in the order of
// most to least likely succeed.
func (r *dockerFetcher) getV2URLPaths(ctx context.Context, desc ocispec.Descriptor) ([]string, error) {
var urls []string
if len(desc.URLs) > 0 {
// handle fetch via external urls.
for _, u := range desc.URLs {
log.G(ctx).WithField("url", u).Debug("adding alternative url")
urls = append(urls, u)
}
}
switch desc.MediaType {
case images.MediaTypeDockerSchema2Manifest, images.MediaTypeDockerSchema2ManifestList,
images.MediaTypeDockerSchema1Manifest,
ocispec.MediaTypeImageManifest, ocispec.MediaTypeImageIndex:
urls = append(urls, r.url(path.Join("manifests", desc.Digest.String())))
}
// always fallback to attempting to get the object out of the blobs store.
urls = append(urls, r.url(path.Join("blobs", desc.Digest.String())))
return urls, nil
}