receiver: Add hot reload for relabel configs (#124)

Author: abhijith-db

cmd/thanos/receive.go

@@ -27,12 +27,10 @@ import (
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/model/labels"
-	"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"
 
 	"github.com/thanos-io/objstore"
 	"github.com/thanos-io/objstore/client"
@@ -231,14 +229,11 @@ func runReceive(
 		return errors.Wrapf(err, "migrate legacy storage in %v to default tenant %v", conf.dataDir, conf.defaultTenantID)
 	}
 
-	relabelContentYaml, err := conf.relabelConfigPath.Content()
+	relabeller, err := receive.NewRelabeller(conf.relabelConfigPath, reg, logger, conf.relabelConfigReloadTimer)
+
 	if err != nil {
 		return errors.Wrap(err, "get content of relabel configuration")
 	}
-	var relabelConfig []*relabel.Config
-	if err := yaml.Unmarshal(relabelContentYaml, &relabelConfig); err != nil {
-		return errors.Wrap(err, "parse relabel configuration")
-	}
 
 	dbs := receive.NewMultiTSDB(
 		conf.dataDir,
@@ -286,30 +281,47 @@ func runReceive(
 	}
 
 	webHandler := receive.NewHandler(log.With(logger, "component", "receive-handler"), &receive.Options{
-		Writer:               writer,
-		ListenAddress:        conf.rwAddress,
-		Registry:             reg,
-		Endpoint:             conf.endpoint,
-		TenantHeader:         conf.tenantHeader,
-		TenantField:          conf.tenantField,
-		DefaultTenantID:      conf.defaultTenantID,
-		ReplicaHeader:        conf.replicaHeader,
-		ReplicationFactor:    conf.replicationFactor,
-		RelabelConfigs:       relabelConfig,
-		ReceiverMode:         receiveMode,
-		Tracer:               tracer,
-		TLSConfig:            rwTLSConfig,
-		SplitTenantLabelName: conf.splitTenantLabelName,
-		DialOpts:             dialOpts,
-		ForwardTimeout:       time.Duration(*conf.forwardTimeout),
-		MaxBackoff:           time.Duration(*conf.maxBackoff),
-		TSDBStats:            dbs,
-		Limiter:              limiter,
-
+		Writer:                  writer,
+		ListenAddress:           conf.rwAddress,
+		Registry:                reg,
+		Endpoint:                conf.endpoint,
+		TenantHeader:            conf.tenantHeader,
+		TenantField:             conf.tenantField,
+		DefaultTenantID:         conf.defaultTenantID,
+		ReplicaHeader:           conf.replicaHeader,
+		ReplicationFactor:       conf.replicationFactor,
+		Relabeller:              relabeller,
+		ReceiverMode:            receiveMode,
+		Tracer:                  tracer,
+		TLSConfig:               rwTLSConfig,
+		SplitTenantLabelName:    conf.splitTenantLabelName,
+		DialOpts:                dialOpts,
+		ForwardTimeout:          time.Duration(*conf.forwardTimeout),
+		MaxBackoff:              time.Duration(*conf.maxBackoff),
+		TSDBStats:               dbs,
+		Limiter:                 limiter,
 		AsyncForwardWorkerCount: conf.asyncForwardWorkerCount,
 		ReplicationProtocol:     receive.ReplicationProtocol(conf.replicationProtocol),
 	})
 
+	{
+		if relabeller.CanReload() {
+			ctx, cancel := context.WithCancel(context.Background())
+			g.Add(func() error {
+				level.Debug(logger).Log("msg", "relabel config initialized with file watcher.")
+				if err := relabeller.StartConfigReloader(ctx); err != nil {
+					level.Error(logger).Log("msg", "initializing relabel config reloading.", "err", err)
+					return err
+				}
+				level.Info(logger).Log("msg", "relabel config reloading initialized.")
+				<-ctx.Done()
+				return nil
+			}, func(error) {
+				cancel()
+			})
+		}
+	}
+
 	grpcProbe := prober.NewGRPC()
 	httpProbe := prober.NewHTTP()
 	statusProber := prober.Combine(
@@ -974,8 +986,9 @@ type receiveConfig struct {
 	ignoreBlockSize       bool
 	allowOutOfOrderUpload bool
 
-	reqLogConfig      *extflag.PathOrContent
-	relabelConfigPath *extflag.PathOrContent
+	reqLogConfig             *extflag.PathOrContent
+	relabelConfigPath        *extflag.PathOrContent
+	relabelConfigReloadTimer time.Duration
 
 	writeLimitsConfig       *extflag.PathOrContent
 	storeRateLimits         store.SeriesSelectLimits
@@ -1073,6 +1086,8 @@ func (rc *receiveConfig) registerFlag(cmd extkingpin.FlagClause) {
 	rc.maxBackoff = extkingpin.ModelDuration(cmd.Flag("receive-forward-max-backoff", "Maximum backoff for each forward fan-out request").Default("5s").Hidden())
 
 	rc.relabelConfigPath = extflag.RegisterPathOrContent(cmd, "receive.relabel-config", "YAML file that contains relabeling configuration.", extflag.WithEnvSubstitution())
+	cmd.Flag("receive.relabel-config-reload-timer", "Minimum amount of time to pass for the relabel configuration to be reloaded. Helps to avoid excessive reloads.").
+		Default("0s").Hidden().DurationVar(&rc.relabelConfigReloadTimer)
 
 	rc.tsdbMinBlockDuration = extkingpin.ModelDuration(cmd.Flag("tsdb.min-block-duration", "Min duration for local TSDB blocks").Default("2h").Hidden())
 

pkg/receive/handler.go

@@ -112,7 +112,7 @@ type Options struct {
 	DialOpts                []grpc.DialOption
 	ForwardTimeout          time.Duration
 	MaxBackoff              time.Duration
-	RelabelConfigs          []*relabel.Config
+	Relabeller              *Relabeller
 	TSDBStats               TSDBStats
 	Limiter                 *Limiter
 	AsyncForwardWorkerCount uint
@@ -1104,13 +1104,14 @@ func (h *Handler) RemoteWrite(ctx context.Context, r *storepb.WriteRequest) (*st
 
 // relabel relabels the time series labels in the remote write request.
 func (h *Handler) relabel(wreq *prompb.WriteRequest) {
-	if len(h.options.RelabelConfigs) == 0 {
+	relabelConfigs := h.options.Relabeller.RelabelConfig()
+	if len(relabelConfigs) == 0 {
 		return
 	}
 	timeSeries := make([]prompb.TimeSeries, 0, len(wreq.Timeseries))
 	for _, ts := range wreq.Timeseries {
 		var keep bool
-		lbls, keep := relabel.Process(labelpb.ZLabelsToPromLabels(ts.Labels), h.options.RelabelConfigs...)
+		lbls, keep := relabel.Process(labelpb.ZLabelsToPromLabels(ts.Labels), relabelConfigs...)
 		if !keep {
 			continue
 		}

pkg/receive/handler_test.go

@@ -1733,7 +1733,7 @@ func TestRelabel(t *testing.T) {
 	} {
 		t.Run(tcase.name, func(t *testing.T) {
 			h := NewHandler(nil, &Options{
-				RelabelConfigs: tcase.relabel,
+				Relabeller: newRelabelerWithConstantConfig(tcase.relabel, nil),
 			})
 
 			h.relabel(&tcase.writeRequest)

pkg/receive/interfaces.go

@@ -0,0 +1,10 @@
+// Copyright (c) The Thanos Authors.
+// Licensed under the Apache License 2.0.
+
+package receive
+
+// fileContent is an interface to avoid a direct dependency on kingpin or extkingpin.
+type fileContent interface {
+	Content() ([]byte, error)
+	Path() string
+}

pkg/receive/limiter.go

@@ -64,12 +64,6 @@ type requestLimiter interface {
 	AllowSamples(tenant string, amount int64) bool
 }
 
-// fileContent is an interface to avoid a direct dependency on kingpin or extkingpin.
-type fileContent interface {
-	Content() ([]byte, error)
-	Path() string
-}
-
 func (l *Limiter) HeadSeriesLimiter() headSeriesLimiter {
 	l.headSeriesLimiterMtx.Lock()
 	defer l.headSeriesLimiterMtx.Unlock()
@@ -143,14 +137,6 @@ func NewLimiterWithOptions(
 				Help:      "How many times the limit configuration failed to reload.",
 			},
 		)
-		limiter.configReloadFailedCounter = promauto.With(limiter.registerer).NewCounter(
-			prometheus.CounterOpts{
-				Namespace: "thanos",
-				Subsystem: "receive",
-				Name:      "limits_config_reload_err_total",
-				Help:      "How many times the limit configuration failed to reload.",
-			},
-		)
 		limiter.maxPendingRequestLimitHit = promauto.With(limiter.registerer).NewCounter(
 			prometheus.CounterOpts{
 				Name: "thanos_receive_max_pending_write_request_limit_hit_total",

pkg/receive/relabeller.go

@@ -0,0 +1,147 @@
+// Copyright (c) The Thanos Authors.
+// Licensed under the Apache License 2.0.
+
+package receive
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/go-kit/log"
+	"github.com/go-kit/log/level"
+	"github.com/pkg/errors"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+	"github.com/prometheus/prometheus/model/relabel"
+	"github.com/thanos-io/thanos/pkg/extkingpin"
+	"go.uber.org/atomic"
+	"gopkg.in/yaml.v2"
+)
+
+// Relabeller is responsible for managing the configuration and initialization of
+// different types that apply relabel configurations to the Receive instance.
+// The new config is atomically swapped in, so we don't need to use locks.
+type Relabeller struct {
+	configPathOrContent       fileContent
+	relabelConfigs            *atomic.Pointer[RelabelConfig]
+	logger                    log.Logger
+	configReloadCounter       prometheus.Counter
+	configReloadFailedCounter prometheus.Counter
+	configReloadTimer         time.Duration
+}
+
+// RelabelConfig is a collection of relabel configurations.
+type RelabelConfig []*relabel.Config
+
+// NewRelabeller creates a new relabeller and loads the configuration to make sure loading is possible.
+func NewRelabeller(configFile fileContent, reg prometheus.Registerer, logger log.Logger, configReloadTimer time.Duration) (*Relabeller, error) {
+	var relabelConfigs atomic.Pointer[RelabelConfig]
+	relabelConfigs.Store(&RelabelConfig{})
+	relabeller := &Relabeller{
+		configPathOrContent: configFile,
+		relabelConfigs:      &relabelConfigs,
+		logger:              logger,
+		configReloadTimer:   configReloadTimer,
+	}
+
+	if reg != nil {
+		relabeller.configReloadCounter = promauto.With(reg).NewCounter(
+			prometheus.CounterOpts{
+				Namespace: "thanos",
+				Subsystem: "receive",
+				Name:      "relabel_config_reload_total",
+				Help:      "How many times the relabel configuration was reloaded",
+			},
+		)
+		relabeller.configReloadFailedCounter = promauto.With(reg).NewCounter(
+			prometheus.CounterOpts{
+				Namespace: "thanos",
+				Subsystem: "receive",
+				Name:      "relabel_config_reload_err_total",
+				Help:      "How many times the relabel configuration failed to reload.",
+			},
+		)
+	}
+	if configFile == nil {
+		return relabeller, nil
+	}
+	relabeller.configPathOrContent = configFile
+	if err := relabeller.loadConfig(); err != nil {
+		return nil, errors.Wrap(err, "load relabel config")
+	}
+	return relabeller, nil
+}
+
+// simply returns the provided config. This is useful for testing.
+func newRelabelerWithConstantConfig(config RelabelConfig, logger log.Logger) *Relabeller {
+	var relabelConfigs atomic.Pointer[RelabelConfig]
+	relabelConfigs.Store(&config)
+	return &Relabeller{nil, &relabelConfigs, logger, nil, nil, 0}
+}
+
+// RelabelConfig returns the current relabel config.
+// This is concurrent safe.
+func (r *Relabeller) RelabelConfig() RelabelConfig {
+	if r == nil {
+		var relabelConfig RelabelConfig
+		return relabelConfig
+	}
+	return *r.relabelConfigs.Load()
+}
+
+// setRelabelConfig sets the relabel config to the provided array.
+// This is concurrent safe.
+func (r *Relabeller) setRelabelConfig(configs RelabelConfig) {
+	r.relabelConfigs.Store(&configs)
+}
+
+func (r *Relabeller) loadConfig() error {
+	relabelContentYaml, err := r.configPathOrContent.Content()
+	if err != nil {
+		return errors.Wrap(err, "getting content of relabel config")
+	}
+	var relabelConfig RelabelConfig
+	if err := yaml.Unmarshal(relabelContentYaml, &relabelConfig); err != nil {
+		return errors.Wrap(err, "parsing relabel config")
+	}
+	r.setRelabelConfig(relabelConfig)
+	return nil
+}
+
+// StartConfigReloader starts the automatic configuration reloader based off of
+// the file indicated by pathOrContent.
+func (r *Relabeller) StartConfigReloader(ctx context.Context) error {
+	if !r.CanReload() {
+		return nil
+	}
+
+	return extkingpin.PathContentReloader(ctx, r.configPathOrContent, r.logger, func() {
+		level.Info(r.logger).Log("msg", "reloading relabel config.")
+
+		if err := r.loadConfig(); err != nil {
+			if failedReload := r.configReloadFailedCounter; failedReload != nil {
+				failedReload.Inc()
+			}
+			errMsg := fmt.Sprintf("error reloading relabel config from %s", r.configPathOrContent.Path())
+			level.Error(r.logger).Log("msg", errMsg, "err", err)
+		}
+		if reloadCounter := r.configReloadCounter; reloadCounter != nil {
+			reloadCounter.Inc()
+		}
+
+	}, r.configReloadTimer)
+}
+
+func (r *Relabeller) CanReload() bool {
+	if r.configReloadTimer == 0 {
+		return false
+	}
+	if r.configPathOrContent == nil {
+		return false
+	}
+	if r.configPathOrContent.Path() == "" {
+		return false
+	}
+	return true
+}