-
-
Notifications
You must be signed in to change notification settings - Fork 0
/
manifest.go
315 lines (276 loc) · 10.8 KB
/
manifest.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
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
package distribution
import (
"context"
"encoding/json"
"fmt"
"io"
"strings"
"github.com/containerd/containerd/content"
cerrdefs "github.com/containerd/containerd/errdefs"
"github.com/containerd/containerd/remotes"
"github.com/containerd/log"
"github.com/distribution/reference"
"github.com/docker/distribution"
"github.com/docker/distribution/manifest/manifestlist"
"github.com/docker/distribution/manifest/schema1"
"github.com/docker/distribution/manifest/schema2"
"github.com/docker/docker/registry"
"github.com/opencontainers/go-digest"
ocispec "github.com/opencontainers/image-spec/specs-go/v1"
"github.com/pkg/errors"
)
// labelDistributionSource describes the source blob comes from.
const labelDistributionSource = "containerd.io/distribution.source"
// This is used by manifestStore to pare down the requirements to implement a
// full distribution.ManifestService, since `Get` is all we use here.
type manifestGetter interface {
Get(ctx context.Context, dgst digest.Digest, options ...distribution.ManifestServiceOption) (distribution.Manifest, error)
Exists(ctx context.Context, dgst digest.Digest) (bool, error)
}
type manifestStore struct {
local ContentStore
remote manifestGetter
}
// ContentStore is the interface used to persist registry blobs
//
// Currently this is only used to persist manifests and manifest lists.
// It is exported because `distribution.Pull` takes one as an argument.
type ContentStore interface {
content.Ingester
content.Provider
Info(ctx context.Context, dgst digest.Digest) (content.Info, error)
Abort(ctx context.Context, ref string) error
Update(ctx context.Context, info content.Info, fieldpaths ...string) (content.Info, error)
}
func makeDistributionSourceLabel(ref reference.Named) (string, string) {
domain := reference.Domain(ref)
if domain == "" {
domain = registry.DefaultNamespace
}
repo := reference.Path(ref)
return fmt.Sprintf("%s.%s", labelDistributionSource, domain), repo
}
// Taken from https://github.com/containerd/containerd/blob/e079e4a155c86f07bbd602fe6753ecacc78198c2/remotes/docker/handler.go#L84-L108
func appendDistributionSourceLabel(originLabel, repo string) string {
repos := []string{}
if originLabel != "" {
repos = strings.Split(originLabel, ",")
}
repos = append(repos, repo)
// use empty string to present duplicate items
for i := 1; i < len(repos); i++ {
tmp, j := repos[i], i-1
for ; j >= 0 && repos[j] >= tmp; j-- {
if repos[j] == tmp {
tmp = ""
}
repos[j+1] = repos[j]
}
repos[j+1] = tmp
}
i := 0
for ; i < len(repos) && repos[i] == ""; i++ {
}
return strings.Join(repos[i:], ",")
}
func hasDistributionSource(label, repo string) bool {
sources := strings.Split(label, ",")
for _, s := range sources {
if s == repo {
return true
}
}
return false
}
func (m *manifestStore) getLocal(ctx context.Context, desc ocispec.Descriptor, ref reference.Named) (distribution.Manifest, error) {
ra, err := m.local.ReaderAt(ctx, desc)
if err != nil {
return nil, errors.Wrap(err, "error getting content store reader")
}
defer ra.Close()
distKey, distRepo := makeDistributionSourceLabel(ref)
info, err := m.local.Info(ctx, desc.Digest)
if err != nil {
return nil, errors.Wrap(err, "error getting content info")
}
if _, ok := ref.(reference.Canonical); ok {
// Since this is specified by digest...
// We know we have the content locally, we need to check if we've seen this content at the specified repository before.
// If we have, we can just return the manifest from the local content store.
// If we haven't, we need to check the remote repository to see if it has the content, otherwise we can end up returning
// a manifest that has never even existed in the remote before.
if !hasDistributionSource(info.Labels[distKey], distRepo) {
log.G(ctx).WithField("ref", ref).Debug("found manifest but no mataching source repo is listed, checking with remote")
exists, err := m.remote.Exists(ctx, desc.Digest)
if err != nil {
return nil, errors.Wrap(err, "error checking if remote exists")
}
if !exists {
return nil, errors.Wrapf(cerrdefs.ErrNotFound, "manifest %v not found", desc.Digest)
}
}
}
// Update the distribution sources since we now know the content exists in the remote.
if info.Labels == nil {
info.Labels = map[string]string{}
}
info.Labels[distKey] = appendDistributionSourceLabel(info.Labels[distKey], distRepo)
if _, err := m.local.Update(ctx, info, "labels."+distKey); err != nil {
log.G(ctx).WithError(err).WithField("ref", ref).Warn("Could not update content distribution source")
}
r := io.NewSectionReader(ra, 0, ra.Size())
data, err := io.ReadAll(r)
if err != nil {
return nil, errors.Wrap(err, "error reading manifest from content store")
}
manifest, _, err := distribution.UnmarshalManifest(desc.MediaType, data)
if err != nil {
return nil, errors.Wrap(err, "error unmarshaling manifest from content store")
}
return manifest, nil
}
func (m *manifestStore) getMediaType(ctx context.Context, desc ocispec.Descriptor) (string, error) {
ra, err := m.local.ReaderAt(ctx, desc)
if err != nil {
return "", errors.Wrap(err, "error getting reader to detect media type")
}
defer ra.Close()
mt, err := detectManifestMediaType(ra)
if err != nil {
return "", errors.Wrap(err, "error detecting media type")
}
return mt, nil
}
func (m *manifestStore) Get(ctx context.Context, desc ocispec.Descriptor, ref reference.Named) (distribution.Manifest, error) {
l := log.G(ctx)
if desc.MediaType == "" {
// When pulling by digest we will not have the media type on the
// descriptor since we have not made a request to the registry yet
//
// We already have the digest, so we only lookup locally... by digest.
//
// Let's try to detect the media type so we can have a good ref key
// here. We may not even have the content locally, and this is fine, but
// if we do we should determine that.
mt, err := m.getMediaType(ctx, desc)
if err != nil && !cerrdefs.IsNotFound(err) {
l.WithError(err).Warn("Error looking up media type of content")
}
desc.MediaType = mt
}
key := remotes.MakeRefKey(ctx, desc)
// Here we open a writer to the requested content. This both gives us a
// reference to write to if indeed we need to persist it and increments the
// ref count on the content.
w, err := m.local.Writer(ctx, content.WithDescriptor(desc), content.WithRef(key))
if err != nil {
if cerrdefs.IsAlreadyExists(err) {
var manifest distribution.Manifest
if manifest, err = m.getLocal(ctx, desc, ref); err == nil {
return manifest, nil
}
}
// always fallback to the remote if there is an error with the local store
}
if w != nil {
defer w.Close()
}
l.WithError(err).Debug("Fetching manifest from remote")
manifest, err := m.remote.Get(ctx, desc.Digest)
if err != nil {
if err := m.local.Abort(ctx, key); err != nil {
l.WithError(err).Warn("Error while attempting to abort content ingest")
}
return nil, err
}
if w != nil {
// if `w` is nil here, something happened with the content store, so don't bother trying to persist.
if err := m.Put(ctx, manifest, desc, w, ref); err != nil {
if err := m.local.Abort(ctx, key); err != nil {
l.WithError(err).Warn("error aborting content ingest")
}
l.WithError(err).Warn("Error persisting manifest")
}
}
return manifest, nil
}
func (m *manifestStore) Put(ctx context.Context, manifest distribution.Manifest, desc ocispec.Descriptor, w content.Writer, ref reference.Named) error {
mt, payload, err := manifest.Payload()
if err != nil {
return err
}
desc.Size = int64(len(payload))
desc.MediaType = mt
if _, err = w.Write(payload); err != nil {
return errors.Wrap(err, "error writing manifest to content store")
}
distKey, distSource := makeDistributionSourceLabel(ref)
if err := w.Commit(ctx, desc.Size, desc.Digest, content.WithLabels(map[string]string{
distKey: distSource,
})); err != nil {
return errors.Wrap(err, "error committing manifest to content store")
}
return nil
}
func detectManifestMediaType(ra content.ReaderAt) (string, error) {
dt := make([]byte, ra.Size())
if _, err := ra.ReadAt(dt, 0); err != nil {
return "", err
}
return detectManifestBlobMediaType(dt)
}
// This is used when the manifest store does not know the media type of a sha it
// was told to get. This would currently only happen when pulling by digest.
// The media type is needed so the blob can be unmarshalled properly.
func detectManifestBlobMediaType(dt []byte) (string, error) {
var mfst struct {
MediaType string `json:"mediaType"`
Manifests json.RawMessage `json:"manifests"` // oci index, manifest list
Config json.RawMessage `json:"config"` // schema2 Manifest
Layers json.RawMessage `json:"layers"` // schema2 Manifest
FSLayers json.RawMessage `json:"fsLayers"` // schema1 Manifest
}
if err := json.Unmarshal(dt, &mfst); err != nil {
return "", err
}
// We may have a media type specified in the json, in which case that should be used.
// Docker types should generally have a media type set.
// OCI (golang) types do not have a `mediaType` defined, and it is optional in the spec.
//
// `distribution.UnmarshalManifest`, which is used to unmarshal this for real, checks these media type values.
// If the specified media type does not match it will error, and in some cases (docker media types) it is required.
// So pretty much if we don't have a media type we can fall back to OCI.
// This does have a special fallback for schema1 manifests just because it is easy to detect.
switch mfst.MediaType {
case schema2.MediaTypeManifest, ocispec.MediaTypeImageManifest:
if mfst.Manifests != nil || mfst.FSLayers != nil {
return "", fmt.Errorf(`media-type: %q should not have "manifests" or "fsLayers"`, mfst.MediaType)
}
return mfst.MediaType, nil
case manifestlist.MediaTypeManifestList, ocispec.MediaTypeImageIndex:
if mfst.Config != nil || mfst.Layers != nil || mfst.FSLayers != nil {
return "", fmt.Errorf(`media-type: %q should not have "config", "layers", or "fsLayers"`, mfst.MediaType)
}
return mfst.MediaType, nil
case schema1.MediaTypeManifest:
if mfst.Manifests != nil || mfst.Layers != nil {
return "", fmt.Errorf(`media-type: %q should not have "manifests" or "layers"`, mfst.MediaType)
}
return mfst.MediaType, nil
default:
if mfst.MediaType != "" {
return mfst.MediaType, nil
}
}
switch {
case mfst.FSLayers != nil && mfst.Manifests == nil && mfst.Layers == nil && mfst.Config == nil:
return schema1.MediaTypeManifest, nil
case mfst.Config != nil && mfst.Manifests == nil && mfst.FSLayers == nil,
mfst.Layers != nil && mfst.Manifests == nil && mfst.FSLayers == nil:
return ocispec.MediaTypeImageManifest, nil
case mfst.Config == nil && mfst.Layers == nil && mfst.FSLayers == nil:
// fallback to index
return ocispec.MediaTypeImageIndex, nil
}
return "", errors.New("media-type: cannot determine")
}