Method MetricsForLabelMatchersStream

pkg/ingester/ingester.go:2130–2173

(req *client.MetricsForLabelMatchersRequest, stream client.Ingester_MetricsForLabelMatchersStreamServer)

}

func (i *Ingester) MetricsForLabelMatchersStream(req *client.MetricsForLabelMatchersRequest, stream client.Ingester_MetricsForLabelMatchersStreamServer) (err error) {
	defer recoverIngester(i.logger, &err)

	ctx := stream.Context()
	userID, userErr := users.TenantID(ctx)
	if userErr != nil {
		return userErr
	}

	// Set pprof labels for profiling
	pprof.Do(ctx, pprof.Labels("user", userID), func(ctx context.Context) {
		result := &client.MetricsForLabelMatchersStreamResponse{}

		var cleanup func()
		cleanup, err = i.metricsForLabelMatchersCommon(ctx, req, func(l labels.Labels) error {
			result.Metric = append(result.Metric, &cortexpb.Metric{
				Labels: cortexpb.FromLabelsToLabelAdapters(l),
			})

			if len(result.Metric) >= metadataStreamBatchSize {
				err := client.SendMetricsForLabelMatchersStream(stream, result)
				if err != nil {
					return err
				}
				result.Metric = result.Metric[:0]
			}
			return nil
		})
		defer cleanup()
		if err != nil {
			return
		}

		// Send last batch
		if len(result.Metric) > 0 {
			err = client.SendMetricsForLabelMatchersStream(stream, result)
			if err != nil {
				return
			}
		}
	})

	return err
}

// metricsForLabelMatchersCommon returns all the metrics which match a set of matchers.
// this should be used by MetricsForLabelMatchers and MetricsForLabelMatchersStream.