Mirroristas/etcd
2
0
Fork 0
mirror of https://github.com/etcd-io/etcd.git synced 2024-09-27 06:25:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #16129 from serathius/robustness-fix-patch-txn-onfailure

Browse Source 
Robustness fix patch txn onfailure
This commit is contained in:
Marek Siarkowicz 2023-06-24 17:32:31 +02:00 committed by GitHub
commit 31b20ef40f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 389 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
tests/robustness/validate/patch_history.go
View File

@ -21,7 +21,34 @@ import (
"go.etcd.io/etcd/tests/v3/robustness/traffic"
)
func patchedOperationHistory(reports []traffic.ClientReport) []porcupine.Operation {
allOperations := operations(reports)
uniqueEvents := uniqueWatchEvents(reports)
return patchOperationsWithWatchEvents(allOperations, uniqueEvents)
}
func operations(reports []traffic.ClientReport) []porcupine.Operation {
var ops []porcupine.Operation
for _, r := range reports {
ops = append(ops, r.OperationHistory.Operations()...)
}
return ops
}
func uniqueWatchEvents(reports []traffic.ClientReport) map[model.Event]traffic.TimedWatchEvent {
persisted := map[model.Event]traffic.TimedWatchEvent{}
for _, r := range reports {
for _, resp := range r.Watch {
for _, event := range resp.Events {
persisted[event.Event] = traffic.TimedWatchEvent{Time: resp.Time, WatchEvent: event}
}
}
}
return persisted
}
func patchOperationsWithWatchEvents(operations []porcupine.Operation, watchEvents map[model.Event]traffic.TimedWatchEvent) []porcupine.Operation {
newOperations := make([]porcupine.Operation, 0, len(operations))
lastObservedOperation := lastOperationObservedInWatch(operations, watchEvents)
@ -41,8 +68,8 @@ func patchOperationsWithWatchEvents(operations []porcupine.Operation, watchEvent
newOperations = append(newOperations, op)
continue
}
if hasNonUniqueWriteOperation(request.Txn) && !hasUniqueWriteOperation(request.Txn) {
// Leave operation as it is as we cannot match non-unique operations to watch events.
if !canBeDiscarded(request.Txn) {
// Leave operation as it is as we cannot discard it.
newOperations = append(newOperations, op)
continue
}
@ -84,8 +111,16 @@ func matchWatchEvent(request *model.TxnRequest, watchEvents map[model.Event]traf
return nil
}
func hasNonUniqueWriteOperation(request *model.TxnRequest) bool {
for _, etcdOp := range request.OperationsOnSuccess {
func canBeDiscarded(request *model.TxnRequest) bool {
return operationsCanBeDiscarded(request.OperationsOnSuccess) && operationsCanBeDiscarded(request.OperationsOnFailure)
}
func operationsCanBeDiscarded(ops []model.EtcdOperation) bool {
return hasUniqueWriteOperation(ops) || !hasWriteOperation(ops)
}
func hasWriteOperation(ops []model.EtcdOperation) bool {
for _, etcdOp := range ops {
if etcdOp.Type == model.PutOperation || etcdOp.Type == model.DeleteOperation {
return true
}
@ -93,8 +128,8 @@ func hasNonUniqueWriteOperation(request *model.TxnRequest) bool {
return false
}
func hasUniqueWriteOperation(request *model.TxnRequest) bool {
for _, etcdOp := range request.OperationsOnSuccess {
func hasUniqueWriteOperation(ops []model.EtcdOperation) bool {
for _, etcdOp := range ops {
if etcdOp.Type == model.PutOperation {
return true
}

347
tests/robustness/validate/patch_history_test.go Normal file
View File

@ -0,0 +1,347 @@
// Copyright 2023 The etcd Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package validate
import (
"errors"
"testing"
"time"
"go.etcd.io/etcd/api/v3/etcdserverpb"
clientv3 "go.etcd.io/etcd/client/v3"
"go.etcd.io/etcd/tests/v3/robustness/identity"
"go.etcd.io/etcd/tests/v3/robustness/model"
"go.etcd.io/etcd/tests/v3/robustness/traffic"
)
func TestPatchHistory(t *testing.T) {
for _, tc := range []struct {
name string
historyFunc func(baseTime time.Time, h *model.AppendableHistory)
event model.Event
expectRemains bool
}{
{
name: "successful range remains",
historyFunc: func(baseTime time.Time, h *model.AppendableHistory) {
start := time.Since(baseTime)
time.Sleep(time.Nanosecond)
stop := time.Since(baseTime)
h.AppendRange("key", "", 0, 0, start, stop, &clientv3.GetResponse{})
},
expectRemains: true,
},
{
name: "successful put remains",
historyFunc: func(baseTime time.Time, h *model.AppendableHistory) {
start := time.Since(baseTime)
time.Sleep(time.Nanosecond)
stop := time.Since(baseTime)
h.AppendPut("key", "value", start, stop, &clientv3.PutResponse{}, nil)
},
expectRemains: true,
},
{
name: "failed put remains if there is a matching event",
historyFunc: func(baseTime time.Time, h *model.AppendableHistory) {
start := time.Since(baseTime)
time.Sleep(time.Nanosecond)
stop := time.Since(baseTime)
h.AppendPut("key", "value", start, stop, nil, errors.New("failed"))
},
event: model.Event{
Type: model.PutOperation,
Key: "key",
Value: model.ToValueOrHash("value"),
},
expectRemains: true,
},
{
name: "failed put is dropped if event has different key",
historyFunc: func(baseTime time.Time, h *model.AppendableHistory) {
start := time.Since(baseTime)
time.Sleep(time.Nanosecond)
stop := time.Since(baseTime)
h.AppendPut("key1", "value", start, stop, nil, errors.New("failed"))
},
event: model.Event{
Type: model.PutOperation,
Key: "key2",
Value: model.ToValueOrHash("value"),
},
expectRemains: false,
},
{
name: "failed put is dropped if event has different value",
historyFunc: func(baseTime time.Time, h *model.AppendableHistory) {
start := time.Since(baseTime)
time.Sleep(time.Nanosecond)
stop := time.Since(baseTime)
h.AppendPut("key", "value1", start, stop, nil, errors.New("failed"))
},
event: model.Event{
Type: model.PutOperation,
Key: "key",
Value: model.ToValueOrHash("value2"),
},
expectRemains: false,
},
{
name: "failed put with lease remains if there is a matching event",
historyFunc: func(baseTime time.Time, h *model.AppendableHistory) {
start := time.Since(baseTime)
time.Sleep(time.Nanosecond)
stop := time.Since(baseTime)
h.AppendPutWithLease("key", "value", 123, start, stop, nil, errors.New("failed"))
},
event: model.Event{
Type: model.PutOperation,
Key: "key",
Value: model.ToValueOrHash("value"),
},
expectRemains: true,
},
{
name: "failed put is dropped",
historyFunc: func(baseTime time.Time, h *model.AppendableHistory) {
start := time.Since(baseTime)
time.Sleep(time.Nanosecond)
stop := time.Since(baseTime)
h.AppendPut("key", "value", start, stop, nil, errors.New("failed"))
},
expectRemains: false,
},
{
name: "failed put with lease is dropped",
historyFunc: func(baseTime time.Time, h *model.AppendableHistory) {
start := time.Since(baseTime)
time.Sleep(time.Nanosecond)
stop := time.Since(baseTime)
h.AppendPutWithLease("key", "value", 123, start, stop, nil, errors.New("failed"))
},
expectRemains: false,
},
{
name: "successful delete remains",
historyFunc: func(baseTime time.Time, h *model.AppendableHistory) {
start := time.Since(baseTime)
time.Sleep(time.Nanosecond)
stop := time.Since(baseTime)
h.AppendDelete("key", start, stop, &clientv3.DeleteResponse{}, nil)
},
expectRemains: true,
},
{
name: "failed delete remains",
historyFunc: func(baseTime time.Time, h *model.AppendableHistory) {
start := time.Since(baseTime)
time.Sleep(time.Nanosecond)
stop := time.Since(baseTime)
h.AppendDelete("key", start, stop, nil, errors.New("failed"))
},
expectRemains: true,
},
{
name: "successful empty txn remains",
historyFunc: func(baseTime time.Time, h *model.AppendableHistory) {
start := time.Since(baseTime)
time.Sleep(time.Nanosecond)
stop := time.Since(baseTime)
h.AppendTxn(nil, []clientv3.Op{}, []clientv3.Op{}, start, stop, &clientv3.TxnResponse{}, nil)
},
expectRemains: true,
},
{
name: "failed empty txn is dropped",
historyFunc: func(baseTime time.Time, h *model.AppendableHistory) {
start := time.Since(baseTime)
time.Sleep(time.Nanosecond)
stop := time.Since(baseTime)
h.AppendTxn(nil, []clientv3.Op{}, []clientv3.Op{}, start, stop, nil, errors.New("failed"))
},
expectRemains: false,
},
{
name: "failed txn put is dropped",
historyFunc: func(baseTime time.Time, h *model.AppendableHistory) {
start := time.Since(baseTime)
time.Sleep(time.Nanosecond)
stop := time.Since(baseTime)
h.AppendTxn(nil, []clientv3.Op{clientv3.OpPut("key", "value")}, []clientv3.Op{}, start, stop, nil, errors.New("failed"))
},
expectRemains: false,
},
{
name: "failed txn put remains if there is a matching event",
historyFunc: func(baseTime time.Time, h *model.AppendableHistory) {
start := time.Since(baseTime)
time.Sleep(time.Nanosecond)
stop := time.Since(baseTime)
h.AppendTxn(nil, []clientv3.Op{clientv3.OpPut("key", "value")}, []clientv3.Op{}, start, stop, nil, errors.New("failed"))
},
event: model.Event{
Type: model.PutOperation,
Key: "key",
Value: model.ToValueOrHash("value"),
},
expectRemains: true,
},
{
name: "failed txn delete remains",
historyFunc: func(baseTime time.Time, h *model.AppendableHistory) {
start := time.Since(baseTime)
time.Sleep(time.Nanosecond)
stop := time.Since(baseTime)
h.AppendTxn(nil, []clientv3.Op{clientv3.OpDelete("key")}, []clientv3.Op{}, start, stop, nil, errors.New("failed"))
},
expectRemains: true,
},
{
name: "successful txn put/delete remains",
historyFunc: func(baseTime time.Time, h *model.AppendableHistory) {
start := time.Since(baseTime)
time.Sleep(time.Nanosecond)
stop := time.Since(baseTime)
h.AppendTxn(nil, []clientv3.Op{clientv3.OpPut("key", "value")}, []clientv3.Op{clientv3.OpDelete("key")}, start, stop, &clientv3.TxnResponse{}, nil)
},
expectRemains: true,
},
{
name: "failed txn put/delete remains",
historyFunc: func(baseTime time.Time, h *model.AppendableHistory) {
start := time.Since(baseTime)
time.Sleep(time.Nanosecond)
stop := time.Since(baseTime)
h.AppendTxn(nil, []clientv3.Op{clientv3.OpPut("key", "value")}, []clientv3.Op{clientv3.OpDelete("key")}, start, stop, nil, errors.New("failed"))
},
expectRemains: true,
},
{
name: "failed txn delete/put remains",
historyFunc: func(baseTime time.Time, h *model.AppendableHistory) {
start := time.Since(baseTime)
time.Sleep(time.Nanosecond)
stop := time.Since(baseTime)
h.AppendTxn(nil, []clientv3.Op{clientv3.OpDelete("key")}, []clientv3.Op{clientv3.OpPut("key", "value")}, start, stop, nil, errors.New("failed"))
},
expectRemains: true,
},
{
name: "failed txn empty/put is dropped",
historyFunc: func(baseTime time.Time, h *model.AppendableHistory) {
start := time.Since(baseTime)
time.Sleep(time.Nanosecond)
stop := time.Since(baseTime)
h.AppendTxn(nil, []clientv3.Op{}, []clientv3.Op{clientv3.OpPut("key", "value")}, start, stop, nil, errors.New("failed"))
},
expectRemains: false,
},
{
name: "failed txn empty/put remains if there is a matching event",
historyFunc: func(baseTime time.Time, h *model.AppendableHistory) {
start := time.Since(baseTime)
time.Sleep(time.Nanosecond)
stop := time.Since(baseTime)
h.AppendTxn(nil, []clientv3.Op{clientv3.OpPut("key", "value")}, []clientv3.Op{}, start, stop, nil, errors.New("failed"))
},
event: model.Event{
Type: model.PutOperation,
Key: "key",
Value: model.ToValueOrHash("value"),
},
expectRemains: true,
},
{
name: "failed txn empty/delete remains",
historyFunc: func(baseTime time.Time, h *model.AppendableHistory) {
start := time.Since(baseTime)
time.Sleep(time.Nanosecond)
stop := time.Since(baseTime)
h.AppendTxn(nil, []clientv3.Op{}, []clientv3.Op{clientv3.OpDelete("key")}, start, stop, nil, errors.New("failed"))
},
expectRemains: true,
},
{
name: "failed txn put&delete is dropped",
historyFunc: func(baseTime time.Time, h *model.AppendableHistory) {
start := time.Since(baseTime)
time.Sleep(time.Nanosecond)
stop := time.Since(baseTime)
h.AppendTxn(nil, []clientv3.Op{clientv3.OpPut("key", "value1"), clientv3.OpDelete("key")}, []clientv3.Op{}, start, stop, nil, errors.New("failed"))
},
expectRemains: false,
},
{
name: "failed txn empty/put&delete is dropped",
historyFunc: func(baseTime time.Time, h *model.AppendableHistory) {
start := time.Since(baseTime)
time.Sleep(time.Nanosecond)
stop := time.Since(baseTime)
h.AppendTxn(nil, []clientv3.Op{}, []clientv3.Op{clientv3.OpPut("key", "value1"), clientv3.OpDelete("key")}, start, stop, nil, errors.New("failed"))
},
expectRemains: false,
},
{
name: "failed txn put&delete/put&delete is dropped",
historyFunc: func(baseTime time.Time, h *model.AppendableHistory) {
start := time.Since(baseTime)
time.Sleep(time.Nanosecond)
stop := time.Since(baseTime)
h.AppendTxn(nil, []clientv3.Op{clientv3.OpPut("key", "value1"), clientv3.OpDelete("key")}, []clientv3.Op{clientv3.OpPut("key", "value2"), clientv3.OpDelete("key")}, start, stop, nil, errors.New("failed"))
},
expectRemains: false,
},
} {
t.Run(tc.name, func(t *testing.T) {
baseTime := time.Now()
history := model.NewAppendableHistory(identity.NewIdProvider())
tc.historyFunc(baseTime, history)
time.Sleep(time.Nanosecond)
start := time.Since(baseTime)
time.Sleep(time.Nanosecond)
stop := time.Since(baseTime)
history.AppendPut("tombstone", "true", start, stop, &clientv3.PutResponse{Header: &etcdserverpb.ResponseHeader{Revision: 3}}, nil)
watch := []traffic.WatchResponse{
{
Events: []model.WatchEvent{{Event: tc.event, Revision: 2}},
Revision: 2,
Time: time.Since(baseTime),
},
{
Events: []model.WatchEvent{
{Event: model.Event{
Type: model.PutOperation,
Key: "tombstone",
Value: model.ToValueOrHash("true"),
}, Revision: 3},
},
Revision: 3,
Time: time.Since(baseTime),
},
}
operations := patchedOperationHistory([]traffic.ClientReport{
{
ClientId: 0,
OperationHistory: history.History,
Watch: watch,
},
})
remains := len(operations) == history.Len()
if remains != tc.expectRemains {
t.Errorf("Unexpected remains, got: %v, want: %v", remains, tc.expectRemains)
}
})
}
}

26
tests/robustness/validate/validate.go
View File

@ -17,42 +17,18 @@ package validate
import (
"testing"
"github.com/anishathalye/porcupine"
"go.uber.org/zap"
"go.etcd.io/etcd/tests/v3/robustness/model"
"go.etcd.io/etcd/tests/v3/robustness/traffic"
)
// ValidateAndReturnVisualize returns visualize as porcupine.linearizationInfo used to generate visualization is private.
func ValidateAndReturnVisualize(t *testing.T, lg *zap.Logger, cfg Config, reports []traffic.ClientReport) (visualize func(basepath string) error) {
eventHistory := validateWatch(t, cfg, reports)
allOperations := operations(reports)
watchEvents := uniqueWatchEvents(reports)
patchedOperations := patchOperationsWithWatchEvents(allOperations, watchEvents)
patchedOperations := patchedOperationHistory(reports)
return validateOperationsAndVisualize(t, lg, patchedOperations, eventHistory)
}
func operations(reports []traffic.ClientReport) []porcupine.Operation {
var ops []porcupine.Operation
for _, r := range reports {
ops = append(ops, r.OperationHistory.Operations()...)
}
return ops
}
func uniqueWatchEvents(reports []traffic.ClientReport) map[model.Event]traffic.TimedWatchEvent {
persisted := map[model.Event]traffic.TimedWatchEvent{}
for _, r := range reports {
for _, resp := range r.Watch {
for _, event := range resp.Events {
persisted[event.Event] = traffic.TimedWatchEvent{Time: resp.Time, WatchEvent: event}
}
}
}
return persisted
}
type Config struct {
ExpectRevisionUnique bool
}