feat: add revision number to proxy and layers to identify changes
Cadoles/bouncer/pipeline/head This commit looks good Details

This commit is contained in:
wpetit 2024-06-27 17:03:50 +02:00
parent 3565618335
commit 3c1939f418
15 changed files with 272 additions and 44 deletions

View File

@ -13,6 +13,7 @@ func layerHeaderHints(outputMode format.OutputMode) format.Hints {
format.NewProp("Type", "Type"), format.NewProp("Type", "Type"),
format.NewProp("Enabled", "Enabled"), format.NewProp("Enabled", "Enabled"),
format.NewProp("Weight", "Weight"), format.NewProp("Weight", "Weight"),
format.NewProp("Revision", "Revision"),
}, },
} }
} }
@ -25,6 +26,7 @@ func layerHints(outputMode format.OutputMode) format.Hints {
format.NewProp("Type", "Type"), format.NewProp("Type", "Type"),
format.NewProp("Enabled", "Enabled"), format.NewProp("Enabled", "Enabled"),
format.NewProp("Weight", "Weight"), format.NewProp("Weight", "Weight"),
format.NewProp("Revision", "Revision"),
format.NewProp("Options", "Options"), format.NewProp("Options", "Options"),
format.NewProp("CreatedAt", "CreatedAt", table.WithCompactModeMaxColumnWidth(20)), format.NewProp("CreatedAt", "CreatedAt", table.WithCompactModeMaxColumnWidth(20)),
format.NewProp("UpdatedAt", "UpdatedAt", table.WithCompactModeMaxColumnWidth(20)), format.NewProp("UpdatedAt", "UpdatedAt", table.WithCompactModeMaxColumnWidth(20)),

View File

@ -12,6 +12,7 @@ func proxyHeaderHints(outputMode format.OutputMode) format.Hints {
format.NewProp("Name", "Name"), format.NewProp("Name", "Name"),
format.NewProp("Enabled", "Enabled"), format.NewProp("Enabled", "Enabled"),
format.NewProp("Weight", "Weight"), format.NewProp("Weight", "Weight"),
format.NewProp("Revision", "Revision"),
}, },
} }
} }
@ -25,6 +26,7 @@ func proxyHints(outputMode format.OutputMode) format.Hints {
format.NewProp("To", "To"), format.NewProp("To", "To"),
format.NewProp("Enabled", "Enabled"), format.NewProp("Enabled", "Enabled"),
format.NewProp("Weight", "Weight"), format.NewProp("Weight", "Weight"),
format.NewProp("Revision", "Revision"),
format.NewProp("CreatedAt", "CreatedAt", table.WithCompactModeMaxColumnWidth(20)), format.NewProp("CreatedAt", "CreatedAt", table.WithCompactModeMaxColumnWidth(20)),
format.NewProp("UpdatedAt", "UpdatedAt", table.WithCompactModeMaxColumnWidth(20)), format.NewProp("UpdatedAt", "UpdatedAt", table.WithCompactModeMaxColumnWidth(20)),
}, },

View File

@ -31,8 +31,8 @@ func BenchmarkProxy(b *testing.B) {
Addrs: []string{redisEndpoint}, Addrs: []string{redisEndpoint},
}) })
proxyRepository := redisStore.NewProxyRepository(client) proxyRepository := redisStore.NewProxyRepository(client, redisStore.DefaultTxMaxAttempts, redisStore.DefaultTxBaseDelay)
layerRepository := redisStore.NewLayerRepository(client) layerRepository := redisStore.NewLayerRepository(client, redisStore.DefaultTxMaxAttempts, redisStore.DefaultTxBaseDelay)
backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusOK) w.WriteHeader(http.StatusOK)

View File

