honeymoose
/
OpenSearch
mirror of https://github.com/honeymoose/OpenSearch.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Load metadata at start time not construction time (#46326) 

Today we load the metadata from disk while constructing the node. However there
is no real need to do so, and this commit moves that code to run later while
the node is starting instead.
This commit is contained in:
David Turner 2019-09-10 11:15:10 +01:00 committed by GitHub
parent b40ac6dee7
commit 6c67b53932
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 107 additions and 75 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
server/src/main/java/org/elasticsearch/discovery/DiscoveryModule.java
View File

@ -28,7 +28,6 @@ import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.cluster.routing.RerouteService;
import org.elasticsearch.cluster.routing.allocation.AllocationService;
import org.elasticsearch.cluster.service.ClusterApplier;
import org.elasticsearch.cluster.service.ClusterApplierService;
import org.elasticsearch.cluster.service.MasterService;
import org.elasticsearch.common.Randomness;
import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
@ -152,12 +151,12 @@ public class DiscoveryModule {
if (ZEN2_DISCOVERY_TYPE.equals(discoveryType) || SINGLE_NODE_DISCOVERY_TYPE.equals(discoveryType)) {
discovery = new Coordinator(NODE_NAME_SETTING.get(settings),
settings, clusterSettings,
transportService, namedWriteableRegistry, allocationService, masterService,
() -> gatewayMetaState.getPersistedState(settings, (ClusterApplierService) clusterApplier), seedHostsProvider,
clusterApplier, joinValidators, new Random(Randomness.get().nextLong()), rerouteService, electionStrategy);
transportService, namedWriteableRegistry, allocationService, masterService, gatewayMetaState::getPersistedState,
seedHostsProvider, clusterApplier, joinValidators, new Random(Randomness.get().nextLong()), rerouteService,
electionStrategy);
} else if (ZEN_DISCOVERY_TYPE.equals(discoveryType)) {
discovery = new ZenDiscovery(settings, threadPool, transportService, namedWriteableRegistry, masterService, clusterApplier,
clusterSettings, seedHostsProvider, allocationService, joinValidators, gatewayMetaState, rerouteService);
clusterSettings, seedHostsProvider, allocationService, joinValidators, rerouteService);
} else {
throw new IllegalArgumentException("Unknown discovery type [" + discoveryType + "]");
}

11
server/src/main/java/org/elasticsearch/discovery/zen/ZenDiscovery.java
View File

