GRPC Streaming: Send diff instead of full response (#4062)

* send diff instead of full response Signed-off-by: Joe Elliott <[email protected]> * changelog Signed-off-by: Joe Elliott <[email protected]> * fix test Signed-off-by: Joe Elliott <[email protected]> --------- Signed-off-by: Joe Elliott <[email protected]>
grafana · Sep 6, 2024 · 7421936 · 7421936
1 parent 269caf0
commit 7421936
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 5 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -4,6 +4,7 @@
 * [CHANGE] **BREAKING CHANGE** The dynamic injection of X-Scope-OrgID header for metrics generator remote-writes is changed. If the header is aleady set in per-tenant overrides or global tempo configuration, then it is honored and not overwritten. [#4021](https://github.com/grafana/tempo/pull/4021) (@mdisibio)
 * [CHANGE] **BREAKING CHANGE** Migrate from OpenTracing to OpenTelemetry instrumentation. Removed the `use_otel_tracer` configuration option. Use the OpenTelemetry environment variables to configure the span exporter [#3646](https://github.com/grafana/tempo/pull/3646) (@andreasgerstmayr)
   To continue using the Jaeger exporter, use the following environment variable: `OTEL_TRACES_EXPORTER=jaeger`.
+* [CHANGE] No longer send the final diff in GRPC streaming. Instead we rely on the streamed intermediate results. [#4062](https://github.com/grafana/tempo/pull/4062) (@joe-elliott)
 * [FEATURE] Discarded span logging `log_discarded_spans` [#3957](https://github.com/grafana/tempo/issues/3957) (@dastrobu)
 * [FEATURE] TraceQL support for instrumentation scope [#3967](https://github.com/grafana/tempo/pull/3967) (@ie-pham)
 * [ENHANCEMENT] TraceQL: Attribute iterators collect matched array values [#3867](https://github.com/grafana/tempo/pull/3867) (@electron0zero, @stoewer)

diff --git a/modules/frontend/pipeline/collector_grpc.go b/modules/frontend/pipeline/collector_grpc.go
@@ -63,8 +63,8 @@ func (c GRPCCollector[T]) RoundTrip(req *http.Request) error {
 		return grpcError(err)
 	}
 
-	// send a final, complete response
-	resp, err := c.combiner.GRPCFinal()
+	// send the final diff if there is anything left
+	resp, err := c.combiner.GRPCDiff()
 	if err != nil {
 		return grpcError(err)
 	}

diff --git a/modules/frontend/search_handlers_test.go b/modules/frontend/search_handlers_test.go
@@ -31,6 +31,7 @@ import (
 	"github.com/grafana/tempo/pkg/cache"
 	"github.com/grafana/tempo/pkg/search"
 	"github.com/grafana/tempo/pkg/tempopb"
+	"github.com/grafana/tempo/pkg/traceql"
 	"github.com/grafana/tempo/pkg/util"
 	"github.com/grafana/tempo/pkg/util/test"
 	"github.com/grafana/tempo/tempodb"
@@ -402,15 +403,18 @@ func TestSearchLimitHonored(t *testing.T) {
 			}
 
 			// grpc
-			var actualResp *tempopb.SearchResponse
+			combiner := traceql.NewMetadataCombiner()
 			err = f.streamingSearch(tc.request, newMockStreamingServer(tenant, func(i int, sr *tempopb.SearchResponse) {
-				actualResp = sr
+				// combine
+				for _, t := range sr.Traces {
+					combiner.AddMetadata(t)
+				}
 			}))
 			if tc.badRequest {
 				require.Equal(t, status.Error(codes.InvalidArgument, "adjust limit: limit 20 exceeds max limit 15"), err)
 			} else {
 				require.NoError(t, err)
-				require.Len(t, actualResp.Traces, tc.expectedTraces)
+				require.Equal(t, combiner.Count(), tc.expectedTraces)
 			}
 		})
 	}