Add periodic config/secrets reload (#31)

* Add periodic config/secrets reload

By default it happens every 60s and the interval is configurable.

* Remove config reload mutex

`select` does not run each case concurrently, so there's no need to synchronize here.

* Fix formatting

* Use non-zero config reload interval in test

* Readd tenants with changed OIDC params on config reload

main.go

@@ -16,18 +16,20 @@ import (
 	monitoringv1 "github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring/v1"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/collectors"
-	"github.com/rhobs/obsctl-reloader/pkg/loader"
-	"github.com/rhobs/obsctl-reloader/pkg/loop"
-	"github.com/rhobs/obsctl-reloader/pkg/syncer"
 	"k8s.io/client-go/kubernetes/scheme"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/apiutil"
 	k8sconfig "sigs.k8s.io/controller-runtime/pkg/client/config"
+
+	"github.com/rhobs/obsctl-reloader/pkg/loader"
+	"github.com/rhobs/obsctl-reloader/pkg/loop"
+	"github.com/rhobs/obsctl-reloader/pkg/syncer"
 )
 
 const (
-	obsctlContextAPIName        = "api"
-	defaultSleepDurationSeconds = 15
+	obsctlContextAPIName               = "api"
+	defaultSleepDurationSeconds        = 15
+	defaultConfigReloadIntervalSeconds = 60
 )
 
 type cfg struct {
@@ -39,6 +41,7 @@ type cfg struct {
 	logRulesEnabled      bool
 	logLevel             string
 	listenInternal       string
+	configReloadInterval uint
 }
 
 func setupLogger(logLevel string) log.Logger {
@@ -68,6 +71,7 @@ func parseFlags() *cfg {
 
 	// Common flags.
 	flag.UintVar(&cfg.sleepDurationSeconds, "sleep-duration-seconds", defaultSleepDurationSeconds, "The interval in seconds after which all PrometheusRules are synced to Observatorium API.")
+	flag.UintVar(&cfg.configReloadInterval, "config-reload-interval-seconds", defaultConfigReloadIntervalSeconds, "The interval in seconds for reloading configuration.")
 	flag.StringVar(&cfg.observatoriumURL, "observatorium-api-url", "", "The URL of the Observatorium API to which rules will be synced.")
 	flag.StringVar(&cfg.managedTenants, "managed-tenants", "", "The name of the tenants whose rules should be synced. If there are multiple tenants, ensure they are comma-separated.")
 	flag.StringVar(&cfg.issuerURL, "issuer-url", "", "The OIDC issuer URL, see https://openid.net/specs/openid-connect-discovery-1_0.html#IssuerDiscovery.")
@@ -149,7 +153,7 @@ func main() {
 		reg,
 	)
 	if err := o.InitOrReloadObsctlConfig(); err != nil {
-		level.Error(logger).Log("msg", "error reloading/initializing obsctl config", "error", err)
+		level.Error(logger).Log("msg", "error initializing obsctl config", "error", err)
 		panic(err)
 	}
 
@@ -165,6 +169,7 @@ func main() {
 				o,
 				cfg.logRulesEnabled,
 				cfg.sleepDurationSeconds,
+				cfg.configReloadInterval,
 			)
 		}, func(_ error) {
 			cancel()

main_test.go

@@ -10,6 +10,7 @@ import (
 	"github.com/go-kit/log"
 	lokiv1 "github.com/grafana/loki/operator/apis/loki/v1"
 	monitoringv1 "github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring/v1"
+
 	"github.com/rhobs/obsctl-reloader/pkg/loop"
 )
 
@@ -82,7 +83,7 @@ func TestSyncLoop(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	time.AfterFunc(25*time.Second, func() { cancel() })
 
-	testutil.Ok(t, loop.SyncLoop(ctx, log.NewLogfmtLogger(log.NewSyncWriter(os.Stderr)), rl, rs, true, 5))
+	testutil.Ok(t, loop.SyncLoop(ctx, log.NewLogfmtLogger(log.NewSyncWriter(os.Stderr)), rl, rs, true, 5, 60))
 
 	testutil.Equals(t, 12, rs.setCurrentTenantCnt)
 	testutil.Equals(t, 4, rs.metricsRulesCnt)

pkg/loop/syncloop.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
+
 	"github.com/rhobs/obsctl-reloader/pkg/loader"
 	"github.com/rhobs/obsctl-reloader/pkg/syncer"
 )
@@ -19,9 +20,14 @@ func SyncLoop(
 	o syncer.RulesSyncer,
 	logRulesEnabled bool,
 	sleepDurationSeconds uint,
+	configReloadIntervalSeconds uint,
 ) error {
 	for {
 		select {
+		case <-time.After(time.Duration(configReloadIntervalSeconds) * time.Second):
+			if err := o.InitOrReloadObsctlConfig(); err != nil {
+				level.Error(logger).Log("msg", "error reloading obsctl config", "error", err)
+			}
 		case <-time.After(time.Duration(sleepDurationSeconds) * time.Second):
 			prometheusRules, err := k.GetPrometheusRules()
 			if err != nil {

pkg/syncer/obsctlsyncer.go

@@ -218,6 +218,15 @@ func (o *ObsctlRulesSyncer) InitOrReloadObsctlConfig() error {
 			}
 		}
 
+		existingTenantCfg, foundTenant := o.c.APIs[obsctlContextAPIName].Contexts[tenant]
+		if foundTenant && !o.tenantConfigMatches(existingTenantCfg, tenantCfg) {
+			err := o.c.RemoveTenant(o.logger, tenantCfg.Tenant, obsctlContextAPIName)
+			if err != nil {
+				// We don't really care about the error here, logging only for visibility.
+				level.Info(o.logger).Log("msg", "removing tenant", "tenant", tenant, "error", err)
+			}
+		}
+
 		if err := o.c.AddTenant(o.logger, tenantCfg.Tenant, obsctlContextAPIName, tenantCfg.Tenant, tenantCfg.OIDC); err != nil {
 			level.Error(o.logger).Log("msg", "adding tenant", "tenant", tenant, "error", err)
 			return errors.Wrap(err, "adding tenant to obsctl config")
@@ -227,6 +236,24 @@ func (o *ObsctlRulesSyncer) InitOrReloadObsctlConfig() error {
 	return nil
 }
 
+// tenantConfigMatches checks if two tenant configs are equal. We consider them equal if they have the same tenant name
+// and OIDC config (regardless of any token that might've been already acquired and cached).
+func (o *ObsctlRulesSyncer) tenantConfigMatches(firstConfig, secondConfig config.TenantConfig) bool {
+	if firstConfig.Tenant != secondConfig.Tenant {
+		return false
+	}
+
+	if firstConfig.OIDC != secondConfig.OIDC {
+		return false
+	}
+
+	return firstConfig.OIDC.ClientID == secondConfig.OIDC.ClientID &&
+		firstConfig.OIDC.ClientSecret == secondConfig.OIDC.ClientSecret &&
+		firstConfig.OIDC.Audience == secondConfig.OIDC.Audience &&
+		firstConfig.OIDC.IssuerURL == secondConfig.OIDC.IssuerURL &&
+		firstConfig.OIDC.OfflineAccess == secondConfig.OIDC.OfflineAccess
+}
+
 func (o *ObsctlRulesSyncer) SetCurrentTenant(tenant string) error {
 	if err := o.c.SetCurrentContext(o.logger, obsctlContextAPIName, tenant); err != nil {
 		level.Error(o.logger).Log("msg", "switching context", "tenant", tenant, "error", err)