diff --git a/lib/backend/etcdbk/etcd.go b/lib/backend/etcdbk/etcd.go index c6355e29f6c..ca136e6dd9c 100644 --- a/lib/backend/etcdbk/etcd.go +++ b/lib/backend/etcdbk/etcd.go @@ -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) } diff --git a/lib/backend/etcdbk/etcd_test.go b/lib/backend/etcdbk/etcd_test.go index 19e157905b5..7693f2ed0b6 100644 --- a/lib/backend/etcdbk/etcd_test.go +++ b/lib/backend/etcdbk/etcd_test.go @@ -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") != "" } diff --git a/lib/utils/fncache.go b/lib/utils/fncache.go index 73c4e9be882..53de9e71802 100644 --- a/lib/utils/fncache.go +++ b/lib/utils/fncache.go @@ -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 diff --git a/lib/utils/fncache_test.go b/lib/utils/fncache_test.go index 436dacd73f8..6ecad386837 100644 --- a/lib/utils/fncache_test.go +++ b/lib/utils/fncache_test.go @@ -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()) +}