Query compacted blocks that fall within 2 * BlocklistPoll

CHANGELOG.md

@@ -8,6 +8,7 @@
 * [BUGFIX] Fixes error where Dell ECS cannot list objects. [#561](https://github.com/grafana/tempo/pull/561)
 * [BUGFIX] Fixes listing blocks in S3 when the list is truncated. [#567](https://github.com/grafana/tempo/pull/567)
 * [BUGFIX] Fixes where ingester may leave file open [#570](https://github.com/grafana/tempo/pull/570)
+* [BUGFIX] Fixes a bug where some blocks were not searched due to query sharding and randomness in blocklist poll. [#583](https://github.com/grafana/tempo/pull/583)
 
 ## v0.6.0
 

docs/tempo/website/configuration/_index.md

@@ -78,6 +78,9 @@ querier:
         frontend_address: query-frontend-discovery.default.svc.cluster.local:9095   # the address of the query frontend to connect to, and process queries
 ```
 
+The Querier also queries compacted blocks that fall within (2 * BlocklistPoll) where the value of Blocklist poll duration
+is defined in the storage section below.
+
 ## Compactor
 See [here](https://github.com/grafana/tempo/blob/master/modules/compactor/config.go) for all configuration options.
 

tempodb/tempodb.go

@@ -242,6 +242,15 @@ func (rw *readerWriter) Find(ctx context.Context, tenantID string, id common.ID,
 			copiedBlocklist = append(copiedBlocklist, b)
 		}
 	}
+
+	// if block is compacted within lookback period, include it in search
+	lookback := time.Now().Add(-(2 * rw.cfg.BlocklistPoll))
+	compactedBlocklist := rw.compactedBlockLists[tenantID]
+	for _, c := range compactedBlocklist {
+		if c.CompactedTime.After(lookback) {
+			copiedBlocklist = append(copiedBlocklist, &c.BlockMeta)
+		}
+	}
 	rw.blockListsMtx.Unlock()
 
 	// deliberately placed outside the blocklist mtx unlock

tempodb/tempodb_test.go

@@ -789,3 +789,111 @@ func TestIncludeBlock(t *testing.T) {
 		})
 	}
 }
+
+func TestSearchCompactedBlocks(t *testing.T) {
+	tempDir, err := ioutil.TempDir("/tmp", "")
+	defer os.RemoveAll(tempDir)
+	assert.NoError(t, err, "unexpected error creating temp dir")
+
+	r, w, c, err := New(&Config{
+		Backend: "local",
+		Local: &local.Config{
+			Path: path.Join(tempDir, "traces"),
+		},
+		Block: &encoding.BlockConfig{
+			IndexDownsampleBytes: 17,
+			BloomFP:              .01,
+			Encoding:             backend.EncLZ4_256k,
+		},
+		WAL: &wal.Config{
+			Filepath: path.Join(tempDir, "wal"),
+		},
+		BlocklistPoll: time.Minute,
+	}, log.NewNopLogger())
+	assert.NoError(t, err)
+
+	c.EnableCompaction(&CompactorConfig{
+		ChunkSizeBytes:          10,
+		MaxCompactionRange:      time.Hour,
+		BlockRetention:          0,
+		CompactedBlockRetention: 0,
+	}, &mockSharder{}, &mockOverrides{})
+
+	wal := w.WAL()
+	assert.NoError(t, err)
+
+	head, err := wal.NewBlock(uuid.New(), testTenantID)
+	assert.NoError(t, err)
+
+	// write
+	numMsgs := 10
+	reqs := make([]*tempopb.PushRequest, 0, numMsgs)
+	ids := make([][]byte, 0, numMsgs)
+	for i := 0; i < numMsgs; i++ {
+		id := make([]byte, 16)
+		rand.Read(id)
+		req := test.MakeRequest(rand.Int()%1000, id)
+		reqs = append(reqs, req)
+		ids = append(ids, id)
+
+		bReq, err := proto.Marshal(req)
+		assert.NoError(t, err)
+		err = head.Write(id, bReq)
+		assert.NoError(t, err, "unexpected error writing req")
+	}
+
+	complete, err := w.CompleteBlock(head, &mockSharder{})
+	assert.NoError(t, err)
+
+	blockID := complete.BlockMeta().BlockID.String()
+
+	err = w.WriteBlock(context.Background(), complete)
+	assert.NoError(t, err)
+
+	rw := r.(*readerWriter)
+
+	// poll
+	rw.pollBlocklist()
+
+	// read
+	for i, id := range ids {
+		bFound, err := r.Find(context.Background(), testTenantID, id, blockID, blockID)
+		assert.NoError(t, err)
+
+		out := &tempopb.PushRequest{}
+		err = proto.Unmarshal(bFound[0], out)
+		assert.NoError(t, err)
+
+		assert.True(t, proto.Equal(out, reqs[i]))
+	}
+
+	// compact
+	var blockMetas []*backend.BlockMeta
+	blockMetas = append(blockMetas, complete.BlockMeta())
+	assert.NoError(t, rw.compact(blockMetas, testTenantID))
+
+	// poll
+	rw.pollBlocklist()
+
+	// make sure the block is compacted
+	compactedBlocks, ok := rw.compactedBlockLists[testTenantID]
+	require.True(t, ok)
+	require.Len(t, compactedBlocks, 1)
+	assert.Equal(t, compactedBlocks[0].BlockID.String(), blockID)
+	blocks, ok := rw.blockLists[testTenantID]
+	require.True(t, ok)
+	require.Len(t, blocks, 1)
+	assert.NotEqual(t, blocks[0].BlockID.String(), blockID)
+
+	// find should succeed with completely different guid ranges
+	for i, id := range ids {
+		bFound, err := r.Find(context.Background(), testTenantID, id, BlockIDMin, BlockIDMin)
+		assert.NoError(t, err)
+
+		out := &tempopb.PushRequest{}
+		err = proto.Unmarshal(bFound[0], out)
+		assert.NoError(t, err)
+
+		assert.True(t, proto.Equal(out, reqs[i]))
+	}
+}