feat: improve perf by caching proxy and layers locally
This commit is contained in:
@ -7,6 +7,7 @@ import (
|
||||
|
||||
"forge.cadoles.com/Cadoles/go-proxy"
|
||||
"forge.cadoles.com/Cadoles/go-proxy/wildcard"
|
||||
"forge.cadoles.com/cadoles/bouncer/internal/cache"
|
||||
"forge.cadoles.com/cadoles/bouncer/internal/store"
|
||||
"github.com/pkg/errors"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
@ -17,6 +18,9 @@ type Director struct {
|
||||
proxyRepository store.ProxyRepository
|
||||
layerRepository store.LayerRepository
|
||||
layerRegistry *LayerRegistry
|
||||
|
||||
proxyCache cache.Cache[string, []*store.Proxy]
|
||||
layerCache cache.Cache[string, []*store.Layer]
|
||||
}
|
||||
|
||||
func (d *Director) rewriteRequest(r *http.Request) (*http.Request, error) {
|
||||
@ -88,7 +92,14 @@ MAIN:
|
||||
return r, nil
|
||||
}
|
||||
|
||||
const proxiesCacheKey = "proxies"
|
||||
|
||||
func (d *Director) getProxies(ctx context.Context) ([]*store.Proxy, error) {
|
||||
proxies, exists := d.proxyCache.Get(proxiesCacheKey)
|
||||
if exists {
|
||||
return proxies, nil
|
||||
}
|
||||
|
||||
headers, err := d.proxyRepository.QueryProxy(ctx, store.WithProxyQueryEnabled(true))
|
||||
if err != nil {
|
||||
return nil, errors.WithStack(err)
|
||||
@ -96,7 +107,7 @@ func (d *Director) getProxies(ctx context.Context) ([]*store.Proxy, error) {
|
||||
|
||||
sort.Sort(store.ByProxyWeight(headers))
|
||||
|
||||
proxies := make([]*store.Proxy, 0, len(headers))
|
||||
proxies = make([]*store.Proxy, 0, len(headers))
|
||||
|
||||
for _, h := range headers {
|
||||
if !h.Enabled {
|
||||
@ -111,10 +122,19 @@ func (d *Director) getProxies(ctx context.Context) ([]*store.Proxy, error) {
|
||||
proxies = append(proxies, proxy)
|
||||
}
|
||||
|
||||
d.proxyCache.Set(proxiesCacheKey, proxies)
|
||||
|
||||
return proxies, nil
|
||||
}
|
||||
|
||||
func (d *Director) getLayers(ctx context.Context, proxyName store.ProxyName) ([]*store.Layer, error) {
|
||||
cacheKey := "layers-" + string(proxyName)
|
||||
|
||||
layers, exists := d.layerCache.Get(cacheKey)
|
||||
if exists {
|
||||
return layers, nil
|
||||
}
|
||||
|
||||
headers, err := d.layerRepository.QueryLayers(ctx, proxyName, store.WithLayerQueryEnabled(true))
|
||||
if err != nil {
|
||||
return nil, errors.WithStack(err)
|
||||
@ -122,7 +142,7 @@ func (d *Director) getLayers(ctx context.Context, proxyName store.ProxyName) ([]
|
||||
|
||||
sort.Sort(store.ByLayerWeight(headers))
|
||||
|
||||
layers := make([]*store.Layer, 0, len(headers))
|
||||
layers = make([]*store.Layer, 0, len(headers))
|
||||
|
||||
for _, h := range headers {
|
||||
if !h.Enabled {
|
||||
@ -137,6 +157,8 @@ func (d *Director) getLayers(ctx context.Context, proxyName store.ProxyName) ([]
|
||||
layers = append(layers, layer)
|
||||
}
|
||||
|
||||
d.layerCache.Set(cacheKey, layers)
|
||||
|
||||
return layers, nil
|
||||
}
|
||||
|
||||
@ -240,8 +262,16 @@ func (d *Director) Middleware() proxy.Middleware {
|
||||
}
|
||||
}
|
||||
|
||||
func New(proxyRepository store.ProxyRepository, layerRepository store.LayerRepository, layers ...Layer) *Director {
|
||||
registry := NewLayerRegistry(layers...)
|
||||
func New(proxyRepository store.ProxyRepository, layerRepository store.LayerRepository, funcs ...OptionFunc) *Director {
|
||||
opts := NewOptions(funcs...)
|
||||
|
||||
return &Director{proxyRepository, layerRepository, registry}
|
||||
registry := NewLayerRegistry(opts.Layers...)
|
||||
|
||||
return &Director{
|
||||
proxyRepository: proxyRepository,
|
||||
layerRepository: layerRepository,
|
||||
layerRegistry: registry,
|
||||
proxyCache: opts.ProxyCache,
|
||||
layerCache: opts.LayerCache,
|
||||
}
|
||||
}
|
||||
|
58
internal/proxy/director/options.go
Normal file
58
internal/proxy/director/options.go
Normal file
@ -0,0 +1,58 @@
|
||||
package director
|
||||
|
||||
import (
|
||||
"time"
|
||||
|
||||
"forge.cadoles.com/cadoles/bouncer/internal/cache"
|
||||
"forge.cadoles.com/cadoles/bouncer/internal/cache/memory"
|
||||
"forge.cadoles.com/cadoles/bouncer/internal/cache/ttl"
|
||||
"forge.cadoles.com/cadoles/bouncer/internal/store"
|
||||
)
|
||||
|
||||
type Options struct {
|
||||
Layers []Layer
|
||||
ProxyCache cache.Cache[string, []*store.Proxy]
|
||||
LayerCache cache.Cache[string, []*store.Layer]
|
||||
}
|
||||
|
||||
type OptionFunc func(opts *Options)
|
||||
|
||||
func NewOptions(funcs ...OptionFunc) *Options {
|
||||
opts := &Options{
|
||||
Layers: make([]Layer, 0),
|
||||
ProxyCache: ttl.NewCache(
|
||||
memory.NewCache[string, []*store.Proxy](),
|
||||
memory.NewCache[string, time.Time](),
|
||||
30*time.Second,
|
||||
),
|
||||
LayerCache: ttl.NewCache(
|
||||
memory.NewCache[string, []*store.Layer](),
|
||||
memory.NewCache[string, time.Time](),
|
||||
30*time.Second,
|
||||
),
|
||||
}
|
||||
|
||||
for _, fn := range funcs {
|
||||
fn(opts)
|
||||
}
|
||||
|
||||
return opts
|
||||
}
|
||||
|
||||
func WithLayers(layers ...Layer) OptionFunc {
|
||||
return func(opts *Options) {
|
||||
opts.Layers = layers
|
||||
}
|
||||
}
|
||||
|
||||
func WithProxyCache(cache cache.Cache[string, []*store.Proxy]) OptionFunc {
|
||||
return func(opts *Options) {
|
||||
opts.ProxyCache = cache
|
||||
}
|
||||
}
|
||||
|
||||
func WithLayerCache(cache cache.Cache[string, []*store.Layer]) OptionFunc {
|
||||
return func(opts *Options) {
|
||||
opts.LayerCache = cache
|
||||
}
|
||||
}
|
@ -1,23 +1,27 @@
|
||||
package proxy
|
||||
|
||||
import (
|
||||
"time"
|
||||
|
||||
"forge.cadoles.com/cadoles/bouncer/internal/config"
|
||||
"forge.cadoles.com/cadoles/bouncer/internal/proxy/director"
|
||||
)
|
||||
|
||||
type Option struct {
|
||||
ServerConfig config.ProxyServerConfig
|
||||
RedisConfig config.RedisConfig
|
||||
DirectorLayers []director.Layer
|
||||
ServerConfig config.ProxyServerConfig
|
||||
RedisConfig config.RedisConfig
|
||||
DirectorLayers []director.Layer
|
||||
DirectorCacheTTL time.Duration
|
||||
}
|
||||
|
||||
type OptionFunc func(*Option)
|
||||
|
||||
func defaultOption() *Option {
|
||||
return &Option{
|
||||
ServerConfig: config.NewDefaultProxyServerConfig(),
|
||||
RedisConfig: config.NewDefaultRedisConfig(),
|
||||
DirectorLayers: make([]director.Layer, 0),
|
||||
ServerConfig: config.NewDefaultProxyServerConfig(),
|
||||
RedisConfig: config.NewDefaultRedisConfig(),
|
||||
DirectorLayers: make([]director.Layer, 0),
|
||||
DirectorCacheTTL: 30 * time.Second,
|
||||
}
|
||||
}
|
||||
|
||||
@ -38,3 +42,9 @@ func WithDirectorLayers(layers ...director.Layer) OptionFunc {
|
||||
opt.DirectorLayers = layers
|
||||
}
|
||||
}
|
||||
|
||||
func WithDirectorCacheTTL(ttl time.Duration) OptionFunc {
|
||||
return func(opt *Option) {
|
||||
opt.DirectorCacheTTL = ttl
|
||||
}
|
||||
}
|
||||
|
156
internal/proxy/proxy_test.go
Normal file
156
internal/proxy/proxy_test.go
Normal file
@ -0,0 +1,156 @@
|
||||
package proxy_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"net/http/httputil"
|
||||
"net/url"
|
||||
"os"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"forge.cadoles.com/Cadoles/go-proxy"
|
||||
"forge.cadoles.com/cadoles/bouncer/internal/cache/memory"
|
||||
"forge.cadoles.com/cadoles/bouncer/internal/cache/ttl"
|
||||
"forge.cadoles.com/cadoles/bouncer/internal/proxy/director"
|
||||
"forge.cadoles.com/cadoles/bouncer/internal/store"
|
||||
redisStore "forge.cadoles.com/cadoles/bouncer/internal/store/redis"
|
||||
"github.com/pkg/errors"
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
func BenchmarkProxy(b *testing.B) {
|
||||
redisEndpoint := os.Getenv("BOUNCER_BENCH_REDIS_ADDR")
|
||||
if redisEndpoint == "" {
|
||||
redisEndpoint = "127.0.0.1:6379"
|
||||
}
|
||||
|
||||
client := redis.NewUniversalClient(&redis.UniversalOptions{
|
||||
Addrs: []string{redisEndpoint},
|
||||
})
|
||||
|
||||
proxyRepository := redisStore.NewProxyRepository(client)
|
||||
layerRepository := redisStore.NewLayerRepository(client)
|
||||
|
||||
backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.WriteHeader(http.StatusOK)
|
||||
if _, err := w.Write([]byte("Hello, world.")); err != nil {
|
||||
b.Logf("[ERROR] %+v", errors.WithStack(err))
|
||||
}
|
||||
}))
|
||||
defer backend.Close()
|
||||
|
||||
if err := waitFor(backend.URL, 5*time.Second); err != nil {
|
||||
b.Fatalf("[FATAL] %+v", errors.WithStack(err))
|
||||
}
|
||||
|
||||
b.Logf("started backend '%s'", backend.URL)
|
||||
|
||||
ctx := context.Background()
|
||||
|
||||
proxyName := store.ProxyName(b.Name())
|
||||
|
||||
b.Logf("creating proxy '%s'", proxyName)
|
||||
|
||||
if err := proxyRepository.DeleteProxy(ctx, proxyName); err != nil {
|
||||
b.Fatalf("[FATAL] %+v", errors.WithStack(err))
|
||||
}
|
||||
|
||||
if _, err := proxyRepository.CreateProxy(ctx, proxyName, backend.URL, "*"); err != nil {
|
||||
b.Fatalf("[FATAL] %+v", errors.WithStack(err))
|
||||
}
|
||||
|
||||
if _, err := proxyRepository.UpdateProxy(ctx, proxyName, store.WithProxyUpdateEnabled(true)); err != nil {
|
||||
b.Fatalf("[FATAL] %+v", errors.WithStack(err))
|
||||
}
|
||||
|
||||
director := director.New(
|
||||
proxyRepository, layerRepository,
|
||||
director.WithLayerCache(
|
||||
ttl.NewCache(
|
||||
memory.NewCache[string, []*store.Layer](),
|
||||
memory.NewCache[string, time.Time](),
|
||||
30*time.Second,
|
||||
),
|
||||
),
|
||||
director.WithProxyCache(
|
||||
ttl.NewCache(
|
||||
memory.NewCache[string, []*store.Proxy](),
|
||||
memory.NewCache[string, time.Time](),
|
||||
30*time.Second,
|
||||
),
|
||||
),
|
||||
)
|
||||
|
||||
directorMiddleware := director.Middleware()
|
||||
|
||||
handler := proxy.New(
|
||||
proxy.WithRequestTransformers(
|
||||
director.RequestTransformer(),
|
||||
),
|
||||
proxy.WithResponseTransformers(
|
||||
director.ResponseTransformer(),
|
||||
),
|
||||
proxy.WithReverseProxyFactory(func(ctx context.Context, target *url.URL) *httputil.ReverseProxy {
|
||||
reverse := httputil.NewSingleHostReverseProxy(target)
|
||||
reverse.ErrorHandler = func(w http.ResponseWriter, r *http.Request, err error) {
|
||||
b.Logf("[ERROR] %s", errors.WithStack(err))
|
||||
}
|
||||
return reverse
|
||||
}),
|
||||
)
|
||||
|
||||
server := httptest.NewServer(directorMiddleware(handler))
|
||||
defer server.Close()
|
||||
|
||||
b.Logf("started proxy '%s'", server.URL)
|
||||
|
||||
httpClient := server.Client()
|
||||
|
||||
b.ResetTimer()
|
||||
|
||||
for i := 0; i < b.N; i++ {
|
||||
res, err := httpClient.Get(server.URL)
|
||||
if err != nil {
|
||||
b.Errorf("could not fetch server url: %+v", errors.WithStack(err))
|
||||
}
|
||||
|
||||
body, err := io.ReadAll(res.Body)
|
||||
if err != nil {
|
||||
b.Errorf("could not read response body: %+v", errors.WithStack(err))
|
||||
}
|
||||
|
||||
b.Logf("%s - %v", res.Status, string(body))
|
||||
|
||||
if err := res.Body.Close(); err != nil {
|
||||
b.Errorf("could not close response body: %+v", errors.WithStack(err))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func waitFor(url string, ttl time.Duration) error {
|
||||
var lastErr error
|
||||
timeout := time.After(ttl)
|
||||
for {
|
||||
select {
|
||||
case <-timeout:
|
||||
if lastErr != nil {
|
||||
return lastErr
|
||||
}
|
||||
|
||||
return errors.New("wait timed out")
|
||||
default:
|
||||
res, err := http.Get(url)
|
||||
if err != nil {
|
||||
lastErr = errors.WithStack(err)
|
||||
continue
|
||||
}
|
||||
|
||||
if res.StatusCode >= 200 && res.StatusCode < 400 {
|
||||
return nil
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
@ -11,6 +11,8 @@ import (
|
||||
"time"
|
||||
|
||||
"forge.cadoles.com/Cadoles/go-proxy"
|
||||
"forge.cadoles.com/cadoles/bouncer/internal/cache/memory"
|
||||
"forge.cadoles.com/cadoles/bouncer/internal/cache/ttl"
|
||||
bouncerChi "forge.cadoles.com/cadoles/bouncer/internal/chi"
|
||||
"forge.cadoles.com/cadoles/bouncer/internal/config"
|
||||
"forge.cadoles.com/cadoles/bouncer/internal/proxy/director"
|
||||
@ -25,11 +27,12 @@ import (
|
||||
)
|
||||
|
||||
type Server struct {
|
||||
serverConfig config.ProxyServerConfig
|
||||
redisConfig config.RedisConfig
|
||||
directorLayers []director.Layer
|
||||
proxyRepository store.ProxyRepository
|
||||
layerRepository store.LayerRepository
|
||||
serverConfig config.ProxyServerConfig
|
||||
redisConfig config.RedisConfig
|
||||
directorLayers []director.Layer
|
||||
directorCacheTTL time.Duration
|
||||
proxyRepository store.ProxyRepository
|
||||
layerRepository store.LayerRepository
|
||||
}
|
||||
|
||||
func (s *Server) Start(ctx context.Context) (<-chan net.Addr, <-chan error) {
|
||||
@ -86,7 +89,21 @@ func (s *Server) run(parentCtx context.Context, addrs chan net.Addr, errs chan e
|
||||
director := director.New(
|
||||
s.proxyRepository,
|
||||
s.layerRepository,
|
||||
s.directorLayers...,
|
||||
director.WithLayers(s.directorLayers...),
|
||||
director.WithLayerCache(
|
||||
ttl.NewCache(
|
||||
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,
|
||||
),
|
||||
),
|
||||
)
|
||||
|
||||
if s.serverConfig.HTTP.UseRealIP {
|
||||
@ -184,8 +201,9 @@ func NewServer(funcs ...OptionFunc) *Server {
|
||||
}
|
||||
|
||||
return &Server{
|
||||
serverConfig: opt.ServerConfig,
|
||||
redisConfig: opt.RedisConfig,
|
||||
directorLayers: opt.DirectorLayers,
|
||||
serverConfig: opt.ServerConfig,
|
||||
redisConfig: opt.RedisConfig,
|
||||
directorLayers: opt.DirectorLayers,
|
||||
directorCacheTTL: opt.DirectorCacheTTL,
|
||||
}
|
||||
}
|
||||
|
Reference in New Issue
Block a user