Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 23 additions & 2 deletions .github/workflows/jepsen-test.yml
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,7 @@ jobs:
go mod download
go build -o /tmp/elastickv-bin .
go build -o /tmp/elastickv-admin-bin ./cmd/elastickv-admin
go build -o /tmp/elastickv-raftadmin-bin ./cmd/raftadmin
- name: Warm Leiningen Maven cache
working-directory: jepsen
run: |
Expand All @@ -90,7 +91,8 @@ jobs:
BOOTSTRAP_MEMBERS="n1=127.0.0.1:50051,n2=127.0.0.1:50052,n3=127.0.0.1:50053"
RAFT_REDIS_MAP="127.0.0.1:50051=127.0.0.1:63791,127.0.0.1:50052=127.0.0.1:63792,127.0.0.1:50053=127.0.0.1:63793"
RAFT_S3_MAP="127.0.0.1:50051=127.0.0.1:63901,127.0.0.1:50052=127.0.0.1:63902,127.0.0.1:50053=127.0.0.1:63903"
RAFT_DYNAMO_MAP="127.0.0.1:50051=127.0.0.1:63801,127.0.0.1:50052=127.0.0.1:63802,127.0.0.1:50053=127.0.0.1:63803"
# No raftDynamoMap: DynamoDB workloads must exercise follower gRPC
# forwarding and leader-issued seals instead of HTTP leader proxying.
RAFT_SQS_MAP="127.0.0.1:50051=127.0.0.1:63501,127.0.0.1:50052=127.0.0.1:63502,127.0.0.1:50053=127.0.0.1:63503"

# Encrypted leg: one random owner-only file KEK shared by the
Expand Down Expand Up @@ -132,7 +134,6 @@ jobs:
--raftBootstrapMembers "$BOOTSTRAP_MEMBERS" \
--raftRedisMap "$RAFT_REDIS_MAP" \
--raftS3Map "$RAFT_S3_MAP" \
--raftDynamoMap "$RAFT_DYNAMO_MAP" \
--raftSqsMap "$RAFT_SQS_MAP" \
"${ENC_ARGS[@]}" \
> "/tmp/elastickv-demo-n${node}.log" 2>&1 &
Expand Down Expand Up @@ -183,6 +184,26 @@ jobs:
timeout-minutes: 3
run: |
timeout 120 ~/lein run -m elastickv.dynamodb-workload --local --time-limit 5 --rate 5 --concurrency 5 --dynamo-ports 63801,63802,63803 --host 127.0.0.1
- name: Run DynamoDB seal workload with recorded leader changes
working-directory: jepsen
run: |
printf 'run_sha=%s\n' "$GITHUB_SHA" > /tmp/dynamodb-seal-source.txt
timeout 150 ~/lein run -m elastickv.dynamodb-seal-workload --local \
Comment thread
bootjp marked this conversation as resolved.
Resolved
--time-limit 15 --rate 30 --concurrency 9 --max-txn-length 4 \
--dynamo-ports 63801,63802,63803 --host 127.0.0.1 \
--raftadmin-bin /tmp/elastickv-raftadmin-bin \
--raft-addresses 127.0.0.1:50051,127.0.0.1:50052,127.0.0.1:50053 \
--leader-change-interval 2
- name: Upload DynamoDB seal acceptance history
if: always()
uses: actions/upload-artifact@v7
with:
name: dynamodb-seal-encryption-${{ matrix.encryption }}
path: |
jepsen/store/elastickv-dynamodb-snapshot-seal/
/tmp/dynamodb-seal-source.txt
retention-days: 7
if-no-files-found: ignore
- name: Run DynamoDB per-type Jepsen workloads against elastickv
working-directory: jepsen
timeout-minutes: 10
Expand Down
2 changes: 1 addition & 1 deletion adapter/dynamodb_item_write.go
Original file line number Diff line number Diff line change
Expand Up @@ -495,7 +495,7 @@ func validatePutItemAdminPathKey(in putItemInput, schema *dynamoTableSchema) err

