Compare commits

...

2 Commits

Author SHA1 Message Date
cc5cdcea96 fix: cache ttl interpolation
All checks were successful
Cadoles/bouncer/pipeline/head This commit looks good
2025-03-18 12:20:00 +01:00
1af7248a6f feat: reset proxy cache with sigusr2
All checks were successful
Cadoles/bouncer/pipeline/head This commit looks good
2025-03-18 11:35:15 +01:00
12 changed files with 139 additions and 31 deletions

View File

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

View File

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

View File

@ -28,6 +28,11 @@ func (c *Cache[K, V]) Set(key K, value V) {
c.values.Set(key, value) 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] { 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]{ return &Cache[K, V]{
values: values, values: values,

View File

@ -32,6 +32,8 @@ func RunCommand() *cli.Command {
logger.SetFormat(logger.Format(conf.Logger.Format)) logger.SetFormat(logger.Format(conf.Logger.Format))
logger.SetLevel(logger.Level(conf.Logger.Level)) logger.SetLevel(logger.Level(conf.Logger.Level))
logger.Debug(ctx.Context, "using config", logger.F("config", conf))
projectVersion := ctx.String("projectVersion") projectVersion := ctx.String("projectVersion")
if conf.Proxy.Sentry.DSN != "" { if conf.Proxy.Sentry.DSN != "" {

View File

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

View File

@ -19,7 +19,7 @@ func (is *InterpolatedString) UnmarshalYAML(value *yaml.Node) error {
return errors.WithStack(err) return errors.WithStack(err)
} }
str, err := envsubst.EvalEnv(str) str, err := envsubst.Eval(str, getEnv)
if err != nil { if err != nil {
return errors.WithStack(err) 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) 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 { if err != nil {
return errors.WithStack(err) 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) 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 { if err != nil {
return errors.WithStack(err) 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) 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 { if err != nil {
return errors.WithStack(err) return errors.WithStack(err)
} }
@ -101,9 +101,10 @@ func (ib *InterpolatedBool) UnmarshalYAML(value *yaml.Node) error {
return nil return nil
} }
var getEnv = os.Getenv
type InterpolatedMap struct { type InterpolatedMap struct {
Data map[string]any Data map[string]any
getEnv func(string) string
} }
func (im *InterpolatedMap) UnmarshalYAML(value *yaml.Node) error { 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) 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) interpolated, err := im.interpolateRecursive(data)
if err != nil { if err != nil {
return errors.WithStack(err) return errors.WithStack(err)
@ -140,7 +137,7 @@ func (im InterpolatedMap) interpolateRecursive(data any) (any, error) {
} }
case string: case string:
value, err := envsubst.Eval(typ, im.getEnv) value, err := envsubst.Eval(typ, getEnv)
if err != nil { if err != nil {
return nil, errors.WithStack(err) return nil, errors.WithStack(err)
} }
@ -171,7 +168,7 @@ func (iss *InterpolatedStringSlice) UnmarshalYAML(value *yaml.Node) error {
} }
for index, value := range data { for index, value := range data {
value, err := envsubst.EvalEnv(value) value, err := envsubst.Eval(value, getEnv)
if err != nil { if err != nil {
return errors.WithStack(err) 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) 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 { if err != nil {
return errors.WithStack(err) return errors.WithStack(err)
} }

View File

@ -4,6 +4,7 @@ import (
"fmt" "fmt"
"os" "os"
"testing" "testing"
"time"
"github.com/pkg/errors" "github.com/pkg/errors"
"gopkg.in/yaml.v3" "gopkg.in/yaml.v3"
@ -65,7 +66,7 @@ func TestInterpolatedMap(t *testing.T) {
var interpolatedMap InterpolatedMap var interpolatedMap InterpolatedMap
if tc.Env != nil { if tc.Env != nil {
interpolatedMap.getEnv = func(key string) string { getEnv = func(key string) string {
return tc.Env[key] 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)
}
})
}
}

View File

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

View File

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

View File

@ -99,9 +99,12 @@ const proxiesCacheKey = "proxies"
func (d *Director) getProxies(ctx context.Context) ([]*store.Proxy, error) { func (d *Director) getProxies(ctx context.Context) ([]*store.Proxy, error) {
proxies, exists := d.proxyCache.Get(proxiesCacheKey) proxies, exists := d.proxyCache.Get(proxiesCacheKey)
if exists { if exists {
logger.Debug(ctx, "using cached proxies")
return proxies, nil return proxies, nil
} }
logger.Debug(ctx, "querying fresh proxies")
headers, err := d.proxyRepository.QueryProxy(ctx, store.WithProxyQueryEnabled(true)) headers, err := d.proxyRepository.QueryProxy(ctx, store.WithProxyQueryEnabled(true))
if err != nil { if err != nil {
return nil, errors.WithStack(err) return nil, errors.WithStack(err)
@ -134,9 +137,12 @@ func (d *Director) getLayers(ctx context.Context, proxyName store.ProxyName) ([]
layers, exists := d.layerCache.Get(cacheKey) layers, exists := d.layerCache.Get(cacheKey)
if exists { if exists {
logger.Debug(ctx, "using cached layers")
return layers, nil return layers, nil
} }
logger.Debug(ctx, "querying fresh layers")
headers, err := d.layerRepository.QueryLayers(ctx, proxyName, store.WithLayerQueryEnabled(true)) headers, err := d.layerRepository.QueryLayers(ctx, proxyName, store.WithLayerQueryEnabled(true))
if err != nil { if err != nil {
return nil, errors.WithStack(err) return nil, errors.WithStack(err)

View File

@ -13,8 +13,11 @@ import (
"net/http/httputil" "net/http/httputil"
"net/http/pprof" "net/http/pprof"
"net/url" "net/url"
"os"
"os/signal"
"path/filepath" "path/filepath"
"strconv" "strconv"
"syscall"
"time" "time"
"forge.cadoles.com/Cadoles/go-proxy" "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") logger.Info(ctx, "http server listening")
layerCache, proxyCache, cancel := s.createDirectorCaches(ctx)
defer cancel()
director := director.New( director := director.New(
s.proxyRepository, s.proxyRepository,
s.layerRepository, s.layerRepository,
director.WithLayers(s.directorLayers...), 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), 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") 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 { func (s *Server) createReverseProxy(ctx context.Context, target *url.URL) *httputil.ReverseProxy {
reverseProxy := httputil.NewSingleHostReverseProxy(target) reverseProxy := httputil.NewSingleHostReverseProxy(target)

View File

@ -131,6 +131,12 @@ proxy:
# Les proxys/layers sont mis en cache local pour une durée de 30s # 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 # par défaut. Si les modifications sont rares, vous pouvez augmenter
# cette valeur pour réduire la "pression" sur le serveur Redis. # 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 ttl: 30s
# Configuration du transport HTTP(S) # Configuration du transport HTTP(S)