-
Notifications
You must be signed in to change notification settings - Fork 0
/
pullthroughblobstore.go
146 lines (121 loc) · 4.86 KB
/
pullthroughblobstore.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
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
package server
import (
"fmt"
"io"
"net/http"
"os"
"time"
"github.com/docker/distribution"
"github.com/docker/distribution/context"
"github.com/docker/distribution/digest"
)
// pullthroughBlobStore wraps a distribution.BlobStore and allows remote repositories to serve blobs from remote
// repositories.
type pullthroughBlobStore struct {
distribution.BlobStore
repo *repository
}
var _ distribution.BlobStore = &pullthroughBlobStore{}
// Stat makes a local check for the blob, then falls through to the other servers referenced by
// the image stream and looks for those that have the layer.
func (r *pullthroughBlobStore) Stat(ctx context.Context, dgst digest.Digest) (distribution.Descriptor, error) {
// check the local store for the blob
desc, err := r.BlobStore.Stat(ctx, dgst)
switch {
case err == distribution.ErrBlobUnknown:
// continue on to the code below and look up the blob in a remote store since it is not in
// the local store
case err != nil:
context.GetLogger(ctx).Errorf("Failed to find blob %q: %#v", dgst.String(), err)
fallthrough
default:
return desc, err
}
remoteGetter, found := RemoteBlobGetterFrom(r.repo.ctx)
if !found {
context.GetLogger(ctx).Errorf("pullthroughBlobStore.Stat: failed to retrieve remote getter from context")
return distribution.Descriptor{}, distribution.ErrBlobUnknown
}
return remoteGetter.Stat(ctx, dgst)
}
// ServeBlob attempts to serve the requested digest onto w, using a remote proxy store if necessary.
func (r *pullthroughBlobStore) ServeBlob(ctx context.Context, w http.ResponseWriter, req *http.Request, dgst digest.Digest) error {
// This call should be done without BlobGetterService in the context.
err := r.BlobStore.ServeBlob(ctx, w, req, dgst)
switch {
case err == distribution.ErrBlobUnknown:
// continue on to the code below and look up the blob in a remote store since it is not in
// the local store
case err != nil:
context.GetLogger(ctx).Errorf("Failed to find blob %q: %#v", dgst.String(), err)
fallthrough
default:
return err
}
remoteGetter, found := RemoteBlobGetterFrom(r.repo.ctx)
if !found {
context.GetLogger(ctx).Errorf("pullthroughBlobStore.ServeBlob: failed to retrieve remote getter from context")
return distribution.ErrBlobUnknown
}
desc, err := remoteGetter.Stat(ctx, dgst)
if err != nil {
context.GetLogger(ctx).Errorf("failed to stat digest %q: %v", dgst.String(), err)
return err
}
remoteReader, err := remoteGetter.Open(ctx, dgst)
if err != nil {
context.GetLogger(ctx).Errorf("failure to open remote store for digest %q: %v", dgst.String(), err)
return err
}
defer remoteReader.Close()
context.GetLogger(ctx).Infof("serving blob %s of type %s %d bytes long", dgst.String(), desc.MediaType, desc.Size)
contentHandled, err := serveRemoteContent(w, req, desc, remoteReader)
if err != nil || contentHandled {
return err
}
w.Header().Set("Content-Length", fmt.Sprintf("%d", desc.Size))
_, err = io.CopyN(w, remoteReader, desc.Size)
return err
}
// Get attempts to fetch the requested blob by digest using a remote proxy store if necessary.
func (r *pullthroughBlobStore) Get(ctx context.Context, dgst digest.Digest) ([]byte, error) {
data, originalErr := r.BlobStore.Get(ctx, dgst)
if originalErr == nil {
return data, nil
}
remoteGetter, found := RemoteBlobGetterFrom(r.repo.ctx)
if !found {
context.GetLogger(ctx).Errorf("pullthroughBlobStore.Get: failed to retrieve remote getter from context")
return nil, originalErr
}
return remoteGetter.Get(ctx, dgst)
}
// setResponseHeaders sets the appropriate content serving headers
func setResponseHeaders(w http.ResponseWriter, length int64, mediaType string, digest digest.Digest) {
w.Header().Set("Content-Type", mediaType)
w.Header().Set("Docker-Content-Digest", digest.String())
w.Header().Set("Etag", digest.String())
}
// serveRemoteContent tries to use http.ServeContent for remote content.
func serveRemoteContent(rw http.ResponseWriter, req *http.Request, desc distribution.Descriptor, remoteReader io.ReadSeeker) (bool, error) {
// Set the appropriate content serving headers.
setResponseHeaders(rw, desc.Size, desc.MediaType, desc.Digest)
// Fallback to Copy if request wasn't given.
if req == nil {
return false, nil
}
// Check whether remoteReader is seekable. The remoteReader' Seek method must work: ServeContent uses
// a seek to the end of the content to determine its size.
if _, err := remoteReader.Seek(0, os.SEEK_END); err != nil {
// The remoteReader isn't seekable. It means that the remote response under the hood of remoteReader
// doesn't contain any Content-Range or Content-Length headers. In this case we need to rollback to
// simple Copy.
return false, nil
}
// Move pointer back to begin.
if _, err := remoteReader.Seek(0, os.SEEK_SET); err != nil {
return false, err
}
http.ServeContent(rw, req, desc.Digest.String(), time.Time{}, remoteReader)
return true, nil
}