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

pkg/logql/engine.go

@@ -170,6 +170,18 @@ type query struct {
 	record    bool
 }
 
+func (q *query) resultLength(res promql_parser.Value) int {
+	switch r := res.(type) {
+	case promql.Matrix:
+		return r.TotalSamples()
+	case logqlmodel.Streams:
+		return int(r.Lines())
+	default:
+		level.Error(q.logger).Log("msg", "unknown query result type", "err", fmt.Sprintf("expected promql.Matrix or logqlmodel.Streams but got %T", r))
+		return 0
+	}
+}
+
 // Exec Implements `Query`. It handles instrumentation & defers to Eval.
 func (q *query) Exec(ctx context.Context) (logqlmodel.Result, error) {
 	log, ctx := spanlogger.New(ctx, "query.Exec")
@@ -187,7 +199,7 @@ func (q *query) Exec(ctx context.Context) (logqlmodel.Result, error) {
 
 	queueTime, _ := ctx.Value(httpreq.QueryQueueTimeHTTPHeader).(time.Duration)
 
-	statResult := statsCtx.Result(time.Since(start), queueTime)
+	statResult := statsCtx.Result(time.Since(start), queueTime, q.resultLength(data))
 	statResult.Log(level.Debug(log))
 
 	status := "200"
@@ -202,7 +214,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"
@@ -23,6 +24,8 @@ const (
 	QueryTypeMetric  = "metric"
 	QueryTypeFilter  = "filter"
 	QueryTypeLimited = "limited"
+	QueryTypeLabels  = "labels"
+	QueryTypeSeries  = "series"
 
 	latencyTypeSlow = "slow"
 	latencyTypeFast = "fast"
@@ -74,7 +77,14 @@ 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))
@@ -113,13 +123,13 @@ func RecordMetrics(ctx context.Context, log log.Logger, p Params, status string,
 		"returned_lines", returnedLines,
 		"throughput", strings.Replace(humanize.Bytes(uint64(stats.Summary.BytesProcessedPerSecond)), " ", "", 1),
 		"total_bytes", strings.Replace(humanize.Bytes(uint64(stats.Summary.TotalBytesProcessed)), " ", "", 1),
+		"total_entries", stats.Summary.TotalEntriesReturned,
 		"queue_time", logql_stats.ConvertSecondsToNanoseconds(stats.Summary.QueueTime),
 		"subqueries", stats.Summary.Subqueries,
 	}...)
 
 	logValues = append(logValues, tagsToKeyValues(queryTags)...)
 
-	// we also log queries, useful for troubleshooting slow queries.
 	level.Info(logger).Log(
 		logValues...,
 	)
@@ -138,6 +148,94 @@ func RecordMetrics(ctx context.Context, log log.Logger, p Params, status string,
 	recordUsageStats(queryType, stats)
 }
 
+func RecordLabelQueryMetrics(
+	ctx context.Context,
+	log log.Logger,
+	start, end time.Time,
+	label, status string,
+	stats logql_stats.Result,
+) {
+	var (
+		logger      = util_log.WithContext(ctx, log)
+		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
+	}
+
+	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),
+		"total_entries", stats.Summary.TotalEntriesReturned,
+	)
+
+	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))
+}
+
+func RecordSeriesQueryMetrics(
+	ctx context.Context,
+	log log.Logger,
+	start, end time.Time,
+	match []string,
+	status string,
+	stats logql_stats.Result,
+) {
+	var (
+		logger      = util_log.WithContext(ctx, log)
+		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),
+		"total_entries", stats.Summary.TotalEntriesReturned,
+	)
+
+	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))
+}
+
 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,
