inflight-logging: Add extra metadata to inflight requests logging

Makefile

@@ -42,6 +42,8 @@ BUILD_IMAGE_VERSION ?= 0.31.2
 # Docker image info
 IMAGE_PREFIX ?= grafana
 
+BUILD_IMAGE_PREFIX ?= grafana
+
 IMAGE_TAG ?= $(shell ./tools/image-tag)
 
 # Version info for binaries
@@ -102,7 +104,7 @@ RM := --rm
 TTY := --tty
 
 DOCKER_BUILDKIT ?= 1
-BUILD_IMAGE = BUILD_IMAGE=$(IMAGE_PREFIX)/loki-build-image:$(BUILD_IMAGE_VERSION)
+BUILD_IMAGE = BUILD_IMAGE=$(BUILD_IMAGE_PREFIX)/loki-build-image:$(BUILD_IMAGE_VERSION)
 PUSH_OCI=docker push
 TAG_OCI=docker tag
 ifeq ($(CI), true)

pkg/logql/blocker.go

@@ -9,6 +9,7 @@ import (
 	"github.com/grafana/regexp"
 
 	logutil "github.com/grafana/loki/pkg/util/log"
+	"github.com/grafana/loki/pkg/util/queryutil"
 	"github.com/grafana/loki/pkg/util/validation"
 )
 