@ -31,6 +31,7 @@ import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.ClusterStateTaskConfig;
import org.elasticsearch.cluster.NotMasterException;
import org.elasticsearch.cluster.block.ClusterBlocks;
import org.elasticsearch.cluster.coordination.FailedToCommitClusterStateException;
import org.elasticsearch.cluster.coordination.JoinTaskExecutor;
import org.elasticsearch.cluster.coordination.NoMasterBlockService;
import org.elasticsearch.cluster.coordination.NodeRemovalClusterStateTaskExecutor;
@ -40,7 +41,6 @@ import org.elasticsearch.cluster.routing.RerouteService;
import org.elasticsearch.cluster.routing.allocation.AllocationService;
import org.elasticsearch.cluster.service.ClusterApplier;
import org.elasticsearch.cluster.service.ClusterApplier.ClusterApplyListener;
import org.elasticsearch.cluster.service.ClusterApplierService;
import org.elasticsearch.cluster.service.MasterService;
import org.elasticsearch.common.Priority;
import org.elasticsearch.common.component.AbstractLifecycleComponent;
@ -59,10 +59,8 @@ import org.elasticsearch.core.internal.io.IOUtils;
import org.elasticsearch.discovery.Discovery;
import org.elasticsearch.discovery.DiscoverySettings;
import org.elasticsearch.discovery.DiscoveryStats;
import org.elasticsearch.cluster.coordination.FailedToCommitClusterStateException;
import org.elasticsearch.discovery.SeedHostsProvider;
import org.elasticsearch.discovery.zen.PublishClusterStateAction.IncomingClusterStateListener;
import org.elasticsearch.gateway.GatewayMetaState;
import org.elasticsearch.tasks.Task;
import org.elasticsearch.threadpool.ThreadPool;
import org.elasticsearch.transport.EmptyTransportResponseHandler;
@ -165,8 +163,7 @@ public class ZenDiscovery extends AbstractLifecycleComponent implements Discover
public ZenDiscovery(Settings settings, ThreadPool threadPool, TransportService transportService,
NamedWriteableRegistry namedWriteableRegistry, MasterService masterService, ClusterApplier clusterApplier,
ClusterSettings clusterSettings, SeedHostsProvider hostsProvider, AllocationService allocationService,
Collection<BiConsumer<DiscoveryNode, ClusterState>> onJoinValidators, GatewayMetaState gatewayMetaState,
RerouteService rerouteService) {
Collection<BiConsumer<DiscoveryNode, ClusterState>> onJoinValidators, RerouteService rerouteService) {
this.onJoinValidators = JoinTaskExecutor.addBuiltInJoinValidators(onJoinValidators);
this.masterService = masterService;
this.clusterApplier = clusterApplier;
@ -234,10 +231,6 @@ public class ZenDiscovery extends AbstractLifecycleComponent implements Discover
transportService.registerRequestHandler(
DISCOVERY_REJOIN_ACTION_NAME, ThreadPool.Names.SAME, RejoinClusterRequest::new, new RejoinClusterRequestHandler());
if (clusterApplier instanceof ClusterApplierService) {
((ClusterApplierService) clusterApplier).addLowPriorityApplier(gatewayMetaState);
}
}
// protected to allow overriding in tests

93
server/src/main/java/org/elasticsearch/gateway/GatewayMetaState.java
View File

