#306 - Compactor flush to backend based on buffer size

cmd/tempo/app/app.go

@@ -94,6 +94,11 @@ func (c *Config) CheckConfig() {
 		level.Warn(util.Logger).Log("msg", "compactor.compaction.compacted_block_timeout < storage.trace.blocklist_poll",
 			"explan", "Queriers and Compactors may attempt to read a block that no longer exists")
 	}
+
+	if c.StorageConfig.Trace.Backend == "s3" && c.Compactor.Compactor.FlushSizeBytes < 5242880 {
+		level.Warn(util.Logger).Log("msg", "c.Compactor.Compactor.FlushSizeBytes < 5242880",
+			"explan", "Compaction flush size should be 5MB or higher for S3 backend")
+	}
 }
 
 // App is the root datastructure.

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

@@ -29,6 +29,7 @@ compactor:
     max_compaction_objects: 1000000    # maximum size of compacted blocks
     block_retention: 1h
     compacted_block_retention: 10m
+    flush_size_bytes: 5242880
 
 storage:
   trace:

modules/compactor/config.go

@@ -17,10 +17,11 @@ type Config struct {
 	OverrideRingKey string                      `yaml:"override_ring_key"`
 }
 
-// RegisterFlags registers the flags.
+// RegisterFlagsAndApplyDefaults registers the flags.
 func (cfg *Config) RegisterFlagsAndApplyDefaults(prefix string, f *flag.FlagSet) {
 	cfg.Compactor = tempodb.CompactorConfig{
-		ChunkSizeBytes:          10485760, // 10 MiB
+		ChunkSizeBytes:          10 * 1024 * 1024, // 10 MiB
+		FlushSizeBytes:          30 * 1024 * 1024, // 30 MiB
 		CompactedBlockRetention: time.Hour,
 	}
 

tempodb/compactor.go

@@ -38,7 +38,6 @@ const (
 	inputBlocks  = 2
 	outputBlocks = 1
 
-	recordsPerBatch = 1000
 	compactionCycle = 30 * time.Second
 )
 
@@ -182,7 +181,7 @@ func (rw *readerWriter) compact(blockMetas []*encoding.BlockMeta, tenantID strin
 		lowestBookmark.clear()
 
 		// write partial block
-		if currentBlock.Length()%recordsPerBatch == 0 {
+		if currentBlock.CurrentBufferLength() >= int(rw.compactorCfg.FlushSizeBytes) {
 			tracker, err = appendBlock(rw, tracker, currentBlock)
 			if err != nil {
 				return errors.Wrap(err, "error writing partial block")

tempodb/config.go

@@ -28,6 +28,7 @@ type Config struct {
 
 type CompactorConfig struct {
 	ChunkSizeBytes          uint32        `yaml:"chunk_size_bytes"` // todo: do we need this?
+	FlushSizeBytes          uint32        `yaml:"flush_size_bytes"`
 	MaxCompactionRange      time.Duration `yaml:"compaction_window"`
 	MaxCompactionObjects    int           `yaml:"max_compaction_objects"`
 	BlockRetention          time.Duration `yaml:"block_retention"`

tempodb/wal/compactor_block.go

@@ -65,6 +65,10 @@ func (c *CompactorBlock) CurrentBuffer() []byte {
 	return c.appendBuffer.Bytes()
 }
 
+func (c *CompactorBlock) CurrentBufferLength() int {
+	return c.appendBuffer.Len()
+}
+
 func (c *CompactorBlock) ResetBuffer() {
 	c.appendBuffer.Reset()
 }