Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fix security index auto-create and state recovery race #39582

Merged
merged 12 commits into from
Mar 5, 2019
Merged

Fix security index auto-create and state recovery race #39582

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
9c2e1de
Main
albertzaharovits Mar 1, 2019
4db7e68
test 1
albertzaharovits Mar 1, 2019
99a1a05
Tests
albertzaharovits Mar 1, 2019
ac82dd4
Merge branch 'master' into fix_security_index_recovery_race
albertzaharovits Mar 1, 2019
f734038
Iter
albertzaharovits Mar 3, 2019
1cd74ec
Iter2
albertzaharovits Mar 3, 2019
6363365
back to the frozen state
albertzaharovits Mar 3, 2019
b358996
Revert
albertzaharovits Mar 4, 2019
ea90cc1
Done
albertzaharovits Mar 4, 2019
0851725
Checkstyle
albertzaharovits Mar 4, 2019
9146fdb
Update x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/s…
tvernum Mar 5, 2019
974fc47
Checkstyle
albertzaharovits Mar 5, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 56 additions & 6 deletions ...security/src/main/java/org/elasticsearch/xpack/security/support/SecurityIndexManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,11 +45,13 @@

import java.nio.charset.StandardCharsets;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Predicate;
Expand Down Expand Up @@ -105,6 +107,11 @@ public boolean indexExists() {
return this.indexState.indexExists;
}

public boolean stateRecovered() {
// concrete security index name is a proxy for the cluster state recovery
return this.indexState.concreteIndexName != null;
}

