S3 Backend MVP

.travis.yml

@@ -5,7 +5,7 @@ sudo: required
 services:
   - docker
 
-go_import_path: dinowernli.me/almanac
+go_import_path: github.com/dinowernli/almanac
 
 env:
   global:

Dockerfile

@@ -1,9 +1,9 @@
 FROM golang:1.9 as build
-WORKDIR /go/src/dinowernli.me/almanac
+WORKDIR /go/src/github.com/dinowernli/almanac
 COPY . .
 RUN CGO_ENABLED=0 GOOS=linux go build -o almanac-linux-static -a -ldflags '-extldflags "-static"' cmd/almanac/almanac.go
 
 FROM scratch
-COPY --from=build /go/src/dinowernli.me/almanac/almanac-linux-static .
+COPY --from=build /go/src/github.com/dinowernli/almanac/almanac-linux-static .
 ENTRYPOINT ["./almanac-linux-static"]
 

cmd/almanac/almanac.go

@@ -5,9 +5,9 @@ import (
 	"net/http"
 	"os"
 
-	"dinowernli.me/almanac/pkg/cluster"
-	"dinowernli.me/almanac/pkg/storage"
-	pb_almanac "dinowernli.me/almanac/proto"
+	"github.com/dinowernli/almanac/pkg/cluster"
+	"github.com/dinowernli/almanac/pkg/storage"
+	pb_almanac "github.com/dinowernli/almanac/proto"
 
 	"github.com/alecthomas/kingpin"
 	"github.com/prometheus/client_golang/prometheus"
@@ -21,8 +21,9 @@ const (
 )
 
 var (
-	flagStorageType = kingpin.Flag("storage", "Which kind of storage to use").Default(storage.StorageTypeMemory).Enum(storage.StorageTypeMemory, storage.StorageTypeDisk, storage.StorageTypeGcs)
+	flagStorageType = kingpin.Flag("storage", "Which kind of storage to use").Default(storage.StorageTypeMemory).Enum(storage.StorageTypeMemory, storage.StorageTypeDisk, storage.StorageTypeGcs, storage.StorageTypeS3)
 	flagGcsBucket   = kingpin.Flag("storage.gcs.bucket", "Which gcs bucket to use for storage").Default("almanac-dev").String()
+	flagS3Bucket    = kingpin.Flag("storage.s3.bucket", "Which s3 bucket to use for storage").Default("almanac-dev").String()
 	flagDiskPath    = kingpin.Flag("storage.disk.path", "An existing empty directory to use as root for storage").Default("/tmp/almanac-dev").String()
 
 	flagAppenderPorts = kingpin.Flag("appender_ports", "Which ports to run appenders on").Default("5001", "5002", "5003", "5004", "5005").Ints()
@@ -55,6 +56,7 @@ func main() {
 
 		StorageType: *flagStorageType,
 		GcsBucket:   *flagGcsBucket,
+		S3Bucket:    *flagS3Bucket,
 		DiskPath:    *flagDiskPath,
 	}
 

pkg/cluster/cluster.go

@@ -5,13 +5,13 @@ import (
 	"net"
 	"time"
 
-	"dinowernli.me/almanac/pkg/service/appender"
-	dc "dinowernli.me/almanac/pkg/service/discovery"
-	in "dinowernli.me/almanac/pkg/service/ingester"
-	"dinowernli.me/almanac/pkg/service/janitor"
-	mx "dinowernli.me/almanac/pkg/service/mixer"
-	st "dinowernli.me/almanac/pkg/storage"
-	pb_almanac "dinowernli.me/almanac/proto"
+	"github.com/dinowernli/almanac/pkg/service/appender"
+	dc "github.com/dinowernli/almanac/pkg/service/discovery"
+	in "github.com/dinowernli/almanac/pkg/service/ingester"
+	"github.com/dinowernli/almanac/pkg/service/janitor"
+	mx "github.com/dinowernli/almanac/pkg/service/mixer"
+	st "github.com/dinowernli/almanac/pkg/storage"
+	pb_almanac "github.com/dinowernli/almanac/proto"
 
 	"github.com/sirupsen/logrus"
 	"golang.org/x/net/context"
@@ -30,6 +30,7 @@ type Config struct {
 
 	StorageType string
 	GcsBucket   string
+	S3Bucket    string
 	DiskPath    string
 }
 
@@ -60,6 +61,11 @@ func CreateCluster(ctx context.Context, logger *logrus.Logger, config *Config, a
 		if err != nil {
 			return nil, fmt.Errorf("unable to create gcs storage: %v", err)
 		}
+	} else if config.StorageType == st.StorageTypeS3 {
+		storage, err = st.NewS3Storage(config.S3Bucket)
+		if err != nil {
+			return nil, fmt.Errorf("unable to create s3 storage: %v", err)
+		}
 	} else if config.StorageType == st.StorageTypeDisk {
 		storage, err = st.NewDiskStorage(config.DiskPath)
 		if err != nil {

pkg/http/http.go

@@ -6,8 +6,8 @@ import (
 	"io"
 	"strconv"
 
-	"dinowernli.me/almanac/pkg/http/templates"
-	pb_almanac "dinowernli.me/almanac/proto"
+	"github.com/dinowernli/almanac/pkg/http/templates"
+	pb_almanac "github.com/dinowernli/almanac/proto"
 )
 
 var (

pkg/http/http_test.go

@@ -3,7 +3,7 @@ package http
 import (
 	"testing"
 
-	pb_almanac "dinowernli.me/almanac/proto"
+	pb_almanac "github.com/dinowernli/almanac/proto"
 
 	"github.com/stretchr/testify/assert"
 )

pkg/index/serialization.go

@@ -10,7 +10,7 @@ import (
 	"path/filepath"
 	"strings"
 
-	pb_almanac "dinowernli.me/almanac/proto"
+	pb_almanac "github.com/dinowernli/almanac/proto"
 )
 
 // Serialize returns a proto with the contents of the supplied index.

pkg/service/appender/appender.go

@@ -5,8 +5,8 @@ import (
 	"sync"
 	"time"
 
-	"dinowernli.me/almanac/pkg/storage"
-	pb_almanac "dinowernli.me/almanac/proto"
+	"github.com/dinowernli/almanac/pkg/storage"
+	pb_almanac "github.com/dinowernli/almanac/proto"
 
 	"github.com/sirupsen/logrus"
 	"golang.org/x/net/context"

pkg/service/appender/open_chunk.go

@@ -7,9 +7,9 @@ import (
 	"sync"
 	"time"
 
-	"dinowernli.me/almanac/pkg/index"
-	"dinowernli.me/almanac/pkg/storage"
-	pb_almanac "dinowernli.me/almanac/proto"
+	"github.com/dinowernli/almanac/pkg/index"
+	"github.com/dinowernli/almanac/pkg/storage"
+	pb_almanac "github.com/dinowernli/almanac/proto"
 
 	"golang.org/x/net/context"
 )

pkg/service/appender/open_chunk_test.go

@@ -4,8 +4,8 @@ import (
 	"testing"
 	"time"
 
-	"dinowernli.me/almanac/pkg/storage"
-	pb_almanac "dinowernli.me/almanac/proto"
+	"github.com/dinowernli/almanac/pkg/storage"
+	pb_almanac "github.com/dinowernli/almanac/proto"
 
 	"github.com/stretchr/testify/assert"
 	"golang.org/x/net/context"

pkg/service/discovery/discovery.go

@@ -3,7 +3,7 @@ package discovery
 import (
 	"fmt"
 
-	pb_almanac "dinowernli.me/almanac/proto"
+	pb_almanac "github.com/dinowernli/almanac/proto"
 
 	"google.golang.org/grpc"
 )

pkg/service/ingester/ingester.go

@@ -6,10 +6,10 @@ import (
 	"net/http"
 	"time"
 
-	almHttp "dinowernli.me/almanac/pkg/http"
-	dc "dinowernli.me/almanac/pkg/service/discovery"
-	"dinowernli.me/almanac/pkg/util"
-	pb_almanac "dinowernli.me/almanac/proto"
+	almHttp "github.com/dinowernli/almanac/pkg/http"
+	dc "github.com/dinowernli/almanac/pkg/service/discovery"
+	"github.com/dinowernli/almanac/pkg/util"
+	pb_almanac "github.com/dinowernli/almanac/proto"
 
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/sirupsen/logrus"

pkg/service/janitor/janitor.go

@@ -4,9 +4,9 @@ import (
 	"fmt"
 	"time"
 
-	st "dinowernli.me/almanac/pkg/storage"
-	"dinowernli.me/almanac/pkg/util"
-	pb_almanac "dinowernli.me/almanac/proto"
+	st "github.com/dinowernli/almanac/pkg/storage"
+	"github.com/dinowernli/almanac/pkg/util"
+	pb_almanac "github.com/dinowernli/almanac/proto"
 
 	"github.com/sirupsen/logrus"
 	"golang.org/x/net/context"

pkg/service/janitor/janitor_test.go

@@ -4,8 +4,8 @@ import (
 	"testing"
 	"time"
 
-	st "dinowernli.me/almanac/pkg/storage"
-	pb_almanac "dinowernli.me/almanac/proto"
+	st "github.com/dinowernli/almanac/pkg/storage"
+	pb_almanac "github.com/dinowernli/almanac/proto"
 
 	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"

pkg/service/mixer/heap.go

@@ -3,8 +3,8 @@ package mixer
 import (
 	"fmt"
 
-	st "dinowernli.me/almanac/pkg/storage"
-	pb_almanac "dinowernli.me/almanac/proto"
+	st "github.com/dinowernli/almanac/pkg/storage"
+	pb_almanac "github.com/dinowernli/almanac/proto"
 
 	"golang.org/x/net/context"
 )

pkg/service/mixer/mixer.go

@@ -6,10 +6,10 @@ import (
 	"net/http"
 	"time"
 
-	almHttp "dinowernli.me/almanac/pkg/http"
-	"dinowernli.me/almanac/pkg/service/discovery"
-	"dinowernli.me/almanac/pkg/storage"
-	pb_almanac "dinowernli.me/almanac/proto"
+	almHttp "github.com/dinowernli/almanac/pkg/http"
+	"github.com/dinowernli/almanac/pkg/service/discovery"
+	"github.com/dinowernli/almanac/pkg/storage"
+	pb_almanac "github.com/dinowernli/almanac/proto"
 
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/sirupsen/logrus"

pkg/service/mixer/mixer_test.go

@@ -6,9 +6,9 @@ import (
 	"strings"
 	"testing"
 
-	"dinowernli.me/almanac/pkg/service/discovery"
-	st "dinowernli.me/almanac/pkg/storage"
-	pb_almanac "dinowernli.me/almanac/proto"
+	"github.com/dinowernli/almanac/pkg/service/discovery"
+	st "github.com/dinowernli/almanac/pkg/storage"
+	pb_almanac "github.com/dinowernli/almanac/proto"
 
 	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"

pkg/storage/chunk.go

@@ -6,9 +6,9 @@ import (
 	"sort"
 	"strings"
 
-	"dinowernli.me/almanac/pkg/index"
-	"dinowernli.me/almanac/pkg/util"
-	pb_almanac "dinowernli.me/almanac/proto"
+	"github.com/dinowernli/almanac/pkg/index"
+	"github.com/dinowernli/almanac/pkg/util"
+	pb_almanac "github.com/dinowernli/almanac/proto"
 
 	"golang.org/x/net/context"
 )

pkg/storage/chunk_test.go

@@ -3,7 +3,7 @@ package storage
 import (
 	"testing"
 
-	pb_almanac "dinowernli.me/almanac/proto"
+	pb_almanac "github.com/dinowernli/almanac/proto"
 
 	"github.com/stretchr/testify/assert"
 )

pkg/storage/entry.go

@@ -1,7 +1,7 @@
 package storage
 
 import (
-	pb_almanac "dinowernli.me/almanac/proto"
+	pb_almanac "github.com/dinowernli/almanac/proto"
 )
 
 // OldestEntryFirst is an ordering over log entries by ascending timestamp.

pkg/storage/s3.go

@@ -0,0 +1,84 @@
+package storage
+
+import (
+	"bytes"
+	"io/ioutil"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/session"
+	"github.com/aws/aws-sdk-go/service/s3"
+	"golang.org/x/net/context"
+)
+
+const ()
+
+// newS3Backend returns a new backend implementation backed by the supplied
+// S3 bucket name. Note that with the current interface, the AWS_REGION environment
+// variable must be specified to use this backend.
+func newS3Backend(bucketName string) (*s3Backend, error) {
+	return &s3Backend{bucketName: aws.String(bucketName)}, nil
+}
+
+type s3Backend struct {
+	bucketName *string
+}
+
+func (b *s3Backend) read(_ context.Context, id string) ([]byte, error) {
+	sess := session.Must(session.NewSession())
+
+	s3Client := s3.New(sess)
+	getOutput, err := s3Client.GetObject(&s3.GetObjectInput{
+		Bucket: b.bucketName,
+		Key:    aws.String(id),
+	})
+	if err != nil {
+		return []byte{}, err
+	}
+
+	return ioutil.ReadAll(getOutput.Body)
+}
+
+func (b *s3Backend) write(_ context.Context, id string, contents []byte) error {
+	sess := session.Must(session.NewSession())
+
+	s3Client := s3.New(sess)
+	_, err := s3Client.PutObject(&s3.PutObjectInput{
+		Bucket: b.bucketName,
+		Key:    aws.String(id),
+		Body:   bytes.NewReader(contents),
+	})
+
+	return err
+}
+
+func (b *s3Backend) list(_ context.Context, prefix string) ([]string, error) {
+	sess := session.Must(session.NewSession())
+
+	s3Client := s3.New(sess)
+	listOutput, err := s3Client.ListObjects(&s3.ListObjectsInput{
+		Bucket: b.bucketName,
+		Prefix: aws.String(prefix),
+	})
+	if err != nil {
+		return []string{}, err
+	}
+
+	var keys []string
+	for _, obj := range listOutput.Contents {
+		keys = append(keys, *obj.Key)
+	}
+
+	return keys, nil
+}
+
+func (b *s3Backend) delete(_ context.Context, id string) error {
+	sess := session.Must(session.NewSession())
+
+	s3Client := s3.New(sess)
+	_, err := s3Client.DeleteObject(&s3.DeleteObjectInput{
+		Bucket: b.bucketName,
+		Key:    aws.String(id),
+	})
+
+	return err
+}

pkg/storage/storage.go

@@ -6,8 +6,8 @@ import (
 	"os"
 	"strings"
 
-	"dinowernli.me/almanac/pkg/util"
-	pb_almanac "dinowernli.me/almanac/proto"
+	"github.com/dinowernli/almanac/pkg/util"
+	pb_almanac "github.com/dinowernli/almanac/proto"
 
 	"github.com/golang/protobuf/proto"
 	"github.com/prometheus/client_golang/prometheus"
@@ -18,6 +18,7 @@ const (
 	StorageTypeMemory = "memory"
 	StorageTypeDisk   = "disk"
 	StorageTypeGcs    = "gcs"
+	StorageTypeS3     = "s3"
 
 	chunkPrefix    = "chunk-"
 	chunkTypeLabel = "chunk_type"
@@ -181,6 +182,15 @@ func NewGcsStorage(bucketName string) (*Storage, error) {
 	return newStorage(backend)
 }
 
+// NewS3Storage returns a storage backed by the supplied S£ bucket.
+func NewS3Storage(bucketName string) (*Storage, error) {
+	backend, err := newS3Backend(bucketName)
+	if err != nil {
+		return nil, fmt.Errorf("unable to create S3 backend: %v", err)
+	}
+	return newStorage(backend)
+}
+
 func newStorage(b backend) (*Storage, error) {
 	m, err := newStorageMetrics()
 	if err != nil {