Skip to content

Commit a76824e

Browse files
kimchykimchy
committed
make index not recovered a bock used in both gateways (shared/blob and local)
1 parent 3f8c03d commit a76824e

File tree

5 files changed

+46
-44
lines changed

5 files changed

+46
-44
lines changed

‎modules/elasticsearch/src/main/java/org/elasticsearch/gateway/GatewayService.java

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,8 @@
2727
import org.elasticsearch.cluster.block.ClusterBlocks;
2828
import org.elasticsearch.cluster.metadata.MetaData;
2929
import org.elasticsearch.cluster.node.DiscoveryNodes;
30+
import org.elasticsearch.cluster.routing.IndexRoutingTable;
31+
import org.elasticsearch.common.collect.ImmutableSet;
3032
import org.elasticsearch.common.component.AbstractLifecycleComponent;
3133
import org.elasticsearch.common.inject.Inject;
3234
import org.elasticsearch.common.settings.Settings;
@@ -35,6 +37,7 @@
3537
import org.elasticsearch.threadpool.ThreadPool;
3638

3739
import javax.annotation.Nullable;
40+
import java.util.Map;
3841
import java.util.concurrent.CountDownLatch;
3942
import java.util.concurrent.TimeUnit;
4043
import java.util.concurrent.atomic.AtomicBoolean;
@@ -48,6 +51,7 @@
4851
public class GatewayService extends AbstractLifecycleComponent<GatewayService> implements ClusterStateListener {
4952

5053
public static final ClusterBlock NOT_RECOVERED_FROM_GATEWAY_BLOCK = new ClusterBlock(1, "not recovered from gateway", ClusterBlockLevel.ALL);
54+
public static final ClusterBlock INDEX_NOT_RECOVERED_BLOCK = new ClusterBlock(3, "index not recovered", ClusterBlockLevel.READ_WRITE);
5155

5256
private final Gateway gateway;
5357

@@ -165,6 +169,23 @@ public class GatewayService extends AbstractLifecycleComponent<GatewayService> i
165169
}
166170
});
167171
}
172+
} else {
173+
for (Map.Entry<String, ImmutableSet<ClusterBlock>> entry : event.state().blocks().indices().entrySet()) {
174+
final String index = entry.getKey();
175+
ImmutableSet<ClusterBlock> indexBlocks = entry.getValue();
176+
if (indexBlocks.contains(GatewayService.INDEX_NOT_RECOVERED_BLOCK)) {
177+
IndexRoutingTable indexRoutingTable = event.state().routingTable().index(index);
178+
if (indexRoutingTable != null && indexRoutingTable.allPrimaryShardsActive()) {
179+
clusterService.submitStateUpdateTask("remove-index-block (all primary shards active for [" + index + "])", new ClusterStateUpdateTask() {
180+
@Override public ClusterState execute(ClusterState currentState) {
181+
ClusterBlocks.Builder blocks = ClusterBlocks.builder().blocks(currentState.blocks());
182+
blocks.removeIndexBlock(index, GatewayService.INDEX_NOT_RECOVERED_BLOCK);
183+
return ClusterState.builder().state(currentState).blocks(blocks).build();
184+
}
185+
});
186+
}
187+
}
188+
}
168189
}
169190
}
170191
}

‎modules/elasticsearch/src/main/java/org/elasticsearch/gateway/local/LocalGateway.java

Lines changed: 3 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -22,9 +22,6 @@
2222
import org.elasticsearch.ElasticSearchException;
2323
import org.elasticsearch.action.FailedNodeException;
2424
import org.elasticsearch.cluster.*;
25-
import org.elasticsearch.cluster.block.ClusterBlock;
26-
import org.elasticsearch.cluster.block.ClusterBlockLevel;
27-
import org.elasticsearch.cluster.block.ClusterBlocks;
2825
import org.elasticsearch.cluster.metadata.IndexMetaData;
2926
import org.elasticsearch.cluster.metadata.MetaData;
3027
import org.elasticsearch.cluster.metadata.MetaDataCreateIndexService;
@@ -44,10 +41,10 @@
4441
import org.elasticsearch.env.NodeEnvironment;
4542
import org.elasticsearch.gateway.Gateway;
4643
import org.elasticsearch.gateway.GatewayException;
44+
import org.elasticsearch.gateway.GatewayService;
4745
import org.elasticsearch.index.gateway.local.LocalIndexGatewayModule;
4846