@@ -79,11 +79,64 @@ func TestLogSlowQuery(t *testing.T) {
 			QueueTime:               0.000000002,
 			ExecTime:                25.25,
 			TotalBytesProcessed:     100000,
+			TotalEntriesReturned:    10,
 		},
 	}, logqlmodel.Streams{logproto.Stream{Entries: make([]logproto.Entry, 10)}})
 	require.Equal(t,
 		fmt.Sprintf(
-			"level=info org_id=foo traceID=%s latency=slow query=\"{foo=\\\"bar\\\"} |= \\\"buzz\\\"\" query_type=filter range_type=range length=1h0m0s step=1m0s duration=25.25s status=200 limit=1000 returned_lines=10 throughput=100kB total_bytes=100kB queue_time=2ns subqueries=0 source=logvolhist feature=beta\n",
+			"level=info org_id=foo traceID=%s latency=slow query=\"{foo=\\\"bar\\\"} |= \\\"buzz\\\"\" query_type=filter range_type=range length=1h0m0s step=1m0s duration=25.25s status=200 limit=1000 returned_lines=10 throughput=100kB total_bytes=100kB total_entries=10 queue_time=2ns subqueries=0 source=logvolhist feature=beta\n",
+			sp.Context().(jaeger.SpanContext).SpanID().String(),
+		),
+		buf.String())
+	util_log.Logger = log.NewNopLogger()
+}
+
+func TestLogLabelsQuery(t *testing.T) {
+	buf := bytes.NewBufferString("")
+	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, logger, now.Add(-1*time.Hour), now, "foo", "200", stats.Result{
+		Summary: stats.Summary{
+			BytesProcessedPerSecond: 100000,
+			ExecTime:                25.25,
+			TotalBytesProcessed:     100000,
+			TotalEntriesReturned:    12,
+		},
+	})
+	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 total_entries=12\n",
+			sp.Context().(jaeger.SpanContext).SpanID().String(),
+		),
+		buf.String())
+	util_log.Logger = log.NewNopLogger()
+}
+
+func TestLogSeriesQuery(t *testing.T) {
+	buf := bytes.NewBufferString("")
+	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, logger, 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,
+			TotalEntriesReturned:    10,
+		},
+	})
+	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 total_entries=10\n",
 			sp.Context().(jaeger.SpanContext).SpanID().String(),
 		),
 		buf.String())

pkg/logqlmodel/stats/context.go

@@ -91,7 +91,7 @@ func (c *Context) Reset() {
 }
 
 // Result calculates the summary based on store and ingester data.
-func (c *Context) Result(execTime time.Duration, queueTime time.Duration) Result {
+func (c *Context) Result(execTime time.Duration, queueTime time.Duration, totalEntriesReturned int) Result {
 	r := c.result
 
 	r.Merge(Result{
@@ -101,7 +101,7 @@ func (c *Context) Result(execTime time.Duration, queueTime time.Duration) Result
 		Ingester: c.ingester,
 	})
 
-	r.ComputeSummary(execTime, queueTime)
+	r.ComputeSummary(execTime, queueTime, totalEntriesReturned)
 
 	return r
 }
@@ -125,7 +125,7 @@ func JoinIngesters(ctx context.Context, inc Ingester) {
 }
 
 // ComputeSummary compute the summary of the statistics.
-func (r *Result) ComputeSummary(execTime time.Duration, queueTime time.Duration) {
+func (r *Result) ComputeSummary(execTime time.Duration, queueTime time.Duration, totalEntriesReturned int) {
 	r.Summary.TotalBytesProcessed = r.Querier.Store.Chunk.DecompressedBytes + r.Querier.Store.Chunk.HeadChunkBytes +
 		r.Ingester.Store.Chunk.DecompressedBytes + r.Ingester.Store.Chunk.HeadChunkBytes
 	r.Summary.TotalLinesProcessed = r.Querier.Store.Chunk.DecompressedLines + r.Querier.Store.Chunk.HeadChunkLines +
@@ -140,6 +140,8 @@ func (r *Result) ComputeSummary(execTime time.Duration, queueTime time.Duration)
 	if queueTime != 0 {
 		r.Summary.QueueTime = queueTime.Seconds()
 	}
+
+	r.Summary.TotalEntriesReturned = int64(totalEntriesReturned)
 }
 
 func (s *Store) Merge(m Store) {
@@ -173,7 +175,7 @@ func (r *Result) Merge(m Result) {
 	r.Querier.Merge(m.Querier)
 	r.Ingester.Merge(m.Ingester)
 	r.ComputeSummary(ConvertSecondsToNanoseconds(r.Summary.ExecTime+m.Summary.ExecTime),
-		ConvertSecondsToNanoseconds(r.Summary.QueueTime+m.Summary.QueueTime))
+		ConvertSecondsToNanoseconds(r.Summary.QueueTime+m.Summary.QueueTime), int(r.Summary.TotalEntriesReturned))
 }
 
 // ConvertSecondsToNanoseconds converts time.Duration representation of seconds (float64)