cherry pick to release branch

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

@@ -46,6 +46,7 @@ import (
 	"github.com/thanos-io/thanos/pkg/info"
 	"github.com/thanos-io/thanos/pkg/info/infopb"
 	"github.com/thanos-io/thanos/pkg/logging"
+	"github.com/thanos-io/thanos/pkg/pantheon"
 	"github.com/thanos-io/thanos/pkg/prober"
 	"github.com/thanos-io/thanos/pkg/receive"
 	"github.com/thanos-io/thanos/pkg/runutil"
@@ -58,8 +59,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 +146,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 +309,7 @@ func runReceive(
 		Endpoint:                conf.endpoint,
 		TenantHeader:            conf.tenantHeader,
 		TenantField:             conf.tenantField,
+		ScopeHeader:             conf.scopeHeader,
 		DefaultTenantID:         conf.defaultTenantID,
 		ReplicaHeader:           conf.replicaHeader,
 		ReplicationFactor:       conf.replicationFactor,
@@ -357,6 +380,13 @@ func runReceive(
 		}
 	}
 
+	level.Debug(logger).Log("msg", "setting up Pantheon config")
+	{
+		if err := setupPantheonConfig(g, logger, reg, conf, webHandler); err != nil {
+			return err
+		}
+	}
+
 	level.Debug(logger).Log("msg", "setting up HTTP server")
 	{
 		srv := httpserver.New(logger, reg, comp, httpProbe,
@@ -365,10 +395,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 +492,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 +501,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 +750,57 @@ func setupHashring(g *run.Group,
 	return nil
 }
 
+// setupPantheonConfig sets up the Pantheon cluster configuration watcher if provided.
+func setupPantheonConfig(g *run.Group,
+	logger log.Logger,
+	reg *prometheus.Registry,
+	conf *receiveConfig,
+	webHandler *receive.Handler,
+) error {
+	if conf.pantheonFilePath == "" {
+		return nil
+	}
+
+	cw, err := receive.NewPantheonConfigWatcher(log.With(logger, "component", "pantheon-config-watcher"), reg, conf.pantheonFilePath, *conf.pantheonRefreshInterval)
+	if err != nil {
+		return errors.Wrap(err, "failed to initialize Pantheon config watcher")
+	}
+
+	// Check the Pantheon configuration before running the watcher.
+	if err := cw.ValidateConfig(); err != nil {
+		cw.Stop()
+		return errors.Wrap(err, "failed to validate Pantheon configuration file")
+	}
+
+	updates := make(chan *pantheon.PantheonCluster, 1)
+	ctx, cancel := context.WithCancel(context.Background())
+	g.Add(func() error {
+		return receive.PantheonConfigFromWatcher(ctx, updates, cw)
+	}, func(error) {
+		cancel()
+	})
+
+	cancelConsumer := make(chan struct{})
+	g.Add(func() error {
+		for {
+			select {
+			case c, ok := <-updates:
+				if !ok {
+					return nil
+				}
+				webHandler.SetPantheonCluster(c)
+				level.Info(logger).Log("msg", "Updated Pantheon cluster configuration.")
+			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 +1041,14 @@ type receiveConfig struct {
 	hashringsFileContent string
 	hashringsAlgorithm   string
 
+	pantheonFilePath        string
+	pantheonRefreshInterval *model.Duration
+
 	refreshInterval     *model.Duration
 	endpoint            string
 	tenantHeader        string
 	tenantField         string
+	scopeHeader         string
 	tenantLabelName     string
 	defaultTenantID     string
 	replicaHeader       string
@@ -979,6 +1070,8 @@ type receiveConfig struct {
 	tsdbMemorySnapshotOnShutdown bool
 	tsdbDisableFlushOnShutdown   bool
 	tsdbEnableNativeHistograms   bool
+	tsdbEnableTenantPathPrefix   bool
+	tsdbPathSegmentsBeforeTenant []string
 
 	walCompression       bool
 	noLockFile           bool
@@ -1007,7 +1100,8 @@ type receiveConfig struct {
 	maxPendingGrpcWriteRequests       int
 	lazyRetrievalMaxBufferedResponses int
 
-	featureList *[]string
+	featureList     *[]string
+	noUploadTenants *[]string
 }
 
 func (rc *receiveConfig) registerFlag(cmd extkingpin.FlagClause) {
@@ -1059,12 +1153,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.pantheon-file", "Path to file that contains the Pantheon cluster configuration. A watcher is initialized to watch changes and update the configuration dynamically.").PlaceHolder("<path>").StringVar(&rc.pantheonFilePath)
+
+	rc.pantheonRefreshInterval = extkingpin.ModelDuration(cmd.Flag("receive.pantheon-file-refresh-interval", "Refresh interval to re-read the Pantheon 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 +1244,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 +1284,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)
 }

go.mod

@@ -2,6 +2,8 @@ module github.com/thanos-io/thanos
 
 go 1.23.0
 
+toolchain go1.23.11
+
 require (
 	cloud.google.com/go/storage v1.43.0 // indirect
 	cloud.google.com/go/trace v1.10.12
@@ -80,10 +82,10 @@ require (
 	go.uber.org/atomic v1.11.0
 	go.uber.org/automaxprocs v1.5.3
 	go.uber.org/goleak v1.3.0
-	golang.org/x/crypto v0.32.0
-	golang.org/x/net v0.34.0
-	golang.org/x/sync v0.10.0
-	golang.org/x/text v0.21.0
+	golang.org/x/crypto v0.40.0
+	golang.org/x/net v0.41.0
+	golang.org/x/sync v0.16.0
+	golang.org/x/text v0.27.0
 	golang.org/x/time v0.7.0
 	google.golang.org/api v0.195.0 // indirect
 	google.golang.org/genproto v0.0.0-20240823204242-4ba0660f739c // indirect
@@ -112,7 +114,7 @@ require (
 )
 
 require (
-	capnproto.org/go/capnp/v3 v3.0.0-alpha.30
+	capnproto.org/go/capnp/v3 v3.0.0-alpha.29
 	github.com/Azure/azure-sdk-for-go/sdk/storage/azdatalake v1.4.0
 	github.com/cortexproject/promqlsmith v0.0.0-20240506042652-6cfdd9739a5e
 	github.com/grpc-ecosystem/go-grpc-middleware/providers/prometheus v1.0.1
@@ -125,6 +127,7 @@ require (
 	go.opentelemetry.io/contrib/propagators/autoprop v0.54.0
 	go4.org/intern v0.0.0-20230525184215-6c62f75575cb
 	golang.org/x/exp v0.0.0-20240613232115-7f521ea00fb8
+	gopkg.in/natefinch/lumberjack.v2 v2.2.1
 )
 
 require github.com/dgryski/go-metro v0.0.0-20200812162917-85c65e2d0165 // indirect
@@ -142,7 +145,7 @@ require (
 	github.com/go-openapi/runtime v0.27.1 // indirect
 	github.com/goccy/go-json v0.10.3 // indirect
 	github.com/godbus/dbus/v5 v5.0.4 // indirect
-	github.com/golang-jwt/jwt/v5 v5.2.1 // indirect
+	github.com/golang-jwt/jwt/v5 v5.3.0 // indirect
 	github.com/google/s2a-go v0.1.8 // indirect
 	github.com/huaweicloud/huaweicloud-sdk-go-obs v3.23.3+incompatible // indirect
 	github.com/jcchavezs/porto v0.1.0 // indirect
@@ -179,7 +182,7 @@ require (
 	github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751 // indirect
 	github.com/alicebob/gopher-json v0.0.0-20200520072559-a9ecdc9d1d3a // indirect
 	github.com/aliyun/aliyun-oss-go-sdk v2.2.2+incompatible // indirect
-	github.com/armon/go-radix v1.0.0 // indirect
+	github.com/armon/go-radix v1.0.0
 	github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2 // indirect
 	github.com/aws/aws-sdk-go v1.55.5 // indirect
 	github.com/aws/aws-sdk-go-v2 v1.16.0 // indirect
@@ -275,10 +278,10 @@ require (
 	go.opentelemetry.io/otel/metric v1.31.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.3.1 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
-	golang.org/x/mod v0.21.0 // indirect
-	golang.org/x/oauth2 v0.23.0 // indirect
-	golang.org/x/sys v0.29.0 // indirect
-	golang.org/x/tools v0.24.0 // indirect
+	golang.org/x/mod v0.25.0 // indirect
+	golang.org/x/oauth2 v0.30.0 // indirect
+	golang.org/x/sys v0.34.0 // indirect
+	golang.org/x/tools v0.34.0 // indirect
 	gonum.org/v1/gonum v0.15.0 // indirect
 	google.golang.org/protobuf v1.35.1
 	howett.net/plist v0.0.0-20181124034731-591f970eefbb // indirect