tests: Separate request type from Txn operation type
Signed-off-by: Marek Siarkowicz <siarkowicz@google.com>dependabot/go_modules/go.uber.org/atomic-1.10.0
parent
ee566c492b
commit
844ac9c76d
|
@ -78,7 +78,7 @@ func (c *recordingClient) Delete(ctx context.Context, key string) error {
|
|||
return nil
|
||||
}
|
||||
|
||||
func (c *recordingClient) Txn(ctx context.Context, key, expectedValue, newValue string) error {
|
||||
func (c *recordingClient) CompareAndSet(ctx context.Context, key, expectedValue, newValue string) error {
|
||||
callTime := time.Now()
|
||||
txn := c.client.Txn(ctx)
|
||||
var cmp clientv3.Cmp
|
||||
|
|
|
@ -46,13 +46,19 @@ var (
|
|||
minimalQPS: 100,
|
||||
maximalQPS: 200,
|
||||
clientCount: 8,
|
||||
traffic: readWriteSingleKey{keyCount: 4, leaseTTL: DefaultLeaseTTL, writes: []opChance{{operation: model.Put, chance: 50}, {operation: model.Delete, chance: 10}, {operation: model.PutWithLease, chance: 10}, {operation: model.LeaseRevoke, chance: 10}, {operation: model.Txn, chance: 20}}},
|
||||
traffic: readWriteSingleKey{keyCount: 4, leaseTTL: DefaultLeaseTTL, writes: []requestChance{
|
||||
{operation: Put, chance: 50},
|
||||
{operation: Delete, chance: 10},
|
||||
{operation: PutWithLease, chance: 10},
|
||||
{operation: LeaseRevoke, chance: 10},
|
||||
{operation: CompareAndSet, chance: 20},
|
||||
}},
|
||||
}
|
||||
HighTrafficPut = trafficConfig{
|
||||
minimalQPS: 200,
|
||||
maximalQPS: 1000,
|
||||
clientCount: 12,
|
||||
traffic: readWriteSingleKey{keyCount: 4, leaseTTL: DefaultLeaseTTL, writes: []opChance{{operation: model.Put, chance: 100}}},
|
||||
traffic: readWriteSingleKey{keyCount: 4, leaseTTL: DefaultLeaseTTL, writes: []requestChance{{operation: Put, chance: 100}}},
|
||||
}
|
||||
)
|
||||
|
||||
|
@ -193,13 +199,14 @@ func patchOperationBasedOnWatchEvents(operations []porcupine.Operation, watchEve
|
|||
lastObservedOperation := lastOperationObservedInWatch(operations, persisted)
|
||||
|
||||
for _, op := range operations {
|
||||
request := op.Input.(model.EtcdRequest)
|
||||
resp := op.Output.(model.EtcdResponse)
|
||||
if resp.Err == nil || op.Call > lastObservedOperation.Call {
|
||||
// No need to patch successfully requests and cannot patch requests outside observed window.
|
||||
if resp.Err == nil || op.Call > lastObservedOperation.Call || request.Type != model.Txn {
|
||||
// Cannot patch those requests.
|
||||
newOperations = append(newOperations, op)
|
||||
continue
|
||||
}
|
||||
event, hasUniqueWriteOperation := matchWatchEvent(op, persisted)
|
||||
event := matchWatchEvent(request.Txn, persisted)
|
||||
if event != nil {
|
||||
// Set revision and time based on watchEvent.
|
||||
op.Return = event.Time.UnixNano()
|
||||
|
@ -210,7 +217,7 @@ func patchOperationBasedOnWatchEvents(operations []porcupine.Operation, watchEve
|
|||
newOperations = append(newOperations, op)
|
||||
continue
|
||||
}
|
||||
if hasWriteOperation(op) && !hasUniqueWriteOperation {
|
||||
if hasNonUniqueWriteOperation(request.Txn) && !hasUniqueWriteOperation(request.Txn) {
|
||||
// Leave operation as it is as we cannot match non-unique operations to watch events.
|
||||
newOperations = append(newOperations, op)
|
||||
continue
|
||||
|
@ -224,7 +231,11 @@ func lastOperationObservedInWatch(operations []porcupine.Operation, watchEvents
|
|||
var maxCallTime int64
|
||||
var lastOperation porcupine.Operation
|
||||
for _, op := range operations {
|
||||
event, _ := matchWatchEvent(op, watchEvents)
|
||||
request := op.Input.(model.EtcdRequest)
|
||||
if request.Type != model.Txn {
|
||||
continue
|
||||
}
|
||||
event := matchWatchEvent(request.Txn, watchEvents)
|
||||
if event != nil && op.Call > maxCallTime {
|
||||
maxCallTime = op.Call
|
||||
lastOperation = op
|
||||
|
@ -233,33 +244,35 @@ func lastOperationObservedInWatch(operations []porcupine.Operation, watchEvents
|
|||
return lastOperation
|
||||
}
|
||||
|
||||
func matchWatchEvent(op porcupine.Operation, watchEvents map[model.EtcdOperation]watchEvent) (event *watchEvent, hasUniqueWriteOperation bool) {
|
||||
request := op.Input.(model.EtcdRequest)
|
||||
func matchWatchEvent(request *model.TxnRequest, watchEvents map[model.EtcdOperation]watchEvent) *watchEvent {
|
||||
for _, etcdOp := range request.Ops {
|
||||
if model.IsWrite(etcdOp.Type) && model.IsUnique(etcdOp.Type) {
|
||||
// We expect all put to be unique as they write unique value.
|
||||
hasUniqueWriteOperation = true
|
||||
opType := etcdOp.Type
|
||||
if opType == model.PutWithLease {
|
||||
opType = model.Put
|
||||
}
|
||||
if etcdOp.Type == model.Put {
|
||||
// Remove LeaseID which is not exposed in watch.
|
||||
event, ok := watchEvents[model.EtcdOperation{
|
||||
Type: opType,
|
||||
Type: etcdOp.Type,
|
||||
Key: etcdOp.Key,
|
||||
Value: etcdOp.Value,
|
||||
}]
|
||||
if ok {
|
||||
return &event, hasUniqueWriteOperation
|
||||
return &event
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil, hasUniqueWriteOperation
|
||||
return nil
|
||||
}
|
||||
|
||||
func hasWriteOperation(op porcupine.Operation) bool {
|
||||
request := op.Input.(model.EtcdRequest)
|
||||
func hasNonUniqueWriteOperation(request *model.TxnRequest) bool {
|
||||
for _, etcdOp := range request.Ops {
|
||||
if model.IsWrite(etcdOp.Type) {
|
||||
if etcdOp.Type == model.Put || etcdOp.Type == model.Delete {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func hasUniqueWriteOperation(request *model.TxnRequest) bool {
|
||||
for _, etcdOp := range request.Ops {
|
||||
if etcdOp.Type == model.Put {
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
|
|
@ -194,11 +194,11 @@ func (h *AppendableHistory) appendFailed(request EtcdRequest, start time.Time, e
|
|||
}
|
||||
|
||||
func getRequest(key string) EtcdRequest {
|
||||
return EtcdRequest{Ops: []EtcdOperation{{Type: Get, Key: key}}}
|
||||
return EtcdRequest{Type: Txn, Txn: &TxnRequest{Ops: []EtcdOperation{{Type: Get, Key: key}}}}
|
||||
}
|
||||
|
||||
func getResponse(value string, revision int64) EtcdResponse {
|
||||
return EtcdResponse{OpsResult: []EtcdOperationResult{{Value: value}}, Revision: revision}
|
||||
return EtcdResponse{Txn: &TxnResponse{OpsResult: []EtcdOperationResult{{Value: value}}}, Revision: revision}
|
||||
}
|
||||
|
||||
func failedResponse(err error) EtcdResponse {
|
||||
|
@ -210,23 +210,23 @@ func unknownResponse(revision int64) EtcdResponse {
|
|||
}
|
||||
|
||||
func putRequest(key, value string) EtcdRequest {
|
||||
return EtcdRequest{Ops: []EtcdOperation{{Type: Put, Key: key, Value: value}}}
|
||||
return EtcdRequest{Type: Txn, Txn: &TxnRequest{Ops: []EtcdOperation{{Type: Put, Key: key, Value: value}}}}
|
||||
}
|
||||
|
||||
func putResponse(revision int64) EtcdResponse {
|
||||
return EtcdResponse{OpsResult: []EtcdOperationResult{{}}, Revision: revision}
|
||||
return EtcdResponse{Txn: &TxnResponse{OpsResult: []EtcdOperationResult{{}}}, Revision: revision}
|
||||
}
|
||||
|
||||
func deleteRequest(key string) EtcdRequest {
|
||||
return EtcdRequest{Ops: []EtcdOperation{{Type: Delete, Key: key}}}
|
||||
return EtcdRequest{Type: Txn, Txn: &TxnRequest{Ops: []EtcdOperation{{Type: Delete, Key: key}}}}
|
||||
}
|
||||
|
||||
func deleteResponse(deleted int64, revision int64) EtcdResponse {
|
||||
return EtcdResponse{OpsResult: []EtcdOperationResult{{Deleted: deleted}}, Revision: revision}
|
||||
return EtcdResponse{Txn: &TxnResponse{OpsResult: []EtcdOperationResult{{Deleted: deleted}}}, Revision: revision}
|
||||
}
|
||||
|
||||
func txnRequest(key, expectValue, newValue string) EtcdRequest {
|
||||
return EtcdRequest{Conds: []EtcdCondition{{Key: key, ExpectedValue: expectValue}}, Ops: []EtcdOperation{{Type: Put, Key: key, Value: newValue}}}
|
||||
return EtcdRequest{Type: Txn, Txn: &TxnRequest{Conds: []EtcdCondition{{Key: key, ExpectedValue: expectValue}}, Ops: []EtcdOperation{{Type: Put, Key: key, Value: newValue}}}}
|
||||
}
|
||||
|
||||
func txnResponse(succeeded bool, revision int64) EtcdResponse {
|
||||
|
@ -234,27 +234,27 @@ func txnResponse(succeeded bool, revision int64) EtcdResponse {
|
|||
if succeeded {
|
||||
result = []EtcdOperationResult{{}}
|
||||
}
|
||||
return EtcdResponse{OpsResult: result, TxnResult: !succeeded, Revision: revision}
|
||||
return EtcdResponse{Txn: &TxnResponse{OpsResult: result, TxnResult: !succeeded}, Revision: revision}
|
||||
}
|
||||
|
||||
func putWithLeaseRequest(key, value string, leaseID int64) EtcdRequest {
|
||||
return EtcdRequest{Ops: []EtcdOperation{{Type: PutWithLease, Key: key, Value: value, LeaseID: leaseID}}}
|
||||
return EtcdRequest{Type: Txn, Txn: &TxnRequest{Ops: []EtcdOperation{{Type: Put, Key: key, Value: value, LeaseID: leaseID}}}}
|
||||
}
|
||||
|
||||
func leaseGrantRequest(leaseID int64) EtcdRequest {
|
||||
return EtcdRequest{Ops: []EtcdOperation{{Type: LeaseGrant, LeaseID: leaseID}}}
|
||||
return EtcdRequest{Type: LeaseGrant, LeaseGrant: &LeaseGrantRequest{LeaseID: leaseID}}
|
||||
}
|
||||
|
||||
func leaseGrantResponse(revision int64) EtcdResponse {
|
||||
return EtcdResponse{OpsResult: []EtcdOperationResult{{}}, Revision: revision}
|
||||
return EtcdResponse{LeaseGrant: &LeaseGrantReponse{}, Revision: revision}
|
||||
}
|
||||
|
||||
func leaseRevokeRequest(leaseID int64) EtcdRequest {
|
||||
return EtcdRequest{Ops: []EtcdOperation{{Type: LeaseRevoke, LeaseID: leaseID}}}
|
||||
return EtcdRequest{Type: LeaseRevoke, LeaseRevoke: &LeaseRevokeRequest{LeaseID: leaseID}}
|
||||
}
|
||||
|
||||
func leaseRevokeResponse(revision int64) EtcdResponse {
|
||||
return EtcdResponse{OpsResult: []EtcdOperationResult{{}}, Revision: revision}
|
||||
return EtcdResponse{LeaseRevoke: &LeaseRevokeResponse{}, Revision: revision}
|
||||
}
|
||||
|
||||
type History struct {
|
||||
|
|
|
@ -17,22 +17,17 @@ package model
|
|||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"github.com/anishathalye/porcupine"
|
||||
"reflect"
|
||||
"strings"
|
||||
|
||||
"github.com/anishathalye/porcupine"
|
||||
)
|
||||
|
||||
type OperationType string
|
||||
|
||||
const (
|
||||
Get OperationType = "get"
|
||||
Put OperationType = "put"
|
||||
Delete OperationType = "delete"
|
||||
Txn OperationType = "txn"
|
||||
PutWithLease OperationType = "putWithLease"
|
||||
LeaseGrant OperationType = "leaseGrant"
|
||||
LeaseRevoke OperationType = "leaseRevoke"
|
||||
Get OperationType = "get"
|
||||
Put OperationType = "put"
|
||||
Delete OperationType = "delete"
|
||||
)
|
||||
|
||||
var Etcd = porcupine.Model{
|
||||
|
@ -57,15 +52,22 @@ var Etcd = porcupine.Model{
|
|||
},
|
||||
}
|
||||
|
||||
func IsWrite(t OperationType) bool {
|
||||
return t == Put || t == Delete || t == PutWithLease || t == LeaseRevoke || t == LeaseGrant
|
||||
}
|
||||
type RequestType string
|
||||
|
||||
func IsUnique(t OperationType) bool {
|
||||
return t == Put || t == PutWithLease
|
||||
}
|
||||
const (
|
||||
Txn RequestType = "txn"
|
||||
LeaseGrant RequestType = "leaseGrant"
|
||||
LeaseRevoke RequestType = "leaseRevoke"
|
||||
)
|
||||
|
||||
type EtcdRequest struct {
|
||||
Type RequestType
|
||||
LeaseGrant *LeaseGrantRequest
|
||||
LeaseRevoke *LeaseRevokeRequest
|
||||
Txn *TxnRequest
|
||||
}
|
||||
|
||||
type TxnRequest struct {
|
||||
Conds []EtcdCondition
|
||||
Ops []EtcdOperation
|
||||
}
|
||||
|
@ -82,14 +84,32 @@ type EtcdOperation struct {
|
|||
LeaseID int64
|
||||
}
|
||||
|
||||
type LeaseGrantRequest struct {
|
||||
LeaseID int64
|
||||
}
|
||||
type LeaseRevokeRequest struct {
|
||||
LeaseID int64
|
||||
}
|
||||
|
||||
type EtcdResponse struct {
|
||||
Err error
|
||||
Revision int64
|
||||
ResultUnknown bool
|
||||
TxnResult bool
|
||||
OpsResult []EtcdOperationResult
|
||||
Txn *TxnResponse
|
||||
LeaseGrant *LeaseGrantReponse
|
||||
LeaseRevoke *LeaseRevokeResponse
|
||||
}
|
||||
|
||||
type TxnResponse struct {
|
||||
TxnResult bool
|
||||
OpsResult []EtcdOperationResult
|
||||
}
|
||||
|
||||
type LeaseGrantReponse struct {
|
||||
LeaseID int64
|
||||
}
|
||||
type LeaseRevokeResponse struct{}
|
||||
|
||||
func Match(r1, r2 EtcdResponse) bool {
|
||||
return ((r1.ResultUnknown || r2.ResultUnknown) && (r1.Revision == r2.Revision)) || reflect.DeepEqual(r1, r2)
|
||||
}
|
||||
|
@ -115,12 +135,38 @@ type EtcdState struct {
|
|||
}
|
||||
|
||||
func describeEtcdRequestResponse(request EtcdRequest, response EtcdResponse) string {
|
||||
prefix := describeEtcdOperations(request.Ops)
|
||||
if len(request.Conds) != 0 {
|
||||
prefix = fmt.Sprintf("if(%s).then(%s)", describeEtcdConditions(request.Conds), prefix)
|
||||
}
|
||||
return fmt.Sprintf("%s -> %s", describeEtcdRequest(request), describeEtcdResponse(request, response))
|
||||
}
|
||||
|
||||
return fmt.Sprintf("%s -> %s", prefix, describeEtcdResponse(request.Ops, response))
|
||||
func describeEtcdResponse(request EtcdRequest, response EtcdResponse) string {
|
||||
if response.Err != nil {
|
||||
return fmt.Sprintf("err: %q", response.Err)
|
||||
}
|
||||
if response.ResultUnknown {
|
||||
return fmt.Sprintf("unknown, rev: %d", response.Revision)
|
||||
}
|
||||
if request.Type == Txn {
|
||||
return fmt.Sprintf("%s, rev: %d", describeTxnResponse(request.Txn, response.Txn), response.Revision)
|
||||
} else {
|
||||
return fmt.Sprintf("ok, rev: %d", response.Revision)
|
||||
}
|
||||
}
|
||||
|
||||
func describeEtcdRequest(request EtcdRequest) string {
|
||||
switch request.Type {
|
||||
case Txn:
|
||||
describeOperations := describeEtcdOperations(request.Txn.Ops)
|
||||
if len(request.Txn.Conds) != 0 {
|
||||
return fmt.Sprintf("if(%s).then(%s)", describeEtcdConditions(request.Txn.Conds), describeOperations)
|
||||
}
|
||||
return describeOperations
|
||||
case LeaseGrant:
|
||||
return fmt.Sprintf("leaseGrant(%d)", request.LeaseGrant.LeaseID)
|
||||
case LeaseRevoke:
|
||||
return fmt.Sprintf("leaseRevoke(%d)", request.LeaseRevoke.LeaseID)
|
||||
default:
|
||||
return fmt.Sprintf("<! unknown request type: %q !>", request.Type)
|
||||
}
|
||||
}
|
||||
|
||||
func describeEtcdConditions(conds []EtcdCondition) string {
|
||||
|
@ -139,21 +185,14 @@ func describeEtcdOperations(ops []EtcdOperation) string {
|
|||
return strings.Join(opsDescription, ", ")
|
||||
}
|
||||
|
||||
func describeEtcdResponse(ops []EtcdOperation, response EtcdResponse) string {
|
||||
if response.Err != nil {
|
||||
return fmt.Sprintf("err: %q", response.Err)
|
||||
}
|
||||
if response.ResultUnknown {
|
||||
return fmt.Sprintf("unknown, rev: %d", response.Revision)
|
||||
}
|
||||
func describeTxnResponse(request *TxnRequest, response *TxnResponse) string {
|
||||
if response.TxnResult {
|
||||
return fmt.Sprintf("txn failed, rev: %d", response.Revision)
|
||||
return fmt.Sprintf("txn failed")
|
||||
}
|
||||
respDescription := make([]string, len(response.OpsResult))
|
||||
for i := range response.OpsResult {
|
||||
respDescription[i] = describeEtcdOperationResponse(ops[i].Type, response.OpsResult[i])
|
||||
respDescription[i] = describeEtcdOperationResponse(request.Ops[i].Type, response.OpsResult[i])
|
||||
}
|
||||
respDescription = append(respDescription, fmt.Sprintf("rev: %d", response.Revision))
|
||||
return strings.Join(respDescription, ", ")
|
||||
}
|
||||
|
||||
|
@ -162,17 +201,12 @@ func describeEtcdOperation(op EtcdOperation) string {
|
|||
case Get:
|
||||
return fmt.Sprintf("get(%q)", op.Key)
|
||||
case Put:
|
||||
return fmt.Sprintf("put(%q, %q)", op.Key, op.Value)
|
||||
if op.LeaseID != 0 {
|
||||
return fmt.Sprintf("put(%q, %q, %d)", op.Key, op.Value, op.LeaseID)
|
||||
}
|
||||
return fmt.Sprintf("put(%q, %q, nil)", op.Key, op.Value)
|
||||
case Delete:
|
||||
return fmt.Sprintf("delete(%q)", op.Key)
|
||||
case Txn:
|
||||
return "<! unsupported: nested transaction !>"
|
||||
case LeaseGrant:
|
||||
return fmt.Sprintf("leaseGrant(%d)", op.LeaseID)
|
||||
case LeaseRevoke:
|
||||
return fmt.Sprintf("leaseRevoke(%d)", op.LeaseID)
|
||||
case PutWithLease:
|
||||
return fmt.Sprintf("putWithLease(%q, %q, %d)", op.Key, op.Value, op.LeaseID)
|
||||
default:
|
||||
return fmt.Sprintf("<! unknown op: %q !>", op.Type)
|
||||
}
|
||||
|
@ -189,14 +223,6 @@ func describeEtcdOperationResponse(op OperationType, resp EtcdOperationResult) s
|
|||
return fmt.Sprintf("ok")
|
||||
case Delete:
|
||||
return fmt.Sprintf("deleted: %d", resp.Deleted)
|
||||
case Txn:
|
||||
return "<! unsupported: nested transaction !>"
|
||||
case LeaseGrant:
|
||||
return fmt.Sprintf("ok")
|
||||
case LeaseRevoke:
|
||||
return fmt.Sprintf("ok")
|
||||
case PutWithLease:
|
||||
return fmt.Sprintf("ok")
|
||||
default:
|
||||
return fmt.Sprintf("<! unknown op: %q !>", op)
|
||||
}
|
||||
|
@ -226,31 +252,34 @@ func initState(request EtcdRequest, response EtcdResponse) EtcdState {
|
|||
KeyLeases: map[string]int64{},
|
||||
Leases: map[int64]EtcdLease{},
|
||||
}
|
||||
if response.TxnResult {
|
||||
return state
|
||||
}
|
||||
for i, op := range request.Ops {
|
||||
opResp := response.OpsResult[i]
|
||||
switch op.Type {
|
||||
case Get:
|
||||
if opResp.Value != "" {
|
||||
state.KeyValues[op.Key] = opResp.Value
|
||||
}
|
||||
case Put:
|
||||
state.KeyValues[op.Key] = op.Value
|
||||
case Delete:
|
||||
case PutWithLease:
|
||||
//nop here since lease wont be there
|
||||
case LeaseGrant:
|
||||
lease := EtcdLease{
|
||||
LeaseID: op.LeaseID,
|
||||
Keys: map[string]struct{}{},
|
||||
}
|
||||
state.Leases[op.LeaseID] = lease
|
||||
case LeaseRevoke:
|
||||
default:
|
||||
panic("Unknown operation")
|
||||
switch request.Type {
|
||||
case Txn:
|
||||
if response.Txn.TxnResult {
|
||||
return state
|
||||
}
|
||||
for i, op := range request.Txn.Ops {
|
||||
opResp := response.Txn.OpsResult[i]
|
||||
switch op.Type {
|
||||
case Get:
|
||||
if opResp.Value != "" {
|
||||
state.KeyValues[op.Key] = opResp.Value
|
||||
}
|
||||
case Put:
|
||||
state.KeyValues[op.Key] = op.Value
|
||||
case Delete:
|
||||
default:
|
||||
panic("Unknown operation")
|
||||
}
|
||||
}
|
||||
case LeaseGrant:
|
||||
lease := EtcdLease{
|
||||
LeaseID: request.LeaseGrant.LeaseID,
|
||||
Keys: map[string]struct{}{},
|
||||
}
|
||||
state.Leases[request.LeaseGrant.LeaseID] = lease
|
||||
case LeaseRevoke:
|
||||
default:
|
||||
panic(fmt.Sprintf("Unknown request type: %v", request.Type))
|
||||
}
|
||||
return state
|
||||
}
|
||||
|
@ -278,81 +307,84 @@ func applyRequest(states PossibleStates, request EtcdRequest, response EtcdRespo
|
|||
|
||||
// applyRequestToSingleState handles a successful request, returning updated state and response it would generate.
|
||||
func applyRequestToSingleState(s EtcdState, request EtcdRequest) (EtcdState, EtcdResponse) {
|
||||
success := true
|
||||
for _, cond := range request.Conds {
|
||||
if val := s.KeyValues[cond.Key]; val != cond.ExpectedValue {
|
||||
success = false
|
||||
break
|
||||
}
|
||||
}
|
||||
if !success {
|
||||
return s, EtcdResponse{Revision: s.Revision, TxnResult: true}
|
||||
}
|
||||
newKVs := map[string]string{}
|
||||
for k, v := range s.KeyValues {
|
||||
newKVs[k] = v
|
||||
}
|
||||
s.KeyValues = newKVs
|
||||
opResp := make([]EtcdOperationResult, len(request.Ops))
|
||||
increaseRevision := false
|
||||
|
||||
for i, op := range request.Ops {
|
||||
switch op.Type {
|
||||
case Get:
|
||||
opResp[i].Value = s.KeyValues[op.Key]
|
||||
case Put:
|
||||
s.KeyValues[op.Key] = op.Value
|
||||
increaseRevision = true
|
||||
s = detachFromOldLease(s, op.Key)
|
||||
case Delete:
|
||||
if _, ok := s.KeyValues[op.Key]; ok {
|
||||
delete(s.KeyValues, op.Key)
|
||||
increaseRevision = true
|
||||
s = detachFromOldLease(s, op.Key)
|
||||
opResp[i].Deleted = 1
|
||||
switch request.Type {
|
||||
case Txn:
|
||||
success := true
|
||||
for _, cond := range request.Txn.Conds {
|
||||
if val := s.KeyValues[cond.Key]; val != cond.ExpectedValue {
|
||||
success = false
|
||||
break
|
||||
}
|
||||
case PutWithLease:
|
||||
if _, ok := s.Leases[op.LeaseID]; ok {
|
||||
//handle put op.
|
||||
}
|
||||
if !success {
|
||||
return s, EtcdResponse{Revision: s.Revision, Txn: &TxnResponse{TxnResult: true}}
|
||||
}
|
||||
opResp := make([]EtcdOperationResult, len(request.Txn.Ops))
|
||||
increaseRevision := false
|
||||
for i, op := range request.Txn.Ops {
|
||||
switch op.Type {
|
||||
case Get:
|
||||
opResp[i].Value = s.KeyValues[op.Key]
|
||||
case Put:
|
||||
_, leaseExists := s.Leases[op.LeaseID]
|
||||
if op.LeaseID != 0 && !leaseExists {
|
||||
break
|
||||
}
|
||||
s.KeyValues[op.Key] = op.Value
|
||||
increaseRevision = true
|
||||
s = detachFromOldLease(s, op.Key)
|
||||
s = attachToNewLease(s, op.LeaseID, op.Key)
|
||||
}
|
||||
case LeaseRevoke:
|
||||
//Delete the keys attached to the lease
|
||||
keyDeleted := false
|
||||
for key, _ := range s.Leases[op.LeaseID].Keys {
|
||||
//same as delete.
|
||||
if _, ok := s.KeyValues[key]; ok {
|
||||
if !keyDeleted {
|
||||
keyDeleted = true
|
||||
}
|
||||
delete(s.KeyValues, key)
|
||||
delete(s.KeyLeases, key)
|
||||
if leaseExists {
|
||||
s = attachToNewLease(s, op.LeaseID, op.Key)
|
||||
}
|
||||
case Delete:
|
||||
if _, ok := s.KeyValues[op.Key]; ok {
|
||||
delete(s.KeyValues, op.Key)
|
||||
increaseRevision = true
|
||||
s = detachFromOldLease(s, op.Key)
|
||||
opResp[i].Deleted = 1
|
||||
}
|
||||
default:
|
||||
panic("unsupported operation")
|
||||
}
|
||||
//delete the lease
|
||||
delete(s.Leases, op.LeaseID)
|
||||
if keyDeleted {
|
||||
increaseRevision = true
|
||||
}
|
||||
case LeaseGrant:
|
||||
lease := EtcdLease{
|
||||
LeaseID: op.LeaseID,
|
||||
Keys: map[string]struct{}{},
|
||||
}
|
||||
s.Leases[op.LeaseID] = lease
|
||||
default:
|
||||
panic("unsupported operation")
|
||||
}
|
||||
if increaseRevision {
|
||||
s.Revision += 1
|
||||
}
|
||||
return s, EtcdResponse{Txn: &TxnResponse{OpsResult: opResp}, Revision: s.Revision}
|
||||
case LeaseGrant:
|
||||
lease := EtcdLease{
|
||||
LeaseID: request.LeaseGrant.LeaseID,
|
||||
Keys: map[string]struct{}{},
|
||||
}
|
||||
s.Leases[request.LeaseGrant.LeaseID] = lease
|
||||
return s, EtcdResponse{Revision: s.Revision, LeaseGrant: &LeaseGrantReponse{}}
|
||||
case LeaseRevoke:
|
||||
//Delete the keys attached to the lease
|
||||
keyDeleted := false
|
||||
for key, _ := range s.Leases[request.LeaseRevoke.LeaseID].Keys {
|
||||
//same as delete.
|
||||
if _, ok := s.KeyValues[key]; ok {
|
||||
if !keyDeleted {
|
||||
keyDeleted = true
|
||||
}
|
||||
delete(s.KeyValues, key)
|
||||
delete(s.KeyLeases, key)
|
||||
}
|
||||
}
|
||||
//delete the lease
|
||||
delete(s.Leases, request.LeaseRevoke.LeaseID)
|
||||
if keyDeleted {
|
||||
s.Revision += 1
|
||||
}
|
||||
return s, EtcdResponse{Revision: s.Revision, LeaseRevoke: &LeaseRevokeResponse{}}
|
||||
default:
|
||||
panic(fmt.Sprintf("Unknown request type: %v", request.Type))
|
||||
}
|
||||
|
||||
if increaseRevision {
|
||||
s.Revision += 1
|
||||
}
|
||||
|
||||
return s, EtcdResponse{OpsResult: opResp, Revision: s.Revision}
|
||||
}
|
||||
|
||||
func detachFromOldLease(s EtcdState, key string) EtcdState {
|
||||
|
|
|
@ -562,17 +562,22 @@ func TestModelDescribe(t *testing.T) {
|
|||
{
|
||||
req: putRequest("key3", "3"),
|
||||
resp: putResponse(3),
|
||||
expectDescribe: `put("key3", "3") -> ok, rev: 3`,
|
||||
expectDescribe: `put("key3", "3", nil) -> ok, rev: 3`,
|
||||
},
|
||||
{
|
||||
req: putWithLeaseRequest("key3b", "3b", 3),
|
||||
resp: putResponse(3),
|
||||
expectDescribe: `put("key3b", "3b", 3) -> ok, rev: 3`,
|
||||
},
|
||||
{
|
||||
req: putRequest("key4", "4"),
|
||||
resp: failedResponse(errors.New("failed")),
|
||||
expectDescribe: `put("key4", "4") -> err: "failed"`,
|
||||
expectDescribe: `put("key4", "4", nil) -> err: "failed"`,
|
||||
},
|
||||
{
|
||||
req: putRequest("key4b", "4b"),
|
||||
resp: unknownResponse(42),
|
||||
expectDescribe: `put("key4b", "4b") -> unknown, rev: 42`,
|
||||
expectDescribe: `put("key4b", "4b", nil) -> unknown, rev: 42`,
|
||||
},
|
||||
{
|
||||
req: deleteRequest("key5"),
|
||||
|
@ -587,17 +592,17 @@ func TestModelDescribe(t *testing.T) {
|
|||
{
|
||||
req: txnRequest("key7", "7", "77"),
|
||||
resp: txnResponse(false, 7),
|
||||
expectDescribe: `if(key7=="7").then(put("key7", "77")) -> txn failed, rev: 7`,
|
||||
expectDescribe: `if(key7=="7").then(put("key7", "77", nil)) -> txn failed, rev: 7`,
|
||||
},
|
||||
{
|
||||
req: txnRequest("key8", "8", "88"),
|
||||
resp: txnResponse(true, 8),
|
||||
expectDescribe: `if(key8=="8").then(put("key8", "88")) -> ok, rev: 8`,
|
||||
expectDescribe: `if(key8=="8").then(put("key8", "88", nil)) -> ok, rev: 8`,
|
||||
},
|
||||
{
|
||||
req: txnRequest("key9", "9", "99"),
|
||||
resp: failedResponse(errors.New("failed")),
|
||||
expectDescribe: `if(key9=="9").then(put("key9", "99")) -> err: "failed"`,
|
||||
expectDescribe: `if(key9=="9").then(put("key9", "99", nil)) -> err: "failed"`,
|
||||
},
|
||||
}
|
||||
for _, tc := range tcs {
|
||||
|
|
|
@ -24,7 +24,6 @@ import (
|
|||
|
||||
"go.etcd.io/etcd/api/v3/mvccpb"
|
||||
"go.etcd.io/etcd/tests/v3/linearizability/identity"
|
||||
"go.etcd.io/etcd/tests/v3/linearizability/model"
|
||||
)
|
||||
|
||||
var (
|
||||
|
@ -32,18 +31,29 @@ var (
|
|||
RequestTimeout = 40 * time.Millisecond
|
||||
)
|
||||
|
||||
type TrafficRequestType string
|
||||
|
||||
const (
|
||||
Get TrafficRequestType = "get"
|
||||
Put TrafficRequestType = "put"
|
||||
Delete TrafficRequestType = "delete"
|
||||
PutWithLease TrafficRequestType = "putWithLease"
|
||||
LeaseRevoke TrafficRequestType = "leaseRevoke"
|
||||
CompareAndSet TrafficRequestType = "compareAndSet"
|
||||
)
|
||||
|
||||
type Traffic interface {
|
||||
Run(ctx context.Context, clientId int, c *recordingClient, limiter *rate.Limiter, ids identity.Provider, lm identity.LeaseIdStorage)
|
||||
}
|
||||
|
||||
type readWriteSingleKey struct {
|
||||
keyCount int
|
||||
writes []opChance
|
||||
writes []requestChance
|
||||
leaseTTL int64
|
||||
}
|
||||
|
||||
type opChance struct {
|
||||
operation model.OperationType
|
||||
type requestChance struct {
|
||||
operation TrafficRequestType
|
||||
chance int
|
||||
}
|
||||
|
||||
|
@ -80,18 +90,18 @@ func (t readWriteSingleKey) Write(ctx context.Context, c *recordingClient, limit
|
|||
writeCtx, cancel := context.WithTimeout(ctx, RequestTimeout)
|
||||
|
||||
var err error
|
||||
switch t.pickWriteOperation() {
|
||||
case model.Put:
|
||||
switch t.pickWriteRequest() {
|
||||
case Put:
|
||||
err = c.Put(writeCtx, key, newValue)
|
||||
case model.Delete:
|
||||
case Delete:
|
||||
err = c.Delete(writeCtx, key)
|
||||
case model.Txn:
|
||||
case CompareAndSet:
|
||||
var expectValue string
|
||||
if len(lastValues) != 0 {
|
||||
expectValue = string(lastValues[0].Value)
|
||||
}
|
||||
err = c.Txn(writeCtx, key, expectValue, newValue)
|
||||
case model.PutWithLease:
|
||||
err = c.CompareAndSet(writeCtx, key, expectValue, newValue)
|
||||
case PutWithLease:
|
||||
leaseId := lm.LeaseId(cid)
|
||||
if leaseId == 0 {
|
||||
leaseId, err = c.LeaseGrant(writeCtx, t.leaseTTL)
|
||||
|
@ -105,7 +115,7 @@ func (t readWriteSingleKey) Write(ctx context.Context, c *recordingClient, limit
|
|||
err = c.PutWithLease(putCtx, key, newValue, leaseId)
|
||||
putCancel()
|
||||
}
|
||||
case model.LeaseRevoke:
|
||||
case LeaseRevoke:
|
||||
leaseId := lm.LeaseId(cid)
|
||||
if leaseId != 0 {
|
||||
err = c.LeaseRevoke(writeCtx, leaseId)
|
||||
|
@ -124,7 +134,7 @@ func (t readWriteSingleKey) Write(ctx context.Context, c *recordingClient, limit
|
|||
return err
|
||||
}
|
||||
|
||||
func (t readWriteSingleKey) pickWriteOperation() model.OperationType {
|
||||
func (t readWriteSingleKey) pickWriteRequest() TrafficRequestType {
|
||||
sum := 0
|
||||
for _, op := range t.writes {
|
||||
sum += op.chance
|
||||
|
|
Loading…
Reference in New Issue