func (d *DynamoDBServer) commitItemWrite(ctx context.Context, readTimestamp kv.ReadTimestamp, req *kv.OperationGroup[kv.OP]) error {
dispatchCtx := readTimestamp.WithDispatchVoucher(ctx)
_, err := kv.DispatchWithReadTimestamp(dispatchCtx, d.coordinator, req)
_, err := d.dispatchWriteWithReadTimestamp(dispatchCtx, req)
if err != nil {
return errors.WithStack(err)
}
Expand Down
3 changes: 2 additions & 1 deletion adapter/dynamodb_locks.go
Original file line number Diff line number Diff line change
Expand Up @@ -391,8 +391,9 @@ func (d *DynamoDBServer) tryTxnWriteTimestamp(ctx context.Context, label string,
var resp *kv.CoordinateResponse
if sealTS == 0 {
resp, err = kv.DispatchWithLeaderIssuedCommitTS(floor.WithDispatchVoucher(ctx), d.coordinator, request)
err = normalizeDynamoWriteDispatchError(err)
} else {
resp, err = kv.DispatchWithReadTimestamp(floor.WithDispatchVoucher(ctx), d.coordinator, request)
resp, err = d.dispatchWriteWithReadTimestamp(floor.WithDispatchVoucher(ctx), request)
}
if err != nil {
return kv.ReadTimestamp{}, errors.WithStack(err)
Expand Down
6 changes: 3 additions & 3 deletions adapter/dynamodb_migration.go
Original file line number Diff line number Diff line change
Expand Up @@ -97,7 +97,7 @@ func (d *DynamoDBServer) startLegacyTableKeyMigration(
}
req.StartTS = readTS
dispatchCtx := readTimestamp.WithDispatchVoucher(ctx)
if _, err := kv.DispatchWithReadTimestamp(dispatchCtx, d.coordinator, req); err != nil {
if _, err := d.dispatchWriteWithReadTimestamp(dispatchCtx, req); err != nil {
return errors.WithStack(err)
}
return nil
Expand Down Expand Up @@ -186,7 +186,7 @@ func (d *DynamoDBServer) migrateLegacyItem(
req.StartTS = readTS
req.ReadKeys = append(req.ReadKeys, sealKey)
dispatchCtx := readTimestamp.WithDispatchVoucher(ctx)
if _, err := kv.DispatchWithReadTimestamp(dispatchCtx, d.coordinator, req); err == nil {
if _, err := d.dispatchWriteWithReadTimestamp(dispatchCtx, req); err == nil {
return nil
} else if !isRetryableTransactWriteError(err) {
return errors.WithStack(err)
Expand Down Expand Up @@ -338,7 +338,7 @@ func (d *DynamoDBServer) finalizeLegacyTableMigration(ctx context.Context, schem
},
}
dispatchCtx := readTimestamp.WithDispatchVoucher(ctx)
if _, err := kv.DispatchWithReadTimestamp(dispatchCtx, d.coordinator, req); err != nil {
if _, err := d.dispatchWriteWithReadTimestamp(dispatchCtx, req); err != nil {
return errors.WithStack(err)
}
d.launchDeletedTableCleanup(cleanup, schema.TableName, oldGeneration)
Expand Down
7 changes: 4 additions & 3 deletions adapter/dynamodb_schema.go
Original file line number Diff line number Diff line change
Expand Up @@ -205,7 +205,7 @@ func (d *DynamoDBServer) createTableWithRetry(ctx context.Context, tableName str
}
req.StartTS = readTS
dispatchCtx := readTimestamp.WithDispatchVoucher(ctx)
if _, err := kv.DispatchWithReadTimestamp(dispatchCtx, d.coordinator, req); err == nil {
if _, err := d.dispatchWriteWithReadTimestamp(dispatchCtx, req); err == nil {
return nil
} else if !isRetryableTransactWriteError(err) {
return errors.WithStack(err)
Expand Down Expand Up @@ -327,7 +327,7 @@ func (d *DynamoDBServer) deleteTableWithRetry(ctx context.Context, tableName str
},
}
dispatchCtx := readTimestamp.WithDispatchVoucher(ctx)
if _, err := kv.DispatchWithReadTimestamp(dispatchCtx, d.coordinator, req); err != nil {
if _, err := d.dispatchWriteWithReadTimestamp(dispatchCtx, req); err != nil {
if !isRetryableTransactWriteError(err) {
return errors.WithStack(err)
}
Expand Down Expand Up @@ -386,6 +386,7 @@ func (d *DynamoDBServer) dispatchDeletedTableCleanupPrefix(ctx context.Context,
{Op: kv.DelPrefix, Key: prefix},
},
})
err = normalizeDynamoWriteDispatchError(err)
if err == nil {
return nil
}
Expand All @@ -412,7 +413,7 @@ func (d *DynamoDBServer) dispatchDeleteBatch(ctx context.Context, keys [][]byte)
}
_, err := d.coordinator.Dispatch(ctx, req)
if err != nil {
return errors.WithStack(err)
return errors.WithStack(normalizeDynamoWriteDispatchError(err))
}
return nil
}
Expand Down
102 changes: 102 additions & 0 deletions adapter/dynamodb_snapshot_seal_benchmark_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,102 @@
package adapter

import (
"context"
"fmt"
"sync/atomic"
"testing"

"github.com/bootjp/elastickv/kv"
"github.com/bootjp/elastickv/store"
"github.com/cockroachdb/errors"
"github.com/stretchr/testify/require"
)

type sealBenchmarkCoordinator struct {
kv.Coordinator
seals atomic.Uint64
conflicts atomic.Uint64
}

func (c *sealBenchmarkCoordinator) Dispatch(ctx context.Context, req *kv.OperationGroup[kv.OP]) (*kv.CoordinateResponse, error) {
if isDynamoSnapshotSealRequest(req) {
c.seals.Add(1)
}
resp, err := c.Coordinator.Dispatch(ctx, req)
if errors.Is(err, store.ErrWriteConflict) {
c.conflicts.Add(1)
}
return resp, err
}

// A single durable Raft voter isolates the extra seal/OCC cost. This is not a
// networked quorum or a production throughput claim. Separate frontends avoid
// serializing the workload with a shared server's item-lock stripes.
func BenchmarkDynamoDBSnapshotSealRealRaft(b *testing.B) {
for _, mode := range []string{"independent", "hot_item", "collision"} {
b.Run(mode, func(b *testing.B) {
ctx := context.Background()
st := store.NewMVCCStore()
defer st.Close()
server, _, stop := sealRaftRestartableServer(b, b.TempDir(), st)
defer stop()
schema := dedupItemTable()
require.NoError(b, server.createTableWithRetry(ctx, schema.TableName, schema))
keys := sealBenchmarkKeys(b, schema, mode)
for _, key := range keys {
_, err := server.putItemWithRetry(ctx, putItemInput{
TableName: schema.TableName, Item: map[string]attributeValue{"pk": newStringAttributeValue(key)},
})
require.NoError(b, err)
}
coord := &sealBenchmarkCoordinator{Coordinator: server.coordinator}
var worker atomic.Uint64
b.ReportAllocs()
b.ResetTimer()
b.RunParallel(func(pb *testing.PB) {
local := NewDynamoDBServer(nil, st, coord)
key := keys[(worker.Add(1)-1)%uint64(len(keys))]
in := putItemInput{TableName: schema.TableName, Item: map[string]attributeValue{
"pk": newStringAttributeValue(key), "value": newStringAttributeValue("constant-size replacement"),
}}
for pb.Next() {
if _, err := local.putItemWithRetry(ctx, in); err != nil {
b.Error(err)
return
}
}
})
b.StopTimer()
b.ReportMetric(float64(coord.seals.Load())/float64(b.N), "seals/op")
b.ReportMetric(float64(coord.conflicts.Load())/float64(b.N), "conflicts/op")
})
}
}

func sealBenchmarkKeys(b *testing.B, schema *dynamoTableSchema, mode string) []string {
b.Helper()
if mode == "hot_item" {
return []string{"hot"}
}
const count = 4
buckets := make(map[string][]string)
independent := make([]string, 0, count)
for i := range 1 << 20 {
key := fmt.Sprintf("seal-bench-%d", i)
item, err := schema.itemKeyFromAttributes(map[string]attributeValue{"pk": newStringAttributeValue(key)})
require.NoError(b, err)
seal := string(dynamoItemSnapshotSealKey(schema.TableName, schema.Generation, item))
if mode == "independent" && len(buckets[seal]) == 0 {
independent = append(independent, key)
if len(independent) == count {
return independent
}
}
buckets[seal] = append(buckets[seal], key)
if mode == "collision" && len(buckets[seal]) == count {
return buckets[seal]
}
}
b.Fatal("could not construct seal bucket fixture")
return nil
}
Loading
Loading