opensource/teleport
3
0
Fork 0
mirror of https://github.com/gravitational/teleport.git synced 2026-09-24 16:17:11 +08:00
Code Issues Packages Projects Releases Wiki Activity

bucket etcd leases

Browse Source
This commit is contained in:
Forrest Marshall
2022-09-27 15:01:32 -07:00
committed by Forrest
parent 7559be43cb
commit ad4be5ae44
4 changed files with 226 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/backend/etcdbk/etcd.go
+95 -25
View File
@@ -135,15 +135,17 @@ var (
type EtcdBackend struct {
nodes []string
*log.Entry
cfg *Config
client *clientv3.Client
cancelC chan bool
stopC chan bool
clock clockwork.Clock
buf *backend.CircularBuffer
ctx context.Context
cancel context.CancelFunc
watchDone chan struct{}
cfg *Config
client *clientv3.Client
cancelC chan bool
stopC chan bool
clock clockwork.Clock
buf *backend.CircularBuffer
leaseBucket time.Duration
leaseCache *utils.FnCache
ctx context.Context
cancel context.CancelFunc
watchDone chan struct{}
}
// Config represents JSON config for etcd backend
@@ -186,8 +188,42 @@ func GetName() string {
// keep this here to test interface conformance
var _ backend.Backend = &EtcdBackend{}
// Option is an etcd backend functional option (used in tests).
type Option func(*options)
type options struct {
leaseBucket time.Duration
clock clockwork.Clock
}
// LeaseBucket overrides the default lease bucketing size
func LeaseBucket(d time.Duration) Option {
return func(opts *options) {
opts.leaseBucket = d
}
}
// Clock overrides the default clockwork.Clock
func Clock(clock clockwork.Clock) Option {
return func(opts *options) {
opts.clock = clock
}
}
// New returns new instance of Etcd-powered backend
func New(ctx context.Context, params backend.Params) (*EtcdBackend, error) {
func New(ctx context.Context, params backend.Params, opts ...Option) (*EtcdBackend, error) {
var options options
for _, opt := range opts {
opt(&options)
}
if options.leaseBucket == 0 {
options.leaseBucket = time.Second * 10
}
if options.clock == nil {
options.clock = clockwork.NewRealClock()
}
err := metrics.RegisterPrometheusCollectors(prometheusCollectors...)
if err != nil {
return nil, trace.Wrap(err)
@@ -210,17 +246,31 @@ func New(ctx context.Context, params backend.Params) (*EtcdBackend, error) {
backend.BufferCapacity(cfg.BufferSize),
)
closeCtx, cancel := context.WithCancel(ctx)
leaseCache, err := utils.NewFnCache(utils.FnCacheConfig{
TTL: utils.SeventhJitter(time.Minute * 2),
Context: closeCtx,
Clock: options.clock,
ReloadOnErr: true,
CleanupInterval: utils.SeventhJitter(time.Minute * 2),
})
if err != nil {
return nil, trace.Wrap(err)
}
b := &EtcdBackend{
Entry: log.WithFields(log.Fields{trace.Component: GetName()}),
cfg: cfg,
nodes: cfg.Nodes,
cancelC: make(chan bool, 1),
stopC: make(chan bool, 1),
clock: clockwork.NewRealClock(),
cancel: cancel,
ctx: closeCtx,
watchDone: make(chan struct{}),
buf: buf,
Entry: log.WithFields(log.Fields{trace.Component: GetName()}),
cfg: cfg,
nodes: cfg.Nodes,
cancelC: make(chan bool, 1),
stopC: make(chan bool, 1),
clock: options.clock,
cancel: cancel,
ctx: closeCtx,
watchDone: make(chan struct{}),
buf: buf,
leaseBucket: utils.SeventhJitter(options.leaseBucket),
leaseCache: leaseCache,
}
// Check that the etcd nodes are at least the minimum version supported
@@ -744,17 +794,37 @@ func (b *EtcdBackend) DeleteRange(ctx context.Context, startKey, endKey []byte)
}
func (b *EtcdBackend) setupLease(ctx context.Context, item backend.Item, lease *backend.Lease, opts *[]clientv3.OpOption) error {
ttl := b.ttl(item.Expires)
elease, err := b.client.Grant(ctx, seconds(ttl))
// in order to reduce excess redundant lease generation, we bucket expiry times
// to the nearest multiple of 10s and then grant one lease per bucket. Too many
// leases can cause problems for etcd at scale.
// TODO(fspmarshall): make bucket size configurable.
bucket := roundUp(item.Expires, b.leaseBucket)
leaseID, err := utils.FnCacheGet(ctx, b.leaseCache, bucket, func(ctx context.Context) (clientv3.LeaseID, error) {
ttl := b.ttl(bucket)
elease, err := b.client.Grant(ctx, seconds(ttl))
if err != nil {
return 0, convertErr(err)
}
return elease.ID, nil
})
if err != nil {
return convertErr(err)
return trace.Wrap(err)
}
*opts = []clientv3.OpOption{clientv3.WithLease(elease.ID)}
lease.ID = int64(elease.ID)
*opts = []clientv3.OpOption{clientv3.WithLease(leaseID)}
lease.ID = int64(leaseID)
lease.Key = item.Key
return nil
}
// roundUp rounds up time t to the nearest multiple of duration d.
func roundUp(t time.Time, d time.Duration) time.Time {
r := t.Round(d)
if t.After(r) {
r = r.Add(d)
}
return r
}
func (b *EtcdBackend) ttl(expires time.Time) time.Duration {
return backend.TTL(b.clock, expires)
}
lib/backend/etcdbk/etcd_test.go
+55 -4
View File
@@ -19,6 +19,7 @@ package etcdbk
import (
"context"
"encoding/base64"
"fmt"
"os"
"testing"
"time"
@@ -51,6 +52,10 @@ var commonEtcdParams = backend.Params{
"tls_ca_file": "../../../examples/etcd/certs/ca-cert.pem",
}
var commonEtcdOptions = []Option{
LeaseBucket(time.Second), // tests are more picky about expiry granularity
}
func TestEtcd(t *testing.T) {
if !etcdTestEnabled() {
t.Skip("This test requires etcd, start it with examples/etcd/start-etcd.sh and set TELEPORT_ETCD_TEST=yes")
@@ -69,7 +74,7 @@ func TestEtcd(t *testing.T) {
// No need to check target backend - all Etcd backends create by this test
// point to the same datastore.
bk, err := New(context.Background(), commonEtcdParams)
bk, err := New(context.Background(), commonEtcdParams, commonEtcdOptions...)
if err != nil {
return nil, nil, err
}
@@ -93,7 +98,7 @@ func TestPrefix(t *testing.T) {
ctx := context.Background()
// Given an etcd backend with a minimal configuration...
unprefixedUut, err := New(context.Background(), commonEtcdParams)
unprefixedUut, err := New(context.Background(), commonEtcdParams, commonEtcdOptions...)
require.NoError(t, err)
defer unprefixedUut.Close()
@@ -104,7 +109,7 @@ func TestPrefix(t *testing.T) {
}
cfg["prefix"] = customPrefix
prefixedUut, err := New(context.Background(), cfg)
prefixedUut, err := New(context.Background(), cfg, commonEtcdOptions...)
require.NoError(t, err)
defer prefixedUut.Close()
@@ -176,7 +181,7 @@ func TestCompareAndSwapOversizedValue(t *testing.T) {
"tls_ca_file": "../../../examples/etcd/certs/ca-cert.pem",
"dial_timeout": 500 * time.Millisecond,
"etcd_max_client_msg_size_bytes": maxClientMsgSize,
})
}, commonEtcdOptions...)
require.NoError(t, err)
defer bk.Close()
prefix := test.MakePrefix()
@@ -192,6 +197,52 @@ func TestCompareAndSwapOversizedValue(t *testing.T) {
require.Regexp(t, ".*ResourceExhausted.*", err)
}
func TestLeaseBucketing(t *testing.T) {
const pfx = "lease-bucket-test"
const count = 40
if !etcdTestEnabled() {
t.Skip("This test requires etcd, start it with examples/etcd/start-etcd.sh and set TELEPORT_ETCD_TEST=yes")
}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
var opts []Option
opts = append(opts, commonEtcdOptions...)
opts = append(opts, LeaseBucket(time.Second*2))
bk, err := New(ctx, commonEtcdParams, opts...)
require.NoError(t, err)
defer bk.Close()
for i := 0; i < count; i++ {
_, err := bk.Put(ctx, backend.Item{
Key: backend.Key(pfx, fmt.Sprintf("%d", i)),
Value: []byte(fmt.Sprintf("val-%d", i)),
Expires: time.Now().Add(time.Minute),
})
require.NoError(t, err)
time.Sleep(time.Millisecond * 200)
}
start := backend.Key(pfx, "")
rslt, err := bk.GetRange(ctx, start, backend.RangeEnd(start), backend.NoLimit)
require.NoError(t, err)
require.Len(t, rslt.Items, count)
leases := make(map[int64]struct{})
for _, item := range rslt.Items {
leases[item.LeaseID] = struct{}{}
}
// ensure that we averaged more than 1 item per lease, but
// also spanned more than one bucket.
require.Greater(t, len(leases), 1)
require.Less(t, len(leases), count/2)
}
func etcdTestEnabled() bool {
return os.Getenv("TELEPORT_ETCD_TEST") != ""
}
lib/utils/fncache.go
+23 -4
View File
@@ -46,10 +46,12 @@ type FnCache struct {
entries map[any]*fnCacheEntry
}
// cleanupMultiplier is an arbitrary multiplier used to derive the schedule for
// periodic lazy cleanup of expired entries. This cache is meant to be used to
// cleanupMultiplier is an arbitrary multiplier used to derive the default interval
// for periodic lazy cleanup of expired entries. This cache is typically used to
// store a small number of regularly read keys, so most old values aught to be
// removed upon subsequent reads of the same key.
// removed upon subsequent reads of the same key. If the cache is being used in a
// context where keys might become regularly orphaned (no longer read), then a
// custom CleanupInterval should be provided.
const cleanupMultiplier time.Duration = 16
type FnCacheConfig struct {
@@ -60,6 +62,16 @@ type FnCacheConfig struct {
// Context is the context used to cancel the cache. All loadfns
// will be provided this context.
Context context.Context
// ReloadOnErr causes entries to be reloaded immediately if
// the currently loaded value is an error. Note that all concurrent
// requests registered before load completes still observe the
// same error. This option is only really useful for longer TTLs.
ReloadOnErr bool
// CleanupInterval is the interval at which cleanups occur (defaults to
// 16x the supplied TTL). Longer cleanup intervals are appropriate for
// caches where keys are unlikely to become orphaned. Shorter cleanup
// intervals should be used when keys regularly become orphaned.
CleanupInterval time.Duration
}
func (c *FnCacheConfig) CheckAndSetDefaults() error {
@@ -75,6 +87,10 @@ func (c *FnCacheConfig) CheckAndSetDefaults() error {
c.Context = context.Background()
}
if c.CleanupInterval <= 0 {
c.CleanupInterval = c.TTL * cleanupMultiplier
}
return nil
}
@@ -149,7 +165,7 @@ func (c *FnCache) get(ctx context.Context, key any, loadfn func(ctx context.Cont
// check if we need to perform periodic cleanup
if now.After(c.nextCleanup) {
c.removeExpiredLocked(now)
c.nextCleanup = now.Add(c.cfg.TTL * cleanupMultiplier)
c.nextCleanup = now.Add(c.cfg.CleanupInterval)
}
entry := c.entries[key]
@@ -160,6 +176,9 @@ func (c *FnCache) get(ctx context.Context, key any, loadfn func(ctx context.Cont
select {
case <-entry.loaded:
needsReload = now.After(entry.t.Add(c.cfg.TTL))
if c.cfg.ReloadOnErr && entry.e != nil {
needsReload = true
}
default:
// reload is already in progress
needsReload = false
lib/utils/fncache_test.go
+53
View File
@@ -345,3 +345,56 @@ func TestFnCacheContext(t *testing.T) {
})
require.ErrorIs(t, err, ErrFnCacheClosed)
}
func TestFnCacheReloadOnErr(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
cache, err := NewFnCache(FnCacheConfig{
TTL: time.Minute,
ReloadOnErr: true,
})
require.NoError(t, err)
happy := atomic.NewInt64(0)
sad := atomic.NewInt64(0)
// test synchronous case, all sad path loads should result in
// calls to loadfn.
for i := 0; i < 100; i++ {
FnCacheGet(ctx, cache, "happy", func(ctx context.Context) (string, error) {
happy.Inc()
return "yay!", nil
})
FnCacheGet(ctx, cache, "sad", func(ctx context.Context) (string, error) {
sad.Inc()
return "", fmt.Errorf("uh-oh")
})
}
require.Equal(t, int64(1), happy.Load())
require.Equal(t, int64(100), sad.Load())
// test concurrent case. some "sad" loads should overlap now.
var wg sync.WaitGroup
for i := 0; i < 100; i++ {
wg.Add(2)
go func() {
defer wg.Done()
FnCacheGet(ctx, cache, "happy", func(ctx context.Context) (string, error) {
happy.Inc()
return "yay!", nil
})
}()
go func() {
defer wg.Done()
FnCacheGet(ctx, cache, "sad", func(ctx context.Context) (string, error) {
sad.Inc()
return "", fmt.Errorf("uh-oh")
})
}()
}
require.Equal(t, int64(1), happy.Load())
require.Greater(t, int64(200), sad.Load())
}