Wait for cluster to recover before resolving index template #99797
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,5 @@ | ||
| pr: 99797 | ||
| summary: Wait for cluster to recover before resolving index template | ||
| area: CRUD | ||
| type: bug | ||
| issues: [] |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -191,9 +191,47 @@ public static <Response extends ReplicationResponse & WriteResponse> ActionListe | |
|
|
||
| @Override | ||
| protected void doExecute(Task task, BulkRequest bulkRequest, ActionListener<BulkResponse> listener) { | ||
| final long startTime = relativeTime(); | ||
| final ClusterState initialState = clusterService.state(); | ||
| final ClusterBlockException blockException = initialState.blocks().globalBlockedException(ClusterBlockLevel.WRITE); | ||
| if (blockException != null) { | ||
| if (false == blockException.retryable()) { | ||
| listener.onFailure(blockException); | ||
| return; | ||
| } | ||
| logger.trace("cluster is blocked, waiting for it to recover", blockException); | ||
| final ClusterStateObserver clusterStateObserver = new ClusterStateObserver( | ||
| initialState, | ||
| clusterService, | ||
| bulkRequest.timeout(), | ||
| logger, | ||
| threadPool.getThreadContext() | ||
| ); | ||
| clusterStateObserver.waitForNextChange(new ClusterStateObserver.Listener() { | ||
| @Override | ||
| public void onNewClusterState(ClusterState state) { | ||
| doExecuteOnWriteThreadPool(task, bulkRequest, startTime, listener); | ||
| } | ||
|
|
||
| @Override | ||
| public void onClusterServiceClose() { | ||
| listener.onFailure(new NodeClosedException(clusterService.localNode())); | ||
| } | ||
|
|
||
| @Override | ||
| public void onTimeout(TimeValue timeout) { | ||
| listener.onFailure(blockException); | ||
| } | ||
| }, newState -> false == newState.blocks().hasGlobalBlockWithLevel(ClusterBlockLevel.WRITE)); | ||
| } else { | ||
| doExecuteOnWriteThreadPool(task, bulkRequest, startTime, listener); | ||
| } | ||
| } | ||
|
|
||
| private void doExecuteOnWriteThreadPool(Task task, BulkRequest bulkRequest, long startTime, ActionListener<BulkResponse> listener) { | ||
ywangd marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| /* | ||
| * This is called on the Transport thread and sometimes on the cluster state applier thread, | ||
| * so we can check the indexing memory pressure *quickly* but we don't want to keep the transport | ||
| * thread busy. Then, as soon as we have the indexing pressure in we fork | ||
| * to one of the write thread pools. We do this because juggling the | ||
| * bulk request can get expensive for a few reasons: | ||
|
|
@@ -209,19 +247,24 @@ protected void doExecute(Task task, BulkRequest bulkRequest, ActionListener<Bulk | |
| final int indexingOps = bulkRequest.numberOfActions(); | ||
| final long indexingBytes = bulkRequest.ramBytesUsed(); | ||
| final boolean isOnlySystem = isOnlySystem(bulkRequest, clusterService.state().metadata().getIndicesLookup(), systemIndices); | ||
| final String executorName = isOnlySystem ? Names.SYSTEM_WRITE : Names.WRITE; | ||
| final Releasable releasable = indexingPressure.markCoordinatingOperationStarted(indexingOps, indexingBytes, isOnlySystem); | ||
|
There was a problem hiding this comment. Ah wait, sorry, we need to do this check first, on the transport worker, before we start waiting for cluster states. Otherwise we might just pile up far too much work in memory waiting for the cluster recovery. But that is a little tricky because we need to compute There was a problem hiding this comment. Sorry I don't quite follow. It seems to me that when the code reaches here, the cluster state is either recovered successfully or we had no need to wait for recover at all. So there is no more waiting after this line? There was a problem hiding this comment. Yes that's right, but it's the waiting before this line that's a problem. Before we run this line we have the bulk request in memory but we aren't tracking it in the indexing pressure subsystem. If we receive lots of indexing requests while the cluster is recovering then we will try to hold them all in memory, rejecting none of them and will eventually just OOM. There was a problem hiding this comment. Makes sense. Thanks a lot for the explanation! Updated in 812154a There was a problem hiding this comment. Btw, I didn't change how There was a problem hiding this comment. I had some concerns about whether There was a problem hiding this comment. Thanks checking! I didn't think that was possible. |
||
| final ActionListener<BulkResponse> releasingListener = ActionListener.runBefore(listener, releasable::close); | ||
| threadPool.executor(Names.WRITE).execute(new ActionRunnable<>(releasingListener) { | ||
| @Override | ||
| protected void doRun() { | ||
| doInternalExecute(task, bulkRequest, executorName, startTime, releasingListener); | ||
| } | ||
| }); | ||
| } | ||
|
|
||
|
There was a problem hiding this comment. I think preserving the original start time is a good change but it seems unrelated to the rest of this PR, and deserves a test of its own too. Could you separate that out into a different PR? There was a problem hiding this comment. Thanks for catching it. This is a bigger change than I initially thought because today the startTime is reset after ingestService completes the processing. My previous change would have made it include the ingestService processing time (but not the request forward time). I have reverted this change. A follow-up is better. Thanks! |
||
| protected void doInternalExecute( | ||
| Task task, | ||
| BulkRequest bulkRequest, | ||
| String executorName, | ||
| long startTime, | ||
| ActionListener<BulkResponse> listener | ||
| ) { | ||
| final AtomicArray<BulkItemResponse> responses = new AtomicArray<>(bulkRequest.requests.size()); | ||
|
|
||
| boolean hasIndexRequestsWithPipelines = false; | ||
|
|
@@ -256,7 +299,7 @@ protected void doInternalExecute(Task task, BulkRequest bulkRequest, String exec | |
| assert arePipelinesResolved : bulkRequest; | ||
| } | ||
| if (clusterService.localNode().isIngestNode()) { | ||
| processBulkIndexIngestRequest(task, bulkRequest, executorName, startTime, l); | ||
| } else { | ||
| ingestForwarder.forwardIngestRequest(BulkAction.INSTANCE, bulkRequest, l); | ||
| } | ||
|
|
@@ -759,6 +802,7 @@ private void processBulkIndexIngestRequest( | |
| Task task, | ||
| BulkRequest original, | ||
| String executorName, | ||
| long startTime, | ||
| ActionListener<BulkResponse> listener | ||
| ) { | ||
| final long ingestStartTimeInNanos = System.nanoTime(); | ||
|
|
@@ -788,7 +832,7 @@ private void processBulkIndexIngestRequest( | |
| ActionRunnable<BulkResponse> runnable = new ActionRunnable<>(actionListener) { | ||
| @Override | ||
| protected void doRun() { | ||
| doInternalExecute(task, bulkRequest, executorName, startTime, actionListener); | ||
| } | ||
|
|
||
| @Override | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Hmm I think it's not necessary to skip this validation, it's only checking that the master is elected. At least, the test should work whether we wait like that or not, so maybe return
randomBoolean()to indicate we're not relying on it.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
That's good know. Thanks. Updated as suggested.