@@ -43,7 +44,7 @@ func (qb *queryBlocker) isBlocked(ctx context.Context, tenant string) bool {
 	for _, b := range blocks {
 
 		if b.Hash > 0 {
-			if b.Hash == HashedQuery(query) {
+			if b.Hash == queryutil.HashedQuery(query) {
 				level.Warn(logger).Log("msg", "query blocker matched with hash policy", "hash", b.Hash, "query", query)
 				return qb.block(b, typ, logger)
 			}

pkg/logql/engine.go

@@ -34,6 +34,7 @@ import (
 	"github.com/grafana/loki/pkg/util/constants"
 	"github.com/grafana/loki/pkg/util/httpreq"
 	logutil "github.com/grafana/loki/pkg/util/log"
+	"github.com/grafana/loki/pkg/util/queryutil"
 	"github.com/grafana/loki/pkg/util/server"
 	"github.com/grafana/loki/pkg/util/spanlogger"
 	"github.com/grafana/loki/pkg/util/validation"
@@ -219,7 +220,7 @@ func (q *query) Exec(ctx context.Context) (logqlmodel.Result, error) {
 	)
 
 	if q.logExecQuery {
-		queryHash := HashedQuery(q.params.Query())
+		queryHash := queryutil.HashedQuery(q.params.Query())
 		if GetRangeType(q.params) == InstantType {
 			level.Info(logutil.WithContext(ctx, q.logger)).Log("msg", "executing query", "type", "instant", "query", q.params.Query(), "query_hash", queryHash)
 		} else {

pkg/logql/metrics.go

@@ -2,7 +2,6 @@ package logql
 
 import (
 	"context"
-	"hash/fnv"
 	"strconv"
 	"strings"
 	"time"
@@ -22,6 +21,7 @@ import (
 	"github.com/grafana/loki/pkg/util/constants"
 	"github.com/grafana/loki/pkg/util/httpreq"
 	util_log "github.com/grafana/loki/pkg/util/log"
+	"github.com/grafana/loki/pkg/util/queryutil"
 	"github.com/grafana/loki/pkg/util/spanlogger"
 )
 
@@ -120,7 +120,7 @@ func RecordRangeAndInstantQueryMetrics(
 	logValues = append(logValues, []interface{}{
 		"latency", latencyType, // this can be used to filter log lines.
 		"query", p.Query(),
-		"query_hash", HashedQuery(p.Query()),
+		"query_hash", queryutil.HashedQuery(p.Query()),
 		"query_type", queryType,
 		"range_type", rt,
 		"length", p.End().Sub(p.Start()),
@@ -187,12 +187,6 @@ func RecordRangeAndInstantQueryMetrics(
 	recordUsageStats(queryType, stats)
 }
 
-func HashedQuery(query string) uint32 {
-	h := fnv.New32()
-	_, _ = h.Write([]byte(query))
-	return h.Sum32()
-}
-
 func RecordLabelQueryMetrics(
 	ctx context.Context,
 	log log.Logger,
@@ -225,7 +219,7 @@ func RecordLabelQueryMetrics(
 		"status", status,
 		"label", label,
 		"query", query,
-		"query_hash", HashedQuery(query),
+		"query_hash", queryutil.HashedQuery(query),
 		"total_entries", stats.Summary.TotalEntriesReturned,
 	)
 
@@ -276,7 +270,7 @@ func RecordSeriesQueryMetrics(ctx context.Context, log log.Logger, start, end ti
 		"duration", time.Duration(int64(stats.Summary.ExecTime*float64(time.Second))),
 		"status", status,
 		"match", PrintMatches(match),
-		"query_hash", HashedQuery(PrintMatches(match)),
+		"query_hash", queryutil.HashedQuery(PrintMatches(match)),
 		"total_entries", stats.Summary.TotalEntriesReturned)
 
 	if shard != nil {
@@ -316,7 +310,7 @@ func RecordStatsQueryMetrics(ctx context.Context, log log.Logger, start, end tim
 		"duration", time.Duration(int64(stats.Summary.ExecTime*float64(time.Second))),
 		"status", status,
 		"query", query,
-		"query_hash", HashedQuery(query),
+		"query_hash", queryutil.HashedQuery(query),
 		"total_entries", stats.Summary.TotalEntriesReturned)
 
 	level.Info(logger).Log(logValues...)
@@ -346,7 +340,7 @@ func RecordVolumeQueryMetrics(ctx context.Context, log log.Logger, start, end ti
 		"latency", latencyType,
 		"query_type", queryType,
 		"query", query,
-		"query_hash", HashedQuery(query),
+		"query_hash", queryutil.HashedQuery(query),
 		"start", start.Format(time.RFC3339Nano),
 		"end", end.Format(time.RFC3339Nano),
 		"start_delta", time.Since(start),

pkg/querier/queryrange/queryrangebase/retry.go

@@ -12,6 +12,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus/promauto"
 
 	util_log "github.com/grafana/loki/pkg/util/log"
+	"github.com/grafana/loki/pkg/util/queryutil"
 )
 
 type RetryMiddlewareMetrics struct {
@@ -73,6 +74,11 @@ func (r retry) Do(ctx context.Context, req Request) (Response, error) {
 		MaxRetries: 0,
 	}
 	bk := backoff.New(ctx, cfg)
+
+	start := req.GetStart()
+	end := req.GetEnd()
+	query := req.GetQuery()
+
 	for ; tries < r.maxRetries; tries++ {
 		if ctx.Err() != nil {
 			return nil, ctx.Err()
@@ -86,7 +92,19 @@ func (r retry) Do(ctx context.Context, req Request) (Response, error) {
 		httpResp, ok := httpgrpc.HTTPResponseFromError(err)
 		if !ok || httpResp.Code/100 == 5 {
 			lastErr = err
-			level.Error(util_log.WithContext(ctx, r.log)).Log("msg", "error processing request", "try", tries, "query", req.GetQuery(), "retry_in", bk.NextDelay(), "err", err)
+			level.Error(util_log.WithContext(ctx, r.log)).Log(
+				"msg", "error processing request",
+				"try", tries,
+				"query", query,
+				"query_hash", queryutil.HashedQuery(query),
+				"start", start.Format(time.RFC3339Nano),
+				"end", end.Format(time.RFC3339Nano),
+				"start_delta", time.Since(start),
+				"end_delta", time.Since(end),
+				"length", end.Sub(start),
+				"retry_in", bk.NextDelay(),
+				"err", err,
+			)
 			bk.Wait()
 			continue
 		}

pkg/querier/queryrange/roundtrip.go

@@ -26,6 +26,7 @@ import (
 	"github.com/grafana/loki/pkg/storage/config"
 	"github.com/grafana/loki/pkg/util/constants"
 	logutil "github.com/grafana/loki/pkg/util/log"
+	"github.com/grafana/loki/pkg/util/queryutil"
 )
 
 // Config is the configuration for the queryrange tripperware
@@ -247,8 +248,19 @@ func (r roundTripper) Do(ctx context.Context, req base.Request) (base.Response,
 			return nil, httpgrpc.Errorf(http.StatusBadRequest, err.Error())
 		}
 
-		queryHash := logql.HashedQuery(op.Query)
-		level.Info(logger).Log("msg", "executing query", "type", "range", "query", op.Query, "length", op.EndTs.Sub(op.StartTs), "step", op.Step, "query_hash", queryHash)
+		queryHash := queryutil.HashedQuery(op.Query)
+		level.Info(logger).Log(
+			"msg", "executing query",
+			"type", "range",
+			"query", op.Query,
+			"start", op.StartTs.Format(time.RFC3339Nano),
+			"end", op.EndTs.Format(time.RFC3339Nano),
+			"start_delta", time.Since(op.StartTs),
+			"end_delta", time.Since(op.EndTs),
+			"length", op.EndTs.Sub(op.StartTs),
+			"step", op.Step,
+			"query_hash", queryHash,
+		)
 
 		switch e := expr.(type) {
 		case syntax.SampleExpr:
@@ -296,7 +308,7 @@ func (r roundTripper) Do(ctx context.Context, req base.Request) (base.Response,
 			return nil, httpgrpc.Errorf(http.StatusBadRequest, err.Error())
 		}
 
-		queryHash := logql.HashedQuery(op.Query)
+		queryHash := queryutil.HashedQuery(op.Query)
 		level.Info(logger).Log("msg", "executing query", "type", "instant", "query", op.Query, "query_hash", queryHash)
 
 		switch expr.(type) {

pkg/ruler/compat.go

@@ -24,11 +24,11 @@ import (
 	"github.com/prometheus/prometheus/rules"
 	"github.com/prometheus/prometheus/template"
 
-	"github.com/grafana/loki/pkg/logql"
 	"github.com/grafana/loki/pkg/logql/syntax"
 	ruler "github.com/grafana/loki/pkg/ruler/base"
 	"github.com/grafana/loki/pkg/ruler/rulespb"
 	"github.com/grafana/loki/pkg/ruler/util"
+	"github.com/grafana/loki/pkg/util/queryutil"
 )
 
 // RulesLimits is the one function we need from limits.Overrides, and
@@ -60,7 +60,7 @@ type RulesLimits interface {
 // and passing an altered timestamp.
 func queryFunc(evaluator Evaluator, checker readyChecker, userID string, logger log.Logger) rules.QueryFunc {
 	return func(ctx context.Context, qs string, t time.Time) (promql.Vector, error) {
-		hash := logql.HashedQuery(qs)
+		hash := queryutil.HashedQuery(qs)
 		detail := rules.FromOriginContext(ctx)
 		detailLog := log.With(logger, "rule_name", detail.Name, "rule_type", detail.Kind, "query", qs, "query_hash", hash)
 

pkg/ruler/evaluator_jitter.go

@@ -10,8 +10,8 @@ import (
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
 
-	"github.com/grafana/loki/pkg/logql"
 	"github.com/grafana/loki/pkg/logqlmodel"
+	"github.com/grafana/loki/pkg/util/queryutil"
 )
 
 // EvaluatorWithJitter wraps a given Evaluator. It applies a consistent jitter based on a rule's query string by hashing
@@ -44,7 +44,7 @@ func NewEvaluatorWithJitter(inner Evaluator, maxJitter time.Duration, hasher has
 }
 
 func (e *EvaluatorWithJitter) Eval(ctx context.Context, qs string, now time.Time) (*logqlmodel.Result, error) {
-	logger := log.With(e.logger, "query", qs, "query_hash", logql.HashedQuery(qs))
+	logger := log.With(e.logger, "query", qs, "query_hash", queryutil.HashedQuery(qs))
 	jitter := e.calculateJitter(qs, logger)
 
 	if jitter > 0 {

pkg/ruler/evaluator_remote.go

@@ -36,11 +36,11 @@ import (
 	"google.golang.org/grpc/keepalive"
 
 	"github.com/grafana/loki/pkg/loghttp"
-	"github.com/grafana/loki/pkg/logql"
 	"github.com/grafana/loki/pkg/logqlmodel"
 	"github.com/grafana/loki/pkg/util/build"
 	"github.com/grafana/loki/pkg/util/constants"
 	"github.com/grafana/loki/pkg/util/httpreq"
+	"github.com/grafana/loki/pkg/util/queryutil"
 	"github.com/grafana/loki/pkg/util/spanlogger"
 )
 
@@ -220,7 +220,7 @@ func (r *RemoteEvaluator) query(ctx context.Context, orgID, query string, ts tim
 		args.Set("time", ts.Format(time.RFC3339Nano))
 	}
 	body := []byte(args.Encode())
-	hash := logql.HashedQuery(query)
+	hash := queryutil.HashedQuery(query)
 
 	req := httpgrpc.HTTPRequest{
 		Method: http.MethodPost,

pkg/util/queryutil/query.go

@@ -0,0 +1,10 @@
+package queryutil
+
+import "hash/fnv"
+
+// HashedQuery returns a unique hash value for the given `query`.
+func HashedQuery(query string) uint32 {
+	h := fnv.New32()
+	_, _ = h.Write([]byte(query))
+	return h.Sum32()
+}