feat: prevent burst of proxy/layers update

internal/cache/cache.go

@@ -3,4 +3,5 @@ package cache
 type Cache[K comparable, V any] interface {
 	Get(key K) (V, bool)
 	Set(key K, value V)
+	Clear()
 }

internal/cache/memory/cache.go

@@ -25,6 +25,10 @@ func (c *Cache[K, V]) Set(key K, value V) {
 	c.store.Store(key, value)
 }
 
+func (c *Cache[K, V]) Clear() {
+	c.store.Clear()
+}
+
 func NewCache[K comparable, V any]() *Cache[K, V] {
 	return &Cache[K, V]{
 		store: new(sync.Map),

internal/cache/ttl/cache.go

@@ -28,6 +28,11 @@ func (c *Cache[K, V]) Set(key K, value V) {
 	c.values.Set(key, value)
 }
 
+func (c *Cache[K, V]) Clear() {
+	c.timestamps.Clear()
+	c.values.Clear()
+}
+
 func NewCache[K comparable, V any](values cache.Cache[K, V], timestamps cache.Cache[K, time.Time], ttl time.Duration) *Cache[K, V] {
 	return &Cache[K, V]{
 		values:     values,

internal/command/server/admin/run.go

@@ -12,10 +12,6 @@ import (
 	"gitlab.com/wpetit/goweb/logger"
 )
 
-const (
-	flagPrintDefaultToken = "print-default-token"
-)
-
 func RunCommand() *cli.Command {
 	flags := common.Flags()
 
@@ -32,6 +28,8 @@ func RunCommand() *cli.Command {
 			logger.SetFormat(logger.Format(conf.Logger.Format))
 			logger.SetLevel(logger.Level(conf.Logger.Level))
 
+			logger.Debug(ctx.Context, "using config", logger.F("config", conf))
+
 			projectVersion := ctx.String("projectVersion")
 
 			if conf.Proxy.Sentry.DSN != "" {

internal/command/server/proxy/run.go

@@ -29,6 +29,8 @@ func RunCommand() *cli.Command {
 			logger.SetFormat(logger.Format(conf.Logger.Format))
 			logger.SetLevel(logger.Level(conf.Logger.Level))
 
+			logger.Debug(ctx.Context, "using config", logger.F("config", conf))
+
 			projectVersion := ctx.String("projectVersion")
 
 			if conf.Proxy.Sentry.DSN != "" {
@@ -49,7 +51,7 @@ func RunCommand() *cli.Command {
 				proxy.WithServerConfig(conf.Proxy),
 				proxy.WithRedisConfig(conf.Redis),
 				proxy.WithDirectorLayers(layers...),
-				proxy.WithDirectorCacheTTL(time.Duration(conf.Proxy.Cache.TTL)),
+				proxy.WithDirectorCacheTTL(time.Duration(*conf.Proxy.Cache.TTL)),
 			)
 
 			addrs, srvErrs := srv.Start(ctx.Context)

internal/config/environment.go

@@ -19,7 +19,7 @@ func (is *InterpolatedString) UnmarshalYAML(value *yaml.Node) error {
 		return errors.WithStack(err)
 	}
 
-	str, err := envsubst.EvalEnv(str)
+	str, err := envsubst.Eval(str, getEnv)
 	if err != nil {
 		return errors.WithStack(err)
 	}
@@ -38,7 +38,7 @@ func (ii *InterpolatedInt) UnmarshalYAML(value *yaml.Node) error {
 		return errors.Wrapf(err, "could not decode value '%v' (line '%d') into string", value.Value, value.Line)
 	}
 
-	str, err := envsubst.EvalEnv(str)
+	str, err := envsubst.Eval(str, getEnv)
 	if err != nil {
 		return errors.WithStack(err)
 	}
@@ -62,7 +62,7 @@ func (ifl *InterpolatedFloat) UnmarshalYAML(value *yaml.Node) error {
 		return errors.Wrapf(err, "could not decode value '%v' (line '%d') into string", value.Value, value.Line)
 	}
 
-	str, err := envsubst.EvalEnv(str)
+	str, err := envsubst.Eval(str, getEnv)
 	if err != nil {
 		return errors.WithStack(err)
 	}
@@ -86,7 +86,7 @@ func (ib *InterpolatedBool) UnmarshalYAML(value *yaml.Node) error {
 		return errors.Wrapf(err, "could not decode value '%v' (line '%d') into string", value.Value, value.Line)
 	}
 
-	str, err := envsubst.EvalEnv(str)
+	str, err := envsubst.Eval(str, getEnv)
 	if err != nil {
 		return errors.WithStack(err)
 	}
@@ -101,9 +101,10 @@ func (ib *InterpolatedBool) UnmarshalYAML(value *yaml.Node) error {
 	return nil
 }
 
+var getEnv = os.Getenv
+
 type InterpolatedMap struct {
-	Data   map[string]any
+	Data map[string]any
-	getEnv func(string) string
 }
 
 func (im *InterpolatedMap) UnmarshalYAML(value *yaml.Node) error {
@@ -113,10 +114,6 @@ func (im *InterpolatedMap) UnmarshalYAML(value *yaml.Node) error {
 		return errors.Wrapf(err, "could not decode value '%v' (line '%d') into map", value.Value, value.Line)
 	}
 
-	if im.getEnv == nil {
-		im.getEnv = os.Getenv
-	}
-
 	interpolated, err := im.interpolateRecursive(data)
 	if err != nil {
 		return errors.WithStack(err)
@@ -140,7 +137,7 @@ func (im InterpolatedMap) interpolateRecursive(data any) (any, error) {
 		}
 
 	case string:
-		value, err := envsubst.Eval(typ, im.getEnv)
+		value, err := envsubst.Eval(typ, getEnv)
 		if err != nil {
 			return nil, errors.WithStack(err)
 		}
@@ -171,7 +168,7 @@ func (iss *InterpolatedStringSlice) UnmarshalYAML(value *yaml.Node) error {
 	}
 
 	for index, value := range data {
-		value, err := envsubst.EvalEnv(value)
+		value, err := envsubst.Eval(value, getEnv)
 		if err != nil {
 			return errors.WithStack(err)
 		}
@@ -193,7 +190,7 @@ func (id *InterpolatedDuration) UnmarshalYAML(value *yaml.Node) error {
 		return errors.Wrapf(err, "could not decode value '%v' (line '%d') into string", value.Value, value.Line)
 	}
 
-	str, err := envsubst.EvalEnv(str)
+	str, err := envsubst.Eval(str, getEnv)
 	if err != nil {
 		return errors.WithStack(err)
 	}

internal/config/environment_test.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"os"
 	"testing"
+	"time"
 
 	"github.com/pkg/errors"
 	"gopkg.in/yaml.v3"
@@ -65,7 +66,7 @@ func TestInterpolatedMap(t *testing.T) {
 			var interpolatedMap InterpolatedMap
 
 			if tc.Env != nil {
-				interpolatedMap.getEnv = func(key string) string {
+				getEnv = func(key string) string {
 					return tc.Env[key]
 				}
 			}
@@ -80,3 +81,54 @@ func TestInterpolatedMap(t *testing.T) {
 		})
 	}
 }
+
+func TestInterpolatedDuration(t *testing.T) {
+	type testCase struct {
+		Path   string
+		Env    map[string]string
+		Assert func(t *testing.T, parsed *InterpolatedDuration)
+	}
+
+	testCases := []testCase{
+		{
+			Path: "testdata/environment/interpolated-duration.yml",
+			Env: map[string]string{
+				"MY_DURATION": "30s",
+			},
+			Assert: func(t *testing.T, parsed *InterpolatedDuration) {
+				if e, g := 30*time.Second, parsed; e != time.Duration(*g) {
+					t.Errorf("parsed: expected '%v', got '%v'", e, g)
+				}
+			},
+		},
+	}
+
+	for idx, tc := range testCases {
+		t.Run(fmt.Sprintf("Case #%d", idx), func(t *testing.T) {
+			data, err := os.ReadFile(tc.Path)
+			if err != nil {
+				t.Fatalf("%+v", errors.WithStack(err))
+			}
+
+			if tc.Env != nil {
+				getEnv = func(key string) string {
+					return tc.Env[key]
+				}
+			}
+
+			config := struct {
+				Duration *InterpolatedDuration `yaml:"duration"`
+			}{
+				Duration: NewInterpolatedDuration(-1),
+			}
+
+			if err := yaml.Unmarshal(data, &config); err != nil {
+				t.Fatalf("%+v", errors.WithStack(err))
+			}
+
+			if tc.Assert != nil {
+				tc.Assert(t, config.Duration)
+			}
+		})
+	}
+}

internal/config/proxy_server.go

@@ -113,12 +113,12 @@ func NewDefaultDialConfig() DialConfig {
 }
 
 type CacheConfig struct {
-	TTL InterpolatedDuration `yaml:"ttl"`
+	TTL *InterpolatedDuration `yaml:"ttl"`
 }
 
 func NewDefaultCacheConfig() CacheConfig {
 	return CacheConfig{
-		TTL: *NewInterpolatedDuration(time.Second * 30),
+		TTL: NewInterpolatedDuration(time.Second * 30),
 	}
 }
 

internal/config/testdata/environment/interpolated-duration.yml

@@ -0,0 +1 @@
+duration: ${MY_DURATION}

internal/proxy/director/director.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"net/http"
 	"sort"
+	"sync"
 
 	"forge.cadoles.com/Cadoles/go-proxy"
 	"forge.cadoles.com/Cadoles/go-proxy/wildcard"
@@ -23,6 +24,9 @@ type Director struct {
 	proxyCache cache.Cache[string, []*store.Proxy]
 	layerCache cache.Cache[string, []*store.Layer]
 
+	proxyCacheLock sync.RWMutex
+	layerCacheLock sync.RWMutex
+
 	handleError HandleErrorFunc
 }
 
@@ -99,9 +103,32 @@ const proxiesCacheKey = "proxies"
 func (d *Director) getProxies(ctx context.Context) ([]*store.Proxy, error) {
 	proxies, exists := d.proxyCache.Get(proxiesCacheKey)
 	if exists {
+		logger.Debug(ctx, "using cached proxies")
 		return proxies, nil
 	}
 
+	locked := d.proxyCacheLock.TryLock()
+	if !locked {
+		d.proxyCacheLock.RLock()
+
+		proxies, exists := d.proxyCache.Get(proxiesCacheKey)
+		if exists {
+			d.proxyCacheLock.RUnlock()
+			logger.Debug(ctx, "using cached proxies")
+			return proxies, nil
+		}
+
+		d.proxyCacheLock.RUnlock()
+	}
+
+	if !locked {
+		d.proxyCacheLock.Lock()
+	}
+
+	defer d.proxyCacheLock.Unlock()
+
+	logger.Debug(ctx, "querying fresh proxies")
+
 	headers, err := d.proxyRepository.QueryProxy(ctx, store.WithProxyQueryEnabled(true))
 	if err != nil {
 		return nil, errors.WithStack(err)
@@ -134,9 +161,32 @@ func (d *Director) getLayers(ctx context.Context, proxyName store.ProxyName) ([]
 
 	layers, exists := d.layerCache.Get(cacheKey)
 	if exists {
+		logger.Debug(ctx, "using cached layers")
 		return layers, nil
 	}
 
+	locked := d.layerCacheLock.TryLock()
+	if !locked {
+		d.layerCacheLock.RLock()
+
+		layers, exists := d.layerCache.Get(cacheKey)
+		if exists {
+			d.layerCacheLock.RUnlock()
+			logger.Debug(ctx, "using cached layers")
+			return layers, nil
+		}
+
+		d.layerCacheLock.RUnlock()
+	}
+
+	if !locked {
+		d.layerCacheLock.Lock()
+	}
+
+	defer d.layerCacheLock.Unlock()
+
+	logger.Debug(ctx, "querying fresh layers")
+
 	headers, err := d.layerRepository.QueryLayers(ctx, proxyName, store.WithLayerQueryEnabled(true))
 	if err != nil {
 		return nil, errors.WithStack(err)

internal/proxy/server.go

@@ -13,8 +13,11 @@ import (
 	"net/http/httputil"
 	"net/http/pprof"
 	"net/url"
+	"os"
+	"os/signal"
 	"path/filepath"
 	"strconv"
+	"syscall"
 	"time"
 
 	"forge.cadoles.com/Cadoles/go-proxy"
@@ -94,24 +97,15 @@ func (s *Server) run(parentCtx context.Context, addrs chan net.Addr, errs chan e
 
 	logger.Info(ctx, "http server listening")
 
+	layerCache, proxyCache, cancel := s.createDirectorCaches(ctx)
+	defer cancel()
+
 	director := director.New(
 		s.proxyRepository,
 		s.layerRepository,
 		director.WithLayers(s.directorLayers...),
-		director.WithLayerCache(
+		director.WithLayerCache(layerCache),
-			ttl.NewCache(
+		director.WithProxyCache(proxyCache),
-				memory.NewCache[string, []*store.Layer](),
-				memory.NewCache[string, time.Time](),
-				s.directorCacheTTL,
-			),
-		),
-		director.WithProxyCache(
-			ttl.NewCache(
-				memory.NewCache[string, []*store.Proxy](),
-				memory.NewCache[string, time.Time](),
-				s.directorCacheTTL,
-			),
-		),
 		director.WithHandleErrorFunc(s.handleError),
 	)
 
@@ -205,6 +199,44 @@ func (s *Server) run(parentCtx context.Context, addrs chan net.Addr, errs chan e
 	logger.Info(ctx, "http server exiting")
 }
 
+func (s *Server) createDirectorCaches(ctx context.Context) (*ttl.Cache[string, []*store.Layer], *ttl.Cache[string, []*store.Proxy], func()) {
+	layerCache := ttl.NewCache(
+		memory.NewCache[string, []*store.Layer](),
+		memory.NewCache[string, time.Time](),
+		s.directorCacheTTL,
+	)
+
+	proxyCache := ttl.NewCache(
+		memory.NewCache[string, []*store.Proxy](),
+		memory.NewCache[string, time.Time](),
+		s.directorCacheTTL,
+	)
+
+	sig := make(chan os.Signal, 1)
+
+	signal.Notify(sig, syscall.SIGUSR2)
+
+	go func() {
+		for {
+			_, ok := <-sig
+			if !ok {
+				return
+			}
+
+			logger.Info(ctx, "received sigusr2 signal, clearing proxies and layers cache")
+
+			layerCache.Clear()
+			proxyCache.Clear()
+		}
+	}()
+
+	cancel := func() {
+		close(sig)
+	}
+
+	return layerCache, proxyCache, cancel
+}
+
 func (s *Server) createReverseProxy(ctx context.Context, target *url.URL) *httputil.ReverseProxy {
 	reverseProxy := httputil.NewSingleHostReverseProxy(target)
 

misc/packaging/common/config.yml

@@ -131,6 +131,12 @@ proxy:
     # Les proxys/layers sont mis en cache local pour une durée de 30s
     # par défaut. Si les modifications sont rares, vous pouvez augmenter
     # cette valeur pour réduire la "pression" sur le serveur Redis.
+    # Il est possible de forcer la réinitialisation du cache en envoyant
+    # le signal SIGUSR2 au processus Bouncer.
+    #
+    # Exemple
+    #
+    # kill -s USR2 $(pgrep bouncer)
     ttl: 30s
 
   # Configuration du transport HTTP(S)