Fix blocking failed queries (#63)

Author: hczhu-db

cmd/thanos/query_frontend.go

@@ -326,7 +326,7 @@ func runQueryFrontend(
 	roundTripper = tripperWare(roundTripper)
 
 	// Create the query frontend transport.
-	handler := transport.NewHandler(*cfg.CortexHandlerConfig, roundTripper, logger, nil)
+	handler := transport.NewHandler(*cfg.CortexHandlerConfig, roundTripper, logger, reg)
 	if cfg.CompressResponses {
 		handler = gzhttp.GzipHandler(handler)
 	}

internal/cortex/frontend/transport/handler.go

@@ -74,6 +74,7 @@ func NewHandler(cfg HandlerConfig, roundTripper http.RoundTripper, log log.Logge
 	}
 
 	if cfg.FailedQueryCacheCapacity > 0 {
+		level.Info(log).Log("msg", "Creating failed query cache", "capacity", cfg.FailedQueryCacheCapacity)
 		FailedQueryCache, errQueryCache := utils.NewFailedQueryCache(cfg.FailedQueryCacheCapacity, reg)
 		if errQueryCache != nil {
 			level.Warn(log).Log(errQueryCache.Error())
@@ -114,7 +115,6 @@ func (f *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	var (
 		stats       *querier_stats.Stats
 		queryString url.Values
-		urlQuery    url.Values
 	)
 
 	// Initialise the stats in the context and make sure it's propagated
@@ -134,14 +134,14 @@ func (f *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	r.Body = http.MaxBytesReader(w, r.Body, f.cfg.MaxBodySize)
 	r.Body = io.NopCloser(io.TeeReader(r.Body, &buf))
 
-	urlQuery = r.URL.Query()
+	queryString = f.parseRequestQueryString(r, buf)
 
 	// Check if query is cached
 	if f.failedQueryCache != nil {
-		cached, message := f.failedQueryCache.QueryHitCache(urlQuery)
+		cached, message := f.failedQueryCache.QueryHitCache(queryString)
 		if cached {
 			w.WriteHeader(http.StatusForbidden)
-			level.Info(util_log.WithContext(r.Context(), f.log)).Log(message)
+			level.Info(util_log.WithContext(r.Context(), f.log)).Log("msg", message)
 			return
 		}
 	}
@@ -152,20 +152,19 @@ func (f *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 
 	if err != nil {
 		writeError(w, err)
-		queryString = f.parseRequestQueryString(r, buf)
 
 		// Update cache for failed queries.
 		if f.failedQueryCache != nil {
-			success, message := f.failedQueryCache.UpdateFailedQueryCache(err, urlQuery)
+			success, message := f.failedQueryCache.UpdateFailedQueryCache(err, queryString, queryResponseTime)
 			if success {
-				level.Info(util_log.WithContext(r.Context(), f.log)).Log(message)
+				level.Info(util_log.WithContext(r.Context(), f.log)).Log("msg", message)
 			} else {
-				level.Debug(util_log.WithContext(r.Context(), f.log)).Log(message)
+				level.Debug(util_log.WithContext(r.Context(), f.log)).Log("msg", message)
 			}
 		}
 
 		if f.cfg.LogFailedQueries {
-			f.reportFailedQuery(r, queryString, err)
+			f.reportFailedQuery(r, queryString, err, queryResponseTime)
 		}
 		return
 	}
@@ -200,7 +199,7 @@ func (f *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
-func (f *Handler) reportFailedQuery(r *http.Request, queryString url.Values, err error) {
+func (f *Handler) reportFailedQuery(r *http.Request, queryString url.Values, err error, queryResponseTime time.Duration) {
 	// NOTE(GiedriusS): see https://github.com/grafana/grafana/pull/60301 for more info.
 	grafanaDashboardUID := "-"
 	if dashboardUID := r.Header.Get("X-Dashboard-Uid"); dashboardUID != "" {
@@ -222,6 +221,7 @@ func (f *Handler) reportFailedQuery(r *http.Request, queryString url.Values, err
 		"error", err.Error(),
 		"grafana_dashboard_uid", grafanaDashboardUID,
 		"grafana_panel_id", grafanaPanelID,
+		"query_response_time", queryResponseTime.String(),
 	}, formatQueryString(queryString)...)
 
 	level.Error(util_log.WithContext(r.Context(), f.log)).Log(logMessage...)

internal/cortex/frontend/transport/utils/utils.go

@@ -10,6 +10,7 @@ import (
 	"net/url"
 	"regexp"
 	"strconv"
+	"time"
 
 	lru "github.com/hashicorp/golang-lru"
 	"github.com/prometheus/client_golang/prometheus"
@@ -22,10 +23,11 @@ var (
 
 // FailedQueryCache Handler holds an instance of FailedQueryCache and calls its methods
 type FailedQueryCache struct {
-	regex        *regexp.Regexp
-	errorExtract *regexp.Regexp
-	lruCache     *lru.Cache
-	cachedHits   *prometheus.CounterVec
+	regex         *regexp.Regexp
+	errorExtract  *regexp.Regexp
+	lruCache      *lru.Cache
+	cachedHits    prometheus.Counter
+	cachedQueries prometheus.Gauge
 }
 
 func NewFailedQueryCache(capacity int, reg prometheus.Registerer) (*FailedQueryCache, error) {
@@ -34,24 +36,32 @@ func NewFailedQueryCache(capacity int, reg prometheus.Registerer) (*FailedQueryC
 	lruCache, err := lru.New(capacity)
 	if err != nil {
 		lruCache = nil
-		err = fmt.Errorf("Failed to create lru cache: %s", err)
+		err = fmt.Errorf("failed to create lru cache: %s", err)
 		return nil, err
 	}
-	cachedHits := promauto.With(reg).NewCounterVec(prometheus.CounterOpts{
+	cachedHits := promauto.With(reg).NewCounter(prometheus.CounterOpts{
+		Namespace: "cortex",
 		Name: "cached_failed_queries_count",
 		Help: "Total number of queries that hit the failed query cache.",
-	}, []string{})
+	})
+	cachedQueries := promauto.With(reg).NewGauge(prometheus.GaugeOpts{
+		Namespace: "cortex",
+		Name: "failed_query_cache_size",
+		Help: "How many queries are cached in the failed query cache.",
+	})
+	cachedQueries.Set(0)
 
 	return &FailedQueryCache{
-		regex:        regex,
-		errorExtract: errorExtract,
-		lruCache:     lruCache,
-		cachedHits:   cachedHits,
+		regex:         regex,
+		errorExtract:  errorExtract,
+		lruCache:      lruCache,
+		cachedHits:    cachedHits,
+		cachedQueries: cachedQueries,
 	}, err
 }
 
 // UpdateFailedQueryCache returns true if query is cached so that callsite can increase counter, returns message as a string for callsite to log outcome
-func (f *FailedQueryCache) updateFailedQueryCache(err error, queryExpressionNormalized string, queryExpressionRangeLength int, lruCache *lru.Cache) (bool, string) {
+func (f *FailedQueryCache) updateFailedQueryCache(err error, queryExpressionNormalized string, queryExpressionRangeLength int) (bool, string) {
 	// Extracting error code from error string.
 	codeExtract := f.errorExtract.FindStringSubmatch(err.Error())
 
@@ -75,25 +85,28 @@ func (f *FailedQueryCache) updateFailedQueryCache(err error, queryExpressionNorm
 		message := createLogMessage("Query not cached due to non-cacheable error code", queryExpressionNormalized, -1, queryExpressionRangeLength, err)
 		return false, message
 	}
+	f.addCacheEntry(queryExpressionNormalized, queryExpressionRangeLength)
+	message := createLogMessage("Cached a failed query", queryExpressionNormalized, -1, queryExpressionRangeLength, err)
+	return true, message
+}
 
+func (f *FailedQueryCache) addCacheEntry(queryExpressionNormalized string, queryExpressionRangeLength int) {
 	// Checks if queryExpression is already in cache, and updates time range length value to min of stored and new value.
-	if contains, _ := lruCache.ContainsOrAdd(queryExpressionNormalized, queryExpressionRangeLength); contains {
-		if oldValue, ok := lruCache.Get(queryExpressionNormalized); ok {
+	if contains, _ := f.lruCache.ContainsOrAdd(queryExpressionNormalized, queryExpressionRangeLength); contains {
+		if oldValue, ok := f.lruCache.Get(queryExpressionNormalized); ok {
 			queryExpressionRangeLength = min(queryExpressionRangeLength, oldValue.(int))
 		}
-		lruCache.Add(queryExpressionNormalized, queryExpressionRangeLength)
+		f.lruCache.Add(queryExpressionNormalized, queryExpressionRangeLength)
 	}
-
-	message := createLogMessage("Cached a failed query", queryExpressionNormalized, -1, queryExpressionRangeLength, err)
-	return true, message
+	f.cachedQueries.Set(float64(f.lruCache.Len()))
 }
 
 // QueryHitCache checks if the lru cache is hit and returns whether to increment counter for cache hits along with appropriate message.
-func queryHitCache(queryExpressionNormalized string, queryExpressionRangeLength int, lruCache *lru.Cache, cachedHits *prometheus.CounterVec) (bool, string) {
+func queryHitCache(queryExpressionNormalized string, queryExpressionRangeLength int, lruCache *lru.Cache, cachedHits prometheus.Counter) (bool, string) {
 	if value, ok := lruCache.Get(queryExpressionNormalized); ok && value.(int) <= queryExpressionRangeLength {
 		cachedQueryRangeSeconds := value.(int)
 		message := createLogMessage("Retrieved query from cache", queryExpressionNormalized, cachedQueryRangeSeconds, queryExpressionRangeLength, nil)
-		cachedHits.WithLabelValues().Inc()
+		cachedHits.Inc()
 		return true, message
 	}
 	return false, ""
@@ -129,27 +142,43 @@ func (f *FailedQueryCache) normalizeQueryString(query url.Values) string {
 func createLogMessage(message string, queryExpressionNormalized string, cachedQueryRangeSeconds int, queryExpressionRangeLength int, err error) string {
 	if err == nil {
 		return fmt.Sprintf(
-			`%s: %s, %s: %s, %s: %d, %s: %d`, "msg", message,
+			`%s, %s: %s, %s: %d, %s: %d`, message,
 			"cached_query", queryExpressionNormalized,
 			"cached_range_seconds", cachedQueryRangeSeconds,
 			"query_range_seconds", queryExpressionRangeLength)
 	}
 	return fmt.Sprintf(
-		`%s: %s, %s: %s, %s: %d, %s: %s`, "msg", message,
+		`%s, %s: %s, %s: %d, %s: %s`, message,
 		"cached_query", queryExpressionNormalized,
 		"query_range_seconds", queryExpressionRangeLength,
 		"cached_error", err)
 }
 
-func (f *FailedQueryCache) UpdateFailedQueryCache(err error, query url.Values) (bool, string) {
+func (f *FailedQueryCache) UpdateFailedQueryCache(err error, query url.Values, queryResponseTime time.Duration) (bool, string) {
 	queryExpressionNormalized := f.normalizeQueryString(query)
 	queryExpressionRangeLength := getQueryRangeSeconds(query)
-	success, message := f.updateFailedQueryCache(err, queryExpressionNormalized, queryExpressionRangeLength, f.lruCache)
+	// TODO(hc.zhu): add a flag for the threshold
+	// The current gateway timeout is 5 minutes, so we cache the failed query running longer than 5 minutes - 10 seconds.
+	if queryResponseTime > time.Second * (60 * 5 - 10) {
+		// Cache long running queries regardless of the error code. The most common case is "context canceled".
+		f.addCacheEntry(queryExpressionNormalized, queryExpressionRangeLength)
+		message := createLogMessage("Cached a failed long running query", queryExpressionNormalized, -1, queryExpressionRangeLength, err)
+		return true, message
+	}
+	if queryExpressionNormalized == "" {
+		// Other APIs don't have "query" parameter e.g., /api/v1/series, /api/v1/labels
+		return false, "Query parameter is empty"
+	}
+	success, message := f.updateFailedQueryCache(err, queryExpressionNormalized, queryExpressionRangeLength)
 	return success, message
 }
 
 func (f *FailedQueryCache) QueryHitCache(query url.Values) (bool, string) {
 	queryExpressionNormalized := f.normalizeQueryString(query)
+	if queryExpressionNormalized == "" {
+		// Other APIs don't have "query" parameter e.g., /api/v1/series, /api/v1/labels
+		return false, "Query parameter is empty"
+	}
 	queryExpressionRangeLength := getQueryRangeSeconds(query)
 	cached, message := queryHitCache(queryExpressionNormalized, queryExpressionRangeLength, f.lruCache, f.cachedHits)
 	return cached, message