BigTable storage backend for Cortex (#468)

* Cortex Bigtable backend. - Currently stores chunks in BigTable only. - Minimal amount of refactoring, had to: - break out the storage client factory into its own package to avoid import loops. - export a few methods on Chunk (ExternalKey and Decode). - Add stack traces to some of the chunk store errors, and log them on queries. - Instrument the BigTable gRPC client. Also, fix some logic in the previous BigTable storage engine: - Return correct part of the row key as the range key - Correctly stop reading bigtable results when trying to read to the end of a 'row'. - Log the range key we read from bigtable. - Chunks are being written before calculating their checksum, which shouldn't happen. - Don't return empty chunks for BigTable backend. * Review feedback * Instrument streaming gRPC BigTable calls too. * Chunks don't nessecarily come from S3 anymore. * When fetching chunks, the should be returned if processingErr == nil.
cortexproject · Jul 13, 2017 · fa9cf5a · fa9cf5a
1 parent aea9137
commit fa9cf5a
Show file tree

Hide file tree

Showing 172 changed files with 27,867 additions and 405 deletions.
diff --git a/Gopkg.lock b/Gopkg.lock
diff --git a/Makefile b/Makefile
@@ -122,6 +122,10 @@ update-gazelle:
 	gazelle -go_prefix github.com/weaveworks/cortex -external vendored \
 		-build_file_name BUILD.bazel
 
+update-vendor:
+	dep ensure && dep prune
+	git status | grep BUILD.bazel | cut -d' ' -f 5 | xargs git checkout HEAD
+
 bazel:
 	bazel build //cmd/...
 

diff --git a/cmd/ingester/BUILD.bazel b/cmd/ingester/BUILD.bazel
@@ -6,6 +6,7 @@ go_library(
     visibility = ["//visibility:private"],
     deps = [
         "//pkg/chunk:go_default_library",
+        "//pkg/chunk/storage:go_default_library",
         "//pkg/ingester:go_default_library",
         "//pkg/ingester/client:go_default_library",
         "//pkg/util:go_default_library",

diff --git a/cmd/ingester/main.go b/cmd/ingester/main.go
@@ -11,6 +11,7 @@ import (
 	"github.com/weaveworks/common/middleware"
 	"github.com/weaveworks/common/server"
 	"github.com/weaveworks/cortex/pkg/chunk"
+	"github.com/weaveworks/cortex/pkg/chunk/storage"
 	"github.com/weaveworks/cortex/pkg/ingester"
 	"github.com/weaveworks/cortex/pkg/ingester/client"
 	"github.com/weaveworks/cortex/pkg/util"
@@ -25,8 +26,8 @@ func main() {
 			},
 		}
 		chunkStoreConfig chunk.StoreConfig
-		storageConfig    chunk.StorageClientConfig
 		schemaConfig     chunk.SchemaConfig
+		storageConfig    storage.Config
 		ingesterConfig   ingester.Config
 	)
 	// Ingester needs to know our gRPC listen port.
@@ -41,7 +42,7 @@ func main() {
 	}
 	defer server.Shutdown()
 
-	storageClient, err := chunk.NewStorageClient(storageConfig, schemaConfig)
+	storageClient, err := storage.NewStorageClient(storageConfig, schemaConfig)
 	if err != nil {
 		log.Fatalf("Error initializing storage client: %v", err)
 	}

diff --git a/cmd/lite/BUILD.bazel b/cmd/lite/BUILD.bazel
@@ -6,6 +6,7 @@ go_library(
     visibility = ["//visibility:private"],
     deps = [
         "//pkg/chunk:go_default_library",
+        "//pkg/chunk/storage:go_default_library",
         "//pkg/distributor:go_default_library",
         "//pkg/ingester:go_default_library",
         "//pkg/ingester/client:go_default_library",

diff --git a/cmd/lite/main.go b/cmd/lite/main.go
@@ -16,6 +16,7 @@ import (
 	"github.com/weaveworks/common/server"
 	"github.com/weaveworks/common/user"
 	"github.com/weaveworks/cortex/pkg/chunk"
+	"github.com/weaveworks/cortex/pkg/chunk/storage"
 	"github.com/weaveworks/cortex/pkg/distributor"
 	"github.com/weaveworks/cortex/pkg/ingester"
 	"github.com/weaveworks/cortex/pkg/ingester/client"
@@ -37,13 +38,14 @@ func main() {
 		distributorConfig distributor.Config
 		ingesterConfig    ingester.Config
 		schemaConfig      chunk.SchemaConfig
-		storageConfig     chunk.StorageClientConfig
-		unauthenticated   bool
+		storageConfig     storage.Config
+
+		unauthenticated bool
 	)
 	// Ingester needs to know our gRPC listen port.
 	ingesterConfig.ListenPort = &serverConfig.GRPCListenPort
 	util.RegisterFlags(&serverConfig, &chunkStoreConfig, &distributorConfig,
-		&ingesterConfig, &schemaConfig, &storageConfig)
+		&ingesterConfig, &storageConfig, &schemaConfig)
 	flag.BoolVar(&unauthenticated, "unauthenticated", false, "Set to true to disable multitenancy.")
 	flag.Parse()
 
@@ -53,7 +55,7 @@ func main() {
 	}
 	defer server.Shutdown()
 
-	storageClient, err := chunk.NewStorageClient(storageConfig, schemaConfig)
+	storageClient, err := storage.NewStorageClient(storageConfig, schemaConfig)
 	if err != nil {
 		log.Fatalf("Error initializing storage client: %v", err)
 	}
@@ -85,7 +87,7 @@ func main() {
 	prometheus.MustRegister(ingester)
 	defer ingester.Shutdown()
 
-	tableClient, err := chunk.NewDynamoDBTableClient(storageConfig.AWSStorageConfig.DynamoDBConfig)
+	tableClient, err := storage.NewTableClient(storageConfig)
 	if err != nil {
 		log.Fatalf("Error initializing DynamoDB table client: %v", err)
 	}

diff --git a/cmd/querier/BUILD.bazel b/cmd/querier/BUILD.bazel
@@ -6,6 +6,7 @@ go_library(
     visibility = ["//visibility:private"],
     deps = [
         "//pkg/chunk:go_default_library",
+        "//pkg/chunk/storage:go_default_library",
         "//pkg/distributor:go_default_library",
         "//pkg/querier:go_default_library",
         "//pkg/ring:go_default_library",

diff --git a/cmd/querier/main.go b/cmd/querier/main.go
@@ -16,6 +16,7 @@ import (
 	"github.com/weaveworks/common/middleware"
 	"github.com/weaveworks/common/server"
 	"github.com/weaveworks/cortex/pkg/chunk"
+	"github.com/weaveworks/cortex/pkg/chunk/storage"
 	"github.com/weaveworks/cortex/pkg/distributor"
 	"github.com/weaveworks/cortex/pkg/querier"
 	"github.com/weaveworks/cortex/pkg/ring"
@@ -33,11 +34,11 @@ func main() {
 		ringConfig        ring.Config
 		distributorConfig distributor.Config
 		chunkStoreConfig  chunk.StoreConfig
-		storageConfig     chunk.StorageClientConfig
 		schemaConfig      chunk.SchemaConfig
+		storageConfig     storage.Config
 	)
 	util.RegisterFlags(&serverConfig, &ringConfig, &distributorConfig,
-		&chunkStoreConfig, &storageConfig, &schemaConfig)
+		&chunkStoreConfig, &schemaConfig, &storageConfig)
 	flag.Parse()
 
 	r, err := ring.New(ringConfig)
@@ -60,7 +61,7 @@ func main() {
 	defer server.Shutdown()
 	server.HTTP.Handle("/ring", r)
 
-	storageClient, err := chunk.NewStorageClient(storageConfig, schemaConfig)
+	storageClient, err := storage.NewStorageClient(storageConfig, schemaConfig)
 	if err != nil {
 		log.Fatalf("Error initializing storage client: %v", err)
 	}

diff --git a/cmd/ruler/BUILD.bazel b/cmd/ruler/BUILD.bazel
@@ -6,6 +6,7 @@ go_library(
     visibility = ["//visibility:private"],
     deps = [
         "//pkg/chunk:go_default_library",
+        "//pkg/chunk/storage:go_default_library",
         "//pkg/distributor:go_default_library",
         "//pkg/ring:go_default_library",
         "//pkg/ruler:go_default_library",

diff --git a/cmd/ruler/main.go b/cmd/ruler/main.go
@@ -10,6 +10,7 @@ import (
 	"github.com/weaveworks/common/middleware"
 	"github.com/weaveworks/common/server"
 	"github.com/weaveworks/cortex/pkg/chunk"
+	"github.com/weaveworks/cortex/pkg/chunk/storage"
 	"github.com/weaveworks/cortex/pkg/distributor"
 	"github.com/weaveworks/cortex/pkg/ring"
 	"github.com/weaveworks/cortex/pkg/ruler"
@@ -28,14 +29,14 @@ func main() {
 		distributorConfig distributor.Config
 		rulerConfig       ruler.Config
 		chunkStoreConfig  chunk.StoreConfig
-		storageConfig     chunk.StorageClientConfig
 		schemaConfig      chunk.SchemaConfig
+		storageConfig     storage.Config
 	)
 	util.RegisterFlags(&serverConfig, &ringConfig, &distributorConfig,
 		&rulerConfig, &chunkStoreConfig, &storageConfig, &schemaConfig)
 	flag.Parse()
 
-	storageClient, err := chunk.NewStorageClient(storageConfig, schemaConfig)
+	storageClient, err := storage.NewStorageClient(storageConfig, schemaConfig)
 	if err != nil {
 		log.Fatalf("Error initializing storage client: %v", err)
 	}

diff --git a/cmd/table-manager/BUILD.bazel b/cmd/table-manager/BUILD.bazel
@@ -6,6 +6,7 @@ go_library(
     visibility = ["//visibility:private"],
     deps = [
         "//pkg/chunk:go_default_library",
+        "//pkg/chunk/storage:go_default_library",
         "//pkg/util:go_default_library",
         "//vendor/github.com/prometheus/common/log:go_default_library",
         "//vendor/github.com/weaveworks/common/middleware:go_default_library",

diff --git a/cmd/table-manager/main.go b/cmd/table-manager/main.go
@@ -9,6 +9,7 @@ import (
 	"github.com/weaveworks/common/middleware"
 	"github.com/weaveworks/common/server"
 	"github.com/weaveworks/cortex/pkg/chunk"
+	"github.com/weaveworks/cortex/pkg/chunk/storage"
 	"github.com/weaveworks/cortex/pkg/util"
 )
 
@@ -21,13 +22,13 @@ func main() {
 			},
 		}
 
-		dynamoDBConfig = chunk.DynamoDBConfig{}
-		schemaConfig   chunk.SchemaConfig
+		storageConfig storage.Config
+		schemaConfig  chunk.SchemaConfig
 	)
-	util.RegisterFlags(&serverConfig, &dynamoDBConfig, &schemaConfig)
+	util.RegisterFlags(&serverConfig, &storageConfig, &schemaConfig)
 	flag.Parse()
 
-	tableClient, err := chunk.NewDynamoDBTableClient(dynamoDBConfig)
+	tableClient, err := storage.NewTableClient(storageConfig)
 	if err != nil {
 		log.Fatalf("Error initializing DynamoDB table client: %v", err)
 	}

diff --git a/pkg/chunk/BUILD.bazel b/pkg/chunk/BUILD.bazel
@@ -32,6 +32,7 @@ go_library(
         "//vendor/github.com/aws/aws-sdk-go/service/s3/s3iface:go_default_library",
         "//vendor/github.com/bradfitz/gomemcache/memcache:go_default_library",
         "//vendor/github.com/golang/snappy:go_default_library",
+        "//vendor/github.com/pkg/errors:go_default_library",
         "//vendor/github.com/prometheus/client_golang/prometheus:go_default_library",
         "//vendor/github.com/prometheus/common/log:go_default_library",
         "//vendor/github.com/prometheus/common/model:go_default_library",

diff --git a/pkg/chunk/aws_storage_client.go b/pkg/chunk/aws_storage_client.go
@@ -7,7 +7,6 @@ import (
 	"io/ioutil"
 	"math/rand"
 	"net/url"
-	"strconv"
 	"strings"
 	"time"
 
@@ -22,7 +21,6 @@ import (
 	"github.com/aws/aws-sdk-go/service/s3/s3iface"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/common/log"
-	"github.com/prometheus/common/model"
 	"golang.org/x/net/context"
 
 	"github.com/weaveworks/common/instrument"
@@ -57,7 +55,7 @@ var (
 		Help:      "Time spent doing DynamoDB requests.",
 
 		// DynamoDB latency seems to range from a few ms to a few sec and is
-		// important.  So use 8 buckets from 64us to 8s.
+		// important.  So use 8 buckets from 128us to 2s.
 		Buckets: prometheus.ExponentialBuckets(0.000128, 4, 8),
 	}, []string{"operation", "status_code"})
 	dynamoConsumedCapacity = prometheus.NewCounterVec(prometheus.CounterOpts{
@@ -430,7 +428,7 @@ func (a awsStorageClient) getS3Chunk(ctx context.Context, chunk Chunk) (Chunk, e
 		var err error
 		resp, err = a.S3.GetObjectWithContext(ctx, &s3.GetObjectInput{
 			Bucket: aws.String(a.bucketName),
-			Key:    aws.String(chunk.externalKey()),
+			Key:    aws.String(chunk.ExternalKey()),
 		})
 		return err
 	})
@@ -442,7 +440,7 @@ func (a awsStorageClient) getS3Chunk(ctx context.Context, chunk Chunk) (Chunk, e
 	if err != nil {
 		return Chunk{}, err
 	}
-	if err := chunk.decode(buf); err != nil {
+	if err := chunk.Decode(buf); err != nil {
 		return Chunk{}, err
 	}
 	return chunk, nil
@@ -456,9 +454,9 @@ func (a awsStorageClient) getDynamoDBChunks(ctx context.Context, chunks []Chunk)
 	outstanding := dynamoDBReadRequest{}
 	chunksByKey := map[string]Chunk{}
 	for _, chunk := range chunks {
-		key := chunk.externalKey()
+		key := chunk.ExternalKey()
 		chunksByKey[key] = chunk
-		tableName := a.chunkTableFor(chunk.From)
+		tableName := a.schemaCfg.ChunkTables.TableFor(chunk.From)
 		outstanding.Add(tableName, key, placeholder)
 	}
 
@@ -542,7 +540,7 @@ func processChunkResponse(response *dynamodb.BatchGetItemOutput, chunksByKey map
 				return nil, fmt.Errorf("Got response from DynamoDB with no value: %+v", item)
 			}
 
-			if err := chunk.decode(buf.B); err != nil {
+			if err := chunk.Decode(buf.B); err != nil {
 				return nil, err
 			}
 
@@ -561,17 +559,17 @@ func (a awsStorageClient) PutChunks(ctx context.Context, chunks []Chunk) error {
 
 	for i := range chunks {
 		// Encode the chunk first - checksum is calculated as a side effect.
-		buf, err := chunks[i].encode()
+		buf, err := chunks[i].Encode()
 		if err != nil {
 			return err
 		}
-		key := chunks[i].externalKey()
+		key := chunks[i].ExternalKey()
 
 		if !a.schemaCfg.ChunkTables.From.IsSet() || chunks[i].From.Before(a.schemaCfg.ChunkTables.From.Time) {
 			s3ChunkKeys = append(s3ChunkKeys, key)
 			s3ChunkBufs = append(s3ChunkBufs, buf)
 		} else {
-			table := a.chunkTableFor(chunks[i].From)
+			table := a.schemaCfg.ChunkTables.TableFor(chunks[i].From)
 			dynamoDBWrites.Add(table, key, placeholder, buf)
 		}
 	}
@@ -588,14 +586,6 @@ func (a awsStorageClient) PutChunks(ctx context.Context, chunks []Chunk) error {
 	return a.BatchWrite(ctx, dynamoDBWrites)
 }
 
-func (a awsStorageClient) chunkTableFor(t model.Time) string {
-	var (
-		periodSecs = int64(a.schemaCfg.ChunkTables.Period / time.Second)
-		table      = t.Unix() / periodSecs
-	)
-	return a.schemaCfg.ChunkTables.Prefix + strconv.Itoa(int(table))
-}
-
 func (a awsStorageClient) putS3Chunks(ctx context.Context, keys []string, bufs [][]byte) error {
 	incomingErrors := make(chan error)
 	for i := range bufs {