Merge all db_main commits to release (#122)

Author: hczhu-db

CHANGELOG.md

@@ -38,6 +38,8 @@ We use *breaking :warning:* to mark changes that are not backward compatible (re
 - [#7885](https://github.com/thanos-io/thanos/pull/7885) Store: Return chunks to the pool after completing a Series call.
 - [#7893](https://github.com/thanos-io/thanos/pull/7893) Sidecar: Fix retrieval of external labels for Prometheus v3.0.0.
 - [#7903](https://github.com/thanos-io/thanos/pull/7903) Query: Fix panic on regex store matchers.
+- [#7915](https://github.com/thanos-io/thanos/pull/7915) Store: Close block series client at the end to not reuse chunk buffer
+- [#7941](https://github.com/thanos-io/thanos/pull/7941) Receive: Fix race condition when adding multiple new tenants, see [issue-7892](https://github.com/thanos-io/thanos/issues/7892).
 
 ### Added
 - [#7763](https://github.com/thanos-io/thanos/pull/7763) Ruler: use native histograms for client latency metrics.

cmd/thanos/compact.go

@@ -436,6 +436,12 @@ func runCompact(
 		level.Info(logger).Log("msg", "retention policy of 1 hour aggregated samples is enabled", "duration", retentionByResolution[compact.ResolutionLevel1h])
 	}
 
+	retentionByTenant, err := compact.ParesRetentionPolicyByTenant(logger, *conf.retentionTenants)
+	if err != nil {
+		level.Error(logger).Log("msg", "failed to parse retention policy by tenant", "err", err)
+		return err
+	}
+
 	var cleanMtx sync.Mutex
 	// TODO(GiedriusS): we could also apply retention policies here but the logic would be a bit more complex.
 	cleanPartialMarked := func() error {
@@ -456,6 +462,17 @@ func runCompact(
 	}
 
 	compactMainFn := func() error {
+		// this should happen before any compaction to remove unnecessary process on backlogs beyond retention.
+		if len(retentionByTenant) != 0 && len(sy.Metas()) == 0 {
+			level.Info(logger).Log("msg", "sync before tenant retention due to no blocks")
+			if err := sy.SyncMetas(ctx); err != nil {
+				return errors.Wrap(err, "sync before tenant retention")
+			}
+		}
+		if err := compact.ApplyRetentionPolicyByTenant(ctx, logger, insBkt, sy.Metas(), retentionByTenant, compactMetrics.blocksMarked.WithLabelValues(metadata.DeletionMarkFilename, metadata.TenantRetentionExpired)); err != nil {
+			return errors.Wrap(err, "retention by tenant failed")
+		}
+
 		if err := compactor.Compact(ctx); err != nil {
 			return errors.Wrap(err, "whole compaction error")
 		}
@@ -726,6 +743,7 @@ type compactConfig struct {
 	objStore                                       extflag.PathOrContent
 	consistencyDelay                               time.Duration
 	retentionRaw, retentionFiveMin, retentionOneHr model.Duration
+	retentionTenants                               *[]string
 	wait                                           bool
 	waitInterval                                   time.Duration
 	disableDownsampling                            bool
@@ -781,6 +799,7 @@ func (cc *compactConfig) registerFlag(cmd extkingpin.FlagClause) {
 		Default("0d").SetValue(&cc.retentionFiveMin)
 	cmd.Flag("retention.resolution-1h", "How long to retain samples of resolution 2 (1 hour) in bucket. Setting this to 0d will retain samples of this resolution forever").
 		Default("0d").SetValue(&cc.retentionOneHr)
+	cc.retentionTenants = cmd.Flag("retention.tenant", "How long to retain samples in bucket per tenant. Setting this to 0d will retain samples of this resolution forever").Strings()
 
 	// TODO(kakkoyun, pgough): https://github.com/thanos-io/thanos/issues/2266.
 	cmd.Flag("wait", "Do not exit after all compactions have been processed and wait for new work.").

cmd/thanos/receive.go

@@ -5,6 +5,7 @@ package main
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"net"
 	"net/http"
@@ -29,6 +30,7 @@ import (
 	"github.com/prometheus/prometheus/model/relabel"
 	"github.com/prometheus/prometheus/tsdb"
 	"github.com/prometheus/prometheus/tsdb/wlog"
+	"github.com/thanos-io/thanos/pkg/store/storepb"
 	"google.golang.org/grpc"
 	"gopkg.in/yaml.v2"
 
@@ -151,6 +153,20 @@ func runReceive(
 		}
 	}
 
+	// Create a matcher converter if specified by command line to cache expensive regex matcher conversions.
+	// Proxy store and TSDB stores of all tenants share a single cache.
+	var matcherConverter *storepb.MatcherConverter
+	if conf.matcherConverterCacheCapacity > 0 {
+		var err error
+		matcherConverter, err = storepb.NewMatcherConverter(conf.matcherConverterCacheCapacity, reg)
+		if err != nil {
+			level.Error(logger).Log("msg", "failed to create matcher converter", "err", err)
+		}
+	}
+	if matcherConverter != nil {
+		multiTSDBOptions = append(multiTSDBOptions, receive.WithMatcherConverter(matcherConverter))
+	}
+
 	rwTLSConfig, err := tls.NewServerConfig(log.With(logger, "protocol", "HTTP"), conf.rwServerCert, conf.rwServerKey, conf.rwServerClientCA, conf.rwServerTlsMinVersion)
 	if err != nil {
 		return err
@@ -252,7 +268,19 @@ func runReceive(
 			return errors.Wrap(err, "parse limit configuration")
 		}
 	}
-	limiter, err := receive.NewLimiter(conf.writeLimitsConfig, reg, receiveMode, log.With(logger, "component", "receive-limiter"), conf.limitsConfigReloadTimer)
+	if conf.maxPendingGrpcWriteRequests > 0 {
+		level.Info(logger).Log("msg", "set max pending gRPC write request in limiter", "max_pending_requests", conf.maxPendingGrpcWriteRequests)
+	}
+	limiter, err := receive.NewLimiterWithOptions(
+		conf.writeLimitsConfig,
+		reg,
+		receiveMode,
+		log.With(logger, "component", "receive-limiter"),
+		conf.limitsConfigReloadTimer,
+		receive.LimiterOptions{
+			MaxPendingRequests: int32(conf.maxPendingGrpcWriteRequests),
+		},
+	)
 	if err != nil {
 		return errors.Wrap(err, "creating limiter")
 	}
@@ -337,6 +365,25 @@ func runReceive(
 				w.WriteHeader(http.StatusOK)
 			}
 		}))
+		srv.Handle("/-/matchers", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			w.WriteHeader(http.StatusOK)
+			if matcherConverter != nil {
+				labelMatchers := matcherConverter.Keys()
+				// Convert the slice to JSON
+				jsonData, err := json.Marshal(labelMatchers)
+				if err != nil {
+					http.Error(w, "Failed to encode JSON", http.StatusInternalServerError)
+					return
+				}
+
+				// Set the Content-Type header and write the response
+				w.Header().Set("Content-Type", "application/json")
+				w.WriteHeader(http.StatusOK)
+				if _, err := w.Write(jsonData); err != nil {
+					level.Error(logger).Log("msg", "failed to write matchers json", "err", err)
+				}
+			}
+		}))
 		g.Add(func() error {
 			statusProber.Healthy()
 			return srv.ListenAndServe()
@@ -359,6 +406,9 @@ func runReceive(
 			store.WithProxyStoreDebugLogging(debugLogging),
 			store.WithoutDedup(),
 		}
+		if matcherConverter != nil {
+			options = append(options, store.WithProxyStoreMatcherConverter(matcherConverter))
+		}
 
 		proxy := store.NewProxyStore(
 			logger,
@@ -370,6 +420,7 @@ func runReceive(
 			store.LazyRetrieval,
 			options...,
 		)
+
 		mts := store.NewLimitedStoreServer(store.NewInstrumentedStoreServer(reg, proxy), reg, conf.storeRateLimits)
 		rw := store.ReadWriteTSDBStore{
 			StoreServer:          mts,
@@ -932,9 +983,11 @@ type receiveConfig struct {
 
 	asyncForwardWorkerCount uint
 
-	numTopMetricsPerTenant       int
-	topMetricsMinimumCardinality uint64
-	topMetricsUpdateInterval     time.Duration
+	numTopMetricsPerTenant        int
+	topMetricsMinimumCardinality  uint64
+	topMetricsUpdateInterval      time.Duration
+	matcherConverterCacheCapacity int
+	maxPendingGrpcWriteRequests   int
 
 	featureList *[]string
 }
@@ -1097,6 +1150,10 @@ func (rc *receiveConfig) registerFlag(cmd extkingpin.FlagClause) {
 		Default("10000").Uint64Var(&rc.topMetricsMinimumCardinality)
 	cmd.Flag("receive.top-metrics-update-interval", "The interval at which the top metrics are updated.").
 		Default("5m").DurationVar(&rc.topMetricsUpdateInterval)
+	cmd.Flag("receive.store-matcher-converter-cache-capacity", "The number of label matchers to cache in the matcher converter for the Store API. Set to 0 to disable to cache. Default is 0.").
+		Default("0").IntVar(&rc.matcherConverterCacheCapacity)
+	cmd.Flag("receive.max-pending-grcp-write-requests", "Reject right away gRPC write requests when this number of requests are pending. Value 0 disables this feature.").
+		Default("0").IntVar(&rc.maxPendingGrpcWriteRequests)
 	rc.featureList = cmd.Flag("enable-feature", "Comma separated experimental feature names to enable. The current list of features is "+metricNamesFilter+".").Default("").Strings()
 }
 

pkg/block/metadata/markers.go

@@ -81,6 +81,8 @@ const (
 	OutOfOrderChunksNoCompactReason = "block-index-out-of-order-chunk"
 	// DownsampleVerticalCompactionNoCompactReason is a reason to not compact overlapping downsampled blocks as it does not make sense e.g. how to vertically compact the average.
 	DownsampleVerticalCompactionNoCompactReason = "downsample-vertical-compaction"
+	// TenantRetentionExpired is a reason to delete block as it's per tenant retention is expired.
+	TenantRetentionExpired = "tenant-retention-expired"
 )
 
 // NoCompactMark marker stores reason of block being excluded from compaction if needed.

pkg/compact/retention.go

@@ -6,19 +6,27 @@ package compact
 import (
 	"context"
 	"fmt"
+	"regexp"
 	"time"
 
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
 	"github.com/oklog/ulid"
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/common/model"
 	"github.com/thanos-io/objstore"
 
 	"github.com/thanos-io/thanos/pkg/block"
 	"github.com/thanos-io/thanos/pkg/block/metadata"
 )
 
+const (
+	// tenantRetentionRegex is the regex pattern for parsing tenant retention.
+	// valid format is `<tenant>:(<yyyy-mm-dd>|<duration>d)` where <duration> > 0.
+	tenantRetentionRegex = `^([\w-]+):((\d{4}-\d{2}-\d{2})|(\d+d))$`
+)
+
 // ApplyRetentionPolicyByResolution removes blocks depending on the specified retentionByResolution based on blocks MaxTime.
 // A value of 0 disables the retention for its resolution.
 func ApplyRetentionPolicyByResolution(
@@ -47,3 +55,87 @@ func ApplyRetentionPolicyByResolution(
 	level.Info(logger).Log("msg", "optional retention apply done")
 	return nil
 }
+
+type RetentionPolicy struct {
+	CutoffDate        time.Time
+	RetentionDuration time.Duration
+}
+
+func (r RetentionPolicy) isExpired(blockMaxTime time.Time) bool {
+	if r.CutoffDate.IsZero() {
+		return time.Now().After(blockMaxTime.Add(r.RetentionDuration))
+	}
+	return r.CutoffDate.After(blockMaxTime)
+}
+
+func ParesRetentionPolicyByTenant(logger log.Logger, retentionTenants []string) (map[string]RetentionPolicy, error) {
+	pattern := regexp.MustCompile(tenantRetentionRegex)
+	retentionByTenant := make(map[string]RetentionPolicy, len(retentionTenants))
+	for _, tenantRetention := range retentionTenants {
+		matches := pattern.FindStringSubmatch(tenantRetention)
+		invalidFormat := errors.Errorf("invalid retention format for tenant: %s, must be `<tenant>:(<yyyy-mm-dd>|<duration>d)`", tenantRetention)
+		if len(matches) != 5 {
+			return nil, errors.Wrapf(invalidFormat, "matched size %d", len(matches))
+		}
+		tenant := matches[1]
+		var policy RetentionPolicy
+		if _, ok := retentionByTenant[tenant]; ok {
+			return nil, errors.Errorf("duplicate retention policy for tenant: %s", tenant)
+		}
+		if cutoffDate, err := time.Parse(time.DateOnly, matches[3]); matches[3] != "" {
+			if err != nil {
+				return nil, errors.Wrapf(invalidFormat, "error parsing cutoff date: %v", err)
+			}
+			policy.CutoffDate = cutoffDate
+		}
+		if duration, err := model.ParseDuration(matches[4]); matches[4] != "" {
+			if err != nil {
+				return nil, errors.Wrapf(invalidFormat, "error parsing duration: %v", err)
+			} else if duration == 0 {
+				return nil, errors.Wrapf(invalidFormat, "duration must be greater than 0")
+			}
+			policy.RetentionDuration = time.Duration(duration)
+		}
+		level.Info(logger).Log("msg", "retention policy for tenant is enabled", "tenant", tenant, "retention policy", fmt.Sprintf("%v", policy))
+		retentionByTenant[tenant] = policy
+	}
+	return retentionByTenant, nil
+}
+
+// ApplyRetentionPolicyByTenant removes blocks depending on the specified retentionByTenant based on blocks MaxTime.
+func ApplyRetentionPolicyByTenant(
+	ctx context.Context,
+	logger log.Logger,
+	bkt objstore.Bucket,
+	metas map[ulid.ULID]*metadata.Meta,
+	retentionByTenant map[string]RetentionPolicy,
+	blocksMarkedForDeletion prometheus.Counter) error {
+	if len(retentionByTenant) == 0 {
+		level.Info(logger).Log("msg", "tenant retention is disabled due to no policy")
+		return nil
+	}
+	level.Info(logger).Log("msg", "start tenant retention", "total", len(metas))
+	deleted, skipped, notExpired := 0, 0, 0
+	for id, m := range metas {
+		policy, ok := retentionByTenant[m.Thanos.GetTenant()]
+		if !ok {
+			skipped++
+			continue
+		}
+		maxTime := time.Unix(m.MaxTime/1000, 0)
+		if policy.isExpired(maxTime) {
+			level.Info(logger).Log("msg", "applying retention: marking block for deletion", "id", id, "maxTime", maxTime.String())
+			if err := block.Delete(ctx, logger, bkt, id); err != nil {
+				level.Error(logger).Log("msg", "failed to delete block", "id", id, "err", err)
+				continue // continue to next block to clean up backlogs
+			} else {
+				blocksMarkedForDeletion.Inc()
+				deleted++
+			}
+		} else {
+			notExpired++
+		}
+	}
+	level.Info(logger).Log("msg", "tenant retention apply done", "deleted", deleted, "skipped", skipped, "notExpired", notExpired)
+	return nil
+}