Method Process

modules/frontend/v1/frontend.go:216–305

Process allows backends to pull requests from the frontend.

(server frontendv1pb.Frontend_ProcessServer)

// Process allows backends to pull requests from the frontend.
func (f *Frontend) Process(server frontendv1pb.Frontend_ProcessServer) error {
	_, querierFeatures, err := getQuerierInfo(server)
	if err != nil {
		return err
	}

	f.connectedQuerierWorkers.Add(1)
	defer f.connectedQuerierWorkers.Add(-1)

	lastUserIndex := queue.FirstUser()

	reqBatch := &requestBatch{}
	batchSize := 1
	if querierSupportsBatching(querierFeatures) {
		batchSize = f.cfg.MaxBatchSize
	}
	for {
		reqSlice := make([]queue.Request, batchSize)
		reqSlice, idx, err := f.requestQueue.GetNextRequestForQuerier(server.Context(), lastUserIndex, reqSlice)
		if err != nil {
			return err
		}
		lastUserIndex = idx

		reqBatch.clear()
		for _, reqWrapper := range reqSlice {
			req := reqWrapper.(*request)

			f.queueDuration.Observe(time.Since(req.enqueueTime).Seconds())
			req.queueSpan.End()

			// only add if not expired
			if req.OriginalContext().Err() != nil {
				continue
			}

			err = reqBatch.add(req)
			if err != nil {
				return fmt.Errorf("unexpected error adding request to batch: %w", err)
			}
		}

		// if all requests are expired then continue requesting jobs for this user. this nicely
		// drains a large expired query for a tenant and allows them to execute a real query
		if reqBatch.len() == 0 {
			lastUserIndex = lastUserIndex.ReuseLastUser()
			continue
		}

		f.actualBatchSize.Observe(float64(reqBatch.len()))

		// Handle the stream sending & receiving on a goroutine so we can
		// monitoring the contexts in a select and cancel things appropriately.
		resps := make(chan *frontendv1pb.ClientToFrontend, 1)
		errs := make(chan error, 1)
		go func() {
			// todo: we are still sending the old Type_HTTP_REQUEST for backwards compat
			// with queriers that don't support the new Type_HTTP_REQUEST_BATCH. this feature