metrics.go support for metadata queries(labels and series)

pkg/logql/engine.go

@@ -202,7 +202,7 @@ func (q *query) Exec(ctx context.Context) (logqlmodel.Result, error) {
 	}
 
 	if q.record {
-		RecordMetrics(ctx, q.logger, q.params, status, statResult, data)
+		RecordRangeAndInstantQueryMetrics(ctx, q.logger, q.params, status, statResult, data)
 	}
 
 	return logqlmodel.Result{

pkg/logql/metrics.go

@@ -3,6 +3,7 @@ package logql
 import (
 	"context"
 	"strings"
+	"time"
 
 	"github.com/dustin/go-humanize"
 	"github.com/go-kit/log"
@@ -13,6 +14,7 @@ import (
 
 	"github.com/grafana/loki/pkg/logql/syntax"
 	"github.com/grafana/loki/pkg/logqlmodel"
+	"github.com/grafana/loki/pkg/logqlmodel/stats"
 	logql_stats "github.com/grafana/loki/pkg/logqlmodel/stats"
 	"github.com/grafana/loki/pkg/usagestats"
 	"github.com/grafana/loki/pkg/util/httpreq"
@@ -23,6 +25,8 @@ const (
 	QueryTypeMetric  = "metric"
 	QueryTypeFilter  = "filter"
 	QueryTypeLimited = "limited"
+	QueryTypeLabels  = "labels"
+	QueryTypeSeries  = "series"
 
 	latencyTypeSlow = "slow"
 	latencyTypeFast = "fast"
@@ -74,7 +78,7 @@ var (
 	linePerSecondLogUsage    = usagestats.NewStatistics("query_log_lines_per_second")
 )
 
-func RecordMetrics(ctx context.Context, log log.Logger, p Params, status string, stats logql_stats.Result, result promql_parser.Value) {
+func RecordRangeAndInstantQueryMetrics(ctx context.Context, log log.Logger, p Params, status string, stats logql_stats.Result, result promql_parser.Value) {
 	var (
 		logger        = util_log.WithContext(ctx, log)
 		rt            = string(GetRangeType(p))
@@ -138,6 +142,95 @@ func RecordMetrics(ctx context.Context, log log.Logger, p Params, status string,
 	recordUsageStats(queryType, stats)
 }
 
+func RecordLabelQueryMetrics(
+	ctx context.Context,
+	start, end time.Time,
+	label, status string,
+	stats stats.Result,
+) {
+	var (
+		logger      = util_log.WithContext(ctx, util_log.Logger)
+		latencyType = latencyTypeFast
+		queryType   = QueryTypeLabels
+	)
+
+	// Tag throughput metric by latency type based on a threshold.
+	// Latency below the threshold is fast, above is slow.
+	if stats.Summary.ExecTime > slowQueryThresholdSecond {
+		latencyType = latencyTypeSlow
+	}
+
+	// we also log queries, useful for troubleshooting slow queries.
+	level.Info(logger).Log(
+		"latency", latencyType,
+		"query_type", queryType,
+		"length", end.Sub(start),
+		"duration", time.Duration(int64(stats.Summary.ExecTime*float64(time.Second))),
+		"status", status,
+		"label", label,
+		"throughput", strings.Replace(humanize.Bytes(uint64(stats.Summary.BytesProcessedPerSecond)), " ", "", 1),
+		"total_bytes", strings.Replace(humanize.Bytes(uint64(stats.Summary.TotalBytesProcessed)), " ", "", 1),
+	)
+
+	bytesPerSecond.WithLabelValues(status, queryType, "", latencyType).
+		Observe(float64(stats.Summary.BytesProcessedPerSecond))
+	execLatency.WithLabelValues(status, queryType, "").
+		Observe(stats.Summary.ExecTime)
+	chunkDownloadLatency.WithLabelValues(status, queryType, "").
+		Observe(stats.ChunksDownloadTime().Seconds())
+	duplicatesTotal.Add(float64(stats.TotalDuplicates()))
+	chunkDownloadedTotal.WithLabelValues(status, queryType, "").
+		Add(float64(stats.TotalChunksDownloaded()))
+	ingesterLineTotal.Add(float64(stats.Ingester.TotalLinesSent))
+
+	// TODO(kavi): Do we need recordUsageStats for labels query?
+}
+
+func RecordSeriesQueryMetrics(
+	ctx context.Context,
+	start, end time.Time,
+	match []string,
+	status string,
+	stats stats.Result,
+) {
+	var (
+		logger      = util_log.WithContext(ctx, util_log.Logger)
+		latencyType = latencyTypeFast
+		queryType   = QueryTypeSeries
+	)
+
+	// Tag throughput metric by latency type based on a threshold.
+	// Latency below the threshold is fast, above is slow.
+	if stats.Summary.ExecTime > slowQueryThresholdSecond {
+		latencyType = latencyTypeSlow
+	}
+
+	// we also log queries, useful for troubleshooting slow queries.
+	level.Info(logger).Log(
+		"latency", latencyType,
+		"query_type", queryType,
+		"length", end.Sub(start),
+		"duration", time.Duration(int64(stats.Summary.ExecTime*float64(time.Second))),
+		"status", status,
+		"match", strings.Join(match, ":"), // not using comma (,) as separator as matcher may already have comma (e.g: `{a="b", c="d"}`)
+		"throughput", strings.Replace(humanize.Bytes(uint64(stats.Summary.BytesProcessedPerSecond)), " ", "", 1),
+		"total_bytes", strings.Replace(humanize.Bytes(uint64(stats.Summary.TotalBytesProcessed)), " ", "", 1),
+	)
+
+	bytesPerSecond.WithLabelValues(status, queryType, "", latencyType).
+		Observe(float64(stats.Summary.BytesProcessedPerSecond))
+	execLatency.WithLabelValues(status, queryType, "").
+		Observe(stats.Summary.ExecTime)
+	chunkDownloadLatency.WithLabelValues(status, queryType, "").
+		Observe(stats.ChunksDownloadTime().Seconds())
+	duplicatesTotal.Add(float64(stats.TotalDuplicates()))
+	chunkDownloadedTotal.WithLabelValues(status, queryType, "").
+		Add(float64(stats.TotalChunksDownloaded()))
+	ingesterLineTotal.Add(float64(stats.Ingester.TotalLinesSent))
+
+	// TODO(kavi): Do we need recordUsageStats for series query?
+}
+
 func recordUsageStats(queryType string, stats logql_stats.Result) {
 	if queryType == QueryTypeMetric {
 		bytePerSecondMetricUsage.Record(float64(stats.Summary.BytesProcessedPerSecond))

pkg/logql/metrics_test.go

@@ -66,7 +66,7 @@ func TestLogSlowQuery(t *testing.T) {
 
 	ctx = context.WithValue(ctx, httpreq.QueryTagsHTTPHeader, "Source=logvolhist,Feature=Beta")
 
-	RecordMetrics(ctx, util_log.Logger, LiteralParams{
+	RecordRangeAndInstantQueryMetrics(ctx, util_log.Logger, LiteralParams{
 		qs:        `{foo="bar"} |= "buzz"`,
 		direction: logproto.BACKWARD,
 		end:       now,
@@ -90,6 +90,56 @@ func TestLogSlowQuery(t *testing.T) {
 	util_log.Logger = log.NewNopLogger()
 }
 
+func TestLogLabelsQuery(t *testing.T) {
+	buf := bytes.NewBufferString("")
+	util_log.Logger = log.NewLogfmtLogger(buf)
+	tr, c := jaeger.NewTracer("foo", jaeger.NewConstSampler(true), jaeger.NewInMemoryReporter())
+	defer c.Close()
+	opentracing.SetGlobalTracer(tr)
+	sp := opentracing.StartSpan("")
+	ctx := opentracing.ContextWithSpan(user.InjectOrgID(context.Background(), "foo"), sp)
+	now := time.Now()
+	RecordLabelQueryMetrics(ctx, now.Add(-1*time.Hour), now, "foo", "200", stats.Result{
+		Summary: stats.Summary{
+			BytesProcessedPerSecond: 100000,
+			ExecTime:                25.25,
+			TotalBytesProcessed:     100000,
+		},
+	})
+	require.Equal(t,
+		fmt.Sprintf(
+			"level=info org_id=foo traceID=%s latency=slow query_type=labels length=1h0m0s duration=25.25s status=200 label=foo throughput=100kB total_bytes=100kB\n",
+			sp.Context().(jaeger.SpanContext).SpanID().String(),
+		),
+		buf.String())
+	util_log.Logger = log.NewNopLogger()
+}
+
+func TestLogSeriesQuery(t *testing.T) {
+	buf := bytes.NewBufferString("")
+	util_log.Logger = log.NewLogfmtLogger(buf)
+	tr, c := jaeger.NewTracer("foo", jaeger.NewConstSampler(true), jaeger.NewInMemoryReporter())
+	defer c.Close()
+	opentracing.SetGlobalTracer(tr)
+	sp := opentracing.StartSpan("")
+	ctx := opentracing.ContextWithSpan(user.InjectOrgID(context.Background(), "foo"), sp)
+	now := time.Now()
+	RecordSeriesQueryMetrics(ctx, now.Add(-1*time.Hour), now, []string{`{container_name=~"prometheus.*", component="server"}`, `{app="loki"}`}, "200", stats.Result{
+		Summary: stats.Summary{
+			BytesProcessedPerSecond: 100000,
+			ExecTime:                25.25,
+			TotalBytesProcessed:     100000,
+		},
+	})
+	require.Equal(t,
+		fmt.Sprintf(
+			"level=info org_id=foo traceID=%s latency=slow query_type=series length=1h0m0s duration=25.25s status=200 match=\"{container_name=~\\\"prometheus.*\\\", component=\\\"server\\\"}:{app=\\\"loki\\\"}\" throughput=100kB total_bytes=100kB\n",
+			sp.Context().(jaeger.SpanContext).SpanID().String(),
+		),
+		buf.String())
+	util_log.Logger = log.NewNopLogger()
+}
+
 func Test_testToKeyValues(t *testing.T) {
 	cases := []struct {
 		name string

pkg/loki/modules.go

@@ -553,7 +553,7 @@ func (t *Loki) initQueryFrontend() (_ services.Service, err error) {
 		httpreq.ExtractQueryTagsMiddleware(),
 		serverutil.RecoveryHTTPMiddleware,
 		t.HTTPAuthMiddleware,
-		queryrange.StatsHTTPMiddleware,
+		queryrange.StatsRangeQueryHTTPMiddleware,
 		serverutil.NewPrepopulateMiddleware(),
 		serverutil.ResponseJSONMiddleware(),
 	).Wrap(frontendHandler)
@@ -564,7 +564,7 @@ func (t *Loki) initQueryFrontend() (_ services.Service, err error) {
 		httpMiddleware := middleware.Merge(
 			httpreq.ExtractQueryTagsMiddleware(),
 			t.HTTPAuthMiddleware,
-			queryrange.StatsHTTPMiddleware,
+			queryrange.StatsRangeQueryHTTPMiddleware,
 		)
 		tailURL, err := url.Parse(t.Cfg.Frontend.TailProxyURL)
 		if err != nil {

pkg/querier/http.go

@@ -19,10 +19,13 @@ import (
 	"github.com/grafana/loki/pkg/logql"
 	"github.com/grafana/loki/pkg/logql/syntax"
 	"github.com/grafana/loki/pkg/logqlmodel"
+	"github.com/grafana/loki/pkg/logqlmodel/stats"
+	"github.com/grafana/loki/pkg/util/httpreq"
 	util_log "github.com/grafana/loki/pkg/util/log"
 	"github.com/grafana/loki/pkg/util/marshal"
 	marshal_legacy "github.com/grafana/loki/pkg/util/marshal/legacy"
 	serverutil "github.com/grafana/loki/pkg/util/server"
+	"github.com/grafana/loki/pkg/util/spanlogger"
 	util_validation "github.com/grafana/loki/pkg/util/validation"
 	"github.com/grafana/loki/pkg/validation"
 )
@@ -200,7 +203,26 @@ func (q *QuerierAPI) LabelHandler(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	log, ctx := spanlogger.New(r.Context(), "query.Label")
+
+	start := time.Now()
+	statsCtx, ctx := stats.NewContext(ctx)
+
 	resp, err := q.querier.Label(r.Context(), req)
+	queueTime, _ := ctx.Value(httpreq.QueryQueueTimeHTTPHeader).(time.Duration)
+
+	// record stats about the label query
+	statResult := statsCtx.Result(time.Since(start), queueTime)
+	statResult.Log(level.Debug(log))
+
+	status := "200"
+	if err != nil {
+		status = "500"
+		// TODO(kavi): adjust status based on all possible errors from `q.Label`
+	}
+
+	logql.RecordLabelQueryMetrics(ctx, *req.Start, *req.End, req.Name, status, statResult)
+
 	if err != nil {
 		serverutil.WriteError(err, w)
 		return
@@ -350,7 +372,25 @@ func (q *QuerierAPI) SeriesHandler(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	log, ctx := spanlogger.New(r.Context(), "query.Series")
+
+	start := time.Now()
+	statsCtx, ctx := stats.NewContext(ctx)
+
 	resp, err := q.querier.Series(r.Context(), req)
+	queueTime, _ := ctx.Value(httpreq.QueryQueueTimeHTTPHeader).(time.Duration)
+
+	// record stats about the label query
+	statResult := statsCtx.Result(time.Since(start), queueTime)
+	statResult.Log(level.Debug(log))
+
+	status := "200"
+	if err != nil {
+		status = "500"
+		// TODO(kavi): adjust status based on all possible errors from `q.Series`
+	}
+
+	logql.RecordSeriesQueryMetrics(ctx, req.Start, req.End, req.Groups, status, statResult)
 	if err != nil {
 		serverutil.WriteError(err, w)
 		return