Blocksconvert suite of tools for converting existing chunks data to blocks

Makefile

@@ -30,7 +30,7 @@ SED ?= $(shell which gsed 2>/dev/null || which sed)
 
 # We don't want find to scan inside a bunch of directories, to accelerate the
 # 'make: Entering directory '/go/src/github.com/cortexproject/cortex' phase.
-DONT_FIND := -name tools -prune -o -name vendor -prune -o -name .git -prune -o -name .cache -prune -o -name .pkg -prune -o -name packaging -prune -o
+DONT_FIND := -name vendor -prune -o -name .git -prune -o -name .cache -prune -o -name .pkg -prune -o -name packaging -prune -o
 
 # Get a list of directories containing Dockerfiles
 DOCKERFILES := $(shell find . $(DONT_FIND) -type f -name 'Dockerfile' -print)
@@ -68,6 +68,7 @@ pkg/ruler/rules/rules.pb.go: pkg/ruler/rules/rules.proto
 pkg/ruler/ruler.pb.go: pkg/ruler/rules/rules.proto
 pkg/ring/kv/memberlist/kv.pb.go: pkg/ring/kv/memberlist/kv.proto
 pkg/chunk/grpc/grpc.pb.go: pkg/chunk/grpc/grpc.proto
+tools/blocksconvert/scheduler.pb.go: tools/blocksconvert/scheduler.proto
 
 all: $(UPTODATE_FILES)
 test: protos

cmd/blocksconvert/Dockerfile

@@ -0,0 +1,9 @@
+FROM       alpine:3.12
+RUN        apk add --no-cache ca-certificates
+COPY       blocksconvert /
+ENTRYPOINT ["/blocksconvert"]
+
+ARG revision
+LABEL org.opencontainers.image.title="blocksconvert" \
+      org.opencontainers.image.source="https://github.com/cortexproject/cortex/tree/master/tools/blocksconvert" \
+      org.opencontainers.image.revision="${revision}"

cmd/blocksconvert/main.go

