Skip to content

Make ILMHistoryStore.putAsync truly async #50403

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

Merged
merged 7 commits into from
Dec 20, 2019
Merged

Make ILMHistoryStore.putAsync truly async #50403

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
d5cc5fa
Make ILMHistoryStore.putAsync truly async
dakrone Dec 19, 2019
6878d90
Merge branch 'master' into ilm-history-async-put
elasticmachine Dec 19, 2019
42c9c84
Remove exception wrapping, add logging guard and URL issue link
dakrone Dec 20, 2019
a8e7460
Merge branch 'master' into ilm-history-async-put
elasticmachine Dec 20, 2019
d0cac4e
Add a sanity time limit of a two minute wait
dakrone Dec 20, 2019
3e2fc23
Unmute recently muted test
dakrone Dec 20, 2019
9946942
Disable ILM history for SLM integ tests
dakrone Dec 20, 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
2 changes: 0 additions & 2 deletions ...qa/multi-node/src/test/java/org/elasticsearch/xpack/ilm/TimeSeriesLifecycleActionsIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1004,7 +1004,6 @@ public void testILMRolloverOnManuallyRolledIndex() throws Exception {
assertBusy(() -> assertTrue(indexExists(thirdIndex)));
}

@AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/50353")
public void testHistoryIsWrittenWithSuccess() throws Exception {
String index = "index";

Expand Down Expand Up @@ -1047,7 +1046,6 @@ public void testHistoryIsWrittenWithSuccess() throws Exception {
assertBusy(() -> assertHistoryIsPresent(policy, index + "-000002", true, "check-rollover-ready"), 30, TimeUnit.SECONDS);
}

@AwaitsFix( bugUrl = "https://github.com/elastic/elasticsearch/issues/50353")
public void testHistoryIsWrittenWithFailure() throws Exception {
String index = "index";

Expand Down
3 changes: 2 additions & 1 deletion x-pack/plugin/ilm/src/main/java/org/elasticsearch/xpack/ilm/IndexLifecycle.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -179,7 +179,8 @@ public Collection<Object> createComponents(Client client, ClusterService cluster
@SuppressWarnings("unused")
ILMHistoryTemplateRegistry ilmTemplateRegistry =
new ILMHistoryTemplateRegistry(settings, clusterService, threadPool, client, xContentRegistry);
ilmHistoryStore.set(new ILMHistoryStore(settings, new OriginSettingClient(client, INDEX_LIFECYCLE_ORIGIN), clusterService));
ilmHistoryStore.set(new ILMHistoryStore(settings, new OriginSettingClient(client, INDEX_LIFECYCLE_ORIGIN),
clusterService, threadPool));
indexLifecycleInitialisationService.set(new IndexLifecycleService(settings, client, clusterService, threadPool,
getClock(), System::currentTimeMillis, xContentRegistry, ilmHistoryStore.get()));
components.add(indexLifecycleInitialisationService.get());
Expand Down
77 changes: 57 additions & 20 deletions x-pack/plugin/ilm/src/main/java/org/elasticsearch/xpack/ilm/history/ILMHistoryStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.logging.log4j.message.ParameterizedMessage;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.ResourceAlreadyExistsException;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.admin.indices.alias.Alias;
Expand All @@ -31,11 +32,13 @@
import org.elasticsearch.common.xcontent.ToXContent;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.threadpool.ThreadPool;

import java.io.Closeable;
import java.io.IOException;
import java.util.Arrays;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

Expand All @@ -54,26 +57,52 @@ public class ILMHistoryStore implements Closeable {
public static final String ILM_HISTORY_INDEX_PREFIX = "ilm-history-" + INDEX_TEMPLATE_VERSION + "-";
public static final String ILM_HISTORY_ALIAS = "ilm-history-" + INDEX_TEMPLATE_VERSION;

private final Client client;
private final ClusterService clusterService;
private final boolean ilmHistoryEnabled;
private final BulkProcessor processor;
private final ThreadPool threadPool;

public ILMHistoryStore(Settings nodeSettings, Client client, ClusterService clusterService) {
this.client = client;
this.clusterService = clusterService;
ilmHistoryEnabled = LIFECYCLE_HISTORY_INDEX_ENABLED_SETTING.get(nodeSettings);
public ILMHistoryStore(Settings nodeSettings, Client client, ClusterService clusterService, ThreadPool threadPool) {
this.ilmHistoryEnabled = LIFECYCLE_HISTORY_INDEX_ENABLED_SETTING.get(nodeSettings);
this.threadPool = threadPool;

this.processor = BulkProcessor.builder(
new OriginSettingClient(client, INDEX_LIFECYCLE_ORIGIN)::bulk,
new BulkProcessor.Listener() {
@Override
public void beforeBulk(long executionId, BulkRequest request) { }
public void beforeBulk(long executionId, BulkRequest request) {
// Prior to actually performing the bulk, we should ensure the index exists, and
// if we were unable to create it or it was in a bad state, we should not
// attempt to index documents.
try {
final CompletableFuture<Boolean> indexCreated = new CompletableFuture<>();
ensureHistoryIndex(client, clusterService.state(), ActionListener.wrap(indexCreated::complete,
ex -> {
logger.warn("failed to create ILM history store index prior to issuing bulk request", ex);
indexCreated.completeExceptionally(ex);
}));
indexCreated.get(2, TimeUnit.MINUTES);
} catch (Exception e) {
logger.warn(new ParameterizedMessage("unable to index the following ILM history items:\n{}",
request.requests().stream()
.filter(dwr -> (dwr instanceof IndexRequest))
.map(dwr -> ((IndexRequest) dwr))
.map(IndexRequest::sourceAsMap)
.map(Object::toString)
.collect(Collectors.joining("\n"))), e);
throw new ElasticsearchException(e);
}
}

@Override
public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
long items = request.numberOfActions();
logger.trace("indexed [{}] items into ILM history index", items);
if (logger.isTraceEnabled()) {
logger.trace("indexed [{}] items into ILM history index [{}]", items,
Arrays.stream(response.getItems())
.map(BulkItemResponse::getIndex)
.distinct()
.collect(Collectors.joining(",")));
}
if (response.hasFailures()) {
Map<String, String> failures = Arrays.stream(response.getItems())
.filter(BulkItemResponse::isFailed)
Expand Down Expand Up @@ -105,18 +134,25 @@ public void putAsync(ILMHistoryItem item) {
LIFECYCLE_HISTORY_INDEX_ENABLED_SETTING.getKey(), item);
return;
}
logger.trace("about to index ILM history item in index [{}]: [{}]", ILM_HISTORY_ALIAS, item);
ensureHistoryIndex(client, clusterService.state(), ActionListener.wrap(createdIndex -> {
try (XContentBuilder builder = XContentFactory.jsonBuilder()) {
item.toXContent(builder, ToXContent.EMPTY_PARAMS);
IndexRequest request = new IndexRequest(ILM_HISTORY_ALIAS).source(builder);
processor.add(request);
} catch (IOException exception) {
logger.error(new ParameterizedMessage("failed to index ILM history item in index [{}]: [{}]",
ILM_HISTORY_ALIAS, item), exception);
}
}, ex -> logger.error(new ParameterizedMessage("failed to ensure ILM history index exists, not indexing history item [{}]",
item), ex)));
logger.trace("queueing ILM history item for indexing [{}]: [{}]", ILM_HISTORY_ALIAS, item);
try (XContentBuilder builder = XContentFactory.jsonBuilder()) {
item.toXContent(builder, ToXContent.EMPTY_PARAMS);
IndexRequest request = new IndexRequest(ILM_HISTORY_ALIAS).source(builder);
// TODO: remove the threadpool wrapping when the .add call is non-blocking
Copy link
Contributor

Choose a reason for hiding this comment

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

Oh, this is not intuitive/obvious. Maybe in a follow-up PR shall we rename the add method to addAndMaybeExecute or something along those lines?

Copy link
Member Author

Choose a reason for hiding this comment

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

In a followup we are going to make the add method not block on execution (but still block on adding to the queue). I opened #50440 for this (and I'll link it in the comment in the code)

Copy link
Contributor

Choose a reason for hiding this comment

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

Brill, thanks @dakrone

// (it can currently execute the bulk request occasionally)
// see: https://github.com/elastic/elasticsearch/issues/50440
threadPool.executor(ThreadPool.Names.GENERIC).execute(() -> {
try {
processor.add(request);
} catch (Exception e) {
logger.error(new ParameterizedMessage("failed add ILM history item to queue for index [{}]: [{}]",
ILM_HISTORY_ALIAS, item), e);
}
});
} catch (IOException exception) {
logger.error(new ParameterizedMessage("failed to queue ILM history item in index [{}]: [{}]",
ILM_HISTORY_ALIAS, item), exception);
}
}

/**
Expand All @@ -134,6 +170,7 @@ static void ensureHistoryIndex(Client client, ClusterState state, ActionListener

if (ilmHistory == null && initialHistoryIndex == null) {
// No alias or index exists with the expected names, so create the index with appropriate alias
logger.debug("creating ILM history index [{}]", initialHistoryIndexName);
client.admin().indices().prepareCreate(initialHistoryIndexName)
.setWaitForActiveShards(1)
.addAlias(new Alias(ILM_HISTORY_ALIAS)
Expand Down
2 changes: 1 addition & 1 deletion x-pack/plugin/ilm/src/test/java/org/elasticsearch/xpack/ilm/IndexLifecycleRunnerTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1118,7 +1118,7 @@ private class NoOpHistoryStore extends ILMHistoryStore {
private final List<ILMHistoryItem> items = new ArrayList<>();

NoOpHistoryStore() {
super(Settings.EMPTY, noopClient, null);
super(Settings.EMPTY, noopClient, null, null);
}

public List<ILMHistoryItem> getItems() {
Expand Down
4 changes: 2 additions & 2 deletions ...ck/plugin/ilm/src/test/java/org/elasticsearch/xpack/ilm/history/ILMHistoryStoreTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,7 +68,7 @@ public void setup() {
threadPool = new TestThreadPool(this.getClass().getName());
client = new VerifyingClient(threadPool);
clusterService = ClusterServiceUtils.createClusterService(threadPool);
historyStore = new ILMHistoryStore(Settings.EMPTY, client, clusterService);
historyStore = new ILMHistoryStore(Settings.EMPTY, client, clusterService, threadPool);
}

@After
Expand All @@ -81,7 +81,7 @@ public void setdown() {

public void testNoActionIfDisabled() throws Exception {
Settings settings = Settings.builder().put(LIFECYCLE_HISTORY_INDEX_ENABLED_SETTING.getKey(), false).build();
try (ILMHistoryStore disabledHistoryStore = new ILMHistoryStore(settings, client, null)) {
try (ILMHistoryStore disabledHistoryStore = new ILMHistoryStore(settings, client, null, threadPool)) {
String policyId = randomAlphaOfLength(5);
final long timestamp = randomNonNegativeLong();
ILMHistoryItem record = ILMHistoryItem.success("index", policyId, timestamp, null, null);
Expand Down
8 changes: 8 additions & 0 deletions ...k/plugin/ilm/src/test/java/org/elasticsearch/xpack/slm/SLMSnapshotBlockingIntegTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
import org.elasticsearch.snapshots.mockstore.MockRepository;
import org.elasticsearch.test.ESIntegTestCase;
import org.elasticsearch.xpack.core.LocalStateCompositeXPackPlugin;
import org.elasticsearch.xpack.core.ilm.LifecycleSettings;
import org.elasticsearch.xpack.core.slm.SnapshotLifecyclePolicy;
import org.elasticsearch.xpack.core.slm.SnapshotLifecyclePolicyItem;
import org.elasticsearch.xpack.core.slm.SnapshotRetentionConfiguration;
Expand Down Expand Up @@ -64,6 +65,13 @@ public class SLMSnapshotBlockingIntegTests extends ESIntegTestCase {
static final String REPO = "my-repo";
List<String> dataNodeNames = null;

@Override
protected Settings nodeSettings(int nodeOrdinal) {
return Settings.builder().put(super.nodeSettings(nodeOrdinal))
.put(LifecycleSettings.LIFECYCLE_HISTORY_INDEX_ENABLED, false)
.build();
}

@Before
public void ensureClusterNodes() {
logger.info("--> starting enough nodes to ensure we have enough to safely stop for tests");
Expand Down