merge db_main into release (#70)

Author: christopherzli

cmd/thanos/query.go

@@ -129,6 +129,8 @@ func registerQuery(app *extkingpin.App) {
 
 	enableDedupMerge := cmd.Flag("query.dedup-merge", "Enable deduplication merge of multiple time series with the same labels.").
 		Default("false").Bool()
+	enableQuorumChunkDedup := cmd.Flag("query.quorum-chunk-dedup", "Enable quorum-based deduplication for chuncks from replicas.").
+		Default("false").Bool()
 
 	instantDefaultMaxSourceResolution := extkingpin.ModelDuration(cmd.Flag("query.instant.default.max_source_resolution", "default value for max_source_resolution for instant queries. If not set, defaults to 0s only taking raw resolution into account. 1h can be a good value if you use instant queries over time ranges that incorporate times outside of your raw-retention.").Default("0s").Hidden())
 
@@ -378,6 +380,7 @@ func registerQuery(app *extkingpin.App) {
 			*tenantLabel,
 			*enableGroupReplicaPartialStrategy,
 			*enableDedupMerge,
+			*enableQuorumChunkDedup,
 		)
 	})
 }
@@ -462,6 +465,7 @@ func runQuery(
 	tenantLabel string,
 	groupReplicaPartialResponseStrategy bool,
 	enableDedupMerge bool,
+	enableQuorumChunkDedup bool,
 ) error {
 	if alertQueryURL == "" {
 		lastColon := strings.LastIndex(httpBindAddr, ":")
@@ -536,6 +540,7 @@ func runQuery(
 	options := []store.ProxyStoreOption{
 		store.WithTSDBSelector(tsdbSelector),
 		store.WithProxyStoreDebugLogging(debugLogging),
+		store.WithQuorumChunkDedup(enableQuorumChunkDedup),
 	}
 
 	var (

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)
 	}

cmd/thanos/receive.go

@@ -286,7 +286,7 @@ func runReceive(
 
 		level.Debug(logger).Log("msg", "setting up TSDB")
 		{
-			if err := startTSDBAndUpload(g, logger, reg, dbs, uploadC, hashringChangedChan, upload, uploadDone, statusProber, bkt, receive.HashringAlgorithm(conf.hashringsAlgorithm)); err != nil {
+			if err := startTSDBAndUpload(g, logger, reg, dbs, uploadC, hashringChangedChan, upload, uploadDone, statusProber, bkt, receive.HashringAlgorithm(conf.hashringsAlgorithm), conf.tsdbDisableFlushOnShutdown); err != nil {
 				return err
 			}
 		}
@@ -578,6 +578,7 @@ func startTSDBAndUpload(g *run.Group,
 	statusProber prober.Probe,
 	bkt objstore.Bucket,
 	hashringAlgorithm receive.HashringAlgorithm,
+	disableFlushOnShutDown bool,
 ) error {
 
 	log.With(logger, "component", "storage")
@@ -603,10 +604,12 @@ func startTSDBAndUpload(g *run.Group,
 		// Before quitting, ensure the WAL is flushed and the DBs are closed.
 		defer func() {
 			level.Info(logger).Log("msg", "shutting down storage")
-			if err := dbs.Flush(); err != nil {
-				level.Error(logger).Log("err", err, "msg", "failed to flush storage")
-			} else {
-				level.Info(logger).Log("msg", "storage is flushed successfully")
+			if !disableFlushOnShutDown {
+				if err := dbs.Flush(); err != nil {
+					level.Error(logger).Log("err", err, "msg", "failed to flush storage")
+				} else {
+					level.Info(logger).Log("msg", "storage is flushed successfully")
+				}
 			}
 			if err := dbs.Close(); err != nil {
 				level.Error(logger).Log("err", err, "msg", "failed to close storage")
@@ -822,6 +825,7 @@ type receiveConfig struct {
 	tsdbMaxBytes                 units.Base2Bytes
 	tsdbWriteQueueSize           int64
 	tsdbMemorySnapshotOnShutdown bool
+	tsdbDisableFlushOnShutdown   bool
 	tsdbEnableNativeHistograms   bool
 
 	walCompression       bool
@@ -958,6 +962,10 @@ func (rc *receiveConfig) registerFlag(cmd extkingpin.FlagClause) {
 		"[EXPERIMENTAL] Enables feature to snapshot in-memory chunks on shutdown for faster restarts.").
 		Default("false").Hidden().BoolVar(&rc.tsdbMemorySnapshotOnShutdown)
 
+	cmd.Flag("tsdb.disable-flush-on-shutdown",
+		"[EXPERIMENTAL] disable flush on receive shutdown for rebuilding mempostings on restart.").
+		Default("false").Hidden().BoolVar(&rc.tsdbDisableFlushOnShutdown)
+
 	cmd.Flag("tsdb.enable-native-histograms",
 		"[EXPERIMENTAL] Enables the ingestion of native histograms.").
 		Default("false").Hidden().BoolVar(&rc.tsdbEnableNativeHistograms)

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())
@@ -113,7 +114,9 @@ func NewHandler(cfg HandlerConfig, roundTripper http.RoundTripper, log log.Logge
 func (f *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	var (
 		stats       *querier_stats.Stats
+		// For failed/slow query logging and query stats.
 		queryString url.Values
+		// For failed query cache
 		urlQuery    url.Values
 	)
 
@@ -134,14 +137,16 @@ 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()
-
 	// Check if query is cached
 	if f.failedQueryCache != nil {
+		// NB: don't call f.parseRequestQueryString(r, buf) before f.roundTripper.RoundTrip(r)
+		//     because the call closes the buffer which has content if the request is a POST with
+		//     form data in body.
+		urlQuery = r.URL.Query()
 		cached, message := f.failedQueryCache.QueryHitCache(urlQuery)
 		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 +157,20 @@ 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, urlQuery, 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)
+			queryString = f.parseRequestQueryString(r, buf)
+			f.reportFailedQuery(r, queryString, err, queryResponseTime)
 		}
 		return
 	}
@@ -200,7 +205,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 +227,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