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
|
/*
* Licensed to Elasticsearch under one or more contributor
* license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright
* ownership. Elasticsearch licenses this file to you 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 org.elasticsearch.cluster.routing.allocation;
import org.elasticsearch.Version;
import org.elasticsearch.cluster.ClusterName;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.EmptyClusterInfoService;
import org.elasticsearch.cluster.health.ClusterHealthStatus;
import org.elasticsearch.cluster.health.ClusterStateHealth;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.cluster.metadata.MetaData;
import org.elasticsearch.cluster.node.DiscoveryNodes;
import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
import org.elasticsearch.cluster.routing.RoutingNode;
import org.elasticsearch.cluster.routing.RoutingTable;
import org.elasticsearch.cluster.routing.ShardRouting;
import org.elasticsearch.cluster.routing.allocation.allocator.BalancedShardsAllocator;
import org.elasticsearch.cluster.routing.allocation.decider.AllocationDecider;
import org.elasticsearch.cluster.routing.allocation.decider.AllocationDeciders;
import org.elasticsearch.cluster.routing.allocation.decider.Decision;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.env.Environment;
import org.elasticsearch.test.ESAllocationTestCase;
import org.elasticsearch.test.gateway.NoopGatewayAllocator;
import java.io.IOException;
import java.util.Collections;
import java.util.Set;
import static org.hamcrest.Matchers.equalTo;
/**
* This class of tests exercise various scenarios of
* primary shard allocation and assert the cluster health
* has the correct status based on those allocation decisions.
*/
public class DecisionsImpactOnClusterHealthTests extends ESAllocationTestCase {
public void testPrimaryShardNoDecisionOnIndexCreation() throws IOException {
final String indexName = "test-idx";
Settings settings = Settings.builder()
.put(Environment.PATH_HOME_SETTING.getKey(), createTempDir().toAbsolutePath().toString())
.build();
AllocationDecider decider = new TestAllocateDecision(Decision.NO);
// if deciders say NO to allocating a primary shard, then the cluster health should be RED
runAllocationTest(
settings, indexName, Collections.singleton(decider), ClusterHealthStatus.RED
);
}
public void testPrimaryShardThrottleDecisionOnIndexCreation() throws IOException {
final String indexName = "test-idx";
Settings settings = Settings.builder()
.put(Environment.PATH_HOME_SETTING.getKey(), createTempDir().toAbsolutePath().toString())
.build();
AllocationDecider decider = new TestAllocateDecision(Decision.THROTTLE) {
// the only allocation decider that implements this is ShardsLimitAllocationDecider and it always
// returns only YES or NO, never THROTTLE
@Override
public Decision canAllocate(RoutingNode node, RoutingAllocation allocation) {
return randomBoolean() ? Decision.YES : Decision.NO;
}
};
// if deciders THROTTLE allocating a primary shard, stay in YELLOW state
runAllocationTest(
settings, indexName, Collections.singleton(decider), ClusterHealthStatus.YELLOW
);
}
public void testPrimaryShardYesDecisionOnIndexCreation() throws IOException {
final String indexName = "test-idx";
Settings settings = Settings.builder()
.put(Environment.PATH_HOME_SETTING.getKey(), createTempDir().toAbsolutePath().toString())
.build();
AllocationDecider decider = new TestAllocateDecision(Decision.YES) {
@Override
public Decision canAllocate(ShardRouting shardRouting, RoutingNode node, RoutingAllocation allocation) {
if (node.getByShardId(shardRouting.shardId()) == null) {
return Decision.YES;
} else {
return Decision.NO;
}
}
};
// if deciders say YES to allocating primary shards, stay in YELLOW state
ClusterState clusterState = runAllocationTest(
settings, indexName, Collections.singleton(decider), ClusterHealthStatus.YELLOW
);
// make sure primaries are initialized
RoutingTable routingTable = clusterState.routingTable();
for (IndexShardRoutingTable indexShardRoutingTable : routingTable.index(indexName)) {
assertTrue(indexShardRoutingTable.primaryShard().initializing());
}
}
private ClusterState runAllocationTest(final Settings settings,
final String indexName,
final Set<AllocationDecider> allocationDeciders,
final ClusterHealthStatus expectedStatus) throws IOException {
final String clusterName = "test-cluster";
final AllocationService allocationService = newAllocationService(settings, allocationDeciders);
logger.info("Building initial routing table");
final int numShards = randomIntBetween(1, 5);
MetaData metaData = MetaData.builder()
.put(IndexMetaData.builder(indexName)
.settings(settings(Version.CURRENT))
.numberOfShards(numShards)
.numberOfReplicas(1))
.build();
RoutingTable routingTable = RoutingTable.builder()
.addAsNew(metaData.index(indexName))
.build();
ClusterState clusterState = ClusterState.builder(new ClusterName(clusterName))
.metaData(metaData)
.routingTable(routingTable)
.build();
logger.info("--> adding nodes");
// we need at least as many nodes as shards for the THROTTLE case, because
// once a shard has been throttled on a node, that node no longer accepts
// any allocations on it
final DiscoveryNodes.Builder discoveryNodes = DiscoveryNodes.builder();
for (int i = 0; i < numShards; i++) {
discoveryNodes.add(newNode("node" + i));
}
clusterState = ClusterState.builder(clusterState).nodes(discoveryNodes).build();
logger.info("--> do the reroute");
routingTable = allocationService.reroute(clusterState, "reroute").routingTable();
clusterState = ClusterState.builder(clusterState).routingTable(routingTable).build();
logger.info("--> assert cluster health");
ClusterStateHealth health = new ClusterStateHealth(clusterState);
assertThat(health.getStatus(), equalTo(expectedStatus));
return clusterState;
}
private static AllocationService newAllocationService(Settings settings, Set<AllocationDecider> deciders) {
return new AllocationService(settings,
new AllocationDeciders(settings, deciders),
NoopGatewayAllocator.INSTANCE,
new BalancedShardsAllocator(settings),
EmptyClusterInfoService.INSTANCE);
}
}
|