/**
* Returns whether the index is on the current format if it exists. If the index does not exist
* we treat the index as up to date as we expect it to be created with the current format.
Expand Down Expand Up @@ -167,8 +174,10 @@ public void clusterChanged(ClusterChangedEvent event) {
this.indexState = newState;

if (newState.equals(previousState) == false) {
for (BiConsumer<State, State> listener : stateChangeListeners) {
listener.accept(previousState, newState);
// point in time iterator
final Iterator<BiConsumer<State, State>> stateListenerIterator = stateChangeListeners.iterator();
while (stateListenerIterator.hasNext()) {
stateListenerIterator.next().accept(previousState, newState);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As far as I can tell this change doesn't actually do anything.
A for loop is just syntactic sugar over iterator() and next(), you haven't change the semantics. The stateListenerIterator is still backed by the same List and is at risk of ConcurrentModificationException issues if a listener is added/removed.

Assuming that's the problem you're trying to solve, then the options come down to:

  1. synchronize whenever we use the list
  2. switch the list implementation to CopyOnWriteArrayList
  3. make a copy of the list before iterating over it

(2) is probably the best option. I don't think we add listeners very often, so making a list copy each time is probably OK, but you'd need to audit the places we modify the list.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As far as I can tell this change doesn't actually do anything.

It is a rewrite that unwraps the syntactic sugar because I find it confusing to use that when the list is concurrently modified.

The list is already of the CopyOnWriteArrayList type.

}
}
}
Expand Down Expand Up @@ -281,7 +290,10 @@ private static Version readMappingVersion(String indexName, MappingMetaData mapp
*/
public void checkIndexVersionThenExecute(final Consumer<Exception> consumer, final Runnable andThen) {
final State indexState = this.indexState; // use a local copy so all checks execute against the same state!
if (indexState.indexExists && indexState.isIndexUpToDate == false) {
if (indexState.concreteIndexName == null) {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

use stateRecovered() == false ?

Copy link
Contributor Author

@albertzaharovits albertzaharovits Mar 3, 2019 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I went with your suggestion, and then reverted. I have kept on checking indexState because this is the "frozen" index state that is used in all if-tests.

// index not recovered from gateway
delayUntilStateRecovered(consumer, () -> checkIndexVersionThenExecute(consumer, andThen));
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This worries me.
We're suddenly adding in a queue of pending tasks to run when the recovery block is removed, but I don't see any analysis on how big that queue could get, and what the impact is on the thread pool when it happens.

We don't do that now. We don't queue if the security index is red, we currently fail if the index is not recovered.
There's not enough information in this PR to know whether or not it's a sensible thing to do, and how we'd protect ourselves from getting into trouble with it (e.g. too many queued tasks, or delayed read + writes because recovery was slow).

From a usability point of view it's a nice idea, but it feels like we're jumping in without thinking it through.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Okay, understood! I just wonder if there is some alternative for "delaying requests" that would be acceptable beyond feature freeze?

} else if (indexState.indexExists && indexState.isIndexUpToDate == false) {
consumer.accept(new IllegalStateException(
"Security index is not on the current version. Security features relying on the index will not be available until " +
"the upgrade API is run on the security index"));
Expand All @@ -297,14 +309,17 @@ public void checkIndexVersionThenExecute(final Consumer<Exception> consumer, fin
public void prepareIndexIfNeededThenExecute(final Consumer<Exception> consumer, final Runnable andThen) {
final State indexState = this.indexState; // use a local copy so all checks execute against the same state!
// TODO we should improve this so we don't fire off a bunch of requests to do the same thing (create or update mappings)
if (indexState.indexExists && indexState.isIndexUpToDate == false) {
if (indexState.concreteIndexName == null) {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

use stateRecovered() == false ?

Copy link
Contributor Author

@albertzaharovits albertzaharovits Mar 3, 2019 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ditto.

I went with your suggestion, and then reverted. I have kept on checking indexState because this is the "frozen" index state that is used in all if-tests.

// index not recovered from gateway
delayUntilStateRecovered(consumer, () -> prepareIndexIfNeededThenExecute(consumer, andThen));
} else if (indexState.indexExists && indexState.isIndexUpToDate == false) {
consumer.accept(new IllegalStateException(
"Security index is not on the current version. Security features relying on the index will not be available until " +
"the upgrade API is run on the security index"));
} else if (indexState.indexExists == false) {
LOGGER.info("security index does not exist. Creating [{}] with alias [{}]", INTERNAL_SECURITY_INDEX, SECURITY_INDEX_NAME);
LOGGER.info("security index does not exist. Creating [{}] with alias [{}]", indexState.concreteIndexName, SECURITY_INDEX_NAME);
Tuple<String, Settings> mappingAndSettings = loadMappingAndSettingsSourceFromTemplate();
CreateIndexRequest request = new CreateIndexRequest(INTERNAL_SECURITY_INDEX)
CreateIndexRequest request = new CreateIndexRequest(indexState.concreteIndexName)
.alias(new Alias(SECURITY_INDEX_NAME))
.mapping("doc", mappingAndSettings.v1(), XContentType.JSON)
.waitForActiveShards(ActiveShardCount.ALL)
Expand Down Expand Up @@ -373,6 +388,41 @@ public static boolean isIndexDeleted(State previousState, State currentState) {
return previousState.indexStatus != null && currentState.indexStatus == null;
}

/**
* Delay the {@code runnable} invocation until cluster state recovered.
*/
private void delayUntilStateRecovered(final Consumer<Exception> consumer, final Runnable runnable) {
// context preserving one shoot runnable
final AtomicBoolean done = new AtomicBoolean(false);
final Runnable delayedRunnable = client.threadPool().getThreadContext().preserveContext(() -> {
if (false == done.get()) {
done.set(true);
runnable.run();
}
});
final BiConsumer<State, State> gatewayRecoveryListener = new BiConsumer<State, State>() {
@Override
public void accept(State prevState, State newState) {
// any cluster state update is a sign that the state recovered
if (newState.concreteIndexName != null) {
stateChangeListeners.remove(this);
client.threadPool().generic().execute(delayedRunnable);
} else {
consumer.accept(new IllegalStateException("State has been recovered, but the security index name is unknown."));
}
}
};
// enqueue and wait for the first cluster state update
stateChangeListeners.add(gatewayRecoveryListener);
// maybe state recovered in the meantime since we last checked
final State indexState = this.indexState;
if (indexState.concreteIndexName != null) {
// state indeed recovered and we _might_ have lost the notification
stateChangeListeners.remove(gatewayRecoveryListener);
delayedRunnable.run();
}
}

/**
* State of the security index.
*/
Expand Down
33 changes: 33 additions & 0 deletions ...ity/src/test/java/org/elasticsearch/xpack/security/support/SecurityIndexManagerTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
import org.elasticsearch.cluster.ClusterChangedEvent;
import org.elasticsearch.cluster.ClusterName;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.block.ClusterBlocks;
import org.elasticsearch.cluster.health.ClusterHealthStatus;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.cluster.metadata.IndexTemplateMetaData;
Expand All @@ -40,8 +41,10 @@
import org.elasticsearch.cluster.service.ClusterService;
import org.elasticsearch.common.UUIDs;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.util.concurrent.EsExecutors;
import org.elasticsearch.common.util.concurrent.ThreadContext;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.gateway.GatewayService;
import org.elasticsearch.index.Index;
import org.elasticsearch.index.shard.ShardId;
import org.elasticsearch.test.ESTestCase;
Expand All @@ -55,6 +58,7 @@
import static org.elasticsearch.xpack.security.support.SecurityIndexManager.SECURITY_TEMPLATE_NAME;
import static org.elasticsearch.xpack.security.support.SecurityIndexManager.TEMPLATE_VERSION_PATTERN;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.is;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;

Expand All @@ -72,6 +76,7 @@ public void setUpManager() {
final Client mockClient = mock(Client.class);
final ThreadPool threadPool = mock(ThreadPool.class);
when(threadPool.getThreadContext()).thenReturn(new ThreadContext(Settings.EMPTY));
when(threadPool.generic()).thenReturn(EsExecutors.newDirectExecutorService());
when(mockClient.threadPool()).thenReturn(threadPool);
when(mockClient.settings()).thenReturn(Settings.EMPTY);
final ClusterService clusterService = mock(ClusterService.class);
Expand Down Expand Up @@ -192,6 +197,32 @@ public void testIndexHealthChangeListeners() throws Exception {
assertEquals(ClusterHealthStatus.GREEN, currentState.get().indexStatus);
}

public void testListeneredNotCalledWhenStateNotRecovered() throws Exception {
final AtomicBoolean listenerCalled = new AtomicBoolean(false);
manager.addIndexStateListener((prev, current) -> {
listenerCalled.set(true);
});
final AtomicBoolean prepareCalled = new AtomicBoolean(false);
manager.prepareIndexIfNeededThenExecute(c -> {}, () -> {
prepareCalled.set(true);
});
final ClusterBlocks.Builder blocks = ClusterBlocks.builder().addGlobalBlock(GatewayService.STATE_NOT_RECOVERED_BLOCK);
// state not recovered
manager.clusterChanged(event(new ClusterState.Builder(CLUSTER_NAME).blocks(blocks)));
assertThat(listenerCalled.get(), is(false));
assertThat(prepareCalled.get(), is(false));
// state still not recovered
manager.clusterChanged(event(new ClusterState.Builder(CLUSTER_NAME).blocks(blocks)));
assertThat(listenerCalled.get(), is(false));
assertThat(prepareCalled.get(), is(false));
// state recovered with index
ClusterState.Builder clusterStateBuilder = createClusterState(INDEX_NAME, TEMPLATE_NAME, SecurityIndexManager.INTERNAL_INDEX_FORMAT);
markShardsAvailable(clusterStateBuilder);
manager.clusterChanged(event(clusterStateBuilder));
assertThat(listenerCalled.get(), is(true));
assertThat(prepareCalled.get(), is(true));
}

public void testIndexOutOfDateListeners() throws Exception {
final AtomicBoolean listenerCalled = new AtomicBoolean(false);
manager.clusterChanged(event(new ClusterState.Builder(CLUSTER_NAME)));
Expand Down Expand Up @@ -236,12 +267,14 @@ private void assertInitialState() {
assertThat(manager.indexExists(), Matchers.equalTo(false));
assertThat(manager.isAvailable(), Matchers.equalTo(false));
assertThat(manager.isMappingUpToDate(), Matchers.equalTo(false));
assertThat(manager.stateRecovered(), Matchers.equalTo(false));
}

private void assertIndexUpToDateButNotAvailable() {
assertThat(manager.indexExists(), Matchers.equalTo(true));
assertThat(manager.isAvailable(), Matchers.equalTo(false));
assertThat(manager.isMappingUpToDate(), Matchers.equalTo(true));
assertThat(manager.stateRecovered(), Matchers.equalTo(true));
}

public static ClusterState.Builder createClusterState(String indexName, String templateName) throws IOException {
Expand Down