Implement polling improvements

.github/workflows/ci.yml

@@ -123,6 +123,9 @@ jobs:
       - name: Test
         run: make test-e2e
 
+      - name: Poller
+        run: make test-integration-poller
+
   integration-tests-serverless:
     name: Test serverless integration e2e suite
     runs-on: ubuntu-latest

Makefile

@@ -129,9 +129,13 @@ test-e2e: tools docker-tempo docker-tempo-query
 test-e2e-serverless: tools docker-tempo docker-serverless
 	$(GOTEST) -v $(GOTEST_OPT) ./integration/e2e/serverless
 
+.PHONY: test-integration-poller
+test-integration-poller: tools
+	$(GOTEST) -v $(GOTEST_OPT) ./integration/poller
+
 # test-all/bench use a docker image so build it first to make sure we're up to date
 .PHONY: test-all
-test-all: test-with-cover test-e2e test-e2e-serverless
+test-all: test-with-cover test-e2e test-e2e-serverless test-integration-poller
 
 .PHONY: test-bench
 test-bench: tools docker-tempo

cmd/tempo-cli/cmd-analyse-blocks.go

@@ -21,7 +21,7 @@ func (cmd *analyseBlocksCmd) Run(ctx *globalOptions) error {
 	}
 
 	// TODO: Parallelize this
-	blocks, err := r.Blocks(context.Background(), cmd.TenantID)
+	blocks, _, err := r.Blocks(context.Background(), cmd.TenantID)
 	if err != nil {
 		return err
 	}

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

