kraken

1
// Copyright (c) 2016-2019 Uber Technologies, Inc.
2
//
3
// Licensed under the Apache License, Version 2.0 (the "License");
4
// you may not use this file except in compliance with the License.
5
// You may obtain a copy of the License at
6
//
7
//     http://www.apache.org/licenses/LICENSE-2.0
8
//
9
// Unless required by applicable law or agreed to in writing, software
10
// distributed under the License is distributed on an "AS IS" BASIS,
11
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12
// See the License for the specific language governing permissions and
13
// limitations under the License.
14
package blobserver
15

16
import (
17
	"io"
18
	"net/http"
19
	"os"
20

21
	"github.com/docker/distribution/uuid"
22
	"github.com/uber/kraken/core"
23
	"github.com/uber/kraken/lib/store"
24
	"github.com/uber/kraken/utils/handler"
25
)
26

27
// uploader executes a chunked upload.
28
type uploader struct {
29
	cas *store.CAStore
30
}
31

32
func newUploader(cas *store.CAStore) *uploader {
33
	return &uploader{cas}
34
}
35

36
func (u *uploader) start(d core.Digest) (uid string, err error) {
37
	if ok, err := blobExists(u.cas, d); err != nil {
38
		return "", err
39
	} else if ok {
40
		return "", handler.ErrorStatus(http.StatusConflict)
41
	}
42
	uid = uuid.Generate().String()
43
	if err := u.cas.CreateUploadFile(uid, 0); err != nil {
44
		return "", handler.Errorf("create upload file: %s", err)
45
	}
46
	return uid, nil
47
}
48

49
func (u *uploader) patch(
50
	d core.Digest, uid string, chunk io.Reader, start, end int64) error {
51

52
	if ok, err := blobExists(u.cas, d); err != nil {
53
		return err
54
	} else if ok {
55
		return handler.ErrorStatus(http.StatusConflict)
56
	}
57
	f, err := u.cas.GetUploadFileReadWriter(uid)
58
	if err != nil {
59
		if os.IsNotExist(err) {
60
			return handler.ErrorStatus(http.StatusNotFound)
61
		}
62
		return handler.Errorf("get upload file: %s", err)
63
	}
64
	defer f.Close()
65
	if _, err := f.Seek(start, 0); err != nil {
66
		return handler.Errorf("seek offset %d: %s", start, err).Status(http.StatusBadRequest)
67
	}
68
	if _, err := io.CopyN(f, chunk, end-start); err != nil {
69
		return handler.Errorf("copy: %s", err)
70
	}
71
	return nil
72
}
73

74
func (u *uploader) commit(d core.Digest, uid string) error {
75
	if err := u.cas.MoveUploadFileToCache(uid, d.Hex()); err != nil {
76
		if os.IsNotExist(err) {
77
			return handler.ErrorStatus(http.StatusNotFound)
78
		}
79
		if os.IsExist(err) {
80
			return handler.ErrorStatus(http.StatusConflict)
81
		}
82
		return handler.Errorf("move upload file to cache: %s", err)
83
	}
84
	return nil
85
}
86