Handle prefixes when listing blocks from S3 and GCS (#3466) (#3502)

* Handle prefixes when listing blocks from S3 fixes #3465 * Handle prefixes when listing blocks from GCS * Add test for prefixes when listing blocks from Azure * Update unit tests to check for actual block IDs instead of just length of the slices Cleanup unit tests * Further refine S3/GCS backend for ListBlocks Brings logic more in line with Azure object parsing. Also has the benefit of handling prefixes without a trailing slash. * Update poller integration test to exercise prefixes * Update e2e test to exercise prefixes * Fix format check error * Fix failing e2e tests * Remove unnecessary prefix permutations from e2e test * Remove unnecessary test config file copy * Ignore lint --------- Co-authored-by: Zach Leslie <[email protected]> (cherry picked from commit 8e6e7fe) Co-authored-by: Ben Foster <[email protected]>
grafana · Mar 19, 2024 · f5ab738 · f5ab738
1 parent ca6818a
commit f5ab738
Show file tree

Hide file tree

Showing 12 changed files with 646 additions and 194 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,6 +1,15 @@
 ## main / unreleased
 
 * [BUGFIX] Fix some instances where spanmetrics histograms could be inconsistent [#3412](https://github.com/grafana/tempo/pull/3412) (@mdisibio)
+* [ENHANCEMENT] Add string interning to TraceQL queries [#3411](https://github.com/grafana/tempo/pull/3411) (@mapno)
+* [ENHANCEMENT] Add new (unsafe) query hints for metrics queries [#3396](https://github.com/grafana/tempo/pull/3396) (@mdisibio)
+* [BUGFIX] Fix metrics query results when filtering and rating on the same attribute [#3428](https://github.com/grafana/tempo/issues/3428) (@mdisibio)
+* [BUGFIX] Fix metrics query results when series contain empty strings or nil values [#3429](https://github.com/grafana/tempo/issues/3429) (@mdisibio)
+* [BUGFIX] Fix metrics query duration check, add per-tenant override for max metrics query duration [#3479](https://github.com/grafana/tempo/issues/3479) (@mdisibio)
+* [BUGFIX] Return unfiltered results when a bad TraceQL query is provided in autocomplete. [#3426](https://github.com/grafana/tempo/pull/3426) (@mapno)
+* [BUGFIX] Correctly handle 429s in GRPC search streaming. [#3469](https://github.com/grafana/tempo/pull/3469) (@joe-ellitot)
+* [BUGFIX] Correctly cancel GRPC and HTTP contexts in the frontend to prevent having to rely on http write timeout. [#3443](https://github.com/grafana/tempo/pull/3443) (@joe-elliott)
+* [BUGFIX] Fix compaction/retention in AWS S3 and GCS when a prefix is configured. [#3465](https://github.com/grafana/tempo/issues/3465) (@bpfoster)
 
 ## v2.4.0-rc.0
 

diff --git a/integration/e2e/config-all-in-one-azurite.yaml b/integration/e2e/config-all-in-one-azurite.yaml
@@ -37,6 +37,7 @@ storage:
  endpoint_suffix: tempo_e2e-azurite:10000
  storage_account_name: "devstoreaccount1"
  storage_account_key: "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw=="
+ prefix: {{ .Prefix }}
  pool:
  max_workers: 10
  queue_depth: 100

diff --git a/integration/e2e/config-all-in-one-gcs.yaml b/integration/e2e/config-all-in-one-gcs.yaml
@@ -36,6 +36,7 @@ storage:
  bucket_name: tempo
  endpoint: https://tempo_e2e-gcs:4443/storage/v1/
  insecure: true
+ prefix: {{ .Prefix }}
  pool:
  max_workers: 10
  queue_depth: 1000

diff --git a/integration/e2e/config-all-in-one-s3.yaml b/integration/e2e/config-all-in-one-s3.yaml
@@ -38,6 +38,7 @@ storage:
  access_key: Cheescake # TODO: use cortex_e2e.MinioAccessKey
  secret_key: supersecret # TODO: use cortex_e2e.MinioSecretKey
  insecure: true
+ prefix: {{ .Prefix }}
  pool:
  max_workers: 10
  queue_depth: 100

diff --git a/integration/e2e/e2e_test.go b/integration/e2e/e2e_test.go
@@ -58,83 +58,103 @@ func TestAllInOne(t *testing.T) {
  },
  }
 
+ storageBackendTestPermutations := []struct {
+ name string
+ prefix string
+ }{
+ {
+ name: "no-prefix",
+ },
+ {
+ name: "prefix",
+ prefix: "a/b/c/",
+ },
+ }
+
  for _, tc := range testBackends {
- t.Run(tc.name, func(t *testing.T) {
- s, err := e2e.NewScenario("tempo_e2e")
- require.NoError(t, err)
- defer s.Close()
+ for _, pc := range storageBackendTestPermutations {
+ t.Run(tc.name+"-"+pc.name, func(t *testing.T) {
+ s, err := e2e.NewScenario("tempo_e2e")
+ require.NoError(t, err)
+ defer s.Close()
 
- // set up the backend
- cfg := app.Config{}
- buff, err := os.ReadFile(tc.configFile)
- require.NoError(t, err)
- err = yaml.UnmarshalStrict(buff, &cfg)
- require.NoError(t, err)
- _, err = backend.New(s, cfg)
- require.NoError(t, err)
+ // copy config template to shared directory and expand template variables
+ tmplConfig := map[string]any{"Prefix": pc.prefix}
+ configFile, err := util.CopyTemplateToSharedDir(s, tc.configFile, "config.yaml", tmplConfig)
+ require.NoError(t, err)
 
- require.NoError(t, util.CopyFileToSharedDir(s, tc.configFile, "config.yaml"))
- tempo := util.NewTempoAllInOne()
- require.NoError(t, s.StartAndWaitReady(tempo))
+ // set up the backend
+ cfg := app.Config{}
+ buff, err := os.ReadFile(configFile)
+ require.NoError(t, err)
+ err = yaml.UnmarshalStrict(buff, &cfg)
+ require.NoError(t, err)
+ _, err = backend.New(s, cfg)
+ require.NoError(t, err)
 
- // Get port for the Jaeger gRPC receiver endpoint
- c, err := util.NewJaegerGRPCClient(tempo.Endpoint(14250))
- require.NoError(t, err)
- require.NotNil(t, c)
+ tempo := util.NewTempoAllInOne()
+ require.NoError(t, s.StartAndWaitReady(tempo))
 
- info := tempoUtil.NewTraceInfo(time.Now(), "")
- require.NoError(t, info.EmitAllBatches(c))
+ // Get port for the Jaeger gRPC receiver endpoint
+ c, err := util.NewJaegerGRPCClient(tempo.Endpoint(14250))
+ require.NoError(t, err)
+ require.NotNil(t, c)
 
- expected, err := info.ConstructTraceFromEpoch()
- require.NoError(t, err)
+  info := tempoUtil.NewTraceInfo(time.Now(), "")
+  require.NoError(t, info.EmitAllBatches(c))
 
- // test metrics
- require.NoError(t, tempo.WaitSumMetrics(e2e.Equals(spanCount(expected)), "tempo_distributor_spans_received_total"))
+  expected, err := info.ConstructTraceFromEpoch()
+  require.NoError(t, err)
 
- // test echo
- assertEcho(t, "http://"+tempo.Endpoint(3200)+"/api/echo")
+  // test metrics
+  require.NoError(t, tempo.WaitSumMetrics(e2e.Equals(spanCount(expected)), "tempo_distributor_spans_received_total"))
 
- apiClient := httpclient.New("http://"+tempo.Endpoint(3200), "")
+ // test echo
+ // nolint:goconst
+ assertEcho(t, "http://"+tempo.Endpoint(3200)+"/api/echo")
 
- // query an in-memory trace
- queryAndAssertTrace(t, apiClient, info)
+ apiClient := httpclient.New("http://"+tempo.Endpoint(3200), "")
 
- // wait trace_idle_time and ensure trace is created in ingester
- require.NoError(t, tempo.WaitSumMetricsWithOptions(e2e.Less(3), []string{"tempo_ingester_traces_created_total"}, e2e.WaitMissingMetrics))
+  // query an in-memory trace
+  queryAndAssertTrace(t, apiClient, info)
 
- // flush trace to backend
- callFlush(t, tempo)
+  // wait trace_idle_time and ensure trace is created in ingester
+  require.NoError(t, tempo.WaitSumMetricsWithOptions(e2e.Less(3), []string{"tempo_ingester_traces_created_total"}, e2e.WaitMissingMetrics))
 
- // search for trace in backend
- util.SearchAndAssertTrace(t, apiClient, info)
- util.SearchTraceQLAndAssertTrace(t, apiClient, info)
+ // flush trace to backend
+ callFlush(t, tempo)
 
- // sleep
- time.Sleep(10 * time.Second)
+ // search for trace in backend
+ util.SearchAndAssertTrace(t, apiClient, info)
+ util.SearchTraceQLAndAssertTrace(t, apiClient, info)
 
- // force clear completed block
- callFlush(t, tempo)
+  // sleep
+  time.Sleep(10 * time.Second)
 
- // test metrics
- require.NoError(t, tempo.WaitSumMetrics(e2e.Equals(1), "tempo_ingester_blocks_flushed_total"))
- require.NoError(t, tempo.WaitSumMetricsWithOptions(e2e.Equals(1), []string{"tempodb_blocklist_length"}, e2e.WaitMissingMetrics))
- require.NoError(t, tempo.WaitSumMetrics(e2e.Equals(3), "tempo_query_frontend_queries_total"))
+ // force clear completed block
+ callFlush(t, tempo)
 
- // query trace - should fetch from backend
- queryAndAssertTrace(t, apiClient, info)
+ // test metrics
+ require.NoError(t, tempo.WaitSumMetrics(e2e.Equals(1), "tempo_ingester_blocks_flushed_total"))
+ require.NoError(t, tempo.WaitSumMetricsWithOptions(e2e.Equals(1), []string{"tempodb_blocklist_length"}, e2e.WaitMissingMetrics))
+ require.NoError(t, tempo.WaitSumMetrics(e2e.Equals(3), "tempo_query_frontend_queries_total"))
 
- // search the backend. this works b/c we're passing a start/end AND setting query ingesters within min/max to 0
- now := time.Now()
- util.SearchAndAssertTraceBackend(t, apiClient, info, now.Add(-20*time.Minute).Unix(), now.Unix())
+ // query trace - should fetch from backend
+ queryAndAssertTrace(t, apiClient, info)
 
- util.SearchAndAsserTagsBackend(t, apiClient, now.Add(-20*time.Minute).Unix(), now.Unix())
+ // search the backend. this works b/c we're passing a start/end AND setting query ingesters within min/max to 0
+ now := time.Now()
+ util.SearchAndAssertTraceBackend(t, apiClient, info, now.Add(-20*time.Minute).Unix(), now.Unix())
 
- // find the trace with streaming. using the http server b/c that's what Grafana will do
- grpcClient, err := util.NewSearchGRPCClient(context.Background(), tempo.Endpoint(3200))
- require.NoError(t, err)
+ util.SearchAndAsserTagsBackend(t, apiClient, now.Add(-20*time.Minute).Unix(), now.Unix())
 
- util.SearchStreamAndAssertTrace(t, context.Background(), grpcClient, info, now.Add(-20*time.Minute).Unix(), now.Unix())
- })
+ // find the trace with streaming. using the http server b/c that's what Grafana will do
+ grpcClient, err := util.NewSearchGRPCClient(context.Background(), tempo.Endpoint(3200))
+ require.NoError(t, err)
+
+ util.SearchStreamAndAssertTrace(t, context.Background(), grpcClient, info, now.Add(-20*time.Minute).Unix(), now.Unix())
+ })
+ }
  }
 }
 
@@ -317,16 +337,20 @@ func TestShutdownDelay(t *testing.T) {
  require.NoError(t, err)
  defer s.Close()
 
+ // copy config template to shared directory and expand template variables
+ tmplConfig := map[string]any{"Prefix": ""}
+ configFile, err := util.CopyTemplateToSharedDir(s, configAllInOneS3, "config.yaml", tmplConfig)
+ require.NoError(t, err)
+
  // set up the backend
  cfg := app.Config{}
- buff, err := os.ReadFile(configAllInOneS3)
+ buff, err := os.ReadFile(configFile)
  require.NoError(t, err)
  err = yaml.UnmarshalStrict(buff, &cfg)
  require.NoError(t, err)
  _, err = backend.New(s, cfg)
  require.NoError(t, err)
 
- require.NoError(t, util.CopyFileToSharedDir(s, configAllInOneS3, "config.yaml"))
  tempo := util.NewTempoAllInOne("-shutdown-delay=5s")
 
  // this line tests confirms that the readiness flag is up

diff --git a/integration/e2e/overrides_test.go b/integration/e2e/overrides_test.go
@@ -49,16 +49,20 @@ func TestOverrides(t *testing.T) {
  require.NoError(t, err)
  defer s.Close()
 
+ // copy config template to shared directory and expand template variables
+ tmplConfig := map[string]any{"Prefix": ""}
+ configFile, err := util.CopyTemplateToSharedDir(s, tc.configFile, "config.yaml", tmplConfig)
+ require.NoError(t, err)
+
  // set up the backend
  cfg := app.Config{}
- buff, err := os.ReadFile(tc.configFile)
+ buff, err := os.ReadFile(configFile)
  require.NoError(t, err)
  err = yaml.UnmarshalStrict(buff, &cfg)
  require.NoError(t, err)
  _, err = backend.New(s, cfg)
  require.NoError(t, err)
 
- require.NoError(t, util.CopyFileToSharedDir(s, tc.configFile, "config.yaml"))
  tempo := util.NewTempoAllInOne()
  require.NoError(t, s.StartAndWaitReady(tempo))