[ARVADOS] created: 171554df860a3f95314f665736ac93a92dd8167b
Git user
git at public.curoverse.com
Wed Sep 28 21:25:52 EDT 2016
at 171554df860a3f95314f665736ac93a92dd8167b (commit)
commit 171554df860a3f95314f665736ac93a92dd8167b
Author: Tom Clegg <tom at curoverse.com>
Date: Wed Sep 28 21:25:46 2016 -0400
9956: Load config from YAML file (WIP)
diff --git a/services/keepstore/azure_blob_volume.go b/services/keepstore/azure_blob_volume.go
index 48cb026..78f9335 100644
--- a/services/keepstore/azure_blob_volume.go
+++ b/services/keepstore/azure_blob_volume.go
@@ -122,6 +122,8 @@ func NewAzureBlobVolume(client storage.Client, containerName string, readonly bo
}
}
+func (v *AzureBlobVolume) Type() string { return "Azure" }
+
// Check returns nil if the volume is usable.
func (v *AzureBlobVolume) Check() error {
ok, err := v.bsClient.ContainerExists(v.containerName)
diff --git a/services/keepstore/keepstore.go b/services/keepstore/keepstore.go
index 48b83de..0846d23 100644
--- a/services/keepstore/keepstore.go
+++ b/services/keepstore/keepstore.go
@@ -230,6 +230,7 @@ func main() {
"trash-check-interval",
24*time.Hour,
"Time duration at which the emptyTrash goroutine will check and delete expired trashed blocks. Default is one day.")
+ flag.Usage = usage
flag.Parse()
diff --git a/services/keepstore/s3_volume.go b/services/keepstore/s3_volume.go
index 1a2a47b..52b6fc7 100644
--- a/services/keepstore/s3_volume.go
+++ b/services/keepstore/s3_volume.go
@@ -11,8 +11,10 @@ import (
"os"
"regexp"
"strings"
+ "sync"
"time"
+ "git.curoverse.com/arvados.git/sdk/go/arvados"
"github.com/AdRoll/goamz/aws"
"github.com/AdRoll/goamz/s3"
)
@@ -49,35 +51,21 @@ func (s *s3VolumeAdder) Set(bucketName string) error {
if s3AccessKeyFile == "" || s3SecretKeyFile == "" {
return fmt.Errorf("-s3-access-key-file and -s3-secret-key-file arguments must given before -s3-bucket-volume")
}
- region, ok := aws.Regions[s3RegionName]
- if s3Endpoint == "" {
- if !ok {
- return fmt.Errorf("unrecognized region %+q; try specifying -s3-endpoint instead", s3RegionName)
- }
- } else {
- if ok {
- return fmt.Errorf("refusing to use AWS region name %+q with endpoint %+q; "+
- "specify empty endpoint (\"-s3-endpoint=\") or use a different region name", s3RegionName, s3Endpoint)
- }
- region = aws.Region{
- Name: s3RegionName,
- S3Endpoint: s3Endpoint,
- }
- }
- var err error
- var auth aws.Auth
- auth.AccessKey, err = readKeyFromFile(s3AccessKeyFile)
- if err != nil {
- return err
- }
- auth.SecretKey, err = readKeyFromFile(s3SecretKeyFile)
- if err != nil {
- return err
- }
if flagSerializeIO {
log.Print("Notice: -serialize is not supported by s3-bucket volumes.")
}
- v := NewS3Volume(auth, region, bucketName, s3RaceWindow, flagReadonly, s3Replication)
+ v := &S3Volume{
+ Bucket: bucketName,
+ AccessKeyFile: s3AccessKeyFile,
+ SecretKeyFile: s3SecretKeyFile,
+ Endpoint: s3Endpoint,
+ Region: s3RegionName,
+ RaceWindow: arvados.Duration(s3RaceWindow),
+ S3Replication: s3Replication,
+ UnsafeDelete: s3UnsafeDelete,
+ ReadOnly: flagReadonly,
+ IndexPageSize: 1000,
+ }
if err := v.Check(); err != nil {
return err
}
@@ -93,6 +81,8 @@ func s3regions() (okList []string) {
}
func init() {
+ VolumeConfigTypes = append(VolumeConfigTypes, func() Volume { return &S3Volume{} })
+
flag.Var(&s3VolumeAdder{&volumes},
"s3-bucket-volume",
"Use the given bucket as a storage volume. Can be given multiple times.")
@@ -135,33 +125,63 @@ func init() {
// S3Volume implements Volume using an S3 bucket.
type S3Volume struct {
- *s3.Bucket
- raceWindow time.Duration
- readonly bool
- replication int
- indexPageSize int
-}
+ AccessKeyFile string
+ SecretKeyFile string
+ Endpoint string
+ Region string
+ Bucket string
+ IndexPageSize int
+ S3Replication int
+ RaceWindow arvados.Duration
+ ReadOnly bool
+ UnsafeDelete bool
-// NewS3Volume returns a new S3Volume using the given auth, region,
-// and bucket name. The replication argument specifies the replication
-// level to report when writing data.
-func NewS3Volume(auth aws.Auth, region aws.Region, bucket string, raceWindow time.Duration, readonly bool, replication int) *S3Volume {
- return &S3Volume{
- Bucket: &s3.Bucket{
- S3: s3.New(auth, region),
- Name: bucket,
- },
- raceWindow: raceWindow,
- readonly: readonly,
- replication: replication,
- indexPageSize: 1000,
- }
+ bucket *s3.Bucket
+
+ startOnce sync.Once
+ err error
}
// Check returns an error if the volume is inaccessible (e.g., config
// error).
func (v *S3Volume) Check() error {
- return nil
+ v.startOnce.Do(v.start)
+ return v.err
+}
+
+func (*S3Volume) Type() string { return "S3" }
+
+func (v *S3Volume) start() {
+ region, ok := aws.Regions[v.Region]
+ if v.Endpoint == "" {
+ if !ok {
+ v.err = fmt.Errorf("unrecognized region %+q; try specifying -s3-endpoint instead", v.Region)
+ return
+ }
+ } else if ok {
+ v.err = fmt.Errorf("refusing to use AWS region name %+q with endpoint %+q; "+
+ "specify empty endpoint (\"-s3-endpoint=\") or use a different region name", v.Region, v.Endpoint)
+ return
+ } else {
+ region = aws.Region{
+ Name: v.Region,
+ S3Endpoint: v.Endpoint,
+ }
+ }
+
+ var auth aws.Auth
+ auth.AccessKey, v.err = readKeyFromFile(v.AccessKeyFile)
+ if v.err != nil {
+ return
+ }
+ auth.SecretKey, v.err = readKeyFromFile(v.SecretKeyFile)
+ if v.err != nil {
+ return
+ }
+ v.bucket = &s3.Bucket{
+ S3: s3.New(auth, region),
+ Name: v.Bucket,
+ }
}
// getReader wraps (Bucket)GetReader.
@@ -170,12 +190,12 @@ func (v *S3Volume) Check() error {
// disappeared in a Trash race, getReader calls fixRace to recover the
// data, and tries again.
func (v *S3Volume) getReader(loc string) (rdr io.ReadCloser, err error) {
- rdr, err = v.Bucket.GetReader(loc)
+ rdr, err = v.bucket.GetReader(loc)
err = v.translateError(err)
if err == nil || !os.IsNotExist(err) {
return
}
- _, err = v.Bucket.Head("recent/"+loc, nil)
+ _, err = v.bucket.Head("recent/"+loc, nil)
err = v.translateError(err)
if err != nil {
// If we can't read recent/X, there's no point in
@@ -186,7 +206,7 @@ func (v *S3Volume) getReader(loc string) (rdr io.ReadCloser, err error) {
err = os.ErrNotExist
return
}
- rdr, err = v.Bucket.GetReader(loc)
+ rdr, err = v.bucket.GetReader(loc)
if err != nil {
log.Printf("warning: reading %s after successful fixRace: %s", loc, err)
err = v.translateError(err)
@@ -223,7 +243,7 @@ func (v *S3Volume) Compare(loc string, expect []byte) error {
// Put writes a block.
func (v *S3Volume) Put(loc string, block []byte) error {
- if v.readonly {
+ if v.ReadOnly {
return MethodDisabledError
}
var opts s3.Options
@@ -234,20 +254,20 @@ func (v *S3Volume) Put(loc string, block []byte) error {
}
opts.ContentMD5 = base64.StdEncoding.EncodeToString(md5)
}
- err := v.Bucket.Put(loc, block, "application/octet-stream", s3ACL, opts)
+ err := v.bucket.Put(loc, block, "application/octet-stream", s3ACL, opts)
if err != nil {
return v.translateError(err)
}
- err = v.Bucket.Put("recent/"+loc, nil, "application/octet-stream", s3ACL, s3.Options{})
+ err = v.bucket.Put("recent/"+loc, nil, "application/octet-stream", s3ACL, s3.Options{})
return v.translateError(err)
}
// Touch sets the timestamp for the given locator to the current time.
func (v *S3Volume) Touch(loc string) error {
- if v.readonly {
+ if v.ReadOnly {
return MethodDisabledError
}
- _, err := v.Bucket.Head(loc, nil)
+ _, err := v.bucket.Head(loc, nil)
err = v.translateError(err)
if os.IsNotExist(err) && v.fixRace(loc) {
// The data object got trashed in a race, but fixRace
@@ -255,27 +275,27 @@ func (v *S3Volume) Touch(loc string) error {
} else if err != nil {
return err
}
- err = v.Bucket.Put("recent/"+loc, nil, "application/octet-stream", s3ACL, s3.Options{})
+ err = v.bucket.Put("recent/"+loc, nil, "application/octet-stream", s3ACL, s3.Options{})
return v.translateError(err)
}
// Mtime returns the stored timestamp for the given locator.
func (v *S3Volume) Mtime(loc string) (time.Time, error) {
- _, err := v.Bucket.Head(loc, nil)
+ _, err := v.bucket.Head(loc, nil)
if err != nil {
return zeroTime, v.translateError(err)
}
- resp, err := v.Bucket.Head("recent/"+loc, nil)
+ resp, err := v.bucket.Head("recent/"+loc, nil)
err = v.translateError(err)
if os.IsNotExist(err) {
// The data object X exists, but recent/X is missing.
- err = v.Bucket.Put("recent/"+loc, nil, "application/octet-stream", s3ACL, s3.Options{})
+ err = v.bucket.Put("recent/"+loc, nil, "application/octet-stream", s3ACL, s3.Options{})
if err != nil {
log.Printf("error: creating %q: %s", "recent/"+loc, err)
return zeroTime, v.translateError(err)
}
log.Printf("info: created %q to migrate existing block to new storage scheme", "recent/"+loc)
- resp, err = v.Bucket.Head("recent/"+loc, nil)
+ resp, err = v.bucket.Head("recent/"+loc, nil)
if err != nil {
log.Printf("error: created %q but HEAD failed: %s", "recent/"+loc, err)
return zeroTime, v.translateError(err)
@@ -292,14 +312,14 @@ func (v *S3Volume) Mtime(loc string) (time.Time, error) {
func (v *S3Volume) IndexTo(prefix string, writer io.Writer) error {
// Use a merge sort to find matching sets of X and recent/X.
dataL := s3Lister{
- Bucket: v.Bucket,
+ Bucket: v.bucket,
Prefix: prefix,
- PageSize: v.indexPageSize,
+ PageSize: v.IndexPageSize,
}
recentL := s3Lister{
- Bucket: v.Bucket,
+ Bucket: v.bucket,
Prefix: "recent/" + prefix,
- PageSize: v.indexPageSize,
+ PageSize: v.IndexPageSize,
}
for data, recent := dataL.First(), recentL.First(); data != nil; data = dataL.Next() {
if data.Key >= "g" {
@@ -346,7 +366,7 @@ func (v *S3Volume) IndexTo(prefix string, writer io.Writer) error {
// Trash a Keep block.
func (v *S3Volume) Trash(loc string) error {
- if v.readonly {
+ if v.ReadOnly {
return MethodDisabledError
}
if t, err := v.Mtime(loc); err != nil {
@@ -358,7 +378,7 @@ func (v *S3Volume) Trash(loc string) error {
if !s3UnsafeDelete {
return ErrS3TrashDisabled
}
- return v.Bucket.Del(loc)
+ return v.bucket.Del(loc)
}
err := v.checkRaceWindow(loc)
if err != nil {
@@ -368,13 +388,13 @@ func (v *S3Volume) Trash(loc string) error {
if err != nil {
return err
}
- return v.translateError(v.Bucket.Del(loc))
+ return v.translateError(v.bucket.Del(loc))
}
// checkRaceWindow returns a non-nil error if trash/loc is, or might
// be, in the race window (i.e., it's not safe to trash loc).
func (v *S3Volume) checkRaceWindow(loc string) error {
- resp, err := v.Bucket.Head("trash/"+loc, nil)
+ resp, err := v.bucket.Head("trash/"+loc, nil)
err = v.translateError(err)
if os.IsNotExist(err) {
// OK, trash/X doesn't exist so we're not in the race
@@ -390,7 +410,7 @@ func (v *S3Volume) checkRaceWindow(loc string) error {
// Can't parse timestamp
return err
}
- safeWindow := t.Add(trashLifetime).Sub(time.Now().Add(v.raceWindow))
+ safeWindow := t.Add(trashLifetime).Sub(time.Now().Add(time.Duration(v.RaceWindow)))
if safeWindow <= 0 {
// We can't count on "touch trash/X" to prolong
// trash/X's lifetime. The new timestamp might not
@@ -408,10 +428,10 @@ func (v *S3Volume) checkRaceWindow(loc string) error {
// (PutCopy returns 200 OK if the request was received, even if the
// copy failed).
func (v *S3Volume) safeCopy(dst, src string) error {
- resp, err := v.Bucket.PutCopy(dst, s3ACL, s3.CopyOptions{
+ resp, err := v.bucket.PutCopy(dst, s3ACL, s3.CopyOptions{
ContentType: "application/octet-stream",
MetadataDirective: "REPLACE",
- }, v.Bucket.Name+"/"+src)
+ }, v.bucket.Name+"/"+src)
err = v.translateError(err)
if err != nil {
return err
@@ -446,7 +466,7 @@ func (v *S3Volume) Untrash(loc string) error {
if err != nil {
return err
}
- err = v.Bucket.Put("recent/"+loc, nil, "application/octet-stream", s3ACL, s3.Options{})
+ err = v.bucket.Put("recent/"+loc, nil, "application/octet-stream", s3ACL, s3.Options{})
return v.translateError(err)
}
@@ -463,19 +483,19 @@ func (v *S3Volume) Status() *VolumeStatus {
// String implements fmt.Stringer.
func (v *S3Volume) String() string {
- return fmt.Sprintf("s3-bucket:%+q", v.Bucket.Name)
+ return fmt.Sprintf("s3-bucket:%+q", v.bucket.Name)
}
// Writable returns false if all future Put, Mtime, and Delete calls
// are expected to fail.
func (v *S3Volume) Writable() bool {
- return !v.readonly
+ return !v.ReadOnly
}
// Replication returns the storage redundancy of the underlying
// device. Configured via command line flag.
func (v *S3Volume) Replication() int {
- return v.replication
+ return v.S3Replication
}
var s3KeepBlockRegexp = regexp.MustCompile(`^[0-9a-f]{32}$`)
@@ -489,7 +509,7 @@ func (v *S3Volume) isKeepBlock(s string) bool {
// there was a race between Put and Trash, fixRace recovers from the
// race by Untrashing the block.
func (v *S3Volume) fixRace(loc string) bool {
- trash, err := v.Bucket.Head("trash/"+loc, nil)
+ trash, err := v.bucket.Head("trash/"+loc, nil)
if err != nil {
if !os.IsNotExist(v.translateError(err)) {
log.Printf("error: fixRace: HEAD %q: %s", "trash/"+loc, err)
@@ -502,7 +522,7 @@ func (v *S3Volume) fixRace(loc string) bool {
return false
}
- recent, err := v.Bucket.Head("recent/"+loc, nil)
+ recent, err := v.bucket.Head("recent/"+loc, nil)
if err != nil {
log.Printf("error: fixRace: HEAD %q: %s", "recent/"+loc, err)
return false
@@ -552,9 +572,9 @@ func (v *S3Volume) EmptyTrash() {
// Use a merge sort to find matching sets of trash/X and recent/X.
trashL := s3Lister{
- Bucket: v.Bucket,
+ Bucket: v.bucket,
Prefix: "trash/",
- PageSize: v.indexPageSize,
+ PageSize: v.IndexPageSize,
}
// Define "ready to delete" as "...when EmptyTrash started".
startT := time.Now()
@@ -571,7 +591,7 @@ func (v *S3Volume) EmptyTrash() {
log.Printf("warning: %s: EmptyTrash: %q: parse %q: %s", v, trash.Key, trash.LastModified, err)
continue
}
- recent, err := v.Bucket.Head("recent/"+loc, nil)
+ recent, err := v.bucket.Head("recent/"+loc, nil)
if err != nil && os.IsNotExist(v.translateError(err)) {
log.Printf("warning: %s: EmptyTrash: found trash marker %q but no %q (%s); calling Untrash", v, trash.Key, "recent/"+loc, err)
err = v.Untrash(loc)
@@ -589,7 +609,7 @@ func (v *S3Volume) EmptyTrash() {
continue
}
if trashT.Sub(recentT) < blobSignatureTTL {
- if age := startT.Sub(recentT); age >= blobSignatureTTL-v.raceWindow {
+ if age := startT.Sub(recentT); age >= blobSignatureTTL-time.Duration(v.RaceWindow) {
// recent/loc is too old to protect
// loc from being Trashed again during
// the raceWindow that starts if we
@@ -602,7 +622,7 @@ func (v *S3Volume) EmptyTrash() {
v.fixRace(loc)
v.Touch(loc)
continue
- } else if _, err := v.Bucket.Head(loc, nil); os.IsNotExist(err) {
+ } else if _, err := v.bucket.Head(loc, nil); os.IsNotExist(err) {
log.Printf("notice: %s: EmptyTrash: detected recent race for %q, calling fixRace", v, loc)
v.fixRace(loc)
continue
@@ -614,7 +634,7 @@ func (v *S3Volume) EmptyTrash() {
if startT.Sub(trashT) < trashLifetime {
continue
}
- err = v.Bucket.Del(trash.Key)
+ err = v.bucket.Del(trash.Key)
if err != nil {
log.Printf("warning: %s: EmptyTrash: deleting %q: %s", v, trash.Key, err)
continue
@@ -622,9 +642,9 @@ func (v *S3Volume) EmptyTrash() {
bytesDeleted += trash.Size
blocksDeleted++
- _, err = v.Bucket.Head(loc, nil)
+ _, err = v.bucket.Head(loc, nil)
if os.IsNotExist(err) {
- err = v.Bucket.Del("recent/" + loc)
+ err = v.bucket.Del("recent/" + loc)
if err != nil {
log.Printf("warning: %s: EmptyTrash: deleting %q: %s", v, "recent/"+loc, err)
}
diff --git a/services/keepstore/s3_volume_test.go b/services/keepstore/s3_volume_test.go
index 6ba3904..b7f0b92 100644
--- a/services/keepstore/s3_volume_test.go
+++ b/services/keepstore/s3_volume_test.go
@@ -127,7 +127,7 @@ func (s *StubbedS3Suite) TestBackendStates(c *check.C) {
return
}
v.serverClock.now = &t
- v.Bucket.Put(key, data, "application/octet-stream", s3ACL, s3.Options{})
+ v.bucket.Put(key, data, "application/octet-stream", s3ACL, s3.Options{})
}
t0 := time.Now()
@@ -286,7 +286,7 @@ func (s *StubbedS3Suite) TestBackendStates(c *check.C) {
// freshAfterEmpty
loc, blk = setupScenario()
v.EmptyTrash()
- _, err = v.Bucket.Head("trash/"+loc, nil)
+ _, err = v.bucket.Head("trash/"+loc, nil)
c.Check(err == nil, check.Equals, scenario.haveTrashAfterEmpty)
if scenario.freshAfterEmpty {
t, err := v.Mtime(loc)
@@ -309,7 +309,7 @@ func (s *StubbedS3Suite) TestBackendStates(c *check.C) {
// PutRaw skips the ContentMD5 test
func (v *TestableS3Volume) PutRaw(loc string, block []byte) {
- err := v.Bucket.Put(loc, block, "application/octet-stream", s3ACL, s3.Options{})
+ err := v.bucket.Put(loc, block, "application/octet-stream", s3ACL, s3.Options{})
if err != nil {
log.Printf("PutRaw: %+v", err)
}
@@ -320,7 +320,7 @@ func (v *TestableS3Volume) PutRaw(loc string, block []byte) {
// while we do this.
func (v *TestableS3Volume) TouchWithDate(locator string, lastPut time.Time) {
v.serverClock.now = &lastPut
- err := v.Bucket.Put("recent/"+locator, nil, "application/octet-stream", s3ACL, s3.Options{})
+ err := v.bucket.Put("recent/"+locator, nil, "application/octet-stream", s3ACL, s3.Options{})
if err != nil {
panic(err)
}
diff --git a/services/keepstore/usage.go b/services/keepstore/usage.go
new file mode 100644
index 0000000..610d6c0
--- /dev/null
+++ b/services/keepstore/usage.go
@@ -0,0 +1,56 @@
+package main
+
+import (
+ "flag"
+ "fmt"
+ "os"
+
+ "github.com/ghodss/yaml"
+)
+
+func usage() {
+ c := DefaultConfig()
+ exampleConfigFile, err := yaml.Marshal(c)
+ if err != nil {
+ panic(err)
+ }
+ fmt.Fprintf(os.Stderr, `
+
+keepstore provides a content-addressed data store backed by a local filesystem or networked storage.
+
+Usage: keepstore -config path/to/keepstore.yml
+
+Options:
+`)
+ flag.PrintDefaults()
+ fmt.Fprintf(os.Stderr, `
+Example config file:
+
+%s
+
+Listen:
+
+ Local port to listen on. Can be "address", "address:port", or
+ ":port", where "address" is a host IP address or name and "port"
+ is a port number or name.
+
+AnonymousTokens:
+
+ Array of tokens to try when a client does not provide a token.
+
+AttachmentOnlyHost:
+
+ Accept credentials, and add "Content-Disposition: attachment"
+ response headers, for requests at this hostname:port.
+
+ This prohibits inline display, which makes it possible to serve
+ untrusted and non-public content from a single origin, i.e.,
+ without wildcard DNS or SSL.
+
+TrustAllContent:
+
+ Serve non-public content from a single origin. Dangerous: read
+ docs before using!
+
+`, exampleConfigFile)
+}
diff --git a/services/keepstore/volume.go b/services/keepstore/volume.go
index 8ae6660..6c52d08 100644
--- a/services/keepstore/volume.go
+++ b/services/keepstore/volume.go
@@ -10,6 +10,10 @@ import (
// for example, a single mounted disk, a RAID array, an Amazon S3 volume,
// etc.
type Volume interface {
+ // Volume type as specified in config file. Examples: "S3",
+ // "Directory".
+ Type() string
+
// Get a block: copy the block data into buf, and return the
// number of bytes copied.
//
diff --git a/services/keepstore/volume_unix.go b/services/keepstore/volume_unix.go
index 5982fb0..0ac6b76 100644
--- a/services/keepstore/volume_unix.go
+++ b/services/keepstore/volume_unix.go
@@ -119,6 +119,8 @@ type UnixVolume struct {
readonly bool
}
+func (v *UnixVolume) Type() string { return "Directory" }
+
// Touch sets the timestamp for the given locator to the current time
func (v *UnixVolume) Touch(loc string) error {
if v.readonly {
-----------------------------------------------------------------------
hooks/post-receive
--
More information about the arvados-commits
mailing list