4947
import java.io.*;
50-
import java.util.Map;
5148
import java.util.Set;
5249
import java.util.concurrent.ExecutorService;
5350
import java.util.concurrent.TimeUnit;
@@ -64,8 +61,6 @@
6461
*/
6562
public class LocalGateway extends AbstractLifecycleComponent<Gateway> implements Gateway, ClusterStateListener {
6663

67-
public static final ClusterBlock INDEX_NOT_RECOVERED_BLOCK = new ClusterBlock(3, "index not recovered (not enough nodes with shards allocated found)", ClusterBlockLevel.READ_WRITE);
68-
6964
private File location;
7065

7166
private final ClusterService clusterService;
@@ -185,7 +180,7 @@ public LocalGatewayStartedShards currentStartedShards() {
185180
createIndexService.createIndex(new MetaDataCreateIndexService.Request("gateway", indexMetaData.index())
186181
.settings(indexMetaData.settings())
187182
.mappingsCompressed(indexMetaData.mappings())
188-
.blocks(ImmutableSet.of(INDEX_NOT_RECOVERED_BLOCK))
183+
.blocks(ImmutableSet.of(GatewayService.INDEX_NOT_RECOVERED_BLOCK))
189184
.timeout(timeValueSeconds(30)),
190185

191186
new MetaDataCreateIndexService.Listener() {
@@ -216,7 +211,7 @@ public LocalGatewayStartedShards currentStartedShards() {
216211
}
217212

218213
@Override public void clusterChanged(final ClusterChangedEvent event) {
219-
// nothing to do until we actually recover from hte gateway
214+
// nothing to do until we actually recover from the gateway
220215
if (!event.state().metaData().recoveredFromGateway()) {
221216
return;
222217
}
@@ -226,26 +221,6 @@ public LocalGatewayStartedShards currentStartedShards() {
226221
return;
227222
}
228223

229-
// go over the indices, if they are blocked, and all are allocated, update the cluster state that it is no longer blocked
230-
if (event.state().nodes().localNodeMaster()) {
231-
for (Map.Entry<String, ImmutableSet<ClusterBlock>> entry : event.state().blocks().indices().entrySet()) {
232-
final String index = entry.getKey();
233-
ImmutableSet<ClusterBlock> indexBlocks = entry.getValue();
234-
if (indexBlocks.contains(INDEX_NOT_RECOVERED_BLOCK)) {
235-
IndexRoutingTable indexRoutingTable = event.state().routingTable().index(index);
236-
if (indexRoutingTable != null && indexRoutingTable.allPrimaryShardsActive()) {
237-
clusterService.submitStateUpdateTask("remove-index-block (all primary shards active for [" + index + "])", new ClusterStateUpdateTask() {
238-
@Override public ClusterState execute(ClusterState currentState) {
239-
ClusterBlocks.Builder blocks = ClusterBlocks.builder().blocks(currentState.blocks());
240-
blocks.removeIndexBlock(index, INDEX_NOT_RECOVERED_BLOCK);
241-
return ClusterState.builder().state(currentState).blocks(blocks).build();
242-
}
243-
});
244-
}
245-
}
246-
}
247-
}
248-
249224
if (event.state().nodes().localNode().masterNode() && event.metaDataChanged()) {
250225
executor.execute(new Runnable() {
251226
@Override public void run() {

‎modules/elasticsearch/src/main/java/org/elasticsearch/gateway/local/LocalGatewayNodeAllocation.java

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@
3737
import org.elasticsearch.common.unit.ByteSizeValue;
3838
import org.elasticsearch.common.unit.TimeValue;
3939
import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
40+
import org.elasticsearch.gateway.GatewayService;
4041
import org.elasticsearch.index.shard.ShardId;
4142
import org.elasticsearch.index.store.StoreFileMetaData;
4243
import org.elasticsearch.indices.store.TransportNodesListShardStoreMetaData;
@@ -86,7 +87,7 @@ public class LocalGatewayNodeAllocation extends NodeAllocation {
8687
}
8788
for (ShardRouting failedShard : allocation.failedShards()) {
8889
IndexRoutingTable indexRoutingTable = allocation.routingTable().index(failedShard.index());
89-
if (!allocation.routingNodes().blocks().hasIndexBlock(indexRoutingTable.index(), LocalGateway.INDEX_NOT_RECOVERED_BLOCK)) {
90+
if (!allocation.routingNodes().blocks().hasIndexBlock(indexRoutingTable.index(), GatewayService.INDEX_NOT_RECOVERED_BLOCK)) {
9091
continue;
9192
}
9293

@@ -151,7 +152,7 @@ public class LocalGatewayNodeAllocation extends NodeAllocation {
151152
// only do the allocation if there is a local "INDEX NOT RECOVERED" block
152153
// we check this here since it helps distinguish between index creation though an API, where the below logic
153154
// should not apply, and when recovering from the gateway, where we should apply this logic
154-
if (!routingNodes.blocks().hasIndexBlock(indexRoutingTable.index(), LocalGateway.INDEX_NOT_RECOVERED_BLOCK)) {
155+
if (!routingNodes.blocks().hasIndexBlock(indexRoutingTable.index(), GatewayService.INDEX_NOT_RECOVERED_BLOCK)) {
155156
continue;
156157
}
157158

‎modules/elasticsearch/src/main/java/org/elasticsearch/gateway/shared/SharedStorageGateway.java

Lines changed: 19 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -25,10 +25,12 @@
2525
import org.elasticsearch.cluster.metadata.MetaData;
2626
import org.elasticsearch.cluster.metadata.MetaDataCreateIndexService;
2727
import org.elasticsearch.common.StopWatch;
28+
import org.elasticsearch.common.collect.ImmutableSet;
2829
import org.elasticsearch.common.component.AbstractLifecycleComponent;
2930
import org.elasticsearch.common.settings.Settings;
3031
import org.elasticsearch.gateway.Gateway;
3132
import org.elasticsearch.gateway.GatewayException;
33+
import org.elasticsearch.gateway.GatewayService;
3234

3335
import java.io.IOException;
3436
import java.util.concurrent.ExecutorService;
@@ -145,17 +147,23 @@ private void updateClusterStateFromGateway(final MetaData fMetaData, final Gatew
145147
// the meta data per index, since we create the index and it will be added automatically
146148
for (final IndexMetaData indexMetaData : fMetaData) {
147149
try {
148-
createIndexService.createIndex(new MetaDataCreateIndexService.Request("gateway", indexMetaData.index()).settings(indexMetaData.settings()).mappingsCompressed(indexMetaData.mappings()).timeout(timeValueSeconds(30)), new MetaDataCreateIndexService.Listener() {
149-
@Override public void onResponse(MetaDataCreateIndexService.Response response) {
150-
if (indicesCounter.decrementAndGet() == 0) {
151-
listener.onSuccess();
152-
}
153-
}
154-
155-
@Override public void onFailure(Throwable t) {
156-
logger.error("failed to create index [{}]", indexMetaData.index(), t);
157-
}
158-
});
150+
createIndexService.createIndex(new MetaDataCreateIndexService.Request("gateway", indexMetaData.index())
151+
.settings(indexMetaData.settings())
152+
.mappingsCompressed(indexMetaData.mappings())
153+
.blocks(ImmutableSet.of(GatewayService.INDEX_NOT_RECOVERED_BLOCK))
154+
.timeout(timeValueSeconds(30)),
155+
156+
new MetaDataCreateIndexService.Listener() {
157+
@Override public void onResponse(MetaDataCreateIndexService.Response response) {
158+
if (indicesCounter.decrementAndGet() == 0) {
159+
listener.onSuccess();
160+
}
161+
}
162+
163+
@Override public void onFailure(Throwable t) {
164+
logger.error("failed to create index [{}]", indexMetaData.index(), t);
165+
}
166+
});
159167
} catch (IOException e) {
160168
logger.error("failed to create index [{}]", indexMetaData.index(), e);
161169
}

‎modules/test/integration/src/test/java/org/elasticsearch/test/integration/gateway/fs/SimpleFsIndexGatewayTests.java

Lines changed: 0 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,4 @@
2424
*/
2525
public class SimpleFsIndexGatewayTests extends AbstractSimpleIndexGatewayTests {
2626

27-
@Override public void testSnapshotOperations() throws Exception {
28-
super.testSnapshotOperations(); //To change body of overridden methods use File | Settings | File Templates.
29-
}
3027
}

0 commit comments

Comments
 (0)