@@ -0,0 +1,102 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"os"
+	"strings"
+
+	"github.com/go-kit/kit/log/level"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/weaveworks/common/server"
+	"github.com/weaveworks/common/signals"
+
+	"github.com/cortexproject/cortex/pkg/cortex"
+	"github.com/cortexproject/cortex/pkg/util"
+	"github.com/cortexproject/cortex/pkg/util/services"
+	"github.com/cortexproject/cortex/tools/blocksconvert"
+	"github.com/cortexproject/cortex/tools/blocksconvert/builder"
+	"github.com/cortexproject/cortex/tools/blocksconvert/scanner"
+	"github.com/cortexproject/cortex/tools/blocksconvert/scheduler"
+)
+
+type Config struct {
+	Target       string
+	ServerConfig server.Config
+
+	SharedConfig    blocksconvert.SharedConfig
+	ScannerConfig   scanner.Config
+	BuilderConfig   builder.Config
+	SchedulerConfig scheduler.Config
+}
+
+func main() {
+	cfg := Config{}
+	flag.StringVar(&cfg.Target, "target", "", "Module to run: Scanner, Scheduler, Builder")
+	cfg.SharedConfig.RegisterFlags(flag.CommandLine)
+	cfg.ScannerConfig.RegisterFlags(flag.CommandLine)
+	cfg.BuilderConfig.RegisterFlags(flag.CommandLine)
+	cfg.SchedulerConfig.RegisterFlags(flag.CommandLine)
+	cfg.ServerConfig.RegisterFlags(flag.CommandLine)
+	flag.Parse()
+
+	util.InitLogger(&cfg.ServerConfig)
+
+	cortex.DisableSignalHandling(&cfg.ServerConfig)
+	serv, err := server.New(cfg.ServerConfig)
+	if err != nil {
+		level.Error(util.Logger).Log("msg", "Unable to initialize server", "err", err.Error())
+		os.Exit(1)
+	}
+
+	cfg.Target = strings.ToLower(cfg.Target)
+
+	registry := prometheus.DefaultRegisterer
+
+	var targetService services.Service
+	switch cfg.Target {
+	case "scanner":
+		targetService, err = scanner.NewScanner(cfg.ScannerConfig, cfg.SharedConfig, util.Logger, registry)
+	case "builder":
+		targetService, err = builder.NewBuilder(cfg.BuilderConfig, cfg.SharedConfig, util.Logger, registry)
+	case "scheduler":
+		targetService, err = scheduler.NewScheduler(cfg.SchedulerConfig, cfg.SharedConfig, util.Logger, registry, serv.HTTP, serv.GRPC)
+	default:
+		err = fmt.Errorf("unknown target")
+	}
+
+	if err != nil {
+		level.Error(util.Logger).Log("msg", "failed to initialize", "err", err)
+		os.Exit(1)
+	}
+
+	servService := cortex.NewServerService(serv, func() []services.Service {
+		return []services.Service{targetService}
+	})
+	servManager, err := services.NewManager(servService, targetService)
+	if err == nil {
+		servManager.AddListener(services.NewManagerListener(nil, nil, func(service services.Service) {
+			servManager.StopAsync()
+		}))
+
+		err = services.StartManagerAndAwaitHealthy(context.Background(), servManager)
+	}
+	if err != nil {
+		level.Error(util.Logger).Log("msg", "Unable to start", "err", err.Error())
+		os.Exit(1)
+	}
+
+	// Setup signal handler and ask service maanger to stop when signal arrives.
+	handler := signals.NewHandler(serv.Log)
+	go func() {
+		handler.Loop()
+		servManager.StopAsync()
+	}()
+
+	// We only wait for target service. If any other service fails, listener will stop it (via manager)
+	if err := targetService.AwaitTerminated(context.Background()); err != nil {
+		level.Error(util.Logger).Log("msg", cfg.Target+" failed", "err", targetService.FailureCase())
+		os.Exit(1)
+	}
+}

pkg/chunk/gcp/bigtable_index_client.go

@@ -116,17 +116,17 @@ func newStorageClientColumnKey(cfg Config, schemaCfg chunk.SchemaConfig, client
 			// We hash the row key and prepend it back to the key for better distribution.
 			// We preserve the existing key to make migrations and o11y easier.
 			if cfg.DistributeKeys {
-				hashValue = hashPrefix(hashValue) + "-" + hashValue
+				hashValue = HashPrefix(hashValue) + "-" + hashValue
 			}
 
 			return hashValue, string(rangeValue)
 		},
 	}
 }
 
-// hashPrefix calculates a 64bit hash of the input string and hex-encodes
+// HashPrefix calculates a 64bit hash of the input string and hex-encodes
 // the result, taking care to zero pad etc.
-func hashPrefix(input string) string {
+func HashPrefix(input string) string {
 	prefix := hashAdd(hashNew(), input)
 	var encodedUint64 [8]byte
 	binary.LittleEndian.PutUint64(encodedUint64[:], prefix)

pkg/chunk/gcp/fixtures.go

@@ -100,7 +100,7 @@ var Fixtures = func() []testutils.Fixture {
 		for _, columnKeyClient := range []bool{true, false} {
 			for _, hashPrefix := range []bool{true, false} {
 				fixtures = append(fixtures, &fixture{
-					name:            fmt.Sprintf("bigtable-columnkey:%v-gcsObjectClient:%v-hashPrefix:%v", columnKeyClient, gcsObjectClient, hashPrefix),
+					name:            fmt.Sprintf("bigtable-columnkey:%v-gcsObjectClient:%v-HashPrefix:%v", columnKeyClient, gcsObjectClient, hashPrefix),
 					columnKeyClient: columnKeyClient,
 					gcsObjectClient: gcsObjectClient,
 					hashPrefix:      hashPrefix,