r.loginov/distribution
1
0
Fork 0
forked from TrueCloudLab/distribution
Code Pull requests Activity

Inline Swift errors handling

Browse source 
Signed-off-by: Sylvain Baubeau <sbaubeau@redhat.com>
This commit is contained in:
Sylvain Baubeau 2015-07-03 12:29:54 +02:00
parent 15d567671b
commit 326c3a9c49
1 changed files with 102 additions and 59 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

157
registry/storage/driver/swift/swift.go
View file

@ -192,17 +192,19 @@ func (d *driver) Name() string {
// GetContent retrieves the content stored at "path" as a []byte.
func (d *driver) GetContent(ctx context.Context, path string) ([]byte, error) {
content, err := d.Conn.ObjectGetBytes(d.Container, d.swiftPath(path))
if err != nil {
return nil, parseError(path, err)
if err == swift.ContainerNotFound || err == swift.ObjectNotFound {
return nil, storagedriver.PathNotFoundError{Path: path}
}
return content, nil
}
// PutContent stores the []byte content at a location designated by "path".
func (d *driver) PutContent(ctx context.Context, path string, contents []byte) error {
err := d.Conn.ObjectPutBytes(d.Container, d.swiftPath(path),
contents, d.getContentType())
return parseError(path, err)
err := d.Conn.ObjectPutBytes(d.Container, d.swiftPath(path), contents, d.getContentType())
if err == swift.ContainerNotFound || err == swift.ObjectNotFound {
return storagedriver.PathNotFoundError{Path: path}
}
return err
}
// ReadStream retrieves an io.ReadCloser for the content stored at "path" with a
@ -212,16 +214,13 @@ func (d *driver) ReadStream(ctx context.Context, path string, offset int64) (io.
headers["Range"] = "bytes=" + strconv.FormatInt(offset, 10) + "-"
file, _, err := d.Conn.ObjectOpen(d.Container, d.swiftPath(path), false, headers)
if err != nil {
if err == swift.ContainerNotFound || err == swift.ObjectNotFound {
return nil, storagedriver.PathNotFoundError{Path: path}
}
if swiftErr, ok := err.(*swift.Error); ok && swiftErr.StatusCode == http.StatusRequestedRangeNotSatisfiable {
return ioutil.NopCloser(bytes.NewReader(nil)), nil
}
return nil, parseError(path, err)
}
return file, nil
return file, err
}
// WriteStream stores the contents of the provided io.Reader at a
@ -257,22 +256,23 @@ func (d *driver) WriteStream(ctx context.Context, path string, offset int64, rea
info, _, err := d.Conn.Object(d.Container, d.swiftPath(path))
if err != nil {
if err == swift.ContainerNotFound || err == swift.ObjectNotFound {
if err == swift.ObjectNotFound {
// Create a object manifest
manifest, err := d.createManifest(path)
if err != nil {
return 0, parseError(path, err)
return 0, err
}
manifest.Close()
} else if err == swift.ContainerNotFound {
return 0, storagedriver.PathNotFoundError{Path: path}
} else {
return 0, parseError(path, err)
return 0, err
}
} else {
// The manifest already exists. Get all the segments
currentLength = info.Bytes
segments, err = d.getAllSegments(path)
if err != nil {
return 0, parseError(path, err)
if segments, err = d.getAllSegments(path); err != nil {
return 0, err
}
}
@ -290,9 +290,13 @@ func (d *driver) WriteStream(ctx context.Context, path string, offset int64, rea
if offset >= currentLength {
for offset-currentLength >= chunkSize {
// Insert a block a zero
d.Conn.ObjectPut(d.Container, getSegment(),
bytes.NewReader(zeroBuf), false, "",
d.getContentType(), nil)
_, err := d.Conn.ObjectPut(d.Container, getSegment(), bytes.NewReader(zeroBuf), false, "", d.getContentType(), nil)
if err != nil {
if err == swift.ContainerNotFound || err == swift.ObjectNotFound {
return 0, storagedriver.PathNotFoundError{Path: getSegment()}
}
return 0, err
}
currentLength += chunkSize
partNumber++
}
@ -304,9 +308,11 @@ func (d *driver) WriteStream(ctx context.Context, path string, offset int64, rea
// data from the beginning of the segment to offset
file, _, err := d.Conn.ObjectOpen(d.Container, getSegment(), false, nil)
if err != nil {
return 0, parseError(getSegment(), err)
if err == swift.ContainerNotFound || err == swift.ObjectNotFound {
return 0, storagedriver.PathNotFoundError{Path: getSegment()}
}
return 0, err
}
defer file.Close()
paddingReader = file
}
@ -321,12 +327,15 @@ func (d *driver) WriteStream(ctx context.Context, path string, offset int64, rea
writeSegment := func(segment string) (finished bool, bytesRead int64, err error) {
currentSegment, err := d.Conn.ObjectCreate(d.Container, segment, false, "", d.getContentType(), nil)
if err != nil {
return false, bytesRead, parseError(path, err)
if err == swift.ContainerNotFound || err == swift.ObjectNotFound {
return false, bytesRead, storagedriver.PathNotFoundError{Path: segment}
}
return false, bytesRead, err
}
n, err := io.Copy(currentSegment, multi)
if err != nil {
return false, bytesRead, parseError(path, err)
return false, bytesRead, err
}
if n > 0 {
@ -342,17 +351,23 @@ func (d *driver) WriteStream(ctx context.Context, path string, offset int64, rea
headers["Range"] = "bytes=" + strconv.FormatInt(cursor+n, 10) + "-" + strconv.FormatInt(cursor+chunkSize, 10)
file, _, err := d.Conn.ObjectOpen(d.Container, d.swiftPath(path), false, headers)
if err != nil {
return false, bytesRead, parseError(path, err)
if err == swift.ContainerNotFound || err == swift.ObjectNotFound {
return false, bytesRead, storagedriver.PathNotFoundError{Path: path}
}
return false, bytesRead, err
}
_, copyErr := io.Copy(currentSegment, file)
if err := file.Close(); err != nil {
return false, bytesRead, parseError(path, err)
if err == swift.ContainerNotFound || err == swift.ObjectNotFound {
return false, bytesRead, storagedriver.PathNotFoundError{Path: path}
}
return false, bytesRead, err
}
if copyErr != nil {
return false, bytesRead, parseError(path, copyErr)
return false, bytesRead, copyErr
}
}
@ -391,6 +406,9 @@ func (d *driver) Stat(ctx context.Context, path string) (storagedriver.FileInfo,
objects, err := d.Conn.ObjectsAll(d.Container, opts)
if err != nil {
if err == swift.ContainerNotFound {
return nil, storagedriver.PathNotFoundError{Path: path}
}
return nil, err
}
@ -407,7 +425,10 @@ func (d *driver) Stat(ctx context.Context, path string) (storagedriver.FileInfo,
// so we need to do a second HEAD request
info, _, err := d.Conn.Object(d.Container, swiftPath)
if err != nil {
return nil, parseError(path, err)
if err == swift.ContainerNotFound || err == swift.ObjectNotFound {
return nil, storagedriver.PathNotFoundError{Path: path}
}
return nil, err
}
fi.IsDir = false
fi.Size = info.Bytes
@ -438,19 +459,20 @@ func (d *driver) List(ctx context.Context, path string) ([]string, error) {
files = append(files, strings.TrimPrefix(strings.TrimSuffix(obj.Name, "/"), d.swiftPath("/")))
}
return files, parseError(path, err)
if err == swift.ContainerNotFound {
return files, storagedriver.PathNotFoundError{Path: path}
}
return files, err
}
// Move moves an object stored at sourcePath to destPath, removing the original
// object.
func (d *driver) Move(ctx context.Context, sourcePath string, destPath string) error {
err := d.Conn.ObjectMove(d.Container, d.swiftPath(sourcePath),
d.Container, d.swiftPath(destPath))
if err != nil {
return parseError(sourcePath, err)
err := d.Conn.ObjectMove(d.Container, d.swiftPath(sourcePath), d.Container, d.swiftPath(destPath))
if err == swift.ContainerNotFound || err == swift.ObjectNotFound {
return storagedriver.PathNotFoundError{Path: sourcePath}
}
return nil
return err
}
// Delete recursively deletes all objects stored at "path" and its subpaths.
@ -461,7 +483,10 @@ func (d *driver) Delete(ctx context.Context, path string) error {
objects, err := d.Conn.ObjectsAll(d.Container, &opts)
if err != nil {
return parseError(path, err)
if err == swift.ContainerNotFound {
return storagedriver.PathNotFoundError{Path: path}
}
return err
}
if d.BulkDeleteSupport {
@ -470,7 +495,10 @@ func (d *driver) Delete(ctx context.Context, path string) error {
filenames[i] = obj.Name
}
if _, err := d.Conn.BulkDelete(d.Container, filenames); err != swift.Forbidden {
return parseError(path, err)
if err == swift.ContainerNotFound || err == swift.ObjectNotFound {
return storagedriver.PathNotFoundError{Path: path}
}
return err
}
}
@ -485,30 +513,46 @@ func (d *driver) Delete(ctx context.Context, path string) error {
segContainer := components[0]
segments, err := d.getAllSegments(components[1])
if err != nil {
return parseError(obj.Name, err)
return err
}
for _, s := range segments {
if err := d.Conn.ObjectDelete(segContainer, s.Name); err != nil {
return parseError(s.Name, err)
if err == swift.ContainerNotFound || err == swift.ObjectNotFound {
return storagedriver.PathNotFoundError{Path: s.Name}
}
return err
}
}
}
} else {
return parseError(obj.Name, err)
if err == swift.ContainerNotFound || err == swift.ObjectNotFound {
return storagedriver.PathNotFoundError{Path: obj.Name}
}
return err
}
if err := d.Conn.ObjectDelete(d.Container, obj.Name); err != nil {
return parseError(obj.Name, err)
if err == swift.ContainerNotFound || err == swift.ObjectNotFound {
return storagedriver.PathNotFoundError{Path: obj.Name}
}
return err
}
}
if _, err := d.Stat(ctx, path); err == nil {
return parseError(path, d.Conn.ObjectDelete(d.Container, d.swiftPath(path)))
} else if len(objects) == 0 {
_, _, err = d.Conn.Object(d.Container, d.swiftPath(path))
if err == nil {
if err := d.Conn.ObjectDelete(d.Container, d.swiftPath(path)); err != nil {
if err == swift.ContainerNotFound || err == swift.ObjectNotFound {
return storagedriver.PathNotFoundError{Path: path}
}
return err
}
} else if err == swift.ObjectNotFound && len(objects) == 0 {
return storagedriver.PathNotFoundError{Path: path}
} else if err == swift.ContainerNotFound {
return storagedriver.PathNotFoundError{Path: path}
}
return nil
}
@ -531,14 +575,21 @@ func (d *driver) getContentType() string {
}
func (d *driver) getAllSegments(path string) ([]swift.Object, error) {
return d.Conn.ObjectsAll(d.Container, &swift.ObjectsOpts{Prefix: d.swiftSegmentPath(path)})
segments, err := d.Conn.ObjectsAll(d.Container, &swift.ObjectsOpts{Prefix: d.swiftSegmentPath(path)})
if err == swift.ContainerNotFound {
return nil, storagedriver.PathNotFoundError{Path: path}
}
return segments, err
}
func (d *driver) createManifest(path string) (*swift.ObjectCreateFile, error) {
headers := make(swift.Headers)
headers["X-Object-Manifest"] = d.Container + "/" + d.swiftSegmentPath(path)
return d.Conn.ObjectCreate(d.Container, d.swiftPath(path), false, "",
d.getContentType(), headers)
file, err := d.Conn.ObjectCreate(d.Container, d.swiftPath(path), false, "", d.getContentType(), headers)
if err == swift.ContainerNotFound {
return nil, storagedriver.PathNotFoundError{Path: path}
}
return file, err
}
func detectBulkDelete(authURL string) (bulkDelete bool) {
@ -553,11 +604,3 @@ func detectBulkDelete(authURL string) (bulkDelete bool) {
}
return
}
func parseError(path string, err error) error {
if err == swift.ContainerNotFound || err == swift.ObjectNotFound {
return storagedriver.PathNotFoundError{Path: path}
}
return err
}