Method process

modules/querier/worker/frontend_processor.go:95–147

process loops processing requests on an established stream.

(c frontendv1pb.Frontend_ProcessClient)

// process loops processing requests on an established stream.
func (fp *frontendProcessor) process(c frontendv1pb.Frontend_ProcessClient) error {
	// Build a child context so we can cancel a query when the stream is closed.
	ctx, cancel := context.WithCancel(c.Context())
	defer cancel()

	for {
		request, err := c.Recv()
		if err != nil {
			return err
		}

		switch request.Type {
		case frontendv1pb.Type_HTTP_REQUEST:
			// Handle the request on a "background" goroutine, so we go back to
			// blocking on c.Recv().  This allows us to detect the stream closing
			// and cancel the query.  We don't actually handle queries in parallel
			// here, as we're running in lock step with the server - each Recv is
			// paired with a Send.
			go func() {
				resp := fp.runRequest(ctx, request.HttpRequest)
				err := fp.handleSendError(c.Send(&frontendv1pb.ClientToFrontend{
					HttpResponse: resp,
				}))
				if err != nil {
					level.Error(fp.log).Log("msg", "error running requests", "err", err)
				}
			}()

		case frontendv1pb.Type_GET_ID:
			err := fp.handleSendError(c.Send(&frontendv1pb.ClientToFrontend{
				ClientID: fp.querierID,
				Features: int32(frontendv1pb.Feature_REQUEST_BATCHING),
			}))
			if err != nil {
				return err
			}

		case frontendv1pb.Type_HTTP_REQUEST_BATCH:
			go func() {
				resp := fp.runRequests(ctx, request.HttpRequestBatch)
				err := fp.handleSendError(c.Send(&frontendv1pb.ClientToFrontend{
					HttpResponseBatch: resp,
				}))
				if err != nil {
					level.Error(fp.log).Log("msg", "error running  batched requests", "err", err)
				}
			}()

		default:
			return fmt.Errorf("unknown request type: %v", request.Type)
		}
	}
}

func (fp *frontendProcessor) runRequests(ctx context.Context, requests []*httpgrpc.HTTPRequest) []*httpgrpc.HTTPResponse {
	wg := sync.WaitGroup{}