forked from openshift/origin
-
Notifications
You must be signed in to change notification settings - Fork 0
/
layerstore.go
190 lines (151 loc) · 4.48 KB
/
layerstore.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
package storage
import (
"strings"
"time"
"code.google.com/p/go-uuid/uuid"
"github.com/docker/distribution"
ctxu "github.com/docker/distribution/context"
"github.com/docker/distribution/digest"
"github.com/docker/distribution/manifest"
storagedriver "github.com/docker/distribution/registry/storage/driver"
)
type layerStore struct {
repository *repository
}
func (ls *layerStore) Exists(digest digest.Digest) (bool, error) {
ctxu.GetLogger(ls.repository.ctx).Debug("(*layerStore).Exists")
// Because this implementation just follows blob links, an existence check
// is pretty cheap by starting and closing a fetch.
_, err := ls.Fetch(digest)
if err != nil {
switch err.(type) {
case distribution.ErrUnknownLayer:
return false, nil
}
return false, err
}
return true, nil
}
func (ls *layerStore) Fetch(dgst digest.Digest) (distribution.Layer, error) {
ctxu.GetLogger(ls.repository.ctx).Debug("(*layerStore).Fetch")
bp, err := ls.path(dgst)
if err != nil {
return nil, err
}
fr, err := newFileReader(ls.repository.driver, bp)
if err != nil {
return nil, err
}
return &layerReader{
fileReader: *fr,
digest: dgst,
}, nil
}
func (ls *layerStore) Delete(dgst digest.Digest) error {
lp, err := ls.linkPath(dgst)
if err != nil {
return err
}
lp = strings.TrimSuffix(lp, "/link")
return ls.repository.driver.Delete(lp)
}
// Upload begins a layer upload, returning a handle. If the layer upload
// is already in progress or the layer has already been uploaded, this
// will return an error.
func (ls *layerStore) Upload() (distribution.LayerUpload, error) {
ctxu.GetLogger(ls.repository.ctx).Debug("(*layerStore).Upload")
// NOTE(stevvooe): Consider the issues with allowing concurrent upload of
// the same two layers. Should it be disallowed? For now, we allow both
// parties to proceed and the the first one uploads the layer.
uuid := uuid.New()
startedAt := time.Now().UTC()
path, err := ls.repository.pm.path(uploadDataPathSpec{
name: ls.repository.Name(),
uuid: uuid,
})
if err != nil {
return nil, err
}
startedAtPath, err := ls.repository.pm.path(uploadStartedAtPathSpec{
name: ls.repository.Name(),
uuid: uuid,
})
if err != nil {
return nil, err
}
// Write a startedat file for this upload
if err := ls.repository.driver.PutContent(startedAtPath, []byte(startedAt.Format(time.RFC3339))); err != nil {
return nil, err
}
return ls.newLayerUpload(uuid, path, startedAt)
}
// Resume continues an in progress layer upload, returning the current
// state of the upload.
func (ls *layerStore) Resume(uuid string) (distribution.LayerUpload, error) {
ctxu.GetLogger(ls.repository.ctx).Debug("(*layerStore).Resume")
startedAtPath, err := ls.repository.pm.path(uploadStartedAtPathSpec{
name: ls.repository.Name(),
uuid: uuid,
})
if err != nil {
return nil, err
}
startedAtBytes, err := ls.repository.driver.GetContent(startedAtPath)
if err != nil {
switch err := err.(type) {
case storagedriver.PathNotFoundError:
return nil, distribution.ErrLayerUploadUnknown
default:
return nil, err
}
}
startedAt, err := time.Parse(time.RFC3339, string(startedAtBytes))
if err != nil {
return nil, err
}
path, err := ls.repository.pm.path(uploadDataPathSpec{
name: ls.repository.Name(),
uuid: uuid,
})
if err != nil {
return nil, err
}
return ls.newLayerUpload(uuid, path, startedAt)
}
// newLayerUpload allocates a new upload controller with the given state.
func (ls *layerStore) newLayerUpload(uuid, path string, startedAt time.Time) (distribution.LayerUpload, error) {
fw, err := newFileWriter(ls.repository.driver, path)
if err != nil {
return nil, err
}
lw := &layerWriter{
layerStore: ls,
uuid: uuid,
startedAt: startedAt,
bufferedFileWriter: *fw,
}
lw.setupResumableDigester()
return lw, nil
}
func (ls *layerStore) linkPath(dgst digest.Digest) (string, error) {
return ls.repository.registry.pm.path(layerLinkPathSpec{name: ls.repository.Name(), digest: dgst})
}
func (ls *layerStore) path(dgst digest.Digest) (string, error) {
// We must traverse this path through the link to enforce ownership.
layerLinkPath, err := ls.linkPath(dgst)
if err != nil {
return "", err
}
blobPath, err := ls.repository.blobStore.resolve(layerLinkPath)
if err != nil {
switch err := err.(type) {
case storagedriver.PathNotFoundError:
return "", distribution.ErrUnknownLayer{
FSLayer: manifest.FSLayer{BlobSum: dgst},
}
default:
return "", err
}
}
return blobPath, nil
}