1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
|
// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
package com.yahoo.documentapi.messagebus.protocol.test.storagepolicy;
import com.yahoo.document.BucketId;
import com.yahoo.document.BucketIdFactory;
import com.yahoo.document.DocumentId;
import com.yahoo.documentapi.messagebus.protocol.DocumentProtocol;
import com.yahoo.documentapi.messagebus.protocol.ContentPolicy;
import com.yahoo.messagebus.routing.RoutingNode;
import com.yahoo.vdslib.distribution.RandomGen;
import com.yahoo.vdslib.state.ClusterState;
import com.yahoo.vdslib.state.Node;
import com.yahoo.vdslib.state.NodeState;
import com.yahoo.vdslib.state.NodeType;
import com.yahoo.vdslib.state.State;
import java.util.Map;
import java.util.TreeMap;
import java.util.regex.Pattern;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
public abstract class Simulator extends ContentPolicyTestEnvironment {
enum FailureType {
TRANSIENT_ERROR,
FATAL_ERROR,
OLD_CLUSTER_STATE,
RESET_CLUSTER_STATE,
RESET_CLUSTER_STATE_NO_GOOD_NODES,
NODE_NOT_IN_SLOBROK
};
private int getIdealTarget(String idString, String clusterState) {
DocumentId did = new DocumentId(idString);
BucketIdFactory factory = new BucketIdFactory();
BucketId bid = factory.getBucketId(did);
try{
return policyFactory.getLastParameters().createDistribution(null).getIdealDistributorNode(new ClusterState(clusterState), bid, "uim");
} catch (Exception e) {
throw new RuntimeException(e);
}
}
public class BadNode {
private int index;
private double failureRate = 1.0;
private FailureType failureType;
private ClusterState badClusterState;
private boolean downInCurrentState = false;
public BadNode(int index, FailureType type) {
this.index = index;
this.failureType = type;
}
public BadNode setFailureRate(double rate) { failureRate = rate; return this; }
public BadNode setBadClusterState(ClusterState state) {
badClusterState = state;
if (debug) System.err.println("Setting bad cluster state in distributor " + index + ": " + state);
return this;
}
public BadNode setDownInCurrentState() { downInCurrentState = true; return this; }
public int getIndex() { return index; }
public FailureType getFailureType() { return failureType; }
public double getFailureRate() { return failureRate; }
public ClusterState getBadClusterState() { return badClusterState; }
public boolean isSetDownInCurrentState() { return downInCurrentState; }
}
public class PersistentFailureTestParameters {
private ClusterState initialClusterState;
private ClusterState currentClusterState;
private int totalRequests = 200;
private int parallellRequests = 10;
private Map<Integer, BadNode> badnodes = new TreeMap<Integer, BadNode>();
private boolean newNode = false;
public PersistentFailureTestParameters() {
try{
initialClusterState = new ClusterState("version:2 bits:16 distributor:9");
currentClusterState = initialClusterState.clone();
currentClusterState.setVersion(3);
} catch (Exception e) { throw new RuntimeException(e); }
}
public PersistentFailureTestParameters setTotalRequests(int count) { totalRequests = count; return this; }
public PersistentFailureTestParameters setParallellRequests(int count) { parallellRequests = count; return this; }
public PersistentFailureTestParameters addBadNode(BadNode node) { badnodes.put(node.getIndex(), node); return this; }
public PersistentFailureTestParameters newNodeAdded() {
currentClusterState.setNodeState(new Node(NodeType.DISTRIBUTOR, 9), new NodeState(NodeType.DISTRIBUTOR, State.UP));
newNode = true;
return this;
}
public void validate() {
// To simplify looping, ensure we do complete loops
assertTrue(totalRequests % parallellRequests == 0);
// Node that in some tests are used as new node up, cannot be a bad node
assertTrue(!badnodes.containsKey(9));
for (BadNode node : badnodes.values()) {
ClusterState badClusterState = currentClusterState.clone();
int nodesToSetDown = 0;
if (node.getFailureType() == FailureType.OLD_CLUSTER_STATE) {
badClusterState.setVersion(1);
nodesToSetDown = 4;
} else if (node.getFailureType() == FailureType.RESET_CLUSTER_STATE) {
badClusterState.setVersion(5);
nodesToSetDown = 4;
} else if (node.getFailureType() == FailureType.RESET_CLUSTER_STATE_NO_GOOD_NODES) {
badClusterState.setVersion(5);
nodesToSetDown = -1;
} else {
badClusterState = null;
}
if (badClusterState != null) {
int setDown = 0;
for (int i=0; i<10; ++i) {
if (nodesToSetDown != -1 && setDown >= nodesToSetDown) break;
if (badnodes.containsKey(i)) continue;
badClusterState.setNodeState(new Node(NodeType.DISTRIBUTOR, i), new NodeState(NodeType.DISTRIBUTOR, State.DOWN));
++setDown;
}
if (nodesToSetDown > 0 && nodesToSetDown != setDown) throw new IllegalStateException("Failed to set down " + nodesToSetDown + " nodes");
node.setBadClusterState(badClusterState);
}
if (node.isSetDownInCurrentState()) {
currentClusterState.setNodeState(new Node(NodeType.DISTRIBUTOR, node.getIndex()), new NodeState(NodeType.DISTRIBUTOR, State.DOWN));
}
}
if (debug) System.err.println("Using initial state " + initialClusterState);
if (debug) System.err.println("Using current state " + currentClusterState);
}
public int getTotalRequests() { return totalRequests; }
public int getParallellRequests() { return parallellRequests; }
public Map<Integer, BadNode> getBadNodes() { return badnodes; }
public boolean isNewNodeAdded() { return newNode; }
public ClusterState getInitialClusterState() { return initialClusterState; }
public ClusterState getCurrentClusterState(Integer distributor) {
if (distributor != null && badnodes.containsKey(distributor) && badnodes.get(distributor).getBadClusterState() != null) {
return badnodes.get(distributor).getBadClusterState();
}
return currentClusterState;
}
}
public void runSimulation(String expected, PersistentFailureTestParameters params) {
params.validate();
// Set nodes in slobrok
setClusterNodes(new int[]{ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9});
for (BadNode node : params.getBadNodes().values()) {
if (node.getFailureType() == FailureType.NODE_NOT_IN_SLOBROK) removeNode(node.getIndex());
}
{
RoutingNode target = select();
replyWrongDistribution(target, "foo", null, params.getInitialClusterState().toString());
}
RandomGen randomizer = new RandomGen(432121);
int[] correctnode = new int[2],
wrongnode = new int[2],
failed = new int[2],
worked = new int[2],
downnode = new int[2];
for (int step = 0, steps = (params.getTotalRequests() / params.getParallellRequests()); step < steps; ++step) {
int half = (step < steps / 2 ? 0 : 1);
if (debug) System.err.println("Starting step " + step + " in half " + half);
String[] docId = new String[params.getParallellRequests()];
RoutingNode[] targets = new RoutingNode[params.getParallellRequests()];
for (int i=0; i<params.getParallellRequests(); ++i) {
docId[i] = "id:ns:testdoc::" + (step * params.getParallellRequests() + i);
frame.setMessage(createMessage(docId[i]));
targets[i] = select();
}
for (int i=0; i<params.getParallellRequests(); ++i) {
RoutingNode target = targets[i];
int index = getAddress(target).getSecond();
if (!params.getCurrentClusterState(null).getNodeState(new Node(NodeType.DISTRIBUTOR, index)).getState().oneOf(ContentPolicy.owningBucketStates)) {
++downnode[half];
}
BadNode badNode = params.getBadNodes().get(index);
if (getAddress(target).getSecond() == getIdealTarget(docId[i], params.getCurrentClusterState(null).toString())) {
++correctnode[half];
} else {
++wrongnode[half];
}
if (badNode != null && randomizer.nextDouble() < badNode.getFailureRate()) {
++failed[half];
switch (badNode.getFailureType()) {
case TRANSIENT_ERROR: replyError(target, new com.yahoo.messagebus.Error(DocumentProtocol.ERROR_ABORTED, "Transient error")); break;
case FATAL_ERROR: replyError(target, new com.yahoo.messagebus.Error(DocumentProtocol.ERROR_UNPARSEABLE, "Fatal error")); break;
case OLD_CLUSTER_STATE:
case RESET_CLUSTER_STATE:
case RESET_CLUSTER_STATE_NO_GOOD_NODES: replyWrongDistribution(target, "foo", null, params.getCurrentClusterState(index).toString()); break;
case NODE_NOT_IN_SLOBROK: throw new IllegalStateException("This point in code should not be reachable");
}
} else {
++worked[half];
boolean correctTarget = (getAddress(target).getSecond() == getIdealTarget(docId[i], params.getCurrentClusterState(index).toString()));
if (correctTarget) {
replyOk(target);
} else {
replyWrongDistribution(target, "foo", null, params.getCurrentClusterState(index).toString());
}
}
}
}
StringBuilder actual = new StringBuilder();
for (int i=0; i<2; ++i) {
actual.append(i == 0 ? "First " : " Last ")
.append("correctnode ").append(correctnode[i])
.append(", wrongnode ").append(wrongnode[i])
.append(", downnode ").append(downnode[i])
.append(", worked ").append(worked[i])
.append(", failed ").append(failed[i]);
}
if ( ! Pattern.matches(expected, actual.toString())) {
assertEquals(expected, actual.toString());
}
}
}
|