@ -23,12 +23,12 @@ import com.carrotsearch.hppc.cursors.ObjectObjectCursor;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.logging.log4j.message.ParameterizedMessage;
import org.apache.lucene.util.SetOnce;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.Version;
import org.elasticsearch.cluster.ClusterChangedEvent;
import org.elasticsearch.cluster.ClusterName;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.ClusterStateApplier;
import org.elasticsearch.cluster.coordination.CoordinationState;
import org.elasticsearch.cluster.coordination.CoordinationState.PersistedState;
import org.elasticsearch.cluster.coordination.InMemoryPersistedState;
import org.elasticsearch.cluster.metadata.IndexMetaData;
@ -38,12 +38,12 @@ import org.elasticsearch.cluster.metadata.MetaDataIndexUpgradeService;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.cluster.routing.RoutingNode;
import org.elasticsearch.cluster.routing.ShardRouting;
import org.elasticsearch.cluster.service.ClusterApplierService;
import org.elasticsearch.cluster.service.ClusterService;
import org.elasticsearch.common.collect.ImmutableOpenMap;
import org.elasticsearch.common.collect.Tuple;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.discovery.DiscoveryModule;
import org.elasticsearch.index.Index;
import org.elasticsearch.plugins.MetaDataUpgrader;
import org.elasticsearch.transport.TransportService;
@ -71,41 +71,71 @@ import java.util.function.UnaryOperator;
* elected as master, it requests metaData from other master eligible nodes. After that, master node performs re-conciliation on the
* gathered results, re-creates {@link ClusterState} and broadcasts this state to other nodes in the cluster.
*/
public class GatewayMetaState implements ClusterStateApplier, CoordinationState.PersistedState {
public class GatewayMetaState implements PersistedState {
protected static final Logger logger = LogManager.getLogger(GatewayMetaState.class);
private final MetaStateService metaStateService;
private final Settings settings;
private final ClusterService clusterService;
private final TransportService transportService;
//there is a single thread executing updateClusterState calls, hence no volatile modifier
// On master-eligible Zen2 nodes, we use this very object for the PersistedState (so that the state is actually persisted); on other
// nodes we use an InMemoryPersistedState instead and persist using a cluster applier if needed. In all cases it's an error to try and
// use this object as a PersistedState before calling start(). TODO stop implementing PersistedState at the top level.
private final SetOnce<PersistedState> persistedState = new SetOnce<>();
// on master-eligible nodes we call updateClusterState under the Coordinator's mutex; on master-ineligible data nodes we call
// updateClusterState on the (unique) cluster applier thread; on other nodes we never call updateClusterState. In all cases there's no
// need to synchronize access to these variables.
protected Manifest previousManifest;
protected ClusterState previousClusterState;
protected boolean incrementalWrite;
public GatewayMetaState(Settings settings, MetaStateService metaStateService,
MetaDataIndexUpgradeService metaDataIndexUpgradeService, MetaDataUpgrader metaDataUpgrader,
TransportService transportService, ClusterService clusterService) throws IOException {
public GatewayMetaState(Settings settings, MetaStateService metaStateService) {
this.settings = settings;
this.metaStateService = metaStateService;
this.transportService = transportService;
this.clusterService = clusterService;
upgradeMetaData(metaDataIndexUpgradeService, metaDataUpgrader);
initializeClusterState(ClusterName.CLUSTER_NAME_SETTING.get(settings));
incrementalWrite = false;
}
public PersistedState getPersistedState(Settings settings, ClusterApplierService clusterApplierService) {
applyClusterStateUpdaters();
if (DiscoveryNode.isMasterNode(settings) == false) {
// use Zen1 way of writing cluster state for non-master-eligible nodes
// this avoids concurrent manipulating of IndexMetadata with IndicesStore
clusterApplierService.addLowPriorityApplier(this);
return new InMemoryPersistedState(getCurrentTerm(), getLastAcceptedState());
public void start(TransportService transportService, ClusterService clusterService,
MetaDataIndexUpgradeService metaDataIndexUpgradeService, MetaDataUpgrader metaDataUpgrader) {
assert previousClusterState == null : "should only start once, but already have " + previousClusterState;
try {
upgradeMetaData(metaDataIndexUpgradeService, metaDataUpgrader);
initializeClusterState(ClusterName.CLUSTER_NAME_SETTING.get(settings));
} catch (IOException e) {
throw new ElasticsearchException("failed to load metadata", e);
}
incrementalWrite = false;
applyClusterStateUpdaters(transportService, clusterService);
if (DiscoveryModule.DISCOVERY_TYPE_SETTING.get(settings).equals(DiscoveryModule.ZEN_DISCOVERY_TYPE)) {
// only for tests that simulate a mixed Zen1/Zen2 clusters, see Zen1IT
if (isMasterOrDataNode()) {
clusterService.addLowPriorityApplier(this::applyClusterState);
}
persistedState.set(new InMemoryPersistedState(getCurrentTerm(), getLastAcceptedState()));
} else {
if (DiscoveryNode.isMasterNode(settings) == false) {
if (DiscoveryNode.isDataNode(settings)) {
// Master-eligible nodes persist index metadata for all indices regardless of whether they hold any shards or not. It's
// vitally important to the safety of the cluster coordination system that master-eligible nodes persist this metadata
// when _accepting_ the cluster state (i.e. before it is committed). This persistence happens on the generic threadpool.
//
// In contrast, master-ineligible data nodes only persist the index metadata for shards that they hold. When all shards
// of an index are moved off such a node the IndicesStore is responsible for removing the corresponding index directory,
// including the metadata, and does so on the cluster applier thread.
//
// This presents a problem: if a shard is unassigned from a node and then reassigned back to it again then there is a
// race between the IndicesStore deleting the index folder and the CoordinationState concurrently trying to write the
// updated metadata into it. We could probably solve this with careful synchronization, but in fact there is no need.
// The persisted state on master-ineligible data nodes is mostly ignored - it's only there to support dangling index
// imports, which is inherently unsafe anyway. Thus we can safely delay metadata writes on master-ineligible data nodes
// until applying the cluster state, which is what this does:
clusterService.addLowPriorityApplier(this::applyClusterState);
}
persistedState.set(new InMemoryPersistedState(getCurrentTerm(), getLastAcceptedState()));
} else {
persistedState.set(this);
}
}
return this;
}
private void initializeClusterState(ClusterName clusterName) throws IOException {
@ -122,7 +152,7 @@ public class GatewayMetaState implements ClusterStateApplier, CoordinationState.
logger.debug("took {} to load state", TimeValue.timeValueMillis(TimeValue.nsecToMSec(System.nanoTime() - startNS)));
}
public void applyClusterStateUpdaters() {
protected void applyClusterStateUpdaters(TransportService transportService, ClusterService clusterService) {
assert previousClusterState.nodes().getLocalNode() == null : "applyClusterStateUpdaters must only be called once";
assert transportService.getLocalNode() != null : "transport service is not yet started";
@ -181,15 +211,18 @@ public class GatewayMetaState implements ClusterStateApplier, CoordinationState.
return DiscoveryNode.isMasterNode(settings) || DiscoveryNode.isDataNode(settings);
}
public PersistedState getPersistedState() {
final PersistedState persistedState = this.persistedState.get();
assert persistedState != null : "not started";
return persistedState;
}
public MetaData getMetaData() {
return previousClusterState.metaData();
}
@Override
public void applyClusterState(ClusterChangedEvent event) {
if (isMasterOrDataNode() == false) {
return;
}
private void applyClusterState(ClusterChangedEvent event) {
assert isMasterOrDataNode();
if (event.state().blocks().disableStatePersistence()) {
incrementalWrite = false;

16
server/src/main/java/org/elasticsearch/node/Node.java
View File

@ -158,7 +158,6 @@ import org.elasticsearch.usage.UsageService;
import org.elasticsearch.watcher.ResourceWatcherService;
import javax.net.ssl.SNIHostName;
import java.io.BufferedWriter;
import java.io.Closeable;
import java.io.IOException;
@ -483,8 +482,7 @@ public class Node implements Closeable {
).collect(Collectors.toSet());
final TransportService transportService = newTransportService(settings, transport, threadPool,
networkModule.getTransportInterceptor(), localNodeFactory, settingsModule.getClusterSettings(), taskHeaders);
final GatewayMetaState gatewayMetaState = new GatewayMetaState(settings, metaStateService,
metaDataIndexUpgradeService, metaDataUpgrader, transportService, clusterService);
final GatewayMetaState gatewayMetaState = new GatewayMetaState(settings, metaStateService);
final ResponseCollectorService responseCollectorService = new ResponseCollectorService(clusterService);
final SearchTransportService searchTransportService = new SearchTransportService(transportService,
SearchExecutionStatsCollector.makeWrapper(responseCollectorService));
@ -699,14 +697,14 @@ public class Node implements Closeable {
assert transportService.getLocalNode().equals(localNodeFactory.getNode())
: "transportService has a different local node than the factory provided";
injector.getInstance(PeerRecoverySourceService.class).start();
final MetaData onDiskMetadata;
// Load (and maybe upgrade) the metadata stored on disk
final GatewayMetaState gatewayMetaState = injector.getInstance(GatewayMetaState.class);
gatewayMetaState.start(transportService, clusterService,
injector.getInstance(MetaDataIndexUpgradeService.class), injector.getInstance(MetaDataUpgrader.class));
// we load the global state here (the persistent part of the cluster state stored on disk) to
// pass it to the bootstrap checks to allow plugins to enforce certain preconditions based on the recovered state.
if (DiscoveryNode.isMasterNode(settings()) || DiscoveryNode.isDataNode(settings())) {
onDiskMetadata = injector.getInstance(GatewayMetaState.class).getMetaData();
} else {
onDiskMetadata = MetaData.EMPTY_META_DATA;
}
final MetaData onDiskMetadata = gatewayMetaState.getPersistedState().getLastAcceptedState().metaData();
assert onDiskMetadata != null : "metadata is null but shouldn't"; // this is never null
validateNodeBeforeAcceptingRequests(new BootstrapContext(environment, onDiskMetadata), transportService.boundAddress(),
pluginsService.filterPlugins(Plugin.class).stream()

9
server/src/test/java/org/elasticsearch/bwcompat/RecoveryWithUnsupportedIndicesIT.java
View File

@ -28,9 +28,11 @@ import java.util.List;
import org.apache.lucene.util.LuceneTestCase;
import org.apache.lucene.util.TestUtil;
import org.elasticsearch.ExceptionsHelper;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.env.Environment;
import org.elasticsearch.env.NodeEnvironment;
import org.elasticsearch.gateway.CorruptStateException;
import org.elasticsearch.test.ESIntegTestCase;
import static org.hamcrest.Matchers.containsString;
@ -42,7 +44,7 @@ public class RecoveryWithUnsupportedIndicesIT extends ESIntegTestCase {
/**
* Return settings that could be used to start a node that has the given zipped home directory.
*/
protected Settings prepareBackwardsDataDir(Path backwardsIndex) throws IOException {
private Settings prepareBackwardsDataDir(Path backwardsIndex) throws IOException {
Path indexDir = createTempDir();
Path dataDir = indexDir.resolve("data");
try (InputStream stream = Files.newInputStream(backwardsIndex)) {
@ -86,7 +88,8 @@ public class RecoveryWithUnsupportedIndicesIT extends ESIntegTestCase {
logger.info("Checking static index {}", indexName);
Settings nodeSettings = prepareBackwardsDataDir(getDataPath("/indices/bwc").resolve(indexName + ".zip"));
assertThat(expectThrows(Exception.class, () -> internalCluster().startNode(nodeSettings))
.getCause().getCause().getMessage(), containsString("Format version is not supported"));
assertThat(ExceptionsHelper.unwrap(
expectThrows(Exception.class, () -> internalCluster().startNode(nodeSettings)), CorruptStateException.class).getMessage(),
containsString("Format version is not supported"));
}
}

4
server/src/test/java/org/elasticsearch/discovery/zen/ZenDiscoveryUnitTests.java
View File

@ -52,7 +52,6 @@ import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.core.internal.io.IOUtils;
import org.elasticsearch.discovery.zen.PublishClusterStateActionTests.AssertingAckListener;
import org.elasticsearch.discovery.zen.ZenDiscovery.ZenNodeRemovalClusterStateTaskExecutor;
import org.elasticsearch.gateway.GatewayMetaState;
import org.elasticsearch.index.shard.ShardId;
import org.elasticsearch.test.ClusterServiceUtils;
import org.elasticsearch.test.ESTestCase;
@ -369,8 +368,7 @@ public class ZenDiscoveryUnitTests extends ESTestCase {
ZenDiscovery zenDiscovery = new ZenDiscovery(settings, threadPool, service,
new NamedWriteableRegistry(ClusterModule.getNamedWriteables()),
masterService, clusterApplier, clusterSettings, hostsResolver -> Collections.emptyList(),
ESAllocationTestCase.createAllocationService(),
Collections.emptyList(), mock(GatewayMetaState.class), (s, p, r) -> {});
ESAllocationTestCase.createAllocationService(), Collections.emptyList(), (s, p, r) -> {});
zenDiscovery.start();
return zenDiscovery;
}

6
server/src/test/java/org/elasticsearch/gateway/GatewayMetaStatePersistedStateTests.java
View File

@ -63,9 +63,9 @@ public class GatewayMetaStatePersistedStateTests extends ESTestCase {
super.tearDown();
}
private MockGatewayMetaState newGateway() throws IOException {
MockGatewayMetaState gateway = new MockGatewayMetaState(settings, nodeEnvironment, xContentRegistry(), localNode);
gateway.applyClusterStateUpdaters();
private MockGatewayMetaState newGateway() {
final MockGatewayMetaState gateway = new MockGatewayMetaState(settings, nodeEnvironment, xContentRegistry(), localNode);
gateway.start();
return gateway;
}

12
test/framework/src/main/java/org/elasticsearch/cluster/coordination/AbstractCoordinatorTestCase.java
View File

@ -703,8 +703,10 @@ public class AbstractCoordinatorTestCase extends ESTestCase {
if (rarely()) {
nodeEnvironment = newNodeEnvironment();
nodeEnvironments.add(nodeEnvironment);
delegate = new MockGatewayMetaState(Settings.EMPTY, nodeEnvironment, xContentRegistry(), localNode)
.getPersistedState(Settings.EMPTY, null);
final MockGatewayMetaState gatewayMetaState
= new MockGatewayMetaState(Settings.EMPTY, nodeEnvironment, xContentRegistry(), localNode);
gatewayMetaState.start();
delegate = gatewayMetaState.getPersistedState();
} else {
nodeEnvironment = null;
delegate = new InMemoryPersistedState(0L,
@ -734,8 +736,10 @@ public class AbstractCoordinatorTestCase extends ESTestCase {
new Manifest(updatedTerm, manifest.getClusterStateVersion(), manifest.getGlobalGeneration(),
manifest.getIndexGenerations()));
}
delegate = new MockGatewayMetaState(Settings.EMPTY, nodeEnvironment, xContentRegistry(), newLocalNode)
.getPersistedState(Settings.EMPTY, null);
final MockGatewayMetaState gatewayMetaState
= new MockGatewayMetaState(Settings.EMPTY, nodeEnvironment, xContentRegistry(), newLocalNode);
gatewayMetaState.start();
delegate = gatewayMetaState.getPersistedState();
} else {
nodeEnvironment = null;
BytesStreamOutput outStream = new BytesStreamOutput();

16
test/framework/src/main/java/org/elasticsearch/gateway/MockGatewayMetaState.java
View File

@ -28,10 +28,6 @@ import org.elasticsearch.env.NodeEnvironment;
import org.elasticsearch.plugins.MetaDataUpgrader;
import org.elasticsearch.transport.TransportService;
import java.io.IOException;
import static org.mockito.Mockito.mock;
/**
* {@link GatewayMetaState} constructor accepts a lot of arguments.
* It's not always easy / convenient to construct these dependencies.
@ -42,10 +38,8 @@ public class MockGatewayMetaState extends GatewayMetaState {
private final DiscoveryNode localNode;
public MockGatewayMetaState(Settings settings, NodeEnvironment nodeEnvironment,
NamedXContentRegistry xContentRegistry, DiscoveryNode localNode) throws IOException {
super(settings, new MetaStateService(nodeEnvironment, xContentRegistry),
mock(MetaDataIndexUpgradeService.class), mock(MetaDataUpgrader.class),
mock(TransportService.class), mock(ClusterService.class));
NamedXContentRegistry xContentRegistry, DiscoveryNode localNode) {
super(settings, new MetaStateService(nodeEnvironment, xContentRegistry));
this.localNode = localNode;
}
@ -55,8 +49,12 @@ public class MockGatewayMetaState extends GatewayMetaState {
}
@Override
public void applyClusterStateUpdaters() {
public void applyClusterStateUpdaters(TransportService transportService, ClusterService clusterService) {
// Just set localNode here, not to mess with ClusterService and IndicesService mocking
previousClusterState = ClusterStateUpdaters.setLocalNode(previousClusterState, localNode);
}
public void start() {
start(null, null, null, null);
}
}

6
test/framework/src/main/java/org/elasticsearch/test/InternalTestCluster.java
View File

@ -958,10 +958,16 @@ public final class InternalTestCluster extends TestCluster {
}
void startNode() {
boolean success = false;
try {
node.start();
success = true;
} catch (NodeValidationException e) {
throw new RuntimeException(e);
} finally {
if (success == false) {
IOUtils.closeWhileHandlingException(node);
}
}
}