tempo-cli: add command to migrate a tenant

CHANGELOG.md

@@ -6,6 +6,7 @@
 * [FEATURE] Add flag to optionally enable all available Go runtime metrics [#2005](https://github.com/grafana/tempo/pull/2005) (@andreasgerstmayr)
 * [ENHANCEMENT] Metrics generator to make use of counters earlier [#2068](https://github.com/grafana/tempo/pull/2068) (@zalegrala)
 * [ENHANCEMENT] Log when a trace is too large to compact [#2105](https://github.com/grafana/tempo/pull/2105) (@scalalang2)
+* [ENHANCEMENT] tempo-cli: add command to migrate a tenant [#2130](https://github.com/grafana/tempo/pull/2130) (@kvrhdn)
 * [BUGFIX] Suppress logspam in single binary mode when metrics generator is disabled. [#2058](https://github.com/grafana/tempo/pull/2058) (@joe-elliott)
 * [BUGFIX] Error more gracefully while reading some blocks written by an interim commit between 1.5 and 2.0 [#2055](https://github.com/grafana/tempo/pull/2055) (@mdisibio)
 * [BUGFIX] Apply `rate()` to bytes/s panel in tenant's dashboard. [#2081](https://github.com/grafana/tempo/pull/2081) (@mapno)

Makefile

@@ -144,6 +144,10 @@ docker-tempo:
 docker-tempo-debug:
 	COMPONENT=tempo $(MAKE) docker-component-debug
 
+.PHONY: docker-cli
+docker-tempo-cli:
+	COMPONENT=tempo-cli $(MAKE) docker-component
+
 .PHONY: docker-tempo-query
 docker-tempo-query:
 	COMPONENT=tempo-query $(MAKE) docker-component

cmd/tempo-cli/Dockerfile

@@ -0,0 +1,5 @@
+FROM alpine:3.16 as certs
+RUN apk --update add ca-certificates
+ARG TARGETARCH
+COPY bin/linux/tempo-cli-${TARGETARCH} /tempo-cli
+ENTRYPOINT ["/tempo-cli"]

cmd/tempo-cli/cmd-migrate-tenant.go

@@ -0,0 +1,94 @@
+package main
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/dustin/go-humanize"
+	"github.com/pkg/errors"
+
+	"github.com/grafana/tempo/tempodb/backend"
+	"github.com/grafana/tempo/tempodb/encoding"
+)
+
+type migrateTenantCmd struct {
+	SourceConfigFile string `type:"path" required:"" help:"Path to tempo config file for source"`
+
+	SourceTenantID string `arg:"" help:"source tenant-id"`
+	DestTenantID   string `arg:"" help:"dest tenant-id"`
+}
+
+func (cmd *migrateTenantCmd) Run(opts *globalOptions) error {
+	ctx := context.Background()
+
+	readerSource, readerDest, writerDest, err := cmd.setupBackends(opts)
+	if err != nil {
+		return errors.Wrap(err, "setting up backends")
+	}
+	defer func() {
+		readerSource.Shutdown()
+		readerDest.Shutdown()
+	}()
+
+	sourceTenantIndex, err := readerSource.TenantIndex(ctx, cmd.SourceTenantID)
+	if err != nil {
+		return errors.Wrap(err, "reading source tenant index")
+	}
+	fmt.Printf("Blocks in source: %d, compacted: %d\n", len(sourceTenantIndex.Meta), len(sourceTenantIndex.CompactedMeta))
+
+	// TODO create dest directory if it doesn't exist yet?
+
+	blocksDest, err := readerDest.Blocks(ctx, cmd.DestTenantID)
+	if err != nil {
+		return err
+	}
+	fmt.Printf("Blocks in destination: %d\n", len(blocksDest))
+
+	var copiedBlocks, copiedSize uint64
+
+blocks:
+	for _, sourceBlockMeta := range sourceTenantIndex.Meta {
+		// check for collisions
+		for _, uuidDest := range blocksDest {
+			if sourceBlockMeta.BlockID == uuidDest {
+				fmt.Printf("UUID %s exists in source and destination, skipping block\n", sourceBlockMeta.BlockID)
+				continue blocks
+			}
+		}
+
+		// create a copy with destination tenant ID
+		destBlockMeta := *sourceBlockMeta
+		destBlockMeta.TenantID = cmd.DestTenantID
+
+		encoder, err := encoding.FromVersion(sourceBlockMeta.Version)
+		if err != nil {
+			return errors.Wrap(err, "creating encoder from version")
+		}
+
+		err = encoder.MigrateBlock(ctx, sourceBlockMeta, &destBlockMeta, readerSource, writerDest)
+		if err != nil {
+			return errors.Wrap(err, "copying block")
+		}
+
+		copiedBlocks++
+		copiedSize += sourceBlockMeta.Size
+	}
+
+	fmt.Printf("Finished migrating data. Copied %d blocks, %s\n", copiedBlocks, humanize.Bytes(copiedSize))
+	return nil
+}
+
+func (cmd *migrateTenantCmd) setupBackends(optsDest *globalOptions) (readerSource, readerDest backend.Reader, writerDest backend.Writer, err error) {
+	emptyBackendOptions := backendOptions{}
+
+	optsSource := &globalOptions{
+		ConfigFile: cmd.SourceConfigFile,
+	}
+	readerSource, _, _, err = loadBackend(&emptyBackendOptions, optsSource)
+	if err != nil {
+		return
+	}
+
+	readerDest, writerDest, _, err = loadBackend(&emptyBackendOptions, optsDest)
+	return
+}

cmd/tempo-cli/main.go

@@ -5,14 +5,14 @@ import (
 	"fmt"
 	"os"
 
-	"github.com/grafana/tempo/cmd/tempo/app"
-	"github.com/grafana/tempo/tempodb/backend"
-	"github.com/grafana/tempo/tempodb/backend/local"
+	"github.com/alecthomas/kong"
 	"gopkg.in/yaml.v2"
 
-	"github.com/alecthomas/kong"
+	"github.com/grafana/tempo/cmd/tempo/app"
+	"github.com/grafana/tempo/tempodb/backend"
 	"github.com/grafana/tempo/tempodb/backend/azure"
 	"github.com/grafana/tempo/tempodb/backend/gcs"
+	"github.com/grafana/tempo/tempodb/backend/local"
 	"github.com/grafana/tempo/tempodb/backend/s3"
 )
 
@@ -74,6 +74,10 @@ var cli struct {
 	Parquet struct {
 		Convert convertParquet `cmd:"" help:"convert from an existing file to tempodb parquet schema"`
 	} `cmd:""`
+
+	Migrate struct {
+		Tenant migrateTenantCmd `cmd:"" help:"migrate tenant between two backends"`
+	} `cmd:""`
 }
 
 func main() {

docs/sources/tempo/operations/tempo_cli.md

@@ -27,13 +27,19 @@ tempo-cli command [subcommand] -h
 ## Running Tempo CLI
 
 Tempo CLI is currently available as source code. A working Go installation is required to build it. It can be compiled to a native binary and executed normally, or it can be executed using the `go run` command.
+It can be packaged as a Docker container using `make docker-tempo-cli`.
 
 **Example:**
 ```bash
 ./tempo-cli [arguments...]
 go run ./cmd/tempo-cli [arguments...]
 ```
 
+```bash
+make docker-tempo-cli
+docker run docker.io/grafana/tempo-cli [arguments...]
+```
+
 ## Backend options
 
 Tempo CLI connects directly to the storage backend for some commands, meaning that it requires the ability to read from S3, GCS, Azure or file-system storage.
@@ -266,6 +272,7 @@ attempting to determine the impact of changing compression or encoding of column
 
 ```bash
 tempo-cli parquet convert <in file> <out file>
+```
 
 Arguments:
 - `in file` Filename of an existing parquet file containing Tempo trace data
@@ -275,3 +282,24 @@ Arguments:
 ```bash
 tempo-cli parquet convert data.parquet out.parquet
 ```
+
+## Migrate tenant command
+Copy blocks from one backend and tenant to another. Blocks can be copied within the same backend or between two
+different backends. Data format will not be converted but tenant ID in `meta.json` will be rewritten.
+
+```bash
+tempo-cli migrate tenant <source tenant> <dest tenant>
+```
+
+Arguments:
+- `source tenant` Tenant to copy blocks from 
+- `dest tenant` Tenant to copy blocks into
+
+Options:
+- `--source-config-file <value>` Configuration file for the source backend
+- `--config-file <value>` Configuration file for the destination backend
+
+**Example:**
+```bash
+tempo-cli migrate tenant --source-config source.yaml --config-file dest.yaml my-tenant my-other-tenant
+```

tempodb/encoding/v2/block.go

@@ -4,9 +4,10 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/pkg/errors"
+
 	"github.com/grafana/tempo/tempodb/backend"
 	"github.com/grafana/tempo/tempodb/encoding/common"
-	"github.com/pkg/errors"
 )
 
 // writeBlockMeta writes the bloom filter, meta and index to the passed in backend.Writer
@@ -46,29 +47,26 @@ func appendBlockData(ctx context.Context, w backend.Writer, meta *backend.BlockM
 }
 
 // CopyBlock copies a block from one backend to another.   It is done at a low level, all encoding/formatting is preserved.
-func CopyBlock(ctx context.Context, meta *backend.BlockMeta, src backend.Reader, dest backend.Writer) error {
-	blockID := meta.BlockID
-	tenantID := meta.TenantID
-
+func CopyBlock(ctx context.Context, srcMeta, destMeta *backend.BlockMeta, src backend.Reader, dest backend.Writer) error {
 	// Copy streams, efficient but can't cache.
 	copyStream := func(name string) error {
-		reader, size, err := src.StreamReader(ctx, name, blockID, tenantID)
+		reader, size, err := src.StreamReader(ctx, name, srcMeta.BlockID, srcMeta.TenantID)
 		if err != nil {
 			return errors.Wrapf(err, "error reading %s", name)
 		}
 		defer reader.Close()
 
-		return dest.StreamWriter(ctx, name, blockID, tenantID, reader, size)
+		return dest.StreamWriter(ctx, name, destMeta.BlockID, destMeta.TenantID, reader, size)
 	}
 
 	// Read entire object and attempt to cache
 	copy := func(name string) error {
-		b, err := src.Read(ctx, name, blockID, tenantID, true)
+		b, err := src.Read(ctx, name, srcMeta.BlockID, srcMeta.TenantID, true)
 		if err != nil {
 			return errors.Wrapf(err, "error reading %s", name)
 		}
 
-		return dest.Write(ctx, name, blockID, tenantID, b, true)
+		return dest.Write(ctx, name, destMeta.BlockID, destMeta.TenantID, b, true)
 	}
 
 	// Data
@@ -78,7 +76,7 @@ func CopyBlock(ctx context.Context, meta *backend.BlockMeta, src backend.Reader,
 	}
 
 	// Bloom
-	for i := 0; i < common.ValidateShardCount(int(meta.BloomShardCount)); i++ {
+	for i := 0; i < common.ValidateShardCount(int(srcMeta.BloomShardCount)); i++ {
 		err = copy(common.BloomName(i))
 		if err != nil {
 			return err
@@ -92,6 +90,6 @@ func CopyBlock(ctx context.Context, meta *backend.BlockMeta, src backend.Reader,
 	}
 
 	// Meta
-	err = dest.WriteBlockMeta(ctx, meta)
+	err = dest.WriteBlockMeta(ctx, destMeta)
 	return err
 }

tempodb/encoding/v2/encoding.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/google/uuid"
+
 	"github.com/grafana/tempo/pkg/model"
 	"github.com/grafana/tempo/tempodb/backend"
 	"github.com/grafana/tempo/tempodb/encoding/common"
@@ -29,7 +30,11 @@ func (v Encoding) OpenBlock(meta *backend.BlockMeta, r backend.Reader) (common.B
 }
 
 func (v Encoding) CopyBlock(ctx context.Context, meta *backend.BlockMeta, from backend.Reader, to backend.Writer) error {
-	return CopyBlock(ctx, meta, from, to)
+	return CopyBlock(ctx, meta, meta, from, to)
+}
+
+func (v Encoding) MigrateBlock(ctx context.Context, fromMeta, toMeta *backend.BlockMeta, from backend.Reader, to backend.Writer) error {
+	return CopyBlock(ctx, fromMeta, toMeta, from, to)
 }
 
 func (v Encoding) CreateBlock(ctx context.Context, cfg *common.BlockConfig, meta *backend.BlockMeta, i common.Iterator, _ backend.Reader, to backend.Writer) (*backend.BlockMeta, error) {

tempodb/encoding/versioned.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"github.com/google/uuid"
+
 	"github.com/grafana/tempo/tempodb/backend"
 	"github.com/grafana/tempo/tempodb/encoding/common"
 	v2 "github.com/grafana/tempo/tempodb/encoding/v2"
@@ -39,6 +40,9 @@ type VersionedEncoding interface {
 	// CopyBlock from one backend to another.
 	CopyBlock(ctx context.Context, meta *backend.BlockMeta, from backend.Reader, to backend.Writer) error
 
+	// MigrateBlock from one backend and tenant to another.
+	MigrateBlock(ctx context.Context, fromMeta, toMeta *backend.BlockMeta, from backend.Reader, to backend.Writer) error
+
 	// OpenWALBlock opens an existing appendable block for the WAL
 	OpenWALBlock(filename string, path string, ingestionSlack time.Duration, additionalStartSlack time.Duration) (common.WALBlock, error, error)
 

tempodb/encoding/vparquet/copy.go

@@ -4,34 +4,32 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/pkg/errors"
+
 	"github.com/grafana/tempo/tempodb/backend"
 	"github.com/grafana/tempo/tempodb/encoding/common"
-	"github.com/pkg/errors"
 )
 
-func CopyBlock(ctx context.Context, meta *backend.BlockMeta, from backend.Reader, to backend.Writer) error {
-	blockID := meta.BlockID
-	tenantID := meta.TenantID
-
+func CopyBlock(ctx context.Context, fromMeta, toMeta *backend.BlockMeta, from backend.Reader, to backend.Writer) error {
 	// Copy streams, efficient but can't cache.
 	copyStream := func(name string) error {
-		reader, size, err := from.StreamReader(ctx, name, blockID, tenantID)
+		reader, size, err := from.StreamReader(ctx, name, fromMeta.BlockID, fromMeta.TenantID)
 		if err != nil {
 			return errors.Wrapf(err, "error reading %s", name)
 		}
 		defer reader.Close()
 
-		return to.StreamWriter(ctx, name, blockID, tenantID, reader, size)
+		return to.StreamWriter(ctx, name, toMeta.BlockID, toMeta.TenantID, reader, size)
 	}
 
 	// Read entire object and attempt to cache
 	copy := func(name string) error {
-		b, err := from.Read(ctx, name, blockID, tenantID, true)
+		b, err := from.Read(ctx, name, fromMeta.BlockID, fromMeta.TenantID, true)
 		if err != nil {
 			return errors.Wrapf(err, "error reading %s", name)
 		}
 
-		return to.Write(ctx, name, blockID, tenantID, b, true)
+		return to.Write(ctx, name, toMeta.BlockID, toMeta.TenantID, b, true)
 	}
 
 	// Data
@@ -41,15 +39,15 @@ func CopyBlock(ctx context.Context, meta *backend.BlockMeta, from backend.Reader
 	}
 
 	// Bloom
-	for i := 0; i < common.ValidateShardCount(int(meta.BloomShardCount)); i++ {
+	for i := 0; i < common.ValidateShardCount(int(fromMeta.BloomShardCount)); i++ {
 		err = copy(common.BloomName(i))
 		if err != nil {
 			return err
 		}
 	}
 
 	// Meta
-	err = to.WriteBlockMeta(ctx, meta)
+	err = to.WriteBlockMeta(ctx, toMeta)
 	return err
 }
 

tempodb/encoding/vparquet/encoding.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/google/uuid"
+
 	"github.com/grafana/tempo/tempodb/backend"
 	"github.com/grafana/tempo/tempodb/encoding/common"
 )
@@ -27,7 +28,11 @@ func (v Encoding) OpenBlock(meta *backend.BlockMeta, r backend.Reader) (common.B
 }
 
 func (v Encoding) CopyBlock(ctx context.Context, meta *backend.BlockMeta, from backend.Reader, to backend.Writer) error {
-	return CopyBlock(ctx, meta, from, to)
+	return CopyBlock(ctx, meta, meta, from, to)
+}
+
+func (v Encoding) MigrateBlock(ctx context.Context, fromMeta, toMeta *backend.BlockMeta, from backend.Reader, to backend.Writer) error {
+	return CopyBlock(ctx, fromMeta, toMeta, from, to)
 }
 
 func (v Encoding) CreateBlock(ctx context.Context, cfg *common.BlockConfig, meta *backend.BlockMeta, i common.Iterator, r backend.Reader, to backend.Writer) (*backend.BlockMeta, error) {