cherry pick to release branch

CHANGELOG.md

@@ -12,6 +12,8 @@ We use *breaking :warning:* to mark changes that are not backward compatible (re
 
 ### Fixed
 
+- [#8378](https://github.com/thanos-io/thanos/pull/8378): Store: fix the reuse of dirty posting slices
+
 ### Added
 
 ### Changed

cmd/thanos/query.go

@@ -244,6 +244,8 @@ func registerQuery(app *extkingpin.App) {
 	tenantCertField := cmd.Flag("query.tenant-certificate-field", "Use TLS client's certificate field to determine tenant for write requests. Must be one of "+tenancy.CertificateFieldOrganization+", "+tenancy.CertificateFieldOrganizationalUnit+" or "+tenancy.CertificateFieldCommonName+". This setting will cause the query.tenant-header flag value to be ignored.").Default("").Enum("", tenancy.CertificateFieldOrganization, tenancy.CertificateFieldOrganizationalUnit, tenancy.CertificateFieldCommonName)
 	enforceTenancy := cmd.Flag("query.enforce-tenancy", "Enforce tenancy on Query APIs. Responses are returned only if the label value of the configured tenant-label-name and the value of the tenant header matches.").Default("false").Bool()
 	tenantLabel := cmd.Flag("query.tenant-label-name", "Label name to use when enforcing tenancy (if --query.enforce-tenancy is enabled).").Default(tenancy.DefaultTenantLabel).String()
+	exclusiveExternalLabels := cmd.Flag("query.exclusive-external-labels", "Comma-separated list of label names for store matching. If any stores have all of those labels and match the values not using any regex operators, e.g., .*, other stores should be filtered out.").
+		Default("").Strings()
 
 	rewriteAggregationLabelStrategy := cmd.Flag("query.aggregation-label-strategy", "The strategy to use when rewriting aggregation labels. Used during aggregator migration only.").Default(string(query.NoopLabelRewriter)).Hidden().Enum(string(query.NoopLabelRewriter), string(query.UpsertLabelRewriter), string(query.InsertOnlyLabelRewriter))
 	rewriteAggregationLabelTo := cmd.Flag("query.aggregation-label-value-override", "The value override for aggregation label. If set to x, all queries on aggregated metrics will have a `__agg_rule_type__=x` matcher. If empty, this behavior is disabled. Default is empty.").Hidden().Default("").String()
@@ -254,6 +256,9 @@ func registerQuery(app *extkingpin.App) {
 	grpcStoreClientKeepAlivePingInterval := extkingpin.ModelDuration(cmd.Flag("query.grcp-store-client-keep-alive-ping-interval", "This value defines how often a store client sends a keepalive ping on an established gRPC stream. 0 means not to set. NB: a client is keeping a long‐running gRPC stream open. It still has active RPCs on the wire—even if Recv() is not called in a while. Setting PermitWithoutStream=false only stops pings when no streams exist; it does not suppress pings during an open stream").
 		Default("0s"))
 
+	blockQueryMetricsWithoutFilter := cmd.Flag("query.block-query-metrics-without-filter", "Comma-separated list of metric patterns to block queries without sufficient label filters. Helps prevent high-cardinality metric queries.").Default("").String()
+	forwardPartialStrategy := cmd.Flag("query.forward-partial-strategy", "Enable forward partial strategy for queries. This is used for a Queier stacked on the top of other Queiers.").Default("false").Bool()
+
 	var storeRateLimits store.SeriesSelectLimits
 	storeRateLimits.RegisterFlags(cmd)
 
@@ -315,6 +320,16 @@ func registerQuery(app *extkingpin.App) {
 			return err
 		}
 
+		// Parse blocked metric patterns
+		var blockedMetricPatterns []string
+		if *blockQueryMetricsWithoutFilter != "" {
+			blockedMetricPatterns = strings.Split(*blockQueryMetricsWithoutFilter, ",")
+			for i, pattern := range blockedMetricPatterns {
+				blockedMetricPatterns[i] = strings.TrimSpace(pattern)
+			}
+			level.Info(logger).Log("msg", "blocking query metrics without filter feature enabled", "patterns", strings.Join(blockedMetricPatterns, ","))
+		}
+
 		return runQuery(
 			g,
 			logger,
@@ -397,6 +412,9 @@ func registerQuery(app *extkingpin.App) {
 			*rewriteAggregationLabelTo,
 			*lazyRetrievalMaxBufferedResponses,
 			time.Duration(*grpcStoreClientKeepAlivePingInterval),
+			blockedMetricPatterns,
+			*forwardPartialStrategy,
+			*exclusiveExternalLabels,
 		)
 	})
 }
@@ -485,6 +503,9 @@ func runQuery(
 	rewriteAggregationLabelTo string,
 	lazyRetrievalMaxBufferedResponses int,
 	grpcStoreClientKeepAlivePingInterval time.Duration,
+	blockedMetricPatterns []string,
+	forwardPartialStrategy bool,
+	exclusiveExternalLabels []string,
 ) error {
 	comp := component.Query
 	if alertQueryURL == "" {
@@ -580,6 +601,17 @@ func runQuery(
 		store.WithLazyRetrievalMaxBufferedResponsesForProxy(lazyRetrievalMaxBufferedResponses),
 	}
 
+	// Add blocked metric patterns option if specified
+	if len(blockedMetricPatterns) > 0 {
+		options = append(options, store.WithBlockedMetricPatterns(blockedMetricPatterns))
+	}
+	if forwardPartialStrategy {
+		options = append(options, store.WithoutForwardPartialStrategy())
+	}
+	if len(exclusiveExternalLabels) > 0 {
+		options = append(options, store.WithExclusiveExternalLabels(exclusiveExternalLabels))
+	}
+
 	// Parse and sanitize the provided replica labels flags.
 	queryReplicaLabels = strutil.ParseFlagLabels(queryReplicaLabels)
 

cmd/thanos/receive.go

@@ -58,8 +58,9 @@ import (
 )
 
 const (
-	compressionNone   = "none"
-	metricNamesFilter = "metric-names-filter"
+	compressionNone          = "none"
+	metricNamesFilter        = "metric-names-filter"
+	grpcReadinessInterceptor = "grpc-readiness-interceptor"
 )
 
 func registerReceive(app *extkingpin.App) {
@@ -144,11 +145,31 @@ func runReceive(
 	level.Info(logger).Log("mode", receiveMode, "msg", "running receive")
 
 	multiTSDBOptions := []receive.MultiTSDBOption{}
+	var enableGRPCReadinessInterceptor bool
 	for _, feature := range *conf.featureList {
 		if feature == metricNamesFilter {
 			multiTSDBOptions = append(multiTSDBOptions, receive.WithMetricNameFilterEnabled())
 			level.Info(logger).Log("msg", "metric name filter feature enabled")
 		}
+		if feature == grpcReadinessInterceptor {
+			enableGRPCReadinessInterceptor = true
+			level.Info(logger).Log("msg", "gRPC readiness interceptor feature enabled")
+		}
+	}
+
+	if len(*conf.noUploadTenants) > 0 {
+		multiTSDBOptions = append(multiTSDBOptions, receive.WithNoUploadTenants(*conf.noUploadTenants))
+		level.Info(logger).Log("msg", "configured tenants for local storage only", "tenants", strings.Join(*conf.noUploadTenants, ","))
+	}
+
+	if conf.tsdbEnableTenantPathPrefix {
+		multiTSDBOptions = append(multiTSDBOptions, receive.WithTenantPathPrefix())
+		level.Info(logger).Log("msg", "tenant path prefix feature enabled")
+	}
+
+	if len(conf.tsdbPathSegmentsBeforeTenant) > 0 {
+		multiTSDBOptions = append(multiTSDBOptions, receive.WithPathSegmentsBeforeTenant(conf.tsdbPathSegmentsBeforeTenant))
+		level.Info(logger).Log("msg", "tenant path segments before tenant feature enabled", "segments", path.Join(conf.tsdbPathSegmentsBeforeTenant...))
 	}
 
 	// Create a matcher converter if specified by command line to cache expensive regex matcher conversions.
@@ -287,6 +308,7 @@ func runReceive(
 		Endpoint:                conf.endpoint,
 		TenantHeader:            conf.tenantHeader,
 		TenantField:             conf.tenantField,
+		ScopeHeader:             conf.scopeHeader,
 		DefaultTenantID:         conf.defaultTenantID,
 		ReplicaHeader:           conf.replicaHeader,
 		ReplicationFactor:       conf.replicationFactor,
@@ -350,11 +372,22 @@ func runReceive(
 		}
 	}
 
-	level.Debug(logger).Log("msg", "setting up hashring")
-	{
-		if err := setupHashring(g, logger, reg, conf, hashringChangedChan, webHandler, statusProber, enableIngestion, dbs); err != nil {
-			return err
+	// Choose between PantheonV2 unified config or legacy separate configs.
+	if conf.pantheonV2WriterFilePath != "" {
+		level.Debug(logger).Log("msg", "setting up PantheonV2 writer config (unified hashring + pantheon)")
+		{
+			if err := setupPantheonV2WriterConfig(g, logger, reg, conf, hashringChangedChan, webHandler, statusProber, enableIngestion, dbs); err != nil {
+				return err
+			}
 		}
+	} else {
+		level.Debug(logger).Log("msg", "setting up hashring")
+		{
+			if err := setupHashring(g, logger, reg, conf, hashringChangedChan, webHandler, statusProber, enableIngestion, dbs); err != nil {
+				return err
+			}
+		}
+
 	}
 
 	level.Debug(logger).Log("msg", "setting up HTTP server")
@@ -365,10 +398,10 @@ func runReceive(
 			httpserver.WithTLSConfig(*conf.httpTLSConfig),
 		)
 		srv.Handle("/-/downscale", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-			tenants := dbs.GetTenants()
-			n := len(tenants)
+			activeTenants := dbs.GetActiveTenants()
+			n := len(activeTenants)
 			w.Header().Set("Tenant-Count", strconv.Itoa(n))
-			for _, tname := range tenants {
+			for _, tname := range activeTenants {
 				w.Header().Add("Tenants", tname)
 			}
 			if n > 0 {
@@ -462,7 +495,7 @@ func runReceive(
 			info.WithExemplarsInfoFunc(),
 		)
 
-		srv := grpcserver.New(logger, receive.NewUnRegisterer(reg), tracer, grpcLogOpts, logFilterMethods, comp, grpcProbe,
+		grpcOptions := []grpcserver.Option{
 			grpcserver.WithServer(store.RegisterStoreServer(rw, logger)),
 			grpcserver.WithServer(store.RegisterWritableStoreServer(rw)),
 			grpcserver.WithServer(exemplars.RegisterExemplarsServer(exemplars.NewMultiTSDB(dbs.TSDBExemplars))),
@@ -471,7 +504,13 @@ func runReceive(
 			grpcserver.WithGracePeriod(conf.grpcConfig.gracePeriod),
 			grpcserver.WithMaxConnAge(conf.grpcConfig.maxConnectionAge),
 			grpcserver.WithTLSConfig(tlsCfg),
-		)
+		}
+
+		if enableGRPCReadinessInterceptor {
+			grpcOptions = append(grpcOptions, receive.NewReadinessGRPCOptions(httpProbe)...)
+		}
+
+		srv := grpcserver.New(logger, receive.NewUnRegisterer(reg), tracer, grpcLogOpts, logFilterMethods, comp, grpcProbe, grpcOptions...)
 
 		g.Add(
 			func() error {
@@ -714,6 +753,97 @@ func setupHashring(g *run.Group,
 	return nil
 }
 
+// setupPantheonV2WriterConfig sets up the PantheonV2 writer configuration watcher if provided.
+// This replaces both setupHashring and setupPantheonConfig when using the unified config file.
+func setupPantheonV2WriterConfig(g *run.Group,
+	logger log.Logger,
+	reg *prometheus.Registry,
+	conf *receiveConfig,
+	hashringChangedChan chan struct{},
+	webHandler *receive.Handler,
+	statusProber prober.Probe,
+	enableIngestion bool,
+	dbs *receive.MultiTSDB,
+) error {
+	if conf.pantheonV2WriterFilePath == "" {
+		return nil
+	}
+
+	cw, err := receive.NewPantheonV2WriterConfigWatcher(log.With(logger, "component", "pantheonv2-writer-config-watcher"), reg, conf.pantheonV2WriterFilePath, *conf.pantheonV2WriterRefreshInterval)
+	if err != nil {
+		return errors.Wrap(err, "failed to initialize PantheonV2 writer config watcher")
+	}
+
+	// Check the PantheonV2 writer configuration before running the watcher.
+	if err := cw.ValidateConfig(); err != nil {
+		cw.Stop()
+		return errors.Wrap(err, "failed to validate PantheonV2 writer configuration file")
+	}
+
+	updates := make(chan *receive.PantheonV2WriterConfig, 1)
+	ctx, cancel := context.WithCancel(context.Background())
+	g.Add(func() error {
+		return receive.PantheonV2WriterConfigFromWatcher(ctx, updates, cw)
+	}, func(error) {
+		cancel()
+	})
+
+	cancelConsumer := make(chan struct{})
+	algorithm := receive.HashringAlgorithm(conf.hashringsAlgorithm)
+	g.Add(func() error {
+		if enableIngestion {
+			defer close(hashringChangedChan)
+		}
+
+		for {
+			select {
+			case c, ok := <-updates:
+				if !ok {
+					return nil
+				}
+
+				// Update hashring from the config.
+				if len(c.Hashrings) == 0 {
+					webHandler.Hashring(receive.SingleNodeHashring(conf.endpoint))
+					level.Info(logger).Log("msg", "Empty hashring config in PantheonV2 writer config. Set up single node hashring.")
+				} else {
+					h, err := receive.NewMultiHashring(algorithm, conf.replicationFactor, c.Hashrings)
+					if err != nil {
+						return errors.Wrap(err, "unable to create new hashring from PantheonV2 writer config")
+					}
+					webHandler.Hashring(h)
+					level.Info(logger).Log("msg", "Set up hashring from PantheonV2 writer config.")
+				}
+
+				if err := dbs.SetHashringConfig(c.Hashrings); err != nil {
+					return errors.Wrap(err, "failed to set hashring config in MultiTSDB from PantheonV2 writer config")
+				}
+
+				// Update PantheonCluster from the config.
+				if c.PantheonCluster != nil {
+					webHandler.SetPantheonCluster(c.PantheonCluster)
+					level.Info(logger).Log("msg", "Updated Pantheon cluster configuration from PantheonV2 writer config.")
+				}
+
+				// If ingestion is enabled, send a signal to TSDB to flush.
+				if enableIngestion {
+					hashringChangedChan <- struct{}{}
+				} else {
+					// If not, just signal we are ready (this is important during first hashring load)
+					statusProber.Ready()
+				}
+
+			case <-cancelConsumer:
+				return nil
+			}
+		}
+	}, func(err error) {
+		close(cancelConsumer)
+	})
+
+	return nil
+}
+
 // startTSDBAndUpload starts the multi-TSDB and sets up the rungroup to flush the TSDB and reload on hashring change.
 // It also upload blocks to object store, if upload is enabled.
 func startTSDBAndUpload(g *run.Group,
@@ -954,10 +1084,14 @@ type receiveConfig struct {
 	hashringsFileContent string
 	hashringsAlgorithm   string
 
+	pantheonV2WriterFilePath        string
+	pantheonV2WriterRefreshInterval *model.Duration
+
 	refreshInterval     *model.Duration
 	endpoint            string
 	tenantHeader        string
 	tenantField         string
+	scopeHeader         string
 	tenantLabelName     string
 	defaultTenantID     string
 	replicaHeader       string
@@ -979,6 +1113,8 @@ type receiveConfig struct {
 	tsdbMemorySnapshotOnShutdown bool
 	tsdbDisableFlushOnShutdown   bool
 	tsdbEnableNativeHistograms   bool
+	tsdbEnableTenantPathPrefix   bool
+	tsdbPathSegmentsBeforeTenant []string
 
 	walCompression       bool
 	noLockFile           bool
@@ -1007,7 +1143,8 @@ type receiveConfig struct {
 	maxPendingGrpcWriteRequests       int
 	lazyRetrievalMaxBufferedResponses int
 
-	featureList *[]string
+	featureList     *[]string
+	noUploadTenants *[]string
 }
 
 func (rc *receiveConfig) registerFlag(cmd extkingpin.FlagClause) {
@@ -1059,12 +1196,19 @@ func (rc *receiveConfig) registerFlag(cmd extkingpin.FlagClause) {
 	rc.refreshInterval = extkingpin.ModelDuration(cmd.Flag("receive.hashrings-file-refresh-interval", "Refresh interval to re-read the hashring configuration file. (used as a fallback)").
 		Default("5m"))
 
+	cmd.Flag("receive.pantheonv2-writer-file", "Path to file that contains the PantheonV2 writer configuration (hashrings + pantheon cluster). A watcher is initialized to watch changes and update the configuration dynamically. Takes precedence over receive.hashrings-file.").PlaceHolder("<path>").StringVar(&rc.pantheonV2WriterFilePath)
+
+	rc.pantheonV2WriterRefreshInterval = extkingpin.ModelDuration(cmd.Flag("receive.pantheonv2-writer-file-refresh-interval", "Refresh interval to re-read the PantheonV2 writer configuration file. (used as a fallback)").
+		Default("5m"))
+
 	cmd.Flag("receive.local-endpoint", "Endpoint of local receive node. Used to identify the local node in the hashring configuration. If it's empty AND hashring configuration was provided, it means that receive will run in RoutingOnly mode.").StringVar(&rc.endpoint)
 
 	cmd.Flag("receive.tenant-header", "HTTP header to determine tenant for write requests.").Default(tenancy.DefaultTenantHeader).StringVar(&rc.tenantHeader)
 
 	cmd.Flag("receive.tenant-certificate-field", "Use TLS client's certificate field to determine tenant for write requests. Must be one of "+tenancy.CertificateFieldOrganization+", "+tenancy.CertificateFieldOrganizationalUnit+" or "+tenancy.CertificateFieldCommonName+". This setting will cause the receive.tenant-header flag value to be ignored.").Default("").EnumVar(&rc.tenantField, "", tenancy.CertificateFieldOrganization, tenancy.CertificateFieldOrganizationalUnit, tenancy.CertificateFieldCommonName)
 
+	cmd.Flag("receive.scope-header", "HTTP header to determine scope for write requests. Used for Pantheon-based tenant attribution.").Default(tenancy.DefaultScopeHeader).StringVar(&rc.scopeHeader)
+
 	cmd.Flag("receive.default-tenant-id", "Default tenant ID to use when none is provided via a header.").Default(tenancy.DefaultTenant).StringVar(&rc.defaultTenantID)
 
 	cmd.Flag("receive.split-tenant-label-name", "Label name through which the request will be split into multiple tenants. This takes precedence over the HTTP header.").Default("").StringVar(&rc.splitTenantLabelName)
@@ -1143,6 +1287,15 @@ func (rc *receiveConfig) registerFlag(cmd extkingpin.FlagClause) {
 		"[EXPERIMENTAL] Enables the ingestion of native histograms.").
 		Default("false").Hidden().BoolVar(&rc.tsdbEnableNativeHistograms)
 
+	cmd.Flag("tsdb.enable-tenant-path-prefix",
+		"[EXPERIMENTAL] Enables the tenant path prefix for object storage.").
+		Default("false").Hidden().BoolVar(&rc.tsdbEnableTenantPathPrefix)
+
+	cmd.Flag("tsdb.path-segments-before-tenant",
+		"[EXPERIMENTAL] Specifies the path segments before the tenant for object storage."+
+			"Must only be used in combination with tsdb.enable-tenant-path-prefix.").
+		Default("raw").Hidden().StringsVar(&rc.tsdbPathSegmentsBeforeTenant)
+
 	cmd.Flag("writer.intern",
 		"[EXPERIMENTAL] Enables string interning in receive writer, for more optimized memory usage.").
 		Default("false").Hidden().BoolVar(&rc.writerInterning)
@@ -1174,7 +1327,8 @@ func (rc *receiveConfig) registerFlag(cmd extkingpin.FlagClause) {
 		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()
+	rc.featureList = cmd.Flag("enable-feature", "Experimental feature names to enable. The current list of features is "+metricNamesFilter+", "+grpcReadinessInterceptor+". Repeat this flag to enable multiple features.").Strings()
+	rc.noUploadTenants = cmd.Flag("receive.no-upload-tenants", "Tenant IDs/patterns that should only store data locally (no object store upload). Supports exact matches (e.g., 'tenant1') and prefix patterns (e.g., 'prod-*'). Repeat this flag to specify multiple patterns.").Strings()
 	cmd.Flag("receive.lazy-retrieval-max-buffered-responses", "The lazy retrieval strategy can buffer up to this number of responses. This is to limit the memory usage. This flag takes effect only when the lazy retrieval strategy is enabled.").
 		Default("20").IntVar(&rc.lazyRetrievalMaxBufferedResponses)
 }
@@ -1183,7 +1337,7 @@ func (rc *receiveConfig) registerFlag(cmd extkingpin.FlagClause) {
 // This is used to configure this Receiver's forwarding and ingesting behavior at runtime.
 func (rc *receiveConfig) determineMode() receive.ReceiverMode {
 	// Has the user provided some kind of hashring configuration?
-	hashringSpecified := rc.hashringsFileContent != "" || rc.hashringsFilePath != ""
+	hashringSpecified := rc.hashringsFileContent != "" || rc.hashringsFilePath != "" || rc.pantheonV2WriterFilePath != ""
 	// Has the user specified the --receive.local-endpoint flag?
 	localEndpointSpecified := rc.endpoint != ""