Adds support for reloader in Promscale.

Signed-off-by: Harkishen-Singh <[email protected]>

Author: Harkishen-Singh

CHANGELOG.md

@@ -19,6 +19,7 @@ We use the following categories for changes:
 - `-enable-feature=promql-per-step-stats` feature for statistics in PromQL evaluation
 - Add `readinessProbe` in helm chart [#1266]
 - Telemetry for recording rules and alerting [#1424]
+- Ability to reload rules and alerting config [#1426]
 
 ### Fixed
 - Trace query returns empty result when queried with 

pkg/api/reload.go

@@ -0,0 +1,26 @@
+// This file and its contents are licensed under the Apache License 2.0.
+// Please see the included NOTICE for copyright information and
+// LICENSE for a copy of the license.
+
+package api
+
+import (
+	"fmt"
+	"net/http"
+
+	"github.com/timescale/promscale/pkg/log"
+)
+
+func Reload(reload chan<- chan error) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		errCh := make(chan error, 1)
+		// Inline with what Prometheus does for reloading API.
+		// https://github.com/prometheus/prometheus/blob/e166cf402a9d2b838d5fba72d5b60117e9265898/web/web.go#L774
+		reload <- errCh
+		if err := <-errCh; err != nil {
+			log.Error("msg", "failed to reload", "err", err.Error())
+			http.Error(w, fmt.Errorf("failed to reload: %w", err).Error(), http.StatusInternalServerError)
+		}
+		w.Header().Set("Content-Length", "0")
+	}
+}

pkg/api/router.go

@@ -29,7 +29,7 @@ import (
 	"github.com/timescale/promscale/pkg/telemetry"
 )
 
-func GenerateRouter(apiConf *Config, promqlConf *query.Config, client *pgclient.Client, query *jaegerQuery.Query) (*mux.Router, error) {
+func GenerateRouter(apiConf *Config, promqlConf *query.Config, client *pgclient.Client, query *jaegerQuery.Query, reload chan<- chan error) (*mux.Router, error) {
 	var writePreprocessors []parser.Preprocessor
 	if apiConf.HighAvailability {
 		service := ha.NewService(haClient.NewLeaseClient(client.Connection))
@@ -101,6 +101,9 @@ func GenerateRouter(apiConf *Config, promqlConf *query.Config, client *pgclient.
 	router.Path("/healthz").Methods(http.MethodGet, http.MethodOptions, http.MethodHead).HandlerFunc(Health(healthChecker))
 	router.Path(apiConf.TelemetryPath).Methods(http.MethodGet).HandlerFunc(promhttp.Handler().ServeHTTP)
 
+	reloadHandler := timeHandler(metrics.HTTPRequestDuration, "/-/reload", Reload(reload))
+	router.Path("/-/reload").Methods(http.MethodPost).HandlerFunc(reloadHandler)
+
 	jaeger.ExtendQueryAPIs(router, client.Connection, query)
 
 	debugProf := router.PathPrefix("/debug/pprof").Subrouter()

pkg/reloader/reloader.go

@@ -0,0 +1,118 @@
+// This file and its contents are licensed under the Apache License 2.0.
+// Please see the included NOTICE for copyright information and
+// LICENSE for a copy of the license.
+
+package reloader
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"os/signal"
+	"sync"
+	"syscall"
+
+	"github.com/timescale/promscale/pkg/log"
+)
+
+type service struct {
+	name    string
+	payload func() error
+}
+
+type loader struct {
+	ctx      context.Context
+	mux      sync.Mutex
+	services []service
+	webAdmin bool
+	// Signal from an external routine. We accept an error channel to respond
+	// to the http request in case some error happens during reloading.
+	sigExt <-chan chan error
+}
+
+// New creates a new thread-safe loader that can reloads a set of services
+// added through the Add() function.
+//
+// Reloading can be done by:
+//	1. SIGHUP. This is implemented by default
+//	2. External call. This happens on receiving a signal from the given sig.
+//		The loader responds to an error channel after reloading,
+//		indicating whether the reload was successful or not.
+//
+// Services for reloading can be added by Add(). When the reloading
+// starts, all services added are reloaded serially to ensure that
+// any dependent service is reloaded first. This also means that
+// each reload is a sync process and waits for the current service
+// to complete reloading before moving to the next.
+//
+// Note:
+//	1. All services are loaded once after calling Run()
+//	2. Reloading stops if any of the services give out an error while reloading.
+// 	   This is important since the following dependent services might crash on reloading
+//	   if reloading their child do not succeed
+//	2. Loader becomes useless after a context cancel
+func New(ctx context.Context, webAdmin bool, sig <-chan chan error) *loader {
+	return &loader{
+		ctx:      ctx,
+		webAdmin: webAdmin,
+		services: make([]service, 0),
+		sigExt:   sig,
+	}
+}
+
+// Add adds the service to the list of reloadable services.
+// The given update func is called whenever a request to reload
+// is received.
+func (l *loader) Add(serviceName string, update func() error) {
+	l.mux.Lock()
+	defer l.mux.Unlock()
+	l.services = append(l.services, service{serviceName, update})
+}
+
+func (l *loader) Run() error {
+	sighup := make(chan os.Signal, 1)
+	signal.Notify(sighup, syscall.SIGHUP)
+
+	if err := l.reload(); err != nil {
+		return fmt.Errorf("error reloading: %w", err)
+	}
+	for {
+		select {
+		case <-l.ctx.Done():
+			l.services = nil // Make services GC-able.
+			return nil
+		case <-sighup:
+			if err := l.reload(); err != nil {
+				log.Error("msg", "stopping reload", "reason", err.Error())
+			}
+		case errCh := <-l.sigExt:
+			if !l.webAdmin {
+				// Reloads via external calls are secured by web-admin.
+				err := fmt.Errorf("reload received but web admin is disabled. To enable, start Promscale with '-web.enable-admin-api' flag")
+				log.Error("msg", err.Error())
+				errCh <- err
+				continue
+			}
+			if err := l.reload(); err != nil {
+				log.Error("msg", "stopping reload", "reason", err.Error())
+				errCh <- err
+			} else {
+				errCh <- nil
+			}
+		}
+	}
+}
+
+func (l *loader) reload() error {
+	l.mux.Lock()
+	defer l.mux.Unlock()
+
+	// Reload services serially, in the order of their addition.
+	for _, s := range l.services {
+		log.Debug("msg", "reloading...", "service-name", s.name)
+		if err := s.payload(); err != nil {
+			return fmt.Errorf("error reloading '%s'", s.name)
+		}
+	}
+	return nil
+}

pkg/runner/runner.go

@@ -25,6 +25,7 @@ import (
 
 	"github.com/google/uuid"
 	grpc_prometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
+	"github.com/pkg/errors"
 
 	"github.com/jaegertracing/jaeger/plugin/storage/grpc/shared"
 	"github.com/prometheus/client_golang/prometheus"
@@ -36,6 +37,7 @@ import (
 	"github.com/timescale/promscale/pkg/pgclient"
 	"github.com/timescale/promscale/pkg/pgmodel/ingestor/trace"
 	dbMetrics "github.com/timescale/promscale/pkg/pgmodel/metrics/database"
+	"github.com/timescale/promscale/pkg/reloader"
 	"github.com/timescale/promscale/pkg/rules"
 	"github.com/timescale/promscale/pkg/telemetry"
 	"github.com/timescale/promscale/pkg/thanos"
@@ -138,8 +140,9 @@ func Run(cfg *Config) error {
 	}
 
 	jaegerQuery := query.New(client.QuerierConnection, &cfg.TracingCfg)
+	reloadSig := make(chan chan error, 1)
 
-	router, err := api.GenerateRouter(&cfg.APICfg, &cfg.PromQLCfg, client, jaegerQuery)
+	router, err := api.GenerateRouter(&cfg.APICfg, &cfg.PromQLCfg, client, jaegerQuery, reloadSig)
 	if err != nil {
 		log.Error("msg", "aborting startup due to error", "err", fmt.Sprintf("generate router: %s", err.Error()))
 		return fmt.Errorf("generate router: %w", err)
@@ -222,6 +225,10 @@ func Run(cfg *Config) error {
 		},
 	)
 
+	loadCtx, stopLoader := context.WithCancel(context.Background())
+	defer stopLoader()
+	loader := reloader.New(loadCtx, cfg.APICfg.AdminAPIEnabled, reloadSig)
+
 	pendingTelemetry := map[string]string{
 		"rules_enabled":    "false", // Will be written in telemetry table as `promscale_rules_enabled: false`
 		"alerting_enabled": "false", // `promscale_alerting_enabled: false`
@@ -243,15 +250,19 @@ func Run(cfg *Config) error {
 		}
 		cfg.APICfg.Rules = manager
 
+		loader.Add("rules-manager", func() error {
+			err := rules.Validate(&cfg.RulesCfg) // This refreshes the cfg.RulesCfg.PrometheusConfig entry in RulesCfg.
+			if err != nil {
+				return fmt.Errorf("error validating rules-config: %w", err)
+			}
+			return errors.WithMessage(manager.ApplyConfig(cfg.RulesCfg.PrometheusConfig), "error applying Prometheus configuration to rules manager")
+		})
+
 		group.Add(
 			func() error {
-				promCfg := cfg.RulesCfg.PrometheusConfig
-				if err = manager.ApplyConfig(promCfg); err != nil {
-					return fmt.Errorf("error applying Prometheus configuration to rules manager: %w", err)
-				}
 				log.Info("msg", "Started Rule-Manager")
 				return manager.Run()
-			}, func(err error) {
+			}, func(error) {
 				log.Info("msg", "Stopping Rule-Manager")
 				stopRuler()
 			},
@@ -267,6 +278,14 @@ func Run(cfg *Config) error {
 		}
 	}()
 
+	group.Add(
+		func() error {
+			return loader.Run()
+		}, func(error) {
+			stopLoader()
+		},
+	)
+
 	mux := http.NewServeMux()
 	mux.Handle("/", router)
 

pkg/tests/end_to_end_tests/promql_endpoint_integration_test.go

@@ -324,7 +324,7 @@ func buildRouterWithAPIConfig(pool *pgxpool.Pool, cfg *api.Config) (*mux.Router,
 
 	jaegerQuery := jaegerquery.New(pgClient.QuerierConnection, &jaegerquery.DefaultConfig)
 
-	router, err := api.GenerateRouter(cfg, qryCfg, pgClient, jaegerQuery)
+	router, err := api.GenerateRouter(cfg, qryCfg, pgClient, jaegerQuery, nil)
 	if err != nil {
 		return nil, nil, fmt.Errorf("generate router: %w", err)
 	}