@@ -37,7 +37,7 @@ func (cmd *migrateTenantCmd) Run(opts *globalOptions) error {
 
 	// TODO create dest directory if it doesn't exist yet?
 
-	blocksDest, err := readerDest.Blocks(ctx, cmd.DestTenantID)
+	blocksDest, _, err := readerDest.Blocks(ctx, cmd.DestTenantID)
 	if err != nil {
 		return err
 	}

cmd/tempo-cli/cmd-query-blocks.go

@@ -84,13 +84,15 @@ func (cmd *queryBlocksCmd) Run(ctx *globalOptions) error {
 }
 
 func queryBucket(ctx context.Context, r backend.Reader, c backend.Compactor, tenantID string, traceID common.ID) ([]queryResults, error) {
-	blockIDs, err := r.Blocks(context.Background(), tenantID)
+	blockIDs, compactedBlockIDs, err := r.Blocks(context.Background(), tenantID)
 	if err != nil {
 		return nil, err
 	}
 
 	fmt.Println("total blocks to search: ", len(blockIDs))
 
+	blockIDs = append(blockIDs, compactedBlockIDs...)
+
 	// Load in parallel
 	wg := boundedwaitgroup.New(100)
 	resultsCh := make(chan queryResults, len(blockIDs))

cmd/tempo-cli/cmd-query-trace-summary.go

@@ -87,7 +87,7 @@ func sortServiceNames(nameFrequencies map[string]int) PairList {
 }
 
 func queryBucketForSummary(ctx context.Context, percentage float32, r backend.Reader, c backend.Compactor, tenantID string, traceID common.ID) (*TraceSummary, error) {
-	blockIDs, err := r.Blocks(context.Background(), tenantID)
+	blockIDs, _, err := r.Blocks(context.Background(), tenantID)
 	if err != nil {
 		return nil, err
 	}

cmd/tempo-cli/cmd-search.go

@@ -53,7 +53,7 @@ func (cmd *searchBlocksCmd) Run(opts *globalOptions) error {
 
 	ctx := context.Background()
 
-	blockIDs, err := r.Blocks(ctx, cmd.TenantID)
+	blockIDs, _, err := r.Blocks(ctx, cmd.TenantID)
 	if err != nil {
 		return err
 	}

cmd/tempo-cli/shared.go

@@ -56,11 +56,13 @@ type blockStats struct {
 }
 
 func loadBucket(r backend.Reader, c backend.Compactor, tenantID string, windowRange time.Duration, includeCompacted bool) ([]blockStats, error) {
-	blockIDs, err := r.Blocks(context.Background(), tenantID)
+	blockIDs, compactedBlockIDs, err := r.Blocks(context.Background(), tenantID)
 	if err != nil {
 		return nil, err
 	}
 
+	blockIDs = append(blockIDs, compactedBlockIDs...)
+
 	fmt.Println("total blocks: ", len(blockIDs))
 
 	// Load in parallel

cmd/tempo/app/modules.go

@@ -1,6 +1,7 @@
 package app
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"io"
@@ -295,7 +296,7 @@ func (t *App) initQuerier() (services.Service, error) {
 
 	// do not enable polling if this is the single binary. in that case the compactor will take care of polling
 	if t.cfg.Target == Querier {
-		t.store.EnablePolling(nil)
+		t.store.EnablePolling(context.Background(), nil)
 	}
 
 	ingesterRings := []ring.ReadRing{t.readRings[ringIngester]}
@@ -394,7 +395,7 @@ func (t *App) initQueryFrontend() (services.Service, error) {
 	t.Server.HTTP.Handle(addHTTPAPIPrefix(&t.cfg, api.PathSpanMetricsSummary), spanMetricsSummaryHandler)
 
 	// the query frontend needs to have knowledge of the blocks so it can shard search jobs
-	t.store.EnablePolling(nil)
+	t.store.EnablePolling(context.Background(), nil)
 
 	// http query echo endpoint
 	t.Server.HTTP.Handle(addHTTPAPIPrefix(&t.cfg, api.PathEcho), echoHandler())

docs/sources/tempo/configuration/_index.md

@@ -731,6 +731,11 @@ storage:
             # Example: "endpoint: s3.dualstack.us-east-2.amazonaws.com"
             [endpoint: <string>]
 
+            # The number of list calls to make in parallel to the backend per instance.
+            # Adjustments here will impact the polling time, as well as the number of Go routines.
+            # Default is 3
+            [list_blocks_concurrency: <int>]
+
             # optional.
             # By default the region is inferred from the endpoint,
             # but is required for some S3-compatible storage engines.

integration/e2e/backend/backend.go

@@ -17,7 +17,7 @@ import (
 
 const (
 	azuriteImage = "mcr.microsoft.com/azure-storage/azurite"
-	gcsImage     = "fsouza/fake-gcs-server:1.47.3"
+	gcsImage     = "fsouza/fake-gcs-server:1.47.6"
 )
 
 func parsePort(endpoint string) (int, error) {

integration/poller/config-azurite.yaml

@@ -0,0 +1,54 @@
+target: all
+stream_over_http_enabled: true
+
+server:
+  http_listen_port: 3200
+
+query_frontend:
+  search:
+    query_backend_after: 0 # setting these both to 0 will force all range searches to hit the backend
+    query_ingesters_until: 0
+
+distributor:
+  receivers:
+    jaeger:
+      protocols:
+        grpc:
+
+ingester:
+  lifecycler:
+    address: 127.0.0.1
+    ring:
+      kvstore:
+        store: inmemory
+      replication_factor: 1
+    final_sleep: 0s
+  trace_idle_period: 1s
+  max_block_duration: 5s
+  complete_block_timeout: 5s
+  flush_check_period: 1s
+
+storage:
+  trace:
+    blocklist_poll: 1s
+    backend: azure
+    azure:
+      container_name: tempo                    # how to store data in azure
+      endpoint_suffix: tempo_e2e-azurite:10000
+      storage_account_name: "devstoreaccount1"
+      storage_account_key: "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw=="
+    pool:
+      max_workers: 10
+      queue_depth: 100
+
+overrides:
+  user_configurable_overrides:
+    enabled: true
+    poll_interval: 10s
+    client:
+      backend: azure
+      azure:
+        container_name: tempo
+        endpoint_suffix: tempo_e2e-azurite:10000
+        storage_account_name: "devstoreaccount1"
+        storage_account_key: "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw=="

integration/poller/config-gcs.yaml

@@ -0,0 +1,54 @@
+target: all
+stream_over_http_enabled: true
+
+server:
+  http_listen_port: 3200
+
+query_frontend:
+  search:
+    query_ingesters_until: 0 # setting these both to 0 will force all range searches to hit the backend
+    query_backend_after: 0
+
+distributor:
+  receivers:
+    jaeger:
+      protocols:
+        grpc:
+
+ingester:
+  lifecycler:
+    address: 127.0.0.1
+    ring:
+      kvstore:
+        store: inmemory
+      replication_factor: 1
+    final_sleep: 0s
+  trace_idle_period: 1s
+  max_block_duration: 5s
+  complete_block_timeout: 5s
+  flush_check_period: 1s
+
+storage:
+  trace:
+    blocklist_poll: 1s
+    backend: gcs
+    gcs:
+      bucket_name: tempo
+      endpoint: https://tempo_e2e-gcs:4443/storage/v1/
+      insecure: true
+    pool:
+      max_workers: 10
+      queue_depth: 1000
+
+overrides:
+  user_configurable_overrides:
+    enabled: true
+    poll_interval: 10s
+    client:
+      backend: gcs
+      # fsouza/fake-gcs-server does not support versioning
+      confirm_versioning: false
+      gcs:
+        bucket_name: tempo
+        endpoint: https://tempo_e2e-gcs:4443/storage/v1/
+        insecure: true

integration/poller/config-s3.yaml

@@ -0,0 +1,57 @@
+target: all
+stream_over_http_enabled: true
+
+server:
+  http_listen_port: 3200
+
+query_frontend:
+  search:
+    query_backend_after: 0 # setting these both to 0 will force all range searches to hit the backend
+    query_ingesters_until: 0
+
+distributor:
+  receivers:
+    jaeger:
+      protocols:
+        grpc:
+
+ingester:
+  lifecycler:
+    address: 127.0.0.1
+    ring:
+      kvstore:
+        store: inmemory
+      replication_factor: 1
+    final_sleep: 0s
+  trace_idle_period: 1s
+  max_block_duration: 5s
+  complete_block_timeout: 5s
+  flush_check_period: 1s
+
+storage:
+  trace:
+    blocklist_poll: 1s
+    backend: s3
+    s3:
+      bucket: tempo
+      endpoint: tempo-integration-minio-9000:9000  # TODO: this is brittle, fix this eventually
+      access_key: Cheescake # TODO: use cortex_e2e.MinioAccessKey
+      secret_key: supersecret # TODO: use cortex_e2e.MinioSecretKey
+      insecure: true
+    pool:
+      max_workers: 10
+      queue_depth: 100
+
+overrides:
+  user_configurable_overrides:
+    enabled: true
+    poll_interval: 10s
+    client:
+      backend: s3
+      s3:
+        # TODO use separate bucket?
+        bucket: tempo
+        endpoint: tempo-integration-minio-9000:9000  # TODO: this is brittle, fix this eventually
+        access_key: Cheescake # TODO: use cortex_e2e.MinioAccessKey
+        secret_key: supersecret # TODO: use cortex_e2e.MinioSecretKey
+        insecure: true