@ -17,9 +17,9 @@ func NewRedisClient(ctx context.Context, conf config.RedisConfig) redis.Universa
} }
func NewProxyRepository(ctx context.Context, client redis.UniversalClient) (store.ProxyRepository, error) { func NewProxyRepository(ctx context.Context, client redis.UniversalClient) (store.ProxyRepository, error) {
return redisStore.NewProxyRepository(client), nil return redisStore.NewProxyRepository(client, redisStore.DefaultTxMaxAttempts, redisStore.DefaultTxBaseDelay), nil
} }
func NewLayerRepository(ctx context.Context, client redis.UniversalClient) (store.LayerRepository, error) { func NewLayerRepository(ctx context.Context, client redis.UniversalClient) (store.LayerRepository, error) {
return redisStore.NewLayerRepository(client), nil return redisStore.NewLayerRepository(client, redisStore.DefaultTxMaxAttempts, redisStore.DefaultTxBaseDelay), nil
} }

View File

@ -13,6 +13,7 @@ type (
type LayerHeader struct { type LayerHeader struct {
Proxy ProxyName `json:"proxy"` Proxy ProxyName `json:"proxy"`
Name LayerName `json:"name"` Name LayerName `json:"name"`
Revision int `json:"revision"`
Type LayerType `json:"type"` Type LayerType `json:"type"`
Weight int `json:"weight"` Weight int `json:"weight"`

View File

@ -8,7 +8,7 @@ type ProxyName Name
type ProxyHeader struct { type ProxyHeader struct {
Name ProxyName `json:"name"` Name ProxyName `json:"name"`
Revision int `json:"revision"`
Weight int `json:"weight"` Weight int `json:"weight"`
Enabled bool `json:"enabled"` Enabled bool `json:"enabled"`
} }

View File

@ -3,10 +3,18 @@ package redis
import ( import (
"context" "context"
"encoding/json" "encoding/json"
"math/rand"
"strings" "strings"
"time"
"github.com/pkg/errors" "github.com/pkg/errors"
"github.com/redis/go-redis/v9" "github.com/redis/go-redis/v9"
"gitlab.com/wpetit/goweb/logger"
)
var (
DefaultTxMaxAttempts = 20
DefaultTxBaseDelay = 100 * time.Millisecond
) )
type jsonWrapper[T any] struct { type jsonWrapper[T any] struct {
@ -65,6 +73,33 @@ func key(parts ...string) string {
return strings.Join(parts, ":") return strings.Join(parts, ":")
} }
func WithRetry(ctx context.Context, client redis.UniversalClient, key string, fn func(ctx context.Context, tx *redis.Tx) error, maxAttempts int, baseDelay time.Duration) error {
var err error
delay := baseDelay
for attempt := 0; attempt < maxAttempts; attempt++ {
if err = WithTx(ctx, client, key, fn); err != nil {
err = errors.WithStack(err)
logger.Debug(ctx, "redis transaction failed", logger.E(err))
if errors.Is(err, redis.TxFailedErr) {
logger.Debug(ctx, "retrying redis transaction", logger.F("attempts", attempt), logger.F("delay", delay))
time.Sleep(delay)
delay = delay*2 + time.Duration(rand.Int63n(int64(baseDelay)))
continue
}
return err
}
return nil
}
return errors.WithStack(redis.TxFailedErr)
}
func WithTx(ctx context.Context, client redis.UniversalClient, key string, fn func(ctx context.Context, tx *redis.Tx) error) error { func WithTx(ctx context.Context, client redis.UniversalClient, key string, fn func(ctx context.Context, tx *redis.Tx) error) error {
txf := func(tx *redis.Tx) error { txf := func(tx *redis.Tx) error {
if err := fn(ctx, tx); err != nil { if err := fn(ctx, tx); err != nil {

View File

@ -10,6 +10,7 @@ import (
type layerHeaderItem struct { type layerHeaderItem struct {
Proxy string `redis:"proxy"` Proxy string `redis:"proxy"`
Name string `redis:"name"` Name string `redis:"name"`
Revision int `redis:"revision"`
Type string `redis:"type"` Type string `redis:"type"`
Weight int `redis:"weight"` Weight int `redis:"weight"`
@ -20,6 +21,7 @@ func (i *layerHeaderItem) ToLayerHeader() (*store.LayerHeader, error) {
layerHeader := &store.LayerHeader{ layerHeader := &store.LayerHeader{
Proxy: store.ProxyName(i.Proxy), Proxy: store.ProxyName(i.Proxy),
Name: store.LayerName(i.Name), Name: store.LayerName(i.Name),
Revision: i.Revision,
Type: store.LayerType(i.Type), Type: store.LayerType(i.Type),
Weight: i.Weight, Weight: i.Weight,
Enabled: i.Enabled, Enabled: i.Enabled,

View File

@ -15,6 +15,8 @@ const (
type LayerRepository struct { type LayerRepository struct {
client redis.UniversalClient client redis.UniversalClient
txMaxAttempts int
txRetryBaseDelay time.Duration
} }
// CreateLayer implements store.LayerRepository // CreateLayer implements store.LayerRepository
@ -28,6 +30,7 @@ func (r *LayerRepository) CreateLayer(ctx context.Context, proxyName store.Proxy
Name: string(layerName), Name: string(layerName),
Type: string(layerType), Type: string(layerType),
Weight: 0, Weight: 0,
Revision: 0,
Enabled: false, Enabled: false,
}, },
@ -96,7 +99,7 @@ func (r *LayerRepository) GetLayer(ctx context.Context, proxyName store.ProxyNam
key := layerKey(proxyName, layerName) key := layerKey(proxyName, layerName)
var layerItem *layerItem var layerItem *layerItem
err := WithTx(ctx, r.client, key, func(ctx context.Context, tx *redis.Tx) error { err := WithRetry(ctx, r.client, key, func(ctx context.Context, tx *redis.Tx) error {
pItem, err := r.txGetLayerItem(ctx, tx, proxyName, layerName) pItem, err := r.txGetLayerItem(ctx, tx, proxyName, layerName)
if err != nil { if err != nil {
return errors.WithStack(err) return errors.WithStack(err)
@ -105,7 +108,7 @@ func (r *LayerRepository) GetLayer(ctx context.Context, proxyName store.ProxyNam
layerItem = pItem layerItem = pItem
return nil return nil
}) }, r.txMaxAttempts, r.txRetryBaseDelay)
if err != nil { if err != nil {
return nil, errors.WithStack(err) return nil, errors.WithStack(err)
} }
@ -197,7 +200,7 @@ func (r *LayerRepository) UpdateLayer(ctx context.Context, proxyName store.Proxy
key := layerKey(proxyName, layerName) key := layerKey(proxyName, layerName)
var layerItem layerItem var layerItem layerItem
err := WithTx(ctx, r.client, key, func(ctx context.Context, tx *redis.Tx) error { err := WithRetry(ctx, r.client, key, func(ctx context.Context, tx *redis.Tx) error {
item, err := r.txGetLayerItem(ctx, tx, proxyName, layerName) item, err := r.txGetLayerItem(ctx, tx, proxyName, layerName)
if err != nil { if err != nil {
return errors.WithStack(err) return errors.WithStack(err)
@ -216,6 +219,7 @@ func (r *LayerRepository) UpdateLayer(ctx context.Context, proxyName store.Proxy
} }
item.UpdatedAt = wrap(time.Now().UTC()) item.UpdatedAt = wrap(time.Now().UTC())
item.Revision = item.Revision + 1
_, err = tx.TxPipelined(ctx, func(p redis.Pipeliner) error { _, err = tx.TxPipelined(ctx, func(p redis.Pipeliner) error {
p.HMSet(ctx, key, item.layerHeaderItem) p.HMSet(ctx, key, item.layerHeaderItem)
@ -230,7 +234,7 @@ func (r *LayerRepository) UpdateLayer(ctx context.Context, proxyName store.Proxy
layerItem = *item layerItem = *item
return nil return nil
}) }, r.txMaxAttempts, r.txRetryBaseDelay)
if err != nil { if err != nil {
return nil, errors.WithStack(err) return nil, errors.WithStack(err)
} }
@ -243,9 +247,11 @@ func (r *LayerRepository) UpdateLayer(ctx context.Context, proxyName store.Proxy
return layer, nil return layer, nil
} }
func NewLayerRepository(client redis.UniversalClient) *LayerRepository { func NewLayerRepository(client redis.UniversalClient, txMaxAttempts int, txRetryBaseDelay time.Duration) *LayerRepository {
return &LayerRepository{ return &LayerRepository{
client: client, client: client,
txMaxAttempts: txMaxAttempts,
txRetryBaseDelay: txRetryBaseDelay,
} }
} }

View File

@ -7,6 +7,6 @@ import (
) )
func TestLayerRepository(t *testing.T) { func TestLayerRepository(t *testing.T) {
repository := NewLayerRepository(client) repository := NewLayerRepository(client, DefaultTxMaxAttempts, DefaultTxBaseDelay)
testsuite.TestLayerRepository(t, repository) testsuite.TestLayerRepository(t, repository)
} }

View File

@ -9,6 +9,7 @@ import (
type proxyHeaderItem struct { type proxyHeaderItem struct {
Name string `redis:"name"` Name string `redis:"name"`
Revision int `redis:"revision"`
Weight int `redis:"weight"` Weight int `redis:"weight"`
Enabled bool `redis:"enabled"` Enabled bool `redis:"enabled"`
@ -20,6 +21,7 @@ type proxyHeaderItem struct {
func (i *proxyHeaderItem) ToProxyHeader() (*store.ProxyHeader, error) { func (i *proxyHeaderItem) ToProxyHeader() (*store.ProxyHeader, error) {
proxyHeader := &store.ProxyHeader{ proxyHeader := &store.ProxyHeader{
Name: store.ProxyName(i.Name), Name: store.ProxyName(i.Name),
Revision: i.Revision,
Weight: i.Weight, Weight: i.Weight,
Enabled: i.Enabled, Enabled: i.Enabled,
} }

View File

@ -15,6 +15,8 @@ const (
type ProxyRepository struct { type ProxyRepository struct {
client redis.UniversalClient client redis.UniversalClient
txMaxAttempts int
txRetryBaseDelay time.Duration
} }
// GetProxy implements store.ProxyRepository // GetProxy implements store.ProxyRepository
@ -22,7 +24,7 @@ func (r *ProxyRepository) GetProxy(ctx context.Context, name store.ProxyName) (*
key := proxyKey(name) key := proxyKey(name)
var proxyItem *proxyItem var proxyItem *proxyItem
err := WithTx(ctx, r.client, key, func(ctx context.Context, tx *redis.Tx) error { err := WithRetry(ctx, r.client, key, func(ctx context.Context, tx *redis.Tx) error {
pItem, err := r.txGetProxyItem(ctx, tx, name) pItem, err := r.txGetProxyItem(ctx, tx, name)
if err != nil { if err != nil {
return errors.WithStack(err) return errors.WithStack(err)
@ -31,7 +33,7 @@ func (r *ProxyRepository) GetProxy(ctx context.Context, name store.ProxyName) (*
proxyItem = pItem proxyItem = pItem
return nil return nil
}) }, r.txMaxAttempts, r.txRetryBaseDelay)
if err != nil { if err != nil {
return nil, errors.WithStack(err) return nil, errors.WithStack(err)
} }
@ -89,6 +91,7 @@ func (r *ProxyRepository) CreateProxy(ctx context.Context, name store.ProxyName,
CreatedAt: wrap(now), CreatedAt: wrap(now),
UpdatedAt: wrap(now), UpdatedAt: wrap(now),
Weight: 0, Weight: 0,
Revision: 0,
Enabled: false, Enabled: false,
}, },
To: to, To: to,
@ -191,7 +194,7 @@ func (r *ProxyRepository) UpdateProxy(ctx context.Context, name store.ProxyName,
key := proxyKey(name) key := proxyKey(name)
var proxyItem proxyItem var proxyItem proxyItem
err := WithTx(ctx, r.client, key, func(ctx context.Context, tx *redis.Tx) error { err := WithRetry(ctx, r.client, key, func(ctx context.Context, tx *redis.Tx) error {
item, err := r.txGetProxyItem(ctx, tx, name) item, err := r.txGetProxyItem(ctx, tx, name)
if err != nil { if err != nil {
return errors.WithStack(err) return errors.WithStack(err)
@ -214,6 +217,7 @@ func (r *ProxyRepository) UpdateProxy(ctx context.Context, name store.ProxyName,
} }
item.UpdatedAt = wrap(time.Now().UTC()) item.UpdatedAt = wrap(time.Now().UTC())
item.Revision = item.Revision + 1
_, err = tx.TxPipelined(ctx, func(p redis.Pipeliner) error { _, err = tx.TxPipelined(ctx, func(p redis.Pipeliner) error {
p.HMSet(ctx, key, item.proxyHeaderItem) p.HMSet(ctx, key, item.proxyHeaderItem)
@ -228,7 +232,7 @@ func (r *ProxyRepository) UpdateProxy(ctx context.Context, name store.ProxyName,
proxyItem = *item proxyItem = *item
return nil return nil
}) }, r.txMaxAttempts, r.txRetryBaseDelay)
if err != nil { if err != nil {
return nil, errors.WithStack(err) return nil, errors.WithStack(err)
} }
@ -241,9 +245,11 @@ func (r *ProxyRepository) UpdateProxy(ctx context.Context, name store.ProxyName,
return proxy, nil return proxy, nil
} }
func NewProxyRepository(client redis.UniversalClient) *ProxyRepository { func NewProxyRepository(client redis.UniversalClient, txMaxAttempts int, txRetryBaseDelay time.Duration) *ProxyRepository {
return &ProxyRepository{ return &ProxyRepository{
client: client, client: client,
txMaxAttempts: 20,
txRetryBaseDelay: txRetryBaseDelay,
} }
} }

View File

@ -7,6 +7,6 @@ import (
) )
func TestProxyRepository(t *testing.T) { func TestProxyRepository(t *testing.T) {
repository := NewProxyRepository(client) repository := NewProxyRepository(client, DefaultTxMaxAttempts, DefaultTxBaseDelay)
testsuite.TestProxyRepository(t, repository) testsuite.TestProxyRepository(t, repository)
} }

View File

@ -3,6 +3,7 @@ package testsuite
import ( import (
"context" "context"
"reflect" "reflect"
"sync"
"testing" "testing"
"forge.cadoles.com/cadoles/bouncer/internal/store" "forge.cadoles.com/cadoles/bouncer/internal/store"
@ -49,6 +50,10 @@ var layerRepositoryTestCases = []layerRepositoryTestCase{
return errors.Errorf("layer.UpdatedAt should not be zero value") return errors.Errorf("layer.UpdatedAt should not be zero value")
} }
if layer.Revision != 0 {
return errors.Errorf("layer.Revision should be zero")
}
return nil return nil
}, },
}, },
@ -230,6 +235,86 @@ var layerRepositoryTestCases = []layerRepositoryTestCase{
return errors.New("could not find created layer in query results") return errors.New("could not find created layer in query results")
} }
return nil
},
},
{
Name: "Create then update layer",
Do: func(repo store.LayerRepository) error {
ctx := context.Background()
var layerName store.LayerName = "create_then_update_layer"
var proxyName store.ProxyName = store.ProxyName(string(layerName) + "_proxy")
var layerType store.LayerType = "dummy"
var layerOptions store.LayerOptions = store.LayerOptions{}
createdLayer, err := repo.CreateLayer(ctx, proxyName, layerName, layerType, layerOptions)
if err != nil {
return errors.WithStack(err)
}
if e, g := 0, createdLayer.Revision; e != g {
return errors.Errorf("createdLayer.Revision: expected '%v', got '%v'", e, g)
}
updatedLayer, err := repo.UpdateLayer(ctx, proxyName, layerName)
if err != nil {
return errors.Wrap(err, "err should be nil")
}
if e, g := 1, updatedLayer.Revision; e != g {
return errors.Errorf("updatedLayer.Revision: expected '%v', got '%v'", e, g)
}
return nil
},
},
{
Name: "Update layer concurrently",
Do: func(repo store.LayerRepository) error {
ctx := context.Background()
var layerName store.LayerName = "update_layer_concurrently"
var proxyName store.ProxyName = store.ProxyName(string(layerName) + "_proxy")
var layerType store.LayerType = "dummy"
var layerOptions store.LayerOptions = store.LayerOptions{}
createdLayer, err := repo.CreateLayer(ctx, proxyName, layerName, layerType, layerOptions)
if err != nil {
return errors.WithStack(err)
}
if createdLayer.Revision != 0 {
return errors.Errorf("createdLayer.Revision should be zero")
}
var wg sync.WaitGroup
total := 100
wg.Add(total)
for i := 0; i < total; i++ {
go func(i int) {
defer wg.Done()
if _, err := repo.UpdateLayer(ctx, createdLayer.Proxy, createdLayer.Name); err != nil {
panic(errors.Wrap(err, "err should be nil"))
}
}(i)
}
wg.Wait()
layer, err := repo.GetLayer(ctx, createdLayer.Proxy, createdLayer.Name)
if err != nil {
return errors.Wrap(err, "err should be nil")
}
if e, g := total, layer.Revision; e != g {
return errors.Errorf("layer.Revision: expected '%v', got '%v'", e, g)
}
return nil return nil
}, },
}, },

View File

@ -3,6 +3,7 @@ package testsuite
import ( import (
"context" "context"
"reflect" "reflect"
"sync"
"testing" "testing"
"forge.cadoles.com/cadoles/bouncer/internal/store" "forge.cadoles.com/cadoles/bouncer/internal/store"
@ -51,6 +52,10 @@ var proxyRepositoryTestCases = []proxyRepositoryTestCase{
return errors.Errorf("proxy.UpdatedAt should not be zero value") return errors.Errorf("proxy.UpdatedAt should not be zero value")
} }
if proxy.Revision != 0 {
return errors.Errorf("proxy.Revision should be zero")
}
return nil return nil
}, },
}, },
@ -99,6 +104,10 @@ var proxyRepositoryTestCases = []proxyRepositoryTestCase{
return errors.Errorf("foundProxy.UpdatedAt: expected '%v', got '%v'", createdProxy.UpdatedAt, foundProxy.UpdatedAt) return errors.Errorf("foundProxy.UpdatedAt: expected '%v', got '%v'", createdProxy.UpdatedAt, foundProxy.UpdatedAt)
} }
if foundProxy.Revision != 0 {
return errors.Errorf("foundProxy.Revision should be zero")
}
return nil return nil
}, },
}, },
@ -194,6 +203,84 @@ var proxyRepositoryTestCases = []proxyRepositoryTestCase{
return errors.Errorf("err: expected store.ErrAlreadyExists, got '%+v'", err) return errors.Errorf("err: expected store.ErrAlreadyExists, got '%+v'", err)
} }
return nil
},
},
{
Name: "Create then update proxy",
Do: func(repo store.ProxyRepository) error {
ctx := context.Background()
to := "http://example.com"
var name store.ProxyName = "create_then_update_proxy"
createdProxy, err := repo.CreateProxy(ctx, name, to, "127.0.0.1:*", "localhost:*")
if err != nil {
return errors.Wrap(err, "err should be nil")
}
if createdProxy.Revision != 0 {
return errors.Errorf("createdProxy.Revision should be zero")
}
updatedProxy, err := repo.UpdateProxy(ctx, name)
if err != nil {
return errors.Wrap(err, "err should be nil")
}
if e, g := 1, updatedProxy.Revision; e != g {
return errors.Errorf("updatedProxy.Revision: expected '%v', got '%v'", e, g)
}
return nil
},
},
{
Name: "Update proxy concurrently",
Do: func(repo store.ProxyRepository) error {
ctx := context.Background()
to := "http://example.com"
var name store.ProxyName = "update_proxy_concurrently"
createdProxy, err := repo.CreateProxy(ctx, name, to, "127.0.0.1:*", "localhost:*")
if err != nil {
return errors.Wrap(err, "err should be nil")
}
if createdProxy.Revision != 0 {
return errors.Errorf("createdProxy.Revision should be zero")
}
var wg sync.WaitGroup
total := 100
wg.Add(total)
for i := 0; i < total; i++ {
go func(i int) {
defer wg.Done()
if _, err := repo.UpdateProxy(ctx, name); err != nil {
panic(errors.Wrap(err, "err should be nil"))
}
}(i)
}
wg.Wait()
proxy, err := repo.GetProxy(ctx, name)
if err != nil {
return errors.Wrap(err, "err should be nil")
}
if e, g := total, proxy.Revision; e != g {
return errors.Errorf("proxy.Revision: expected '%v', got '%v'", e, g)
}
return nil return nil
}, },
}, },