Prune in-memory blocks from missing tenants

CHANGELOG.md

@@ -7,3 +7,4 @@
 * [ENHANCEMENT] Add command line flags for s3 credentials. [#308](https://github.com/grafana/tempo/pull/308)
 * [BUGFIX] Increase Prometheus `notfound` metric on tempo-vulture. [#301](https://github.com/grafana/tempo/pull/301)
 * [BUGFIX] Return 404 if searching for a tenant id that does not exist in the backend. [#321](https://github.com/grafana/tempo/pull/321)
+* [BUGFIX] Prune in-memory blocks from missing tenants. [#314](https://github.com/grafana/tempo/pull/314)

tempodb/tempodb.go

@@ -383,6 +383,13 @@ func (rw *readerWriter) pollBlocklist() {
 			metricBlocklistErrors.WithLabelValues(tenantID).Inc()
 			level.Error(rw.logger).Log("msg", "error polling blocklist", "tenantID", tenantID, "err", err)
 		}
+		if len(blockIDs) == 0 {
+			rw.blockListsMtx.Lock()
+			delete(rw.blockLists, tenantID)
+			delete(rw.compactedBlockLists, tenantID)
+			rw.blockListsMtx.Unlock()
+			level.Info(rw.logger).Log("msg", "deleted in-memory blocklists", "tenantID", tenantID)
+		}
 
 		interfaceSlice := make([]interface{}, 0, len(blockIDs))
 		for _, id := range blockIDs {

tempodb/tempodb_test.go

@@ -175,6 +175,58 @@ func TestRetention(t *testing.T) {
 	checkBlocklists(t, blockID, 0, 0, rw)
 }
 
+func TestBlockCleanup(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"),
+		},
+		WAL: &wal.Config{
+			Filepath:        path.Join(tempDir, "wal"),
+			IndexDownsample: 17,
+			BloomFP:         .01,
+		},
+		BlocklistPoll: 0,
+	}, log.NewNopLogger())
+	assert.NoError(t, err)
+
+	c.EnableCompaction(&CompactorConfig{
+		ChunkSizeBytes:          10,
+		MaxCompactionRange:      time.Hour,
+		BlockRetention:          0,
+		CompactedBlockRetention: 0,
+	}, &mockSharder{})
+
+	blockID := uuid.New()
+
+	wal := w.WAL()
+	assert.NoError(t, err)
+
+	head, err := wal.NewBlock(blockID, testTenantID)
+	assert.NoError(t, err)
+
+	complete, err := head.Complete(wal, &mockSharder{})
+	assert.NoError(t, err)
+	blockID = complete.BlockMeta().BlockID
+
+	err = w.WriteBlock(context.Background(), complete)
+	assert.NoError(t, err)
+
+	rw := r.(*readerWriter)
+
+	os.RemoveAll(tempDir + "/traces/" + testTenantID)
+
+	// poll
+	rw.pollBlocklist()
+
+	assert.Len(t, rw.blockLists[testTenantID], 0)
+	assert.Len(t, rw.compactedBlockLists[testTenantID], 0)
+}
+
 func checkBlocklists(t *testing.T, expectedID uuid.UUID, expectedB int, expectedCB int, rw *readerWriter) {
 	rw.pollBlocklist()