Merge db_main/ into release (#74)

Author: hczhu-db

internal/cortex/frontend/transport/handler.go

@@ -59,10 +59,12 @@ type Handler struct {
 	failedQueryCache *utils.FailedQueryCache
 
 	// Metrics.
-	querySeconds *prometheus.CounterVec
-	querySeries  *prometheus.CounterVec
-	queryBytes   *prometheus.CounterVec
-	activeUsers  *util.ActiveUsersCleanupService
+	querySeconds     *prometheus.CounterVec
+	querySeries      *prometheus.CounterVec
+	queryBytes       *prometheus.CounterVec
+	activeUsers      *util.ActiveUsersCleanupService
+	slowQueryCount   prometheus.Counter
+	failedQueryCount prometheus.Counter
 }
 
 // NewHandler creates a new frontend handler.
@@ -108,16 +110,24 @@ func NewHandler(cfg HandlerConfig, roundTripper http.RoundTripper, log log.Logge
 		_ = h.activeUsers.StartAsync(context.Background())
 	}
 
+	h.slowQueryCount = promauto.With(reg).NewCounter(prometheus.CounterOpts{
+		Name: "cortex_slow_query_total",
+		Help: "Total number of slow queries detected.",
+	})
+	h.failedQueryCount = promauto.With(reg).NewCounter(prometheus.CounterOpts{
+		Name: "cortex_failed_query_total",
+		Help: "Total number of failed queries detected.",
+	})
 	return h
 }
 
 func (f *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	var (
-		stats       *querier_stats.Stats
+		stats *querier_stats.Stats
 		// For failed/slow query logging and query stats.
 		queryString url.Values
 		// For failed query cache
-		urlQuery    url.Values
+		urlQuery url.Values
 	)
 
 	// Initialise the stats in the context and make sure it's propagated
@@ -206,6 +216,7 @@ func (f *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 }
 
 func (f *Handler) reportFailedQuery(r *http.Request, queryString url.Values, err error, queryResponseTime time.Duration) {
+	f.failedQueryCount.Inc()
 	// NOTE(GiedriusS): see https://github.com/grafana/grafana/pull/60301 for more info.
 	grafanaDashboardUID := "-"
 	if dashboardUID := r.Header.Get("X-Dashboard-Uid"); dashboardUID != "" {
@@ -235,6 +246,7 @@ func (f *Handler) reportFailedQuery(r *http.Request, queryString url.Values, err
 
 // reportSlowQuery reports slow queries.
 func (f *Handler) reportSlowQuery(r *http.Request, responseHeaders http.Header, queryString url.Values, queryResponseTime time.Duration) {
+	f.slowQueryCount.Inc()
 	// NOTE(GiedriusS): see https://github.com/grafana/grafana/pull/60301 for more info.
 	grafanaDashboardUID := "-"
 	if dashboardUID := r.Header.Get("X-Dashboard-Uid"); dashboardUID != "" {

pkg/query/endpointset.go

@@ -551,7 +551,13 @@ func (e *EndpointSet) Update(ctx context.Context) {
 func (e *EndpointSet) updateEndpoint(ctx context.Context, spec *GRPCEndpointSpec, er *endpointRef) {
 	metadata, err := er.Metadata(ctx, infopb.NewInfoClient(er.cc), storepb.NewStoreClient(er.cc))
 	if err != nil {
-		level.Warn(e.logger).Log("msg", "update of endpoint failed", "err", errors.Wrap(err, "getting metadata"), "address", spec.Addr())
+		level.Warn(e.logger).Log(
+			"msg", "update of endpoint failed",
+			"err", errors.Wrap(err, "getting metadata"),
+			"address", spec.Addr(),
+			"group_key", spec.groupKey,
+			"replica_key", spec.replicaKey,
+		)
 	}
 	er.update(e.now, metadata, err)
 }

pkg/receive/multitsdb.go

@@ -141,7 +141,7 @@ func (l *localClient) String() string {
 	mint, maxt := l.store.TimeRange()
 	return fmt.Sprintf(
 		"LabelSets: %v MinTime: %d MaxTime: %d",
-		labelpb.PromLabelSetsToStringN(l.LabelSets(), 10), mint, maxt,
+		labelpb.PromLabelSetsToStringN(l.LabelSets(), 500), mint, maxt,
 	)
 }
 

pkg/store/labelpb/label.go

@@ -303,15 +303,18 @@ func ExtendSortedLabels(lset, extend labels.Labels) labels.Labels {
 }
 
 func PromLabelSetsToString(lsets []labels.Labels) string {
-	return PromLabelSetsToStringN(lsets, 1000000)
+	return PromLabelSetsToStringN(lsets, 500)
 }
 
-func PromLabelSetsToStringN(lsets []labels.Labels, maxNumLabels int) string {
+func PromLabelSetsToStringN(lsets []labels.Labels, maxLength int) string {
+	if len(lsets) == 0 {
+		return ""
+	}
 	s := []string{}
 	for _, ls := range lsets {
 		s = append(s, ls.String())
-		maxNumLabels--
-		if maxNumLabels <= 0 {
+		maxLength -= len(ls.String())
+		if maxLength <= 0 {
 			break
 		}
 	}

pkg/store/proxy.go

@@ -99,6 +99,7 @@ type ProxyStore struct {
 
 type proxyStoreMetrics struct {
 	emptyStreamResponses prometheus.Counter
+	storeFailureCount    *prometheus.CounterVec
 }
 
 func newProxyStoreMetrics(reg prometheus.Registerer) *proxyStoreMetrics {
@@ -108,6 +109,10 @@ func newProxyStoreMetrics(reg prometheus.Registerer) *proxyStoreMetrics {
 		Name: "thanos_proxy_store_empty_stream_responses_total",
 		Help: "Total number of empty responses received.",
 	})
+	m.storeFailureCount = promauto.With(reg).NewCounterVec(prometheus.CounterOpts{
+		Name: "thanos_proxy_store_failure_total",
+		Help: "Total number of store failures.",
+	}, []string{"group", "replica"})
 
 	return &m
 }
@@ -431,7 +436,8 @@ func (s *ProxyStore) Series(originalRequest *storepb.SeriesRequest, srv storepb.
 		if err != nil {
 			// NB: respSet is nil in case of error.
 			level.Error(reqLogger).Log("err", err)
-			level.Warn(s.logger).Log("msg", "Store failure", "group", st.GroupKey(), "replica", st.ReplicaKey())
+			level.Warn(s.logger).Log("msg", "Store failure", "group", st.GroupKey(), "replica", st.ReplicaKey(), "err", err)
+			s.metrics.storeFailureCount.WithLabelValues(st.GroupKey(), st.ReplicaKey()).Inc()
 			bumpCounter(st.GroupKey(), st.ReplicaKey(), failedStores)
 			totalFailedStores++
 			if r.PartialResponseStrategy == storepb.PartialResponseStrategy_GROUP_REPLICA {
@@ -464,14 +470,18 @@ func (s *ProxyStore) Series(originalRequest *storepb.SeriesRequest, srv storepb.
 
 		if resp.GetWarning() != "" {
 			totalFailedStores++
-			level.Error(s.logger).Log("msg", "Series: warning from store", "warning", resp.GetWarning())
+			maxWarningBytes := 2000
+			warning := resp.GetWarning()[:min(maxWarningBytes, len(resp.GetWarning()))]
+			level.Error(s.logger).Log("msg", "Store failure with warning", "warning", warning)
+			// Don't have group/replica keys here, so we can't attribute the warning to a specific store.
+			s.metrics.storeFailureCount.WithLabelValues("", "").Inc()
 			if r.PartialResponseStrategy == storepb.PartialResponseStrategy_GROUP_REPLICA {
 				// TODO: attribute the warning to the store(group key and replica key) that produced it.
 				// Each client streams a sequence of time series, so it's not trivial to attribute the warning to a specific client.
 				if totalFailedStores > 1 {
 					level.Error(reqLogger).Log("msg", "more than one stores have failed")
 					// If we don't know which store has failed, we can tolerate at most one failed store.
-					return status.Error(codes.Aborted, resp.GetWarning())
+					return status.Error(codes.Aborted, warning)
 				}
 			} else if r.PartialResponseDisabled || r.PartialResponseStrategy == storepb.PartialResponseStrategy_ABORT {
 				return status.Error(codes.Aborted, resp.GetWarning())