Compressed Blocks? Compressed Blocks!

.gitignore

@@ -8,3 +8,4 @@
 /cmd/tempo-cli/tempo-cli
 /example/docker-compose/example-data/tempo
 .DS_Store
+/tempodb/encoding/backend_block

cmd/tempo-cli/cmd-list-block.go

@@ -64,12 +64,12 @@ func dumpBlock(r tempodb_backend.Reader, c tempodb_backend.Compactor, tenantID s
 			Version:  unifiedMeta.version,
 			TenantID: tenantID,
 			BlockID:  id,
-		})
+		}, r)
 		if err != nil {
 			return err
 		}
 
-		iter, err := block.Iterator(10*1024*1024, r)
+		iter, err := block.Iterator(10 * 1024 * 1024)
 		if err != nil {
 			return err
 		}

example/docker-compose/etc/tempo-azure.yaml

@@ -33,10 +33,12 @@ compactor:
 storage:
   trace:
     backend: azure                     # backend configuration to use
-    wal:
-      path: /tmp/tempo/wal             # where to store the the wal locally
+    block:
       bloom_filter_false_positive: .05 # bloom filter false positive rate.  lower values create larger filters but fewer false positives
       index_downsample: 10             # number of traces per index record
+      encoding: gzip                   # block encoding/compression.  options: none, gzip, lz4-64k, lz4-256k, lz4-1M, lz4, snappy
+    wal:
+      path: /tmp/tempo/wal             # where to store the the wal locally
     azure:
       container-name: tempo                    # how to store data in azure
       endpoint-suffix: azurite:10000

example/docker-compose/etc/tempo-gcs-fake.yaml

@@ -34,10 +34,12 @@ compactor:
 storage:
   trace:
     backend: gcs                       # backend configuration to use
-    wal:
-      path: /tmp/tempo/wal             # where to store the the wal locally
+    block:
       bloom_filter_false_positive: .05 # bloom filter false positive rate.  lower values create larger filters but fewer false positives
       index_downsample: 10             # number of traces per index record
+      encoding: gzip                   # block encoding/compression. options: none, gzip, lz4-64k, lz4-256k, lz4-1M, lz4, snappy
+    wal:
+      path: /tmp/tempo/wal             # where to store the the wal locally
     gcs:
       bucket_name: tempo
       endpoint: https://gcs:4443/storage/v1/

example/docker-compose/etc/tempo-local.yaml

@@ -33,10 +33,12 @@ compactor:
 storage:
   trace:
     backend: local                     # backend configuration to use
-    wal:
-      path: /tmp/tempo/wal             # where to store the the wal locally
+    block:
       bloom_filter_false_positive: .05 # bloom filter false positive rate.  lower values create larger filters but fewer false positives
       index_downsample: 10             # number of traces per index record
+      encoding: gzip                   # block encoding/compression. options: none, gzip, lz4-64k, lz4-256k, lz4-1M, lz4, snappy
+    wal:
+      path: /tmp/tempo/wal             # where to store the the wal locally
     local:
       path: /tmp/tempo/blocks
     pool:

example/docker-compose/etc/tempo-s3-minio.yaml

@@ -34,10 +34,12 @@ compactor:
 storage:
   trace:
     backend: s3                        # backend configuration to use
-    wal:
-      path: /tmp/tempo/wal             # where to store the the wal locally
+    block:
       bloom_filter_false_positive: .05 # bloom filter false positive rate.  lower values create larger filters but fewer false positives
       index_downsample: 10             # number of traces per index record
+      encoding: gzip                   # block encoding/compression. options: none, gzip, lz4-64k, lz4-256k, lz4-1M, lz4, snappy
+    wal:
+      path: /tmp/tempo/wal             # where to store the the wal locally
     s3:
       bucket: tempo                    # how to store data in s3
       endpoint: minio:9000

go.mod

