Implement stateless mode for Thanos Ruler

cmd/thanos/rule.go

@@ -26,10 +26,12 @@ import (
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/common/route"
+	"github.com/prometheus/prometheus/config"
 	"github.com/prometheus/prometheus/pkg/labels"
 	"github.com/prometheus/prometheus/pkg/relabel"
 	"github.com/prometheus/prometheus/promql"
 	"github.com/prometheus/prometheus/rules"
+	"github.com/prometheus/prometheus/storage"
 	"github.com/prometheus/prometheus/tsdb"
 	"github.com/prometheus/prometheus/util/strutil"
 	"github.com/thanos-io/thanos/pkg/errutil"
@@ -50,6 +52,7 @@ import (
 	"github.com/thanos-io/thanos/pkg/promclient"
 	"github.com/thanos-io/thanos/pkg/query"
 	thanosrules "github.com/thanos-io/thanos/pkg/rules"
+	"github.com/thanos-io/thanos/pkg/rules/remotewrite"
 	"github.com/thanos-io/thanos/pkg/runutil"
 	grpcserver "github.com/thanos-io/thanos/pkg/server/grpc"
 	httpserver "github.com/thanos-io/thanos/pkg/server/http"
@@ -75,6 +78,8 @@ type ruleConfig struct {
 	alertQueryURL          *url.URL
 	alertRelabelConfigYAML []byte
 
+	rwConfig *extflag.PathOrContent
+
 	resendDelay    time.Duration
 	evalInterval   time.Duration
 	ruleFiles      []string
@@ -117,6 +122,8 @@ func registerRule(app *extkingpin.App) {
 	cmd.Flag("eval-interval", "The default evaluation interval to use.").
 		Default("30s").DurationVar(&conf.evalInterval)
 
+	conf.rwConfig = extflag.RegisterPathOrContent(cmd, "remote-write.config", "YAML config for the remote-write server where samples should be sent to (see https://prometheus.io/docs/prometheus/latest/configuration/configuration/#remote_write). This automatically enables stateless mode for ruler and no series will be stored in the ruler's TSDB. If an empty config (or file) is provided, the flag is ignored and ruler is run with its own TSDB.", extflag.WithEnvSubstitution())
+
 	reqLogDecision := cmd.Flag("log.request.decision", "Deprecation Warning - This flag would be soon deprecated, and replaced with `request.logging-config`. Request Logging for logging the start and end of requests. By default this flag is disabled. LogFinishCall: Logs the finish call of the requests. LogStartAndFinishCall: Logs the start and finish call of the requests. NoLogCall: Disable request logging.").Default("").Enum("NoLogCall", "LogFinishCall", "LogStartAndFinishCall", "")
 
 	conf.objStoreConfig = extkingpin.RegisterCommonObjStoreFlags(cmd, "", false)
@@ -319,25 +326,52 @@ func runRule(
 		// Discover and resolve query addresses.
 		addDiscoveryGroups(g, queryClient, conf.query.dnsSDInterval)
 	}
+	var (
+		appendable storage.Appendable
+		queryable  storage.Queryable
+		db         *tsdb.DB
+	)
 
-	db, err := tsdb.Open(conf.dataDir, log.With(logger, "component", "tsdb"), reg, tsdbOpts, nil)
+	rwCfgYAML, err := conf.rwConfig.Content()
 	if err != nil {
-		return errors.Wrap(err, "open TSDB")
+		return err
 	}
 
-	level.Debug(logger).Log("msg", "removing storage lock file if any")
-	if err := removeLockfileIfAny(logger, conf.dataDir); err != nil {
-		return errors.Wrap(err, "remove storage lock files")
-	}
+	if len(rwCfgYAML) > 0 {
+		var rwCfg config.RemoteWriteConfig
+		rwCfg, err = remotewrite.LoadRemoteWriteConfig(rwCfgYAML)
+		if err != nil {
+			return err
+		}
+		walDir := filepath.Join(conf.dataDir, rwCfg.Name)
+		remoteStore, err := remotewrite.NewFanoutStorage(logger, reg, walDir, &rwCfg)
+		if err != nil {
+			return errors.Wrap(err, "set up remote-write store for ruler")
+		}
+		appendable = remoteStore
+		queryable = remoteStore
+	} else {
+		db, err = tsdb.Open(conf.dataDir, log.With(logger, "component", "tsdb"), reg, tsdbOpts, nil)
+		if err != nil {
+			return errors.Wrap(err, "open TSDB")
+		}
 
-	{
-		done := make(chan struct{})
-		g.Add(func() error {
-			<-done
-			return db.Close()
-		}, func(error) {
-			close(done)
-		})
+		level.Debug(logger).Log("msg", "removing storage lock file if any")
+		if err := removeLockfileIfAny(logger, conf.dataDir); err != nil {
+			return errors.Wrap(err, "remove storage lock files")
+		}
+
+		{
+			done := make(chan struct{})
+			g.Add(func() error {
+				<-done
+				return db.Close()
+			}, func(error) {
+				close(done)
+			})
+		}
+		appendable = db
+		queryable = db
 	}
 
 	// Build the Alertmanager clients.
@@ -435,9 +469,9 @@ func runRule(
 			rules.ManagerOptions{
 				NotifyFunc:  notifyFunc,
 				Logger:      logger,
-				Appendable:  db,
+				Appendable:  appendable,
 				ExternalURL: nil,
-				Queryable:   db,
+				Queryable:   queryable,
 				ResendDelay: conf.resendDelay,
 			},
 			queryFuncCreator(logger, queryClients, metrics.duplicatedQuery, metrics.ruleEvalWarnings, conf.query.httpMethod),
@@ -522,7 +556,7 @@ func runRule(
 	)
 
 	// Start gRPC server.
-	{
+	if db != nil {
 		tsdbStore := store.NewTSDBStore(logger, db, component.Rule, conf.lset)
 
 		tlsCfg, err := tls.NewServerConfig(log.With(logger, "protocol", "gRPC"), conf.grpc.tlsSrvCert, conf.grpc.tlsSrvKey, conf.grpc.tlsSrvClientCA)
@@ -547,6 +581,7 @@ func runRule(
 			s.Shutdown(err)
 		})
 	}
+
 	// Start UI & metrics HTTP server.
 	{
 		router := route.New()

docs/components/rule.md

@@ -344,6 +344,26 @@ Flags:
                                  (repeatable).
       --query.sd-interval=5m     Refresh interval to re-read file SD files.
                                  (used as a fallback)
+      --remote-write.config=<content>  
+                                 Alternative to 'remote-write.config-file' flag
+                                 (mutually exclusive). Content of YAML config
+                                 for the remote-write server where samples
+                                 should be sent to (see
+                                 https://prometheus.io/docs/prometheus/latest/configuration/configuration/#remote_write).
+                                 This automatically enables stateless mode for
+                                 ruler and no series will be stored in the
+                                 ruler's TSDB. If an empty config (or file) is
+                                 provided, the flag is ignored and ruler is run
+                                 with its own TSDB.
+      --remote-write.config-file=<file-path>  
+                                 Path to YAML config for the remote-write server
+                                 where samples should be sent to (see
+                                 https://prometheus.io/docs/prometheus/latest/configuration/configuration/#remote_write).
+                                 This automatically enables stateless mode for
+                                 ruler and no series will be stored in the
+                                 ruler's TSDB. If an empty config (or file) is
+                                 provided, the flag is ignored and ruler is run
+                                 with its own TSDB.
       --request.logging-config=<content>  
                                  Alternative to 'request.logging-config-file'
                                  flag (mutually exclusive). Content of YAML file

go.mod

@@ -61,6 +61,7 @@ require (
 	github.com/prometheus/common v0.29.0
 	github.com/prometheus/exporter-toolkit v0.6.0
 	github.com/prometheus/prometheus v1.8.2-0.20210720123808-b1ed4a0a663d
+	github.com/stretchr/testify v1.7.0
 	github.com/tencentyun/cos-go-sdk-v5 v0.7.31
 	github.com/uber/jaeger-client-go v2.29.1+incompatible
 	github.com/uber/jaeger-lib v2.4.1+incompatible

pkg/rules/remotewrite/remotewrite.go

@@ -0,0 +1,43 @@
+// Copyright (c) The Thanos Authors.
+// Licensed under the Apache License 2.0.
+
+package remotewrite
+
+import (
+	"time"
+
+	"github.com/pkg/errors"
+
+	"github.com/go-kit/kit/log"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/prometheus/config"
+	"github.com/prometheus/prometheus/storage"
+	"github.com/prometheus/prometheus/storage/remote"
+	"gopkg.in/yaml.v2"
+)
+
+// LoadRemoteWriteConfig prepares a RemoteWriteConfig instance from a given YAML config.
+func LoadRemoteWriteConfig(configYAML []byte) (config.RemoteWriteConfig, error) {
+	var cfg config.RemoteWriteConfig
+	if err := yaml.Unmarshal(configYAML, &cfg); err != nil {
+		return cfg, err
+	}
+	return cfg, nil
+}
+
+// NewFanoutStorage creates a storage that fans-out to both the WAL and a configured remote storage.
+// The remote storage tails the WAL and sends the metrics it reads using Prometheus' remote_write.
+func NewFanoutStorage(logger log.Logger, reg prometheus.Registerer, walDir string, rwConfig *config.RemoteWriteConfig) (storage.Storage, error) {
+	walStore, err := NewStorage(logger, reg, walDir)
+	if err != nil {
+		return nil, err
+	}
+	remoteStore := remote.NewStorage(logger, reg, walStore.StartTime, walStore.Directory(), 1*time.Minute, nil)
+	if err := remoteStore.ApplyConfig(&config.Config{
+		GlobalConfig:       config.DefaultGlobalConfig,
+		RemoteWriteConfigs: []*config.RemoteWriteConfig{rwConfig},
+	}); err != nil {
+		return nil, errors.Wrap(err, "applying config to remote storage")
+	}
+	return storage.NewFanout(logger, walStore, remoteStore), nil
+}