Function TestTotalJobsIncludesIngester

modules/frontend/search_sharder_test.go:808–873

(t *testing.T)

}

func TestTotalJobsIncludesIngester(t *testing.T) {
	next := pipeline.AsyncRoundTripperFunc[combiner.PipelineResponse](func(_ pipeline.Request) (pipeline.Responses[combiner.PipelineResponse], error) {
		resString, err := (&jsonpb.Marshaler{}).MarshalToString(&tempopb.SearchResponse{
			Metrics: &tempopb.SearchMetrics{},
		})
		require.NoError(t, err)

		return pipeline.NewHTTPToAsyncResponse(&http.Response{
			Body:       io.NopCloser(strings.NewReader(resString)),
			StatusCode: 200,
		}), nil
	})

	o, err := overrides.NewOverrides(overrides.Config{}, nil, prometheus.DefaultRegisterer)
	require.NoError(t, err)

	blockTime := time.Now().Add(-10 * time.Minute).Unix()

	sharder := newAsyncSearchSharder(&mockReader{
		metas: []*backend.BlockMeta{ // one block with 2 records that are each the target bytes per request will force 2 sub queries
			{
				StartTime:    time.Unix(blockTime, 0),
				EndTime:      time.Unix(blockTime, 0),
				Size_:        defaultTargetBytesPerRequest * 2,
				TotalRecords: 2,
				BlockID:      backend.MustParse("00000000-0000-0000-0000-000000000000"),
			},
		},
	}, o, SearchSharderConfig{
		QueryBackendAfter:     5 * time.Minute,
		ConcurrentRequests:    1, // 1 concurrent request to force order
		TargetBytesPerRequest: defaultTargetBytesPerRequest,
		MostRecentShards:      defaultMostRecentShards,
		IngesterShards:        1,
	}, nil, newJobsPerQueryHistogram(), log.NewNopLogger())
	testRT := sharder.Wrap(next)

	// query range straddles the QueryBackendAfter boundary so both backend and ingester are queried
	path := fmt.Sprintf("/?start=%d&end=%d", blockTime-1, time.Now().Unix())
	req := httptest.NewRequest("GET", path, nil)
	ctx := req.Context()
	ctx = user.InjectOrgID(ctx, "blerg")
	req = req.WithContext(ctx)

	resps, err := testRT.RoundTrip(pipeline.NewHTTPRequest(req))
	require.NoError(t, err)
	// find a response with total jobs > . this is the metadata response

	totalJobs := 0
	for {
		res, done, err := resps.Next(context.Background())

		if res.IsMetadata() {
			searchJobResponse := res.(*combiner.SearchJobResponse)
			totalJobs += searchJobResponse.TotalJobs

			break
		}