mirror of
https://github.com/gravitational/teleport.git
synced 2026-09-24 16:17:11 +08:00
Add DeleteBatch scaffolding and tests (#66258)
* Add DeleteBatch scaffolding and tests * update * Deduplicate keys * Add type assertion
This commit is contained in:
+45
-2
@@ -30,6 +30,7 @@ import (
|
||||
"github.com/jonboulle/clockwork"
|
||||
|
||||
"github.com/gravitational/teleport/api/types"
|
||||
"github.com/gravitational/teleport/lib/utils/slices"
|
||||
)
|
||||
|
||||
// Forever means that object TTL will not expire unless deleted
|
||||
@@ -385,17 +386,32 @@ type BatchPutter interface {
|
||||
PutBatch(context.Context, []Item) ([]string, error)
|
||||
}
|
||||
|
||||
// BatchDeleter is an optional interface that backends can implement
|
||||
// to support batched DeleteBatch operations for improved performance when
|
||||
// deleting multiple items at once.
|
||||
type BatchDeleter interface {
|
||||
// DeleteBatch deletes multiple keys from the backend in a single call, in a
|
||||
// way that is equivalent to a loop around multiple invocations of
|
||||
// [backend.Delete], but with the potential to be more efficient or faster,
|
||||
// depending on the implementation. Keys that do not exist are silently
|
||||
// ignored. If an error is returned, it's possible for some of the keys to
|
||||
// have already been deleted.
|
||||
DeleteBatch(context.Context, []Key) error
|
||||
}
|
||||
|
||||
// PutBatch is an implementation of PutBatch that by default calls Put for each item.
|
||||
// Backends can overwrite this behavior providing optimized PutBatch implementation.
|
||||
//
|
||||
// WARNING: Make sure that items have unique keys when calling PutBatch.
|
||||
//
|
||||
// TODO(smallinsky): Move to backend interfaces and make required for backends to
|
||||
// implement batch deletes interfaces.
|
||||
func PutBatch(ctx context.Context, bk Backend, items []Item) ([]string, error) {
|
||||
if v, hasDuplicate := hasDuplicateKeys(items); hasDuplicate {
|
||||
return nil, trace.BadParameter("duplicate key detected in PutBatch: %q", v)
|
||||
}
|
||||
// Many Backend implementations rely on unique keys for correct operation.
|
||||
// Where it is up to the caller to ensure this to remove duplication keys
|
||||
// Just in case we will fallback to single Put calls if duplicates are detected.
|
||||
// Where it is up to the caller to ensure this to remove duplication keys.
|
||||
if v, ok := bk.(BatchPutter); ok {
|
||||
revs, err := v.PutBatch(ctx, items)
|
||||
return revs, trace.Wrap(err)
|
||||
@@ -412,6 +428,33 @@ func PutBatch(ctx context.Context, bk Backend, items []Item) ([]string, error) {
|
||||
return revisions, nil
|
||||
}
|
||||
|
||||
// DeleteBatch deletes multiple keys from the backend. If the backend implements
|
||||
// [BatchDeleter], it delegates to the optimized batch implementation. Otherwise,
|
||||
// it falls back to calling [Backend.Delete] for each key individually.
|
||||
// Keys that do not exist are silently ignored.
|
||||
//
|
||||
// WARNING: Make sure that keys are unique when calling DeleteBatch.
|
||||
//
|
||||
// TODO(smallinsky): Move to backend interfaces and make required for backends to
|
||||
// implement batch deletes interfaces.
|
||||
func DeleteBatch(ctx context.Context, bk Backend, keys []Key) error {
|
||||
keys = slices.DeduplicateKey(keys, func(k Key) string { return k.String() })
|
||||
|
||||
if v, ok := bk.(BatchDeleter); ok {
|
||||
return trace.Wrap(v.DeleteBatch(ctx, keys))
|
||||
}
|
||||
|
||||
for _, key := range keys {
|
||||
if err := bk.Delete(ctx, key); err != nil {
|
||||
if trace.IsNotFound(err) {
|
||||
continue
|
||||
}
|
||||
return trace.Wrap(err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func hasDuplicateKeys(items Items) (string, bool) {
|
||||
seen := make(map[string]struct{})
|
||||
for _, ca := range items {
|
||||
|
||||
@@ -81,6 +81,8 @@ func (r *ReporterConfig) CheckAndSetDefaults() error {
|
||||
}
|
||||
|
||||
var _ Backend = (*Reporter)(nil)
|
||||
var _ BatchDeleter = (*Reporter)(nil)
|
||||
var _ BatchPutter = (*Reporter)(nil)
|
||||
|
||||
// Reporter wraps a Backend implementation and reports
|
||||
// statistics about the backend operations
|
||||
@@ -545,6 +547,32 @@ func (s *Reporter) AtomicWrite(ctx context.Context, condacts []ConditionalAction
|
||||
return
|
||||
}
|
||||
|
||||
// DeleteBatch deletes multiple keys from the backend.
|
||||
func (s *Reporter) DeleteBatch(ctx context.Context, keys []Key) error {
|
||||
ctx, span := s.Tracer.Start(
|
||||
ctx,
|
||||
"backend/DeleteBatch",
|
||||
oteltrace.WithAttributes(
|
||||
attribute.Int("batch_size", len(keys)),
|
||||
),
|
||||
)
|
||||
defer span.End()
|
||||
|
||||
start := s.Clock().Now()
|
||||
err := DeleteBatch(ctx, s.Backend, keys)
|
||||
s.batchWriteLatencies.Observe(s.Clock().Since(start).Seconds())
|
||||
s.batchWriteRequests.Inc()
|
||||
if err != nil {
|
||||
s.batchWriteRequestsFailed.Inc()
|
||||
} else {
|
||||
s.writes.Add(float64(len(keys)))
|
||||
}
|
||||
for _, key := range keys {
|
||||
s.trackRequest(ctx, types.OpDelete, key, Key{})
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
// DeleteRange deletes range of items
|
||||
func (s *Reporter) DeleteRange(ctx context.Context, startKey, endKey Key) error {
|
||||
ctx, span := s.Tracer.Start(
|
||||
|
||||
@@ -75,6 +75,8 @@ func IsKeySafe(key Key) bool {
|
||||
}
|
||||
|
||||
var _ Backend = (*Sanitizer)(nil)
|
||||
var _ BatchDeleter = (*Sanitizer)(nil)
|
||||
var _ BatchPutter = (*Sanitizer)(nil)
|
||||
|
||||
// Sanitizer wraps a [Backend] implementation to make sure all
|
||||
// [Key]s written to the backend are allowed. Retrieval and deletion
|
||||
@@ -170,6 +172,11 @@ func (s *Sanitizer) Delete(ctx context.Context, key Key) error {
|
||||
return s.backend.Delete(ctx, key)
|
||||
}
|
||||
|
||||
// DeleteBatch deletes multiple items from the backend.
|
||||
func (s *Sanitizer) DeleteBatch(ctx context.Context, keys []Key) error {
|
||||
return trace.Wrap(DeleteBatch(ctx, s.backend, keys))
|
||||
}
|
||||
|
||||
// ConditionalDelete deletes the item by key if the revision matches the stored revision.
|
||||
func (s *Sanitizer) ConditionalDelete(ctx context.Context, key Key, revision string) error {
|
||||
return s.backend.ConditionalDelete(ctx, key, revision)
|
||||
|
||||
@@ -0,0 +1,188 @@
|
||||
/*
|
||||
* Teleport
|
||||
* Copyright (C) 2026 Gravitational, Inc.
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as published by
|
||||
* the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
package test
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/gravitational/teleport/api/types"
|
||||
"github.com/gravitational/teleport/lib/backend"
|
||||
)
|
||||
|
||||
func runDeleteBatch(t *testing.T, newBackend Constructor) {
|
||||
t.Helper()
|
||||
|
||||
bk, _, err := newBackend()
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { _ = bk.Close() })
|
||||
|
||||
deleter, ok := bk.(backend.BatchDeleter)
|
||||
if !ok {
|
||||
t.Skip("backend does not implement DeleteBatch; skipping DeleteBatch suite")
|
||||
}
|
||||
|
||||
prefix := MakePrefix()
|
||||
rangeStart := prefix("")
|
||||
rangeEnd := backend.RangeEnd(prefix(""))
|
||||
|
||||
putItems := func(t *testing.T, items []backend.Item) {
|
||||
t.Helper()
|
||||
for _, item := range items {
|
||||
_, err := bk.Put(t.Context(), item)
|
||||
require.NoError(t, err)
|
||||
}
|
||||
}
|
||||
|
||||
newTestItems := func() []backend.Item {
|
||||
return []backend.Item{
|
||||
{Key: prefix("a"), Value: []byte("A"), Expires: time.Now().Add(1 * time.Hour)},
|
||||
{Key: prefix("b"), Value: []byte("B")},
|
||||
{Key: prefix("c"), Value: []byte("C"), Expires: time.Now().Add(2 * time.Hour)},
|
||||
}
|
||||
}
|
||||
|
||||
t.Run("delete batch items should be propagated in event stream", func(t *testing.T) {
|
||||
items := newTestItems()
|
||||
putItems(t, items)
|
||||
|
||||
w, err := bk.NewWatcher(t.Context(), backend.Watch{})
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { w.Close() })
|
||||
|
||||
select {
|
||||
case <-w.Done():
|
||||
t.Fatal("watcher closed immediately")
|
||||
case ev := <-w.Events():
|
||||
require.Equal(t, types.OpInit, ev.Type)
|
||||
}
|
||||
|
||||
keys := make([]backend.Key, 0, len(items))
|
||||
for _, item := range items {
|
||||
keys = append(keys, item.Key)
|
||||
}
|
||||
|
||||
err = deleter.DeleteBatch(t.Context(), keys)
|
||||
require.NoError(t, err)
|
||||
|
||||
got := waitForDeleteEvents(t, w, len(keys), watchEventTimeout)
|
||||
require.Len(t, got, len(keys))
|
||||
for i, key := range keys {
|
||||
require.Equal(t, 0, key.Compare(got[i]))
|
||||
}
|
||||
|
||||
// Confirm nothing remains.
|
||||
res, err := bk.GetRange(t.Context(), rangeStart, rangeEnd, backend.NoLimit)
|
||||
require.NoError(t, err)
|
||||
require.Empty(t, res.Items)
|
||||
})
|
||||
|
||||
t.Run("delete-then-verify-empty", func(t *testing.T) {
|
||||
items := newTestItems()
|
||||
putItems(t, items)
|
||||
|
||||
// Verify items exist.
|
||||
res, err := bk.GetRange(t.Context(), rangeStart, rangeEnd, backend.NoLimit)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, res.Items, len(items))
|
||||
|
||||
keys := make([]backend.Key, 0, len(items))
|
||||
for _, item := range items {
|
||||
keys = append(keys, item.Key)
|
||||
}
|
||||
|
||||
err = deleter.DeleteBatch(t.Context(), keys)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Verify items are gone.
|
||||
res, err = bk.GetRange(t.Context(), rangeStart, rangeEnd, backend.NoLimit)
|
||||
require.NoError(t, err)
|
||||
require.Empty(t, res.Items)
|
||||
})
|
||||
|
||||
t.Run("delete-nonexistent-keys", func(t *testing.T) {
|
||||
keys := []backend.Key{
|
||||
prefix("nonexistent1"),
|
||||
prefix("nonexistent2"),
|
||||
prefix("nonexistent3"),
|
||||
}
|
||||
|
||||
// Deleting keys that don't exist should not error.
|
||||
err := deleter.DeleteBatch(t.Context(), keys)
|
||||
require.NoError(t, err)
|
||||
})
|
||||
|
||||
t.Run("delete-empty-batch", func(t *testing.T) {
|
||||
err := deleter.DeleteBatch(t.Context(), nil)
|
||||
require.NoError(t, err)
|
||||
|
||||
err = deleter.DeleteBatch(t.Context(), []backend.Key{})
|
||||
require.NoError(t, err)
|
||||
})
|
||||
|
||||
t.Run("delete-partial-existing", func(t *testing.T) {
|
||||
items := newTestItems()
|
||||
putItems(t, items)
|
||||
|
||||
// Delete a mix of existing and non-existing keys.
|
||||
keys := []backend.Key{
|
||||
items[0].Key,
|
||||
prefix("nonexistent"),
|
||||
items[2].Key,
|
||||
}
|
||||
|
||||
err := deleter.DeleteBatch(t.Context(), keys)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Only the middle item should remain.
|
||||
res, err := bk.GetRange(t.Context(), rangeStart, rangeEnd, backend.NoLimit)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, res.Items, 1)
|
||||
require.Equal(t, 0, items[1].Key.Compare(res.Items[0].Key))
|
||||
|
||||
require.NoError(t, bk.DeleteRange(t.Context(), rangeStart, rangeEnd))
|
||||
})
|
||||
}
|
||||
|
||||
func waitForDeleteEvents(t *testing.T, w backend.Watcher, wantCount int, timeout time.Duration) []backend.Key {
|
||||
t.Helper()
|
||||
|
||||
var out []backend.Key
|
||||
deadline := time.NewTimer(timeout)
|
||||
defer deadline.Stop()
|
||||
|
||||
for len(out) < wantCount {
|
||||
select {
|
||||
case ev, ok := <-w.Events():
|
||||
if !ok {
|
||||
t.Fatalf("watcher closed before receiving all events: got=%d want=%d", len(out), wantCount)
|
||||
}
|
||||
if ev.Type == types.OpDelete {
|
||||
out = append(out, ev.Item.Key)
|
||||
}
|
||||
case <-deadline.C:
|
||||
t.Fatalf("timed out waiting for delete events: got=%d want=%d", len(out), wantCount)
|
||||
case <-w.Done():
|
||||
t.Fatalf("watcher done before receiving all events: got=%d want=%d", len(out), wantCount)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -202,6 +202,10 @@ func RunBackendComplianceSuite(t *testing.T, newBackend Constructor) {
|
||||
t.Run("PutBatch", func(t *testing.T) {
|
||||
runPutBatch(t, newBackend)
|
||||
})
|
||||
|
||||
t.Run("DeleteBatch", func(t *testing.T) {
|
||||
runDeleteBatch(t, newBackend)
|
||||
})
|
||||
}
|
||||
|
||||
// RequireItems asserts that the supplied `actual` items collection matches
|
||||
|
||||
@@ -29,6 +29,8 @@ import (
|
||||
)
|
||||
|
||||
var _ Backend = (*Wrapper)(nil)
|
||||
var _ BatchDeleter = (*Wrapper)(nil)
|
||||
var _ BatchPutter = (*Wrapper)(nil)
|
||||
|
||||
// Wrapper wraps a Backend implementation that can fail
|
||||
// on demand.
|
||||
@@ -123,6 +125,11 @@ func (s *Wrapper) Delete(ctx context.Context, key Key) error {
|
||||
return s.backend.Delete(ctx, key)
|
||||
}
|
||||
|
||||
// DeleteBatch deletes multiple items from the backend.
|
||||
func (s *Wrapper) DeleteBatch(ctx context.Context, keys []Key) error {
|
||||
return trace.Wrap(DeleteBatch(ctx, s.backend, keys))
|
||||
}
|
||||
|
||||
// ConditionalDelete deletes item by key if revisions match.
|
||||
func (s *Wrapper) ConditionalDelete(ctx context.Context, key Key, revision string) error {
|
||||
return s.backend.ConditionalDelete(ctx, key, revision)
|
||||
|
||||
Reference in New Issue
Block a user