Invalidation automatique du cache #51
@ -5,7 +5,6 @@ import (
|
|||||||
"time"
|
"time"
|
||||||
|
|
||||||
"forge.cadoles.com/cadoles/bouncer/internal/config"
|
"forge.cadoles.com/cadoles/bouncer/internal/config"
|
||||||
"forge.cadoles.com/cadoles/bouncer/internal/lock/redis"
|
|
||||||
"forge.cadoles.com/cadoles/bouncer/internal/schema"
|
"forge.cadoles.com/cadoles/bouncer/internal/schema"
|
||||||
"forge.cadoles.com/cadoles/bouncer/internal/setup"
|
"forge.cadoles.com/cadoles/bouncer/internal/setup"
|
||||||
"forge.cadoles.com/cadoles/bouncer/internal/store"
|
"forge.cadoles.com/cadoles/bouncer/internal/store"
|
||||||
@ -21,10 +20,9 @@ func (s *Server) bootstrapProxies(ctx context.Context) error {
|
|||||||
proxyRepo := s.proxyRepository
|
proxyRepo := s.proxyRepository
|
||||||
layerRepo := s.layerRepository
|
layerRepo := s.layerRepository
|
||||||
|
|
||||||
lockTimeout := time.Duration(s.bootstrapConfig.LockTimeout)
|
lockTimeout := time.Duration(*s.bootstrapConfig.LockTimeout)
|
||||||
locker := redis.NewLocker(s.redisClient, int(s.bootstrapConfig.MaxConnectionRetries))
|
|
||||||
|
|
||||||
err := locker.WithLock(ctx, "bouncer-admin-bootstrap", lockTimeout, func(ctx context.Context) error {
|
err := s.locker.WithLock(ctx, "bouncer-admin-bootstrap", lockTimeout, func(ctx context.Context) error {
|
||||||
logger.Info(ctx, "bootstrapping proxies")
|
logger.Info(ctx, "bootstrapping proxies")
|
||||||
|
|
||||||
for proxyName, proxyConfig := range s.bootstrapConfig.Proxies {
|
for proxyName, proxyConfig := range s.bootstrapConfig.Proxies {
|
||||||
|
@ -5,57 +5,10 @@ import (
|
|||||||
|
|
||||||
"forge.cadoles.com/cadoles/bouncer/internal/integration"
|
"forge.cadoles.com/cadoles/bouncer/internal/integration"
|
||||||
"forge.cadoles.com/cadoles/bouncer/internal/jwk"
|
"forge.cadoles.com/cadoles/bouncer/internal/jwk"
|
||||||
"forge.cadoles.com/cadoles/bouncer/internal/setup"
|
|
||||||
"github.com/pkg/errors"
|
"github.com/pkg/errors"
|
||||||
"gitlab.com/wpetit/goweb/logger"
|
"gitlab.com/wpetit/goweb/logger"
|
||||||
)
|
)
|
||||||
|
|
||||||
func (s *Server) initRepositories(ctx context.Context) error {
|
|
||||||
if err := s.initRedisClient(ctx); err != nil {
|
|
||||||
return errors.WithStack(err)
|
|
||||||
}
|
|
||||||
|
|
||||||
if err := s.initLayerRepository(ctx); err != nil {
|
|
||||||
return errors.WithStack(err)
|
|
||||||
}
|
|
||||||
|
|
||||||
if err := s.initProxyRepository(ctx); err != nil {
|
|
||||||
return errors.WithStack(err)
|
|
||||||
}
|
|
||||||
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (s *Server) initRedisClient(ctx context.Context) error {
|
|
||||||
client := setup.NewSharedClient(s.redisConfig)
|
|
||||||
|
|
||||||
s.redisClient = client
|
|
||||||
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (s *Server) initLayerRepository(ctx context.Context) error {
|
|
||||||
layerRepository, err := setup.NewLayerRepository(ctx, s.redisClient)
|
|
||||||
if err != nil {
|
|
||||||
return errors.WithStack(err)
|
|
||||||
}
|
|
||||||
|
|
||||||
s.layerRepository = layerRepository
|
|
||||||
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (s *Server) initProxyRepository(ctx context.Context) error {
|
|
||||||
proxyRepository, err := setup.NewProxyRepository(ctx, s.redisClient)
|
|
||||||
if err != nil {
|
|
||||||
return errors.WithStack(err)
|
|
||||||
}
|
|
||||||
|
|
||||||
s.proxyRepository = proxyRepository
|
|
||||||
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (s *Server) initPrivateKey(ctx context.Context) error {
|
func (s *Server) initPrivateKey(ctx context.Context) error {
|
||||||
localKey, err := jwk.LoadOrGenerate(string(s.serverConfig.Auth.PrivateKey), jwk.DefaultKeySize)
|
localKey, err := jwk.LoadOrGenerate(string(s.serverConfig.Auth.PrivateKey), jwk.DefaultKeySize)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
@ -3,13 +3,19 @@ package admin
|
|||||||
import (
|
import (
|
||||||
"forge.cadoles.com/cadoles/bouncer/internal/config"
|
"forge.cadoles.com/cadoles/bouncer/internal/config"
|
||||||
"forge.cadoles.com/cadoles/bouncer/internal/integration"
|
"forge.cadoles.com/cadoles/bouncer/internal/integration"
|
||||||
|
"forge.cadoles.com/cadoles/bouncer/internal/lock"
|
||||||
|
"forge.cadoles.com/cadoles/bouncer/internal/store"
|
||||||
)
|
)
|
||||||
|
|
||||||
type Option struct {
|
type Option struct {
|
||||||
BootstrapConfig config.BootstrapConfig
|
BootstrapConfig config.BootstrapConfig
|
||||||
ServerConfig config.AdminServerConfig
|
ServerConfig config.AdminServerConfig
|
||||||
RedisConfig config.RedisConfig
|
|
||||||
Integrations []integration.Integration
|
Integrations []integration.Integration
|
||||||
|
|
||||||
|
ProxyRepository store.ProxyRepository
|
||||||
|
LayerRepository store.LayerRepository
|
||||||
|
|
||||||
|
Locker lock.Locker
|
||||||
}
|
}
|
||||||
|
|
||||||
type OptionFunc func(*Option)
|
type OptionFunc func(*Option)
|
||||||
@ -17,7 +23,6 @@ type OptionFunc func(*Option)
|
|||||||
func defaultOption() *Option {
|
func defaultOption() *Option {
|
||||||
return &Option{
|
return &Option{
|
||||||
ServerConfig: config.NewDefaultAdminServerConfig(),
|
ServerConfig: config.NewDefaultAdminServerConfig(),
|
||||||
RedisConfig: config.NewDefaultRedisConfig(),
|
|
||||||
Integrations: make([]integration.Integration, 0),
|
Integrations: make([]integration.Integration, 0),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -28,12 +33,6 @@ func WithServerConfig(conf config.AdminServerConfig) OptionFunc {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func WithRedisConfig(conf config.RedisConfig) OptionFunc {
|
|
||||||
return func(opt *Option) {
|
|
||||||
opt.RedisConfig = conf
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func WithBootstrapConfig(conf config.BootstrapConfig) OptionFunc {
|
func WithBootstrapConfig(conf config.BootstrapConfig) OptionFunc {
|
||||||
return func(opt *Option) {
|
return func(opt *Option) {
|
||||||
opt.BootstrapConfig = conf
|
opt.BootstrapConfig = conf
|
||||||
@ -45,3 +44,21 @@ func WithIntegrations(integrations ...integration.Integration) OptionFunc {
|
|||||||
opt.Integrations = integrations
|
opt.Integrations = integrations
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func WithLayerRepository(layerRepository store.LayerRepository) OptionFunc {
|
||||||
|
return func(opt *Option) {
|
||||||
|
opt.LayerRepository = layerRepository
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func WithProxyRepository(proxyRepository store.ProxyRepository) OptionFunc {
|
||||||
|
return func(opt *Option) {
|
||||||
|
opt.ProxyRepository = proxyRepository
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func WithLocker(locker lock.Locker) OptionFunc {
|
||||||
|
return func(opt *Option) {
|
||||||
|
opt.Locker = locker
|
||||||
|
}
|
||||||
|
}
|
||||||
|
@ -15,6 +15,7 @@ import (
|
|||||||
"forge.cadoles.com/cadoles/bouncer/internal/config"
|
"forge.cadoles.com/cadoles/bouncer/internal/config"
|
||||||
"forge.cadoles.com/cadoles/bouncer/internal/integration"
|
"forge.cadoles.com/cadoles/bouncer/internal/integration"
|
||||||
"forge.cadoles.com/cadoles/bouncer/internal/jwk"
|
"forge.cadoles.com/cadoles/bouncer/internal/jwk"
|
||||||
|
"forge.cadoles.com/cadoles/bouncer/internal/lock"
|
||||||
"forge.cadoles.com/cadoles/bouncer/internal/store"
|
"forge.cadoles.com/cadoles/bouncer/internal/store"
|
||||||
sentryhttp "github.com/getsentry/sentry-go/http"
|
sentryhttp "github.com/getsentry/sentry-go/http"
|
||||||
"github.com/go-chi/chi/v5"
|
"github.com/go-chi/chi/v5"
|
||||||
@ -22,15 +23,13 @@ import (
|
|||||||
"github.com/go-chi/cors"
|
"github.com/go-chi/cors"
|
||||||
"github.com/pkg/errors"
|
"github.com/pkg/errors"
|
||||||
"github.com/prometheus/client_golang/prometheus/promhttp"
|
"github.com/prometheus/client_golang/prometheus/promhttp"
|
||||||
"github.com/redis/go-redis/v9"
|
|
||||||
"gitlab.com/wpetit/goweb/logger"
|
"gitlab.com/wpetit/goweb/logger"
|
||||||
)
|
)
|
||||||
|
|
||||||
type Server struct {
|
type Server struct {
|
||||||
serverConfig config.AdminServerConfig
|
serverConfig config.AdminServerConfig
|
||||||
redisConfig config.RedisConfig
|
|
||||||
|
|
||||||
redisClient redis.UniversalClient
|
locker lock.Locker
|
||||||
|
|
||||||
integrations []integration.Integration
|
integrations []integration.Integration
|
||||||
|
|
||||||
@ -60,12 +59,6 @@ func (s *Server) run(parentCtx context.Context, addrs chan net.Addr, errs chan e
|
|||||||
ctx, cancel := context.WithCancel(parentCtx)
|
ctx, cancel := context.WithCancel(parentCtx)
|
||||||
defer cancel()
|
defer cancel()
|
||||||
|
|
||||||
if err := s.initRepositories(ctx); err != nil {
|
|
||||||
errs <- errors.WithStack(err)
|
|
||||||
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
if err := s.bootstrapProxies(ctx); err != nil {
|
if err := s.bootstrapProxies(ctx); err != nil {
|
||||||
errs <- errors.WithStack(err)
|
errs <- errors.WithStack(err)
|
||||||
|
|
||||||
@ -231,8 +224,10 @@ func NewServer(funcs ...OptionFunc) *Server {
|
|||||||
|
|
||||||
return &Server{
|
return &Server{
|
||||||
serverConfig: opt.ServerConfig,
|
serverConfig: opt.ServerConfig,
|
||||||
redisConfig: opt.RedisConfig,
|
|
||||||
bootstrapConfig: opt.BootstrapConfig,
|
bootstrapConfig: opt.BootstrapConfig,
|
||||||
integrations: opt.Integrations,
|
integrations: opt.Integrations,
|
||||||
|
proxyRepository: opt.ProxyRepository,
|
||||||
|
layerRepository: opt.LayerRepository,
|
||||||
|
locker: opt.Locker,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -6,6 +6,7 @@ import (
|
|||||||
|
|
||||||
"forge.cadoles.com/cadoles/bouncer/internal/admin"
|
"forge.cadoles.com/cadoles/bouncer/internal/admin"
|
||||||
"forge.cadoles.com/cadoles/bouncer/internal/command/common"
|
"forge.cadoles.com/cadoles/bouncer/internal/command/common"
|
||||||
|
"forge.cadoles.com/cadoles/bouncer/internal/lock/redis"
|
||||||
"forge.cadoles.com/cadoles/bouncer/internal/setup"
|
"forge.cadoles.com/cadoles/bouncer/internal/setup"
|
||||||
"github.com/pkg/errors"
|
"github.com/pkg/errors"
|
||||||
"github.com/urfave/cli/v2"
|
"github.com/urfave/cli/v2"
|
||||||
@ -46,10 +47,26 @@ func RunCommand() *cli.Command {
|
|||||||
return errors.Wrap(err, "could not setup integrations")
|
return errors.Wrap(err, "could not setup integrations")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
redisClient := setup.NewSharedClient(conf.Redis)
|
||||||
|
|
||||||
|
proxyRepository, err := setup.NewProxyRepository(ctx.Context, redisClient)
|
||||||
|
if err != nil {
|
||||||
|
return errors.Wrap(err, "could not initialize proxy repository")
|
||||||
|
}
|
||||||
|
|
||||||
|
layerRepository, err := setup.NewLayerRepository(ctx.Context, redisClient)
|
||||||
|
if err != nil {
|
||||||
|
return errors.Wrap(err, "could not initialize layer repository")
|
||||||
|
}
|
||||||
|
|
||||||
|
locker := redis.NewLocker(redisClient, int(conf.Bootstrap.MaxConnectionRetries))
|
||||||
|
|
||||||
srv := admin.NewServer(
|
srv := admin.NewServer(
|
||||||
admin.WithServerConfig(conf.Admin),
|
admin.WithServerConfig(conf.Admin),
|
||||||
admin.WithRedisConfig(conf.Redis),
|
|
||||||
admin.WithBootstrapConfig(conf.Bootstrap),
|
admin.WithBootstrapConfig(conf.Bootstrap),
|
||||||
|
admin.WithProxyRepository(proxyRepository),
|
||||||
|
admin.WithLayerRepository(layerRepository),
|
||||||
|
admin.WithLocker(locker),
|
||||||
admin.WithIntegrations(integrations...),
|
admin.WithIntegrations(integrations...),
|
||||||
)
|
)
|
||||||
|
|
||||||
|
@ -2,7 +2,10 @@ package proxy
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"os"
|
||||||
|
"os/signal"
|
||||||
"strings"
|
"strings"
|
||||||
|
"syscall"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"forge.cadoles.com/cadoles/bouncer/internal/command/common"
|
"forge.cadoles.com/cadoles/bouncer/internal/command/common"
|
||||||
@ -47,15 +50,44 @@ func RunCommand() *cli.Command {
|
|||||||
return errors.Wrap(err, "could not initialize director layers")
|
return errors.Wrap(err, "could not initialize director layers")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
redisClient := setup.NewSharedClient(conf.Redis)
|
||||||
|
|
||||||
|
proxyRepository, err := setup.NewProxyRepository(ctx.Context, redisClient)
|
||||||
|
if err != nil {
|
||||||
|
return errors.Wrap(err, "could not initialize proxy repository")
|
||||||
|
}
|
||||||
|
|
||||||
|
layerRepository, err := setup.NewLayerRepository(ctx.Context, redisClient)
|
||||||
|
if err != nil {
|
||||||
|
return errors.Wrap(err, "could not initialize layer repository")
|
||||||
|
}
|
||||||
|
|
||||||
srv := proxy.NewServer(
|
srv := proxy.NewServer(
|
||||||
proxy.WithServerConfig(conf.Proxy),
|
proxy.WithServerConfig(conf.Proxy),
|
||||||
proxy.WithRedisConfig(conf.Redis),
|
proxy.WithProxyRepository(proxyRepository),
|
||||||
|
proxy.WithLayerRepository(layerRepository),
|
||||||
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)
|
||||||
|
|
||||||
|
// Clear director's cache on SIGUSR2
|
||||||
|
sig := make(chan os.Signal, 1)
|
||||||
|
signal.Notify(sig, syscall.SIGUSR2)
|
||||||
|
|
||||||
|
go func() {
|
||||||
|
for {
|
||||||
|
select {
|
||||||
|
case <-sig:
|
||||||
|
logger.Info(ctx.Context, "received sigusr2, clearing cache")
|
||||||
|
srv.ClearCache()
|
||||||
|
case <-ctx.Context.Done():
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
select {
|
select {
|
||||||
case addr := <-addrs:
|
case addr := <-addrs:
|
||||||
url := fmt.Sprintf("http://%s", addr.String())
|
url := fmt.Sprintf("http://%s", addr.String())
|
||||||
|
@ -14,7 +14,7 @@ import (
|
|||||||
type BootstrapConfig struct {
|
type BootstrapConfig struct {
|
||||||
Proxies map[store.ProxyName]BootstrapProxyConfig `yaml:"proxies"`
|
Proxies map[store.ProxyName]BootstrapProxyConfig `yaml:"proxies"`
|
||||||
Dir InterpolatedString `yaml:"dir"`
|
Dir InterpolatedString `yaml:"dir"`
|
||||||
LockTimeout InterpolatedDuration `yaml:"lockTimeout"`
|
LockTimeout *InterpolatedDuration `yaml:"lockTimeout"`
|
||||||
MaxConnectionRetries InterpolatedInt `yaml:"maxRetries"`
|
MaxConnectionRetries InterpolatedInt `yaml:"maxRetries"`
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -65,7 +65,7 @@ type BootstrapLayerConfig struct {
|
|||||||
func NewDefaultBootstrapConfig() BootstrapConfig {
|
func NewDefaultBootstrapConfig() BootstrapConfig {
|
||||||
return BootstrapConfig{
|
return BootstrapConfig{
|
||||||
Dir: "",
|
Dir: "",
|
||||||
LockTimeout: *NewInterpolatedDuration(30 * time.Second),
|
LockTimeout: NewInterpolatedDuration(30 * time.Second),
|
||||||
MaxConnectionRetries: 10,
|
MaxConnectionRetries: 10,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -220,3 +220,11 @@ func NewInterpolatedDuration(d time.Duration) *InterpolatedDuration {
|
|||||||
id := InterpolatedDuration(d)
|
id := InterpolatedDuration(d)
|
||||||
return &id
|
return &id
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func Default[T any](value *T, defaultValue *T) *T {
|
||||||
|
if value == nil {
|
||||||
|
return defaultValue
|
||||||
|
}
|
||||||
|
|
||||||
|
return value
|
||||||
|
}
|
||||||
|
@ -16,7 +16,7 @@ func NewDefaultIntegrationsConfig() IntegrationsConfig {
|
|||||||
PrivateKeySecret: "",
|
PrivateKeySecret: "",
|
||||||
PrivateKeySecretNamespace: "",
|
PrivateKeySecretNamespace: "",
|
||||||
ReaderTokenSecret: "",
|
ReaderTokenSecret: "",
|
||||||
LockTimeout: *NewInterpolatedDuration(30 * time.Second),
|
LockTimeout: NewInterpolatedDuration(30 * time.Second),
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -29,5 +29,5 @@ type KubernetesConfig struct {
|
|||||||
ReaderTokenSecretNamespace InterpolatedString `yaml:"readerTokenSecretNamespace"`
|
ReaderTokenSecretNamespace InterpolatedString `yaml:"readerTokenSecretNamespace"`
|
||||||
PrivateKeySecret InterpolatedString `yaml:"privateKeySecret"`
|
PrivateKeySecret InterpolatedString `yaml:"privateKeySecret"`
|
||||||
PrivateKeySecretNamespace InterpolatedString `yaml:"privateKeySecretNamespace"`
|
PrivateKeySecretNamespace InterpolatedString `yaml:"privateKeySecretNamespace"`
|
||||||
LockTimeout InterpolatedDuration `yaml:"lockTimeout"`
|
LockTimeout *InterpolatedDuration `yaml:"lockTimeout"`
|
||||||
}
|
}
|
||||||
|
@ -11,33 +11,36 @@ const (
|
|||||||
type RedisConfig struct {
|
type RedisConfig struct {
|
||||||
Adresses InterpolatedStringSlice `yaml:"addresses"`
|
Adresses InterpolatedStringSlice `yaml:"addresses"`
|
||||||
Master InterpolatedString `yaml:"master"`
|
Master InterpolatedString `yaml:"master"`
|
||||||
ReadTimeout InterpolatedDuration `yaml:"readTimeout"`
|
ReadTimeout *InterpolatedDuration `yaml:"readTimeout"`
|
||||||
WriteTimeout InterpolatedDuration `yaml:"writeTimeout"`
|
WriteTimeout *InterpolatedDuration `yaml:"writeTimeout"`
|
||||||
DialTimeout InterpolatedDuration `yaml:"dialTimeout"`
|
DialTimeout *InterpolatedDuration `yaml:"dialTimeout"`
|
||||||
LockMaxRetries InterpolatedInt `yaml:"lockMaxRetries"`
|
LockMaxRetries InterpolatedInt `yaml:"lockMaxRetries"`
|
||||||
RouteByLatency InterpolatedBool `yaml:"routeByLatency"`
|
RouteByLatency InterpolatedBool `yaml:"routeByLatency"`
|
||||||
ContextTimeoutEnabled InterpolatedBool `yaml:"contextTimeoutEnabled"`
|
ContextTimeoutEnabled InterpolatedBool `yaml:"contextTimeoutEnabled"`
|
||||||
MaxRetries InterpolatedInt `yaml:"maxRetries"`
|
MaxRetries InterpolatedInt `yaml:"maxRetries"`
|
||||||
PingInterval InterpolatedDuration `yaml:"pingInterval"`
|
PingInterval *InterpolatedDuration `yaml:"pingInterval"`
|
||||||
PoolSize InterpolatedInt `yaml:"poolSize"`
|
PoolSize InterpolatedInt `yaml:"poolSize"`
|
||||||
PoolTimeout InterpolatedDuration `yaml:"poolTimeout"`
|
PoolTimeout *InterpolatedDuration `yaml:"poolTimeout"`
|
||||||
MinIdleConns InterpolatedInt `yaml:"minIdleConns"`
|
MinIdleConns InterpolatedInt `yaml:"minIdleConns"`
|
||||||
MaxIdleConns InterpolatedInt `yaml:"maxIdleConns"`
|
MaxIdleConns InterpolatedInt `yaml:"maxIdleConns"`
|
||||||
ConnMaxIdleTime InterpolatedDuration `yaml:"connMaxIdleTime"`
|
ConnMaxIdleTime *InterpolatedDuration `yaml:"connMaxIdleTime"`
|
||||||
ConnMaxLifetime InterpolatedDuration `yaml:"connMaxLifeTime"`
|
ConnMaxLifetime *InterpolatedDuration `yaml:"connMaxLifeTime"`
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewDefaultRedisConfig() RedisConfig {
|
func NewDefaultRedisConfig() RedisConfig {
|
||||||
return RedisConfig{
|
return RedisConfig{
|
||||||
Adresses: InterpolatedStringSlice{"localhost:6379"},
|
Adresses: InterpolatedStringSlice{"localhost:6379"},
|
||||||
Master: "",
|
Master: "",
|
||||||
ReadTimeout: InterpolatedDuration(30 * time.Second),
|
ReadTimeout: NewInterpolatedDuration(30 * time.Second),
|
||||||
WriteTimeout: InterpolatedDuration(30 * time.Second),
|
WriteTimeout: NewInterpolatedDuration(30 * time.Second),
|
||||||
DialTimeout: InterpolatedDuration(30 * time.Second),
|
DialTimeout: NewInterpolatedDuration(30 * time.Second),
|
||||||
|
PoolTimeout: NewInterpolatedDuration(31 * time.Second),
|
||||||
LockMaxRetries: 10,
|
LockMaxRetries: 10,
|
||||||
MaxRetries: 3,
|
MaxRetries: 3,
|
||||||
PingInterval: InterpolatedDuration(30 * time.Second),
|
PingInterval: NewInterpolatedDuration(30 * time.Second),
|
||||||
ContextTimeoutEnabled: true,
|
ContextTimeoutEnabled: true,
|
||||||
RouteByLatency: true,
|
RouteByLatency: true,
|
||||||
|
ConnMaxIdleTime: NewInterpolatedDuration(1 * time.Minute),
|
||||||
|
ConnMaxLifetime: NewInterpolatedDuration(5 * time.Minute),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -1,45 +0,0 @@
|
|||||||
package proxy
|
|
||||||
|
|
||||||
import (
|
|
||||||
"context"
|
|
||||||
|
|
||||||
"forge.cadoles.com/cadoles/bouncer/internal/setup"
|
|
||||||
"github.com/pkg/errors"
|
|
||||||
"github.com/redis/go-redis/v9"
|
|
||||||
)
|
|
||||||
|
|
||||||
func (s *Server) initRepositories(ctx context.Context) error {
|
|
||||||
client := setup.NewSharedClient(s.redisConfig)
|
|
||||||
|
|
||||||
if err := s.initProxyRepository(ctx, client); err != nil {
|
|
||||||
return errors.WithStack(err)
|
|
||||||
}
|
|
||||||
|
|
||||||
if err := s.initLayerRepository(ctx, client); err != nil {
|
|
||||||
return errors.WithStack(err)
|
|
||||||
}
|
|
||||||
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (s *Server) initProxyRepository(ctx context.Context, client redis.UniversalClient) error {
|
|
||||||
proxyRepository, err := setup.NewProxyRepository(ctx, client)
|
|
||||||
if err != nil {
|
|
||||||
return errors.WithStack(err)
|
|
||||||
}
|
|
||||||
|
|
||||||
s.proxyRepository = proxyRepository
|
|
||||||
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (s *Server) initLayerRepository(ctx context.Context, client redis.UniversalClient) error {
|
|
||||||
layerRepository, err := setup.NewLayerRepository(ctx, client)
|
|
||||||
if err != nil {
|
|
||||||
return errors.WithStack(err)
|
|
||||||
}
|
|
||||||
|
|
||||||
s.layerRepository = layerRepository
|
|
||||||
|
|
||||||
return nil
|
|
||||||
}
|
|
@ -5,13 +5,16 @@ import (
|
|||||||
|
|
||||||
"forge.cadoles.com/cadoles/bouncer/internal/config"
|
"forge.cadoles.com/cadoles/bouncer/internal/config"
|
||||||
"forge.cadoles.com/cadoles/bouncer/internal/proxy/director"
|
"forge.cadoles.com/cadoles/bouncer/internal/proxy/director"
|
||||||
|
"forge.cadoles.com/cadoles/bouncer/internal/store"
|
||||||
)
|
)
|
||||||
|
|
||||||
type Option struct {
|
type Option struct {
|
||||||
ServerConfig config.ProxyServerConfig
|
ServerConfig config.ProxyServerConfig
|
||||||
RedisConfig config.RedisConfig
|
|
||||||
DirectorLayers []director.Layer
|
DirectorLayers []director.Layer
|
||||||
DirectorCacheTTL time.Duration
|
DirectorCacheTTL time.Duration
|
||||||
|
|
||||||
|
ProxyRepository store.ProxyRepository
|
||||||
|
LayerRepository store.LayerRepository
|
||||||
}
|
}
|
||||||
|
|
||||||
type OptionFunc func(*Option)
|
type OptionFunc func(*Option)
|
||||||
@ -19,7 +22,6 @@ type OptionFunc func(*Option)
|
|||||||
func defaultOption() *Option {
|
func defaultOption() *Option {
|
||||||
return &Option{
|
return &Option{
|
||||||
ServerConfig: config.NewDefaultProxyServerConfig(),
|
ServerConfig: config.NewDefaultProxyServerConfig(),
|
||||||
RedisConfig: config.NewDefaultRedisConfig(),
|
|
||||||
DirectorLayers: make([]director.Layer, 0),
|
DirectorLayers: make([]director.Layer, 0),
|
||||||
DirectorCacheTTL: 30 * time.Second,
|
DirectorCacheTTL: 30 * time.Second,
|
||||||
}
|
}
|
||||||
@ -31,12 +33,6 @@ func WithServerConfig(conf config.ProxyServerConfig) OptionFunc {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func WithRedisConfig(conf config.RedisConfig) OptionFunc {
|
|
||||||
return func(opt *Option) {
|
|
||||||
opt.RedisConfig = conf
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func WithDirectorLayers(layers ...director.Layer) OptionFunc {
|
func WithDirectorLayers(layers ...director.Layer) OptionFunc {
|
||||||
return func(opt *Option) {
|
return func(opt *Option) {
|
||||||
opt.DirectorLayers = layers
|
opt.DirectorLayers = layers
|
||||||
@ -48,3 +44,15 @@ func WithDirectorCacheTTL(ttl time.Duration) OptionFunc {
|
|||||||
opt.DirectorCacheTTL = ttl
|
opt.DirectorCacheTTL = ttl
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func WithLayerRepository(layerRepository store.LayerRepository) OptionFunc {
|
||||||
|
return func(opt *Option) {
|
||||||
|
opt.LayerRepository = layerRepository
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func WithProxyRepository(proxyRepository store.ProxyRepository) OptionFunc {
|
||||||
|
return func(opt *Option) {
|
||||||
|
opt.ProxyRepository = proxyRepository
|
||||||
|
}
|
||||||
|
}
|
||||||
|
@ -13,14 +13,12 @@ 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"
|
||||||
|
"forge.cadoles.com/cadoles/bouncer/internal/cache"
|
||||||
"forge.cadoles.com/cadoles/bouncer/internal/cache/memory"
|
"forge.cadoles.com/cadoles/bouncer/internal/cache/memory"
|
||||||
"forge.cadoles.com/cadoles/bouncer/internal/cache/ttl"
|
"forge.cadoles.com/cadoles/bouncer/internal/cache/ttl"
|
||||||
bouncerChi "forge.cadoles.com/cadoles/bouncer/internal/chi"
|
bouncerChi "forge.cadoles.com/cadoles/bouncer/internal/chi"
|
||||||
@ -39,11 +37,13 @@ import (
|
|||||||
|
|
||||||
type Server struct {
|
type Server struct {
|
||||||
serverConfig config.ProxyServerConfig
|
serverConfig config.ProxyServerConfig
|
||||||
redisConfig config.RedisConfig
|
|
||||||
directorLayers []director.Layer
|
directorLayers []director.Layer
|
||||||
directorCacheTTL time.Duration
|
|
||||||
proxyRepository store.ProxyRepository
|
proxyRepository store.ProxyRepository
|
||||||
layerRepository store.LayerRepository
|
layerRepository store.LayerRepository
|
||||||
|
|
||||||
|
layerCache cache.Cache[string, []*store.Layer]
|
||||||
|
proxyCache cache.Cache[string, []*store.Proxy]
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *Server) Start(ctx context.Context) (<-chan net.Addr, <-chan error) {
|
func (s *Server) Start(ctx context.Context) (<-chan net.Addr, <-chan error) {
|
||||||
@ -55,6 +55,11 @@ func (s *Server) Start(ctx context.Context) (<-chan net.Addr, <-chan error) {
|
|||||||
return addrs, errs
|
return addrs, errs
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (s *Server) ClearCache() {
|
||||||
|
s.layerCache.Clear()
|
||||||
|
s.proxyCache.Clear()
|
||||||
|
}
|
||||||
|
|
||||||
func (s *Server) run(parentCtx context.Context, addrs chan net.Addr, errs chan error) {
|
func (s *Server) run(parentCtx context.Context, addrs chan net.Addr, errs chan error) {
|
||||||
defer func() {
|
defer func() {
|
||||||
close(errs)
|
close(errs)
|
||||||
@ -64,12 +69,6 @@ func (s *Server) run(parentCtx context.Context, addrs chan net.Addr, errs chan e
|
|||||||
ctx, cancel := context.WithCancel(parentCtx)
|
ctx, cancel := context.WithCancel(parentCtx)
|
||||||
defer cancel()
|
defer cancel()
|
||||||
|
|
||||||
if err := s.initRepositories(ctx); err != nil {
|
|
||||||
errs <- errors.WithStack(err)
|
|
||||||
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
listener, err := net.Listen("tcp", fmt.Sprintf("%s:%d", s.serverConfig.HTTP.Host, s.serverConfig.HTTP.Port))
|
listener, err := net.Listen("tcp", fmt.Sprintf("%s:%d", s.serverConfig.HTTP.Host, s.serverConfig.HTTP.Port))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
errs <- errors.WithStack(err)
|
errs <- errors.WithStack(err)
|
||||||
@ -97,15 +96,12 @@ 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(layerCache),
|
director.WithLayerCache(s.layerCache),
|
||||||
director.WithProxyCache(proxyCache),
|
director.WithProxyCache(s.proxyCache),
|
||||||
director.WithHandleErrorFunc(s.handleError),
|
director.WithHandleErrorFunc(s.handleError),
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -199,44 +195,6 @@ 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)
|
||||||
|
|
||||||
@ -346,8 +304,20 @@ func NewServer(funcs ...OptionFunc) *Server {
|
|||||||
|
|
||||||
return &Server{
|
return &Server{
|
||||||
serverConfig: opt.ServerConfig,
|
serverConfig: opt.ServerConfig,
|
||||||
redisConfig: opt.RedisConfig,
|
|
||||||
directorLayers: opt.DirectorLayers,
|
directorLayers: opt.DirectorLayers,
|
||||||
directorCacheTTL: opt.DirectorCacheTTL,
|
|
||||||
|
proxyRepository: opt.ProxyRepository,
|
||||||
|
layerRepository: opt.LayerRepository,
|
||||||
|
|
||||||
|
layerCache: ttl.NewCache(
|
||||||
|
memory.NewCache[string, []*store.Layer](),
|
||||||
|
memory.NewCache[string, time.Time](),
|
||||||
|
opt.DirectorCacheTTL,
|
||||||
|
),
|
||||||
|
proxyCache: ttl.NewCache(
|
||||||
|
memory.NewCache[string, []*store.Proxy](),
|
||||||
|
memory.NewCache[string, time.Time](),
|
||||||
|
opt.DirectorCacheTTL,
|
||||||
|
),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -39,7 +39,7 @@ func setupKubernetesIntegration(ctx context.Context, conf *config.Config) (*kube
|
|||||||
kubernetes.WithPrivateKeySecretNamespace(string(conf.Integrations.Kubernetes.PrivateKeySecretNamespace)),
|
kubernetes.WithPrivateKeySecretNamespace(string(conf.Integrations.Kubernetes.PrivateKeySecretNamespace)),
|
||||||
kubernetes.WithIssuer(string(conf.Admin.Auth.Issuer)),
|
kubernetes.WithIssuer(string(conf.Admin.Auth.Issuer)),
|
||||||
kubernetes.WithLocker(locker),
|
kubernetes.WithLocker(locker),
|
||||||
kubernetes.WithLockTimeout(time.Duration(conf.Integrations.Kubernetes.LockTimeout)),
|
kubernetes.WithLockTimeout(time.Duration(*conf.Integrations.Kubernetes.LockTimeout)),
|
||||||
)
|
)
|
||||||
|
|
||||||
return integration, nil
|
return integration, nil
|
||||||
|
@ -35,27 +35,32 @@ func newRedisClient(conf config.RedisConfig) redis.UniversalClient {
|
|||||||
client := redis.NewUniversalClient(&redis.UniversalOptions{
|
client := redis.NewUniversalClient(&redis.UniversalOptions{
|
||||||
Addrs: conf.Adresses,
|
Addrs: conf.Adresses,
|
||||||
MasterName: string(conf.Master),
|
MasterName: string(conf.Master),
|
||||||
ReadTimeout: time.Duration(conf.ReadTimeout),
|
ReadTimeout: time.Duration(*conf.ReadTimeout),
|
||||||
WriteTimeout: time.Duration(conf.WriteTimeout),
|
WriteTimeout: time.Duration(*conf.WriteTimeout),
|
||||||
DialTimeout: time.Duration(conf.DialTimeout),
|
DialTimeout: time.Duration(*conf.DialTimeout),
|
||||||
RouteByLatency: bool(conf.RouteByLatency),
|
RouteByLatency: bool(conf.RouteByLatency),
|
||||||
ContextTimeoutEnabled: bool(conf.ContextTimeoutEnabled),
|
ContextTimeoutEnabled: bool(conf.ContextTimeoutEnabled),
|
||||||
MaxRetries: int(conf.MaxRetries),
|
MaxRetries: int(conf.MaxRetries),
|
||||||
PoolSize: int(conf.PoolSize),
|
PoolSize: int(conf.PoolSize),
|
||||||
PoolTimeout: time.Duration(conf.PoolTimeout),
|
PoolTimeout: time.Duration(*conf.PoolTimeout),
|
||||||
MinIdleConns: int(conf.MinIdleConns),
|
MinIdleConns: int(conf.MinIdleConns),
|
||||||
MaxIdleConns: int(conf.MaxIdleConns),
|
MaxIdleConns: int(conf.MaxIdleConns),
|
||||||
ConnMaxIdleTime: time.Duration(conf.ConnMaxIdleTime),
|
ConnMaxIdleTime: time.Duration(*conf.ConnMaxIdleTime),
|
||||||
ConnMaxLifetime: time.Duration(conf.ConnMaxLifetime),
|
ConnMaxLifetime: time.Duration(*conf.ConnMaxLifetime),
|
||||||
})
|
})
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
pingInterval := time.Duration(*conf.PingInterval)
|
||||||
|
|
||||||
|
if pingInterval > 0 {
|
||||||
go func() {
|
go func() {
|
||||||
ctx := logger.With(context.Background(),
|
ctx := logger.With(ctx,
|
||||||
logger.F("adresses", conf.Adresses),
|
logger.F("adresses", conf.Adresses),
|
||||||
logger.F("master", conf.Master),
|
logger.F("master", conf.Master),
|
||||||
)
|
)
|
||||||
|
|
||||||
timer := time.NewTicker(time.Duration(conf.PingInterval))
|
timer := time.NewTicker(pingInterval)
|
||||||
defer timer.Stop()
|
defer timer.Stop()
|
||||||
|
|
||||||
connected := true
|
connected := true
|
||||||
@ -73,6 +78,9 @@ func newRedisClient(conf config.RedisConfig) redis.UniversalClient {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
} else {
|
||||||
|
logger.Warn(ctx, "redis ping interval at 0, disabling")
|
||||||
|
}
|
||||||
|
|
||||||
return client
|
return client
|
||||||
}
|
}
|
||||||
|
Reference in New Issue
Block a user