@@ -13,17 +13,20 @@ require (
 	github.com/gogo/status v1.0.3
 	github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b
 	github.com/golang/protobuf v1.4.3
+	github.com/golang/snappy v0.0.2
 	github.com/google/uuid v1.1.1
 	github.com/gorilla/mux v1.7.4
 	github.com/grafana/loki v1.3.0
 	github.com/grpc-ecosystem/grpc-opentracing v0.0.0-20180507213350-8e809c8a8645
 	github.com/hashicorp/go-hclog v0.14.0
 	github.com/jaegertracing/jaeger v1.18.2-0.20200707061226-97d2319ff2be
 	github.com/jsternberg/zap-logfmt v1.0.0
+	github.com/klauspost/compress v1.10.5
 	github.com/minio/minio-go/v7 v7.0.5
 	github.com/olekukonko/tablewriter v0.0.2
 	github.com/open-telemetry/opentelemetry-proto v0.4.0
 	github.com/opentracing/opentracing-go v1.2.0
+	github.com/pierrec/lz4/v4 v4.1.3
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.8.0
 	github.com/prometheus/client_model v0.2.0

go.sum

@@ -1233,6 +1233,7 @@ github.com/klauspost/compress v1.4.0/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0
 github.com/klauspost/compress v1.9.4/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0guNDohfE1A=
 github.com/klauspost/compress v1.9.5/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0guNDohfE1A=
 github.com/klauspost/compress v1.10.4/go.mod h1:aoV0uJVorq1K+umq18yTdKaF57EivdYsUV+/s2qKfXs=
+github.com/klauspost/compress v1.10.5 h1:7q6vHIqubShURwQz8cQK6yIe/xC3IF0Vm7TGfqjewrc=
 github.com/klauspost/compress v1.10.5/go.mod h1:aoV0uJVorq1K+umq18yTdKaF57EivdYsUV+/s2qKfXs=
 github.com/klauspost/cpuid v0.0.0-20170728055534-ae7887de9fa5/go.mod h1:Pj4uuM528wm8OyEC2QMXAi2YiTZ96dNQPGgoMS4s3ek=
 github.com/klauspost/cpuid v1.2.3/go.mod h1:Pj4uuM528wm8OyEC2QMXAi2YiTZ96dNQPGgoMS4s3ek=
@@ -1576,7 +1577,10 @@ github.com/pierrec/lz4 v0.0.0-20190327172049-315a67e90e41/go.mod h1:3/3N9NVKO0je
 github.com/pierrec/lz4 v1.0.2-0.20190131084431-473cd7ce01a1/go.mod h1:3/3N9NVKO0jef7pBehbT1qWhCMrIgbYNnFAZCqQ5LRc=
 github.com/pierrec/lz4 v2.0.5+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
 github.com/pierrec/lz4 v2.3.1-0.20191115212037-9085dacd1e1e+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
+github.com/pierrec/lz4 v2.4.1+incompatible h1:mFe7ttWaflA46Mhqh+jUfjp2qTbPYxLB2/OyBppH9dg=
 github.com/pierrec/lz4 v2.4.1+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
+github.com/pierrec/lz4/v4 v4.1.3 h1:/dvQpkb0o1pVlSgKNQqfkavlnXaIK+hJ0LXsKRUN9D4=
+github.com/pierrec/lz4/v4 v4.1.3/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
 github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.8.1-0.20171018195549-f15c970de5b7/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=

modules/ingester/ingester.go

@@ -227,7 +227,7 @@ func (i *Ingester) getOrCreateInstance(instanceID string) (*instance, error) {
 	inst, ok = i.instances[instanceID]
 	if !ok {
 		var err error
-		inst, err = newInstance(instanceID, i.limiter, i.store.WAL())
+		inst, err = newInstance(instanceID, i.limiter, i.store)
 		if err != nil {
 			return nil, err
 		}

modules/ingester/ingester_test.go

@@ -14,14 +14,17 @@ import (
 	"github.com/go-kit/kit/log"
 	"github.com/golang/protobuf/proto"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"github.com/weaveworks/common/user"
 
 	"github.com/grafana/tempo/modules/overrides"
 	"github.com/grafana/tempo/modules/storage"
 	"github.com/grafana/tempo/pkg/tempopb"
 	"github.com/grafana/tempo/pkg/util/test"
 	"github.com/grafana/tempo/tempodb"
+	"github.com/grafana/tempo/tempodb/backend"
 	"github.com/grafana/tempo/tempodb/backend/local"
+	"github.com/grafana/tempo/tempodb/encoding"
 	"github.com/grafana/tempo/tempodb/wal"
 )
 
@@ -179,28 +182,31 @@ func TestFlush(t *testing.T) {
 func defaultIngester(t *testing.T, tmpDir string) (*Ingester, []*tempopb.Trace, [][]byte) {
 	ingesterConfig := defaultIngesterTestConfig()
 	limits, err := overrides.NewOverrides(defaultLimitsTestConfig())
-	assert.NoError(t, err, "unexpected error creating overrides")
+	require.NoError(t, err, "unexpected error creating overrides")
 
 	s, err := storage.NewStore(storage.Config{
 		Trace: tempodb.Config{
 			Backend: "local",
 			Local: &local.Config{
 				Path: tmpDir,
 			},
-			WAL: &wal.Config{
-				Filepath:        tmpDir,
+			Block: &encoding.BlockConfig{
 				IndexDownsample: 2,
 				BloomFP:         .01,
+				Encoding:        backend.EncLZ4_1M,
+			},
+			WAL: &wal.Config{
+				Filepath: tmpDir,
 			},
 		},
 	}, log.NewNopLogger())
-	assert.NoError(t, err, "unexpected error store")
+	require.NoError(t, err, "unexpected error store")
 
 	ingester, err := New(ingesterConfig, s, limits)
-	assert.NoError(t, err, "unexpected error creating ingester")
+	require.NoError(t, err, "unexpected error creating ingester")
 
 	err = ingester.starting(context.Background())
-	assert.NoError(t, err, "unexpected error starting ingester")
+	require.NoError(t, err, "unexpected error starting ingester")
 
 	// make some fake traceIDs/requests
 	traces := make([]*tempopb.Trace, 0)
@@ -209,7 +215,7 @@ func defaultIngester(t *testing.T, tmpDir string) (*Ingester, []*tempopb.Trace,
 	for i := 0; i < 10; i++ {
 		id := make([]byte, 16)
 		_, err = rand.Read(id)
-		assert.NoError(t, err)
+		require.NoError(t, err)
 
 		traces = append(traces, test.MakeTrace(10, id))
 		traceIDs = append(traceIDs, id)
@@ -222,7 +228,7 @@ func defaultIngester(t *testing.T, tmpDir string) (*Ingester, []*tempopb.Trace,
 				&tempopb.PushRequest{
 					Batch: batch,
 				})
-			assert.NoError(t, err, "unexpected error pushing")
+			require.NoError(t, err, "unexpected error pushing")
 		}
 	}
 

modules/ingester/instance.go

@@ -20,6 +20,7 @@ import (
 
 	"github.com/grafana/tempo/pkg/tempopb"
 	"github.com/grafana/tempo/pkg/util"
+	"github.com/grafana/tempo/tempodb"
 	"github.com/grafana/tempo/tempodb/encoding"
 	"github.com/grafana/tempo/tempodb/encoding/common"
 	"github.com/grafana/tempo/tempodb/wal"
@@ -63,20 +64,20 @@ type instance struct {
 	tracesCreatedTotal prometheus.Counter
 	bytesWrittenTotal  prometheus.Counter
 	limiter            *Limiter
-	wal                *wal.WAL
+	writer             tempodb.Writer
 
 	hash hash.Hash32
 }
 
-func newInstance(instanceID string, limiter *Limiter, wal *wal.WAL) (*instance, error) {
+func newInstance(instanceID string, limiter *Limiter, writer tempodb.Writer) (*instance, error) {
 	i := &instance{
 		traces: map[uint32]*trace{},
 
 		instanceID:         instanceID,
 		tracesCreatedTotal: metricTracesCreatedTotal.WithLabelValues(instanceID),
 		bytesWrittenTotal:  metricBytesWrittenTotal.WithLabelValues(instanceID),
 		limiter:            limiter,
-		wal:                wal,
+		writer:             writer,
 
 		hash: fnv.New32(),
 	}
@@ -153,7 +154,7 @@ func (i *instance) CutBlockIfReady(maxBlockLifetime time.Duration, maxBlockBytes
 
 		// todo : this should be a queue of blocks to complete with workers
 		go func() {
-			completeBlock, err := i.completingBlock.Complete(i.wal, i)
+			completeBlock, err := i.writer.CompleteBlock(i.completingBlock, i)
 			i.blocksMtx.Lock()
 			defer i.blocksMtx.Unlock()
 
@@ -307,7 +308,7 @@ func (i *instance) tokenForTraceID(id []byte) uint32 {
 // resetHeadBlock() should be called under lock
 func (i *instance) resetHeadBlock() error {
 	var err error
-	i.headBlock, err = i.wal.NewBlock(uuid.New(), i.instanceID)
+	i.headBlock, err = i.writer.WAL().NewBlock(uuid.New(), i.instanceID)
 	i.lastBlockCut = time.Now()
 	return err
 }

modules/ingester/instance_test.go

@@ -9,9 +9,15 @@ import (
 	"testing"
 	"time"
 
+	"github.com/go-kit/kit/log"
 	"github.com/grafana/tempo/modules/overrides"
+	"github.com/grafana/tempo/modules/storage"
 	"github.com/grafana/tempo/pkg/tempopb"
 	"github.com/grafana/tempo/pkg/util/test"
+	"github.com/grafana/tempo/tempodb"
+	"github.com/grafana/tempo/tempodb/backend"
+	"github.com/grafana/tempo/tempodb/backend/local"
+	"github.com/grafana/tempo/tempodb/encoding"
 	"github.com/grafana/tempo/tempodb/wal"
 
 	"github.com/stretchr/testify/assert"
@@ -36,11 +42,9 @@ func TestInstance(t *testing.T) {
 	defer os.RemoveAll(tempDir)
 
 	ingester, _, _ := defaultIngester(t, tempDir)
-	wal := ingester.store.WAL()
-
 	request := test.MakeRequest(10, []byte{})
 
-	i, err := newInstance("fake", limiter, wal)
+	i, err := newInstance("fake", limiter, ingester.store)
 	assert.NoError(t, err, "unexpected error creating new instance")
 	err = i.Push(context.Background(), request)
 	assert.NoError(t, err)
@@ -89,12 +93,10 @@ func TestInstanceFind(t *testing.T) {
 	defer os.RemoveAll(tempDir)
 
 	ingester, _, _ := defaultIngester(t, tempDir)
-	wal := ingester.store.WAL()
-
 	request := test.MakeRequest(10, []byte{})
 	traceID := test.MustTraceID(request)
 
-	i, err := newInstance("fake", limiter, wal)
+	i, err := newInstance("fake", limiter, ingester.store)
 	assert.NoError(t, err, "unexpected error creating new instance")
 	err = i.Push(context.Background(), request)
 	assert.NoError(t, err)
@@ -128,9 +130,8 @@ func TestInstanceDoesNotRace(t *testing.T) {
 	defer os.RemoveAll(tempDir)
 
 	ingester, _, _ := defaultIngester(t, tempDir)
-	wal := ingester.store.WAL()
 
-	i, err := newInstance("fake", limiter, wal)
+	i, err := newInstance("fake", limiter, ingester.store)
 	assert.NoError(t, err, "unexpected error creating new instance")
 
 	end := make(chan struct{})
@@ -197,9 +198,8 @@ func TestInstanceLimits(t *testing.T) {
 	defer os.RemoveAll(tempDir)
 
 	ingester, _, _ := defaultIngester(t, tempDir)
-	wal := ingester.store.WAL()
 
-	i, err := newInstance("fake", limiter, wal)
+	i, err := newInstance("fake", limiter, ingester.store)
 	assert.NoError(t, err, "unexpected error creating new instance")
 
 	type push struct {
@@ -427,19 +427,30 @@ func TestInstanceCutBlockIfReady(t *testing.T) {
 	}
 }
 
-func defaultInstance(t assert.TestingT, tempDir string) *instance {
+func defaultInstance(t assert.TestingT, tmpDir string) *instance {
 	limits, err := overrides.NewOverrides(overrides.Limits{})
 	assert.NoError(t, err, "unexpected error creating limits")
 	limiter := NewLimiter(limits, &ringCountMock{count: 1}, 1)
 
-	wal, err := wal.New(&wal.Config{
-		Filepath:        tempDir,
-		IndexDownsample: 2,
-		BloomFP:         .01,
-	})
-	assert.NoError(t, err, "unexpected error creating wal")
+	s, err := storage.NewStore(storage.Config{
+		Trace: tempodb.Config{
+			Backend: "local",
+			Local: &local.Config{
+				Path: tmpDir,
+			},
+			Block: &encoding.BlockConfig{
+				IndexDownsample: 2,
+				BloomFP:         .01,
+				Encoding:        backend.EncLZ4_1M,
+			},
+			WAL: &wal.Config{
+				Filepath: tmpDir,
+			},
+		},
+	}, log.NewNopLogger())
+	assert.NoError(t, err, "unexpected error creating store")
 
-	instance, err := newInstance("fake", limiter, wal)
+	instance, err := newInstance("fake", limiter, s)
 	assert.NoError(t, err, "unexpected error creating new instance")
 
 	return instance

modules/querier/querier.go

@@ -29,18 +29,6 @@ import (
 )
 
 var (
-	metricQueryReads = promauto.NewHistogramVec(prometheus.HistogramOpts{
-		Namespace: "tempo",
-		Name:      "query_reads",
-		Help:      "count of reads",
-		Buckets:   prometheus.ExponentialBuckets(1, 2, 10),
-	}, []string{"layer"})
-	metricQueryBytesRead = promauto.NewHistogramVec(prometheus.HistogramOpts{
-		Namespace: "tempo",
-		Name:      "query_bytes_read",
-		Help:      "bytes read",
-		Buckets:   prometheus.ExponentialBuckets(1024*1024, 2, 8),
-	}, []string{"layer"})
 	metricIngesterClients = promauto.NewGauge(prometheus.GaugeOpts{
 		Namespace: "tempo",
 		Name:      "querier_ingester_clients",
@@ -194,7 +182,7 @@ func (q *Querier) FindTraceByID(ctx context.Context, req *tempopb.TraceByIDReque
 
 	// if the ingester didn't have it check the store.
 	if completeTrace == nil {
-		foundBytes, metrics, err := q.store.Find(opentracing.ContextWithSpan(ctx, span), userID, req.TraceID, req.BlockStart, req.BlockEnd)
+		foundBytes, err := q.store.Find(opentracing.ContextWithSpan(ctx, span), userID, req.TraceID, req.BlockStart, req.BlockEnd)
 		if err != nil {
 			return nil, errors.Wrap(err, "error querying store in Querier.FindTraceByID")
 		}
@@ -207,12 +195,6 @@ func (q *Querier) FindTraceByID(ctx context.Context, req *tempopb.TraceByIDReque
 
 		span.LogFields(ot_log.String("msg", "found backend trace"), ot_log.Int("len", len(foundBytes)))
 		completeTrace = out
-		metricQueryReads.WithLabelValues("bloom").Observe(float64(metrics.BloomFilterReads.Load()))
-		metricQueryBytesRead.WithLabelValues("bloom").Observe(float64(metrics.BloomFilterBytesRead.Load()))
-		metricQueryReads.WithLabelValues("index").Observe(float64(metrics.IndexReads.Load()))
-		metricQueryBytesRead.WithLabelValues("index").Observe(float64(metrics.IndexBytesRead.Load()))
-		metricQueryReads.WithLabelValues("block").Observe(float64(metrics.BlockReads.Load()))
-		metricQueryBytesRead.WithLabelValues("block").Observe(float64(metrics.BlockBytesRead.Load()))
 	}
 
 	return &tempopb.TraceByIDResponse{