Hi - We are using Elasticsearch 8.6.2 running on Azure AKS and noticed some serious issues lately.
As of last week the week of Jun30 2023, we started noticing huge unresolved search transport queues in our cluster. The queues grow indefinitely until eventually thread pools start getting rejected.
The issue happens sporadically and only seems to occur in prod under high volume. We have a ~48 data node cluster with 106 shards targeted 50GB for shard size.
We did not make any major infra changs when this issue started occuring. I checked the hot_threads output as was suggested elsewhere and it did not inform much on the nodes with queues. I did notice however on the _tasks API there is a huge backlog of tasks that all look like this:
As you can see, the task is marked as 'cancelled' , but it stil persists for over 900 seconds. You would think Elasticsearch would see this and eventually have a circuit breaker that times it out?
There are hundreds and hundreds of these tasks that pile up before the cluster starts falling over.
I also tried taking the query from the task and running it on my own, and it returns in a prompt manner. So i dont think the issue is specific to the query.
We think there are transient/sporadic connection issues occuring that are causing connections to be timed out between nodes and then the task is stuck waiting on a response indefinitely.
Restarting affected data nodes temporarily fixes the issue, but it does not seem to be a sustainable solution, as we have been having to do this throughout the week to prevent prod outages and it has taken a considerable amount of time to just keep the queues at bay.
How can we configure elastic so that if a task is marked ' cancelled' , it will actually go away and these ghost/orphan tasks dont just pile up and kill our cluster?
It's not a config thing, this should just happen. If it doesn't then that's a bug. To investigate further, could you get GET _tasks?detailed and GET _nodes/hot_threads?threads=9999 every minute or so?
Hmm actually having said that I don't think it's a config issue, Linux has a silly default that means that it can take 900+ seconds between a connection drop and a notification to userspace about the connection drop. This page of the manual has more details. You'd see messages about dropped connections in the logs if it was this.
You'd need to be a bit more precise about what you mean by "kill our cluster" tho. The other recent thread on this topic has the zombie tasks consuming a lot of CPU, but that wouldn't happen if they were just waiting for a dead connection to time out. They'd potentially hold on to a lot of heap, causing GC pressure, but wouldn't themselves consume any CPU.
Over time, these queues will rapidly grow and fill up with tasks, eventually causing searches and writes to fail with es-rejected-execution-exception. So thats what i mean by 'kill our cluster' apologies for the poor choice of wording. If you see this example here, the queue rapidly increases and the only we we have found to resolve it so far has been by restarting the data node:
In our metrics we see that the CPU/Memory/GC all look fine on the nodes with queues, but the thread pool size will increase until it gets capped out by presumably the VM limitation. Once that happen, we see spikes in rejected threads and all the indices that are on the affected nodes will say that the shards have failed in the es-rejected-execution-exception logs when searched or written to.
With regards to the TCP Transmission link you gave, I believe this may apply to us? For the node to node communication, we use Azure Internal Load Balancer. This hasn't caused any issues before (3+ years of running in prod), but when troubleshooting this I noticed the idle timeout setting is 4 mins on the load balancer used to facilitate node to node communication. Should we look into enabling transport.ping_schedule as described in that post?
We actually just had a small queue start forming on one of the data nodes, ive posted the relevant output from the GET _tasks/?detailed=true on the node, and the hot_threads from the node below:
hot_threads:
::: {api-es-data-z3-2}{CtydY-1CTPy1x3awIq_iPg}{Pfrk46euSoq7AKgCe72eoQ}{api-es-data-z3-2}{10.95.43.222}{10.95.43.222:9300}{d}{zone=centralus-3, xpack.installed=true, k8s_node_name=aks-deligibilv2x-34581791-vmss000012}
Hot threads at 2023-07-03T18:49:38.315Z, interval=500ms, busiestThreads=3, ignoreIdleThreads=true:
Not at all, but thanks for the extra precision. This is not consistent with lost responses, because the search threadpool queue comprises entirely tasks waiting for a free thread to run them: a task waiting for a response does not wait in this queue.
The hot threads you shared did not include the ?threads=9999 parameter so doesn't have any useful information unfortunately.
I do not think that specific PR has any relationship with this situation. It only pertains to tasks which have a transport-layer timeout (internal to Elasticsearch) and none of the tasks involved in a search have that kind of timeout.
Apologies for the screwup - Here is the hot threads output from a node with a queue as of a few mins ago with the proper setting of ?threads=9999, this may contain more helpful info:
::: {api-es-data-z3-2}{CtydY-1CTPy1x3awIq_iPg}{gCxMPRKlSBazEarOoqS4UA}{api-es-data-z3-2}{10.95.43.214}{10.95.43.214:9300}{d}{zone=centralus-3, xpack.installed=true, k8s_node_name=aks-deligibilv2x-34581791-vmss000012}
Hot threads at 2023-07-03T19:29:15.352Z, interval=500ms, busiestThreads=9999, ignoreIdleThreads=true:
2.0% [cpu=2.0%, other=0.0%] (10ms out of 500ms) cpu usage by thread 'elasticsearch[api-es-data-z3-2][search][T#12]'
3/10 snapshots sharing following 2 elements
java.base@19.0.2/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
java.base@19.0.2/java.lang.Thread.run(Thread.java:1589)
2.0% [cpu=2.0%, other=0.0%] (9.8ms out of 500ms) cpu usage by thread 'elasticsearch[api-es-data-z3-2][search][T#9]'
2/10 snapshots sharing following 2 elements
java.base@19.0.2/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
java.base@19.0.2/java.lang.Thread.run(Thread.java:1589)
1.9% [cpu=1.9%, other=0.0%] (9.4ms out of 500ms) cpu usage by thread 'elasticsearch[api-es-data-z3-2][search][T#19]'
6/10 snapshots sharing following 2 elements
java.base@19.0.2/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
java.base@19.0.2/java.lang.Thread.run(Thread.java:1589)
1.7% [cpu=1.7%, other=0.0%] (8.6ms out of 500ms) cpu usage by thread 'elasticsearch[api-es-data-z3-2][search][T#40]'
9/10 snapshots sharing following 2 elements
java.base@19.0.2/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
java.base@19.0.2/java.lang.Thread.run(Thread.java:1589)
1.7% [cpu=1.7%, other=0.0%] (8.5ms out of 500ms) cpu usage by thread 'elasticsearch[api-es-data-z3-2][search][T#6]'
8/10 snapshots sharing following 2 elements
java.base@19.0.2/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
java.base@19.0.2/java.lang.Thread.run(Thread.java:1589)
1.5% [cpu=1.5%, other=0.0%] (7.3ms out of 500ms) cpu usage by thread 'elasticsearch[api-es-data-z3-2][search][T#27]'
3/10 snapshots sharing following 2 elements
java.base@19.0.2/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
java.base@19.0.2/java.lang.Thread.run(Thread.java:1589)
1.4% [cpu=1.4%, other=0.0%] (7ms out of 500ms) cpu usage by thread 'elasticsearch[api-es-data-z3-2][search][T#25]'
2/10 snapshots sharing following 2 elements
java.base@19.0.2/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
java.base@19.0.2/java.lang.Thread.run(Thread.java:1589)
1.4% [cpu=1.4%, other=0.0%] (6.8ms out of 500ms) cpu usage by thread 'elasticsearch[api-es-data-z3-2][search][T#30]'
5/10 snapshots sharing following 2 elements
java.base@19.0.2/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
java.base@19.0.2/java.lang.Thread.run(Thread.java:1589)
1.1% [cpu=1.1%, other=0.0%] (5.3ms out of 500ms) cpu usage by thread 'elasticsearch[api-es-data-z3-2][search][T#3]'
2/10 snapshots sharing following 2 elements
java.base@19.0.2/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
java.base@19.0.2/java.lang.Thread.run(Thread.java:1589)
0.8% [cpu=0.8%, other=0.0%] (4ms out of 500ms) cpu usage by thread 'elasticsearch[api-es-data-z3-2][search][T#15]'
10/10 snapshots sharing following 2 elements
java.base@19.0.2/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
java.base@19.0.2/java.lang.Thread.run(Thread.java:1589)
0.0% [cpu=0.0%, other=0.0%] (30.5micros out of 500ms) cpu usage by thread 'ticker-schedule-trigger-engine'
10/10 snapshots sharing following 3 elements
java.base@19.0.2/java.lang.Thread.sleep0(Native Method)
java.base@19.0.2/java.lang.Thread.sleep(Thread.java:465)
org.elasticsearch.xpack.watcher.trigger.schedule.engine.TickerScheduleTriggerEngine$Ticker.run(TickerScheduleTriggerEngine.java:195)
0.0% [cpu=0.0%, other=0.0%] (0s out of 500ms) cpu usage by thread 'Connection evictor'
10/10 snapshots sharing following 4 elements
java.base@19.0.2/java.lang.Thread.sleep0(Native Method)
java.base@19.0.2/java.lang.Thread.sleep(Thread.java:465)
org.apache.httpcomponents.httpclient@4.5.13/org.apache.http.impl.client.IdleConnectionEvictor$1.run(IdleConnectionEvictor.java:66)
java.base@19.0.2/java.lang.Thread.run(Thread.java:1589)
0.0% [cpu=0.0%, other=0.0%] (0s out of 500ms) cpu usage by thread 'x-pack-ml-controller-stderr-pump'
10/10 snapshots sharing following 17 elements
java.base@19.0.2/java.io.FileInputStream.readBytes(Native Method)
java.base@19.0.2/java.io.FileInputStream.read(FileInputStream.java:293)
java.base@19.0.2/java.io.BufferedInputStream.read1(BufferedInputStream.java:308)
java.base@19.0.2/java.io.BufferedInputStream.implRead(BufferedInputStream.java:382)
java.base@19.0.2/java.io.BufferedInputStream.read(BufferedInputStream.java:367)
java.base@19.0.2/sun.nio.cs.StreamDecoder.readBytes(StreamDecoder.java:333)
java.base@19.0.2/sun.nio.cs.StreamDecoder.implRead(StreamDecoder.java:376)
java.base@19.0.2/sun.nio.cs.StreamDecoder.lockedRead(StreamDecoder.java:219)
java.base@19.0.2/sun.nio.cs.StreamDecoder.read(StreamDecoder.java:173)
java.base@19.0.2/java.io.InputStreamReader.read(InputStreamReader.java:189)
java.base@19.0.2/java.io.BufferedReader.fill(BufferedReader.java:161)
java.base@19.0.2/java.io.BufferedReader.implReadLine(BufferedReader.java:371)
java.base@19.0.2/java.io.BufferedReader.readLine(BufferedReader.java:348)
java.base@19.0.2/java.io.BufferedReader.readLine(BufferedReader.java:437)
app/org.elasticsearch.server@8.6.2/org.elasticsearch.bootstrap.Spawner.lambda$startPumpThread$3(Spawner.java:108)
app/org.elasticsearch.server@8.6.2/org.elasticsearch.bootstrap.Spawner$$Lambda$361/0x000000080118eb90.run(Unknown Source)
java.base@19.0.2/java.lang.Thread.run(Thread.java:1589)
0.0% [cpu=0.0%, other=0.0%] (0s out of 500ms) cpu usage by thread 'x-pack-ml-controller-stdout-pump'
10/10 snapshots sharing following 17 elements
java.base@19.0.2/java.io.FileInputStream.readBytes(Native Method)
java.base@19.0.2/java.io.FileInputStream.read(FileInputStream.java:293)
java.base@19.0.2/java.io.BufferedInputStream.read1(BufferedInputStream.java:308)
java.base@19.0.2/java.io.BufferedInputStream.implRead(BufferedInputStream.java:382)
java.base@19.0.2/java.io.BufferedInputStream.read(BufferedInputStream.java:367)
java.base@19.0.2/sun.nio.cs.StreamDecoder.readBytes(StreamDecoder.java:333)
java.base@19.0.2/sun.nio.cs.StreamDecoder.implRead(StreamDecoder.java:376)
java.base@19.0.2/sun.nio.cs.StreamDecoder.lockedRead(StreamDecoder.java:219)
java.base@19.0.2/sun.nio.cs.StreamDecoder.read(StreamDecoder.java:173)
java.base@19.0.2/java.io.InputStreamReader.read(InputStreamReader.java:189)
java.base@19.0.2/java.io.BufferedReader.fill(BufferedReader.java:161)
java.base@19.0.2/java.io.BufferedReader.implReadLine(BufferedReader.java:371)
java.base@19.0.2/java.io.BufferedReader.readLine(BufferedReader.java:348)
java.base@19.0.2/java.io.BufferedReader.readLine(BufferedReader.java:437)
app/org.elasticsearch.server@8.6.2/org.elasticsearch.bootstrap.Spawner.lambda$startPumpThread$3(Spawner.java:108)
app/org.elasticsearch.server@8.6.2/org.elasticsearch.bootstrap.Spawner$$Lambda$361/0x000000080118eb90.run(Unknown Source)
java.base@19.0.2/java.lang.Thread.run(Thread.java:1589)
0.0% [cpu=0.0%, other=0.0%] (0s out of 500ms) cpu usage by thread 'process reaper (pid 122)'
10/10 snapshots sharing following 6 elements
java.base@19.0.2/java.lang.ProcessHandleImpl.waitForProcessExit0(Native Method)
java.base@19.0.2/java.lang.ProcessHandleImpl$1.run(ProcessHandleImpl.java:163)
java.base@19.0.2/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
java.base@19.0.2/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
java.base@19.0.2/java.lang.Thread.run(Thread.java:1589)
java.base@19.0.2/jdk.internal.misc.InnocuousThread.run(InnocuousThread.java:186)
0.0% [cpu=0.0%, other=0.0%] (0s out of 500ms) cpu usage by thread 'elasticsearch[api-es-data-z3-2][write][T#18]'
3/10 snapshots sharing following 2 elements
java.base@19.0.2/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
java.base@19.0.2/java.lang.Thread.run(Thread.java:1589)
0.0% [cpu=0.0%, other=0.0%] (0s out of 500ms) cpu usage by thread 'ml-cpp-log-tail-thread'
10/10 snapshots sharing following 6 elements
java.base@19.0.2/java.io.FileInputStream.readBytes(Native Method)
java.base@19.0.2/java.io.FileInputStream.read(FileInputStream.java:269)
org.elasticsearch.xpack.ml.process.logging.CppLogMessageHandler.tailStream(CppLogMessageHandler.java:107)
org.elasticsearch.xpack.ml.process.NativeController.lambda$tailLogsInThread$0(NativeController.java:97)
org.elasticsearch.xpack.ml.process.NativeController$$Lambda$4218/0x0000000801b137a8.run(Unknown Source)
java.base@19.0.2/java.lang.Thread.run(Thread.java:1589)
0.0% [cpu=0.0%, other=0.0%] (0s out of 500ms) cpu usage by thread 'elasticsearch[keepAlive/8.6.2]'
10/10 snapshots sharing following 8 elements
java.base@19.0.2/jdk.internal.misc.Unsafe.park(Native Method)
java.base@19.0.2/java.util.concurrent.locks.LockSupport.park(LockSupport.java:221)
java.base@19.0.2/java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire(AbstractQueuedSynchronizer.java:715)
java.base@19.0.2/java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1047)
java.base@19.0.2/java.util.concurrent.CountDownLatch.await(CountDownLatch.java:230)
app/org.elasticsearch.server@8.6.2/org.elasticsearch.bootstrap.Elasticsearch.lambda$new$2(Elasticsearch.java:428)
app/org.elasticsearch.server@8.6.2/org.elasticsearch.bootstrap.Elasticsearch$$Lambda$5890/0x0000000801f29720.run(Unknown Source)
java.base@19.0.2/java.lang.Thread.run(Thread.java:1589)
0.0% [cpu=0.0%, other=0.0%] (0s out of 500ms) cpu usage by thread 'Thread-1'
10/10 snapshots sharing following 8 elements
java.base@19.0.2/java.io.FileInputStream.readBytes(Native Method)
java.base@19.0.2/java.io.FileInputStream.read(FileInputStream.java:293)
java.base@19.0.2/java.io.BufferedInputStream.fill(BufferedInputStream.java:255)
java.base@19.0.2/java.io.BufferedInputStream.implRead(BufferedInputStream.java:289)
java.base@19.0.2/java.io.BufferedInputStream.read(BufferedInputStream.java:276)
app/org.elasticsearch.server@8.6.2/org.elasticsearch.bootstrap.Elasticsearch.lambda$startCliMonitorThread$0(Elasticsearch.java:346)
app/org.elasticsearch.server@8.6.2/org.elasticsearch.bootstrap.Elasticsearch$$Lambda$6891/0x0000000802116600.run(Unknown Source)
java.base@19.0.2/java.lang.Thread.run(Thread.java:1589)
You mention the transport layer timeouts - we are using an internally built elasticsearch plugin using the esplugin gradle featur - the plugin calls a small rest API living in the elastic cluster's kubernetes cluster on every search. The rest API then queries a smaller index within the same cluster to assist with security/filtering the data from our larger indices.
Would this be something that would fit that classification for the cancelled tasks if it timed out? IE - the plugin indefinitely waiting for a response from the rest API?
Hmm it doesn't, although this time that's the fault of the stack-collapsing behaviour of this API rather than anything you can change. But that's still pretty strange. Could you grab a few full thread dumps of the queueing node with jstack instead?
Are you only interacting with REST APIs, no transport ones? A custom plugin is kind of an all-bets-are-off situation, but I can't think of a way for a plugin to inject transport-layer timeouts into the processing of a search request.
Hmm it doesn't, although this time that's the fault of the stack-collapsing behaviour of this API rather than anything you can change. But that's still pretty strange. Could you grab a few full thread dumps of the queueing node with jstack instead?
Unfortunately one of our team members just bounced the node to resolve the queues, but once another queue forms, which im sure it will, I will try to use jstack on the queueing node and I will ping you with the output.
Are you only interacting with REST APIs, no transport ones? A custom plugin is kind of an all-bets-are-off situation, but I can't think of a way for a plugin to inject transport-layer timeouts into the processing of a search request.
Yep, totally understood that a custom plugin is outside of yall's bounds of support. Not entirely sure about the Transport API question but i dont think so? The flow goes
Search to "Core" Data node ----> Plugin invoked via search ----> Azure Load Balancer -------> call Rest API behind the LB ----> Rest API queries smaller ES cluster via k8s service discovery -------> Return result to plugin -----> Plugin concludes the original search on Core data node
I do think its worth honing in on the Azure load balancer. We use the same load balancer implementation for both the rest API invoked by the plugin, as well as for the elasticsearch node to node communication for each elasticsearch cluster (separate ports and load balancers for each)
I noticed that the documentation you linked explicitly mentioned that network devices that terminate idle connections can cause issues, and im seeing that the load balancer settings for keepIdle time is 4 minutes with a max of 30, and it can't be disabled.
Do you think its worth looking into the transport.ping_schedule config described here?
to try to fight this? Maybe increase the idleTimeout to 30 mins on the azure side and set a ping schedule to be less than the idleTimeout? We havent adjusted these settings since going to prod in 2021, but the circumstances led me here.
At the end of the day though, our primary concern is just clearing up these cancelled tasks. If theres anything we can do on our end to just terminate these things and cause them to go away, we can automate it and prevent the buildup. It doesn't seem that there's much we can do on the user end once cancelled is set to true by the Task Management API, which is already done for all of these.
I'd strongly recommend not putting a load balancer between ES nodes at all. It can cause a lot of trouble and has no meaningful benefits. transport.ping_schedule is kind of a workaround to get clusters out of a hole but it's not a great long-term solution IMO.
I hear you but I can't recommend any action without understanding the problem first.
Oh just to add that I can think of ways that a subtle bug in a plugin could cause the effect you describe, and have seen other plugins with such bugs in the past. I would expect such a bug to be reasonably obvious from the stack dump, if collected at the right time.
I'd strongly recommend not putting a load balancer between ES nodes at all. It can cause a lot of trouble and has no meaningful benefits. transport.ping_schedule is kind of a workaround to get clusters out of a hole but it's not a great long-term solution IMO.
Got it, thanks for the guidance on that point. We will stay away from this config then.
Oh just to add that I can think of ways that a subtle bug in a plugin could cause the effect you describe, and have seen other plugins with such bugs in the past. I would expect such a bug to be reasonably obvious from the stack dump, if collected at the right time.
Now this is encouraging to hear. A data point to probably make you aware of - we were able to significantly improve the issue over the weekend by adding maxIdleTime settings as well as custom error handling to the rest api invoked by our plugin to always return an empty response instead of a 500 status code if there were any exceptions. It was happening all over the place before, and once we did that, the issue became isolated to our cluster that has by far the most search traffic.
That brings me to the plugin implementation itself. I am wondering if perhaps the httpClient configured inside of it is not properly pooling connections, timing them out, or otherwise dealing with them in the proper way.
Are there any ways that I can enable logging on a custom plugin that you know of ? I've tried searching around and haven't found much. Our plugin is written in java 17 and compiled via gradle esplugin - can I add some options to the JVM flags to have it output logs to its respective data node whenever it encounters an exception or whenever some log4j statement gets invoked?
Feel free to disregard the code snippet below if this is too in the weeds for your domain , but this is the call that we think is timing out inside of the plugin and causing the queues:
public HttpAuthorizer(PluginConfig pluginConfig) {
this.pluginConfig = pluginConfig;
this.jsonSerde = new JsonSerde();
PoolingHttpClientConnectionManager poolingConnManager
= new PoolingHttpClientConnectionManager();
poolingConnManager.setMaxTotal(10);
RequestConfig requestConfig = RequestConfig.custom()
.setConnectTimeout(Math.toIntExact(pluginConfig.getConnectTimeoutMs()))
.build();
this.httpClient = HttpClients.custom().setConnectionManager(poolingConnManager)
.setDefaultRequestConfig(requestConfig)
.build();
}
@Override
public Boolean authorize(RecordAuthorizationRequest request) {
String json = jsonSerde.requestToJson(request);
return AccessController.doPrivileged((PrivilegedAction<Boolean>) () -> {
Boolean authorized = false; // Defaults to no access
HttpPost httpRequest = new HttpPost(URI.create(pluginConfig.getUri()));
httpRequest.setHeader("Content-Type", "application/json");
try {
httpRequest.setEntity(new StringEntity(json));
CloseableHttpResponse response;
try {
response = httpClient.execute(httpRequest);
} catch (Exception e) {
throw new SecurityPluginException("Unexpected exception retrieving permission data from data authorization service", e);
}
if (response.getStatusLine().getStatusCode() == 200) {
authorized = jsonSerde.parseResponse(EntityUtils.toString(response.getEntity(), "UTF-8"));
} else {
throw new SecurityPluginException(
"Unexpected http response status "
+ response.getStatusLine().getStatusCode()
+ " returned from authorization service. Response body "
+ response.getEntity().toString()
);
}
}
catch (IOException e) {
throw new SecurityPluginException("Exception forming Elasticsearch Authorization request", e);
}
return authorized;
});
}
Perhaps this implementation would have the issue, one thing that immediately came to my mind when I found this was that there is no requestConnectTimeout or socketTimeout configured on the http client. Our plugin had to switch libraries for this call when upgrading to elasticsearch 8 base which makes me suspect it and want to potentially log/track the connection stats and exceptions if possible.
Is this happening on a thread owned by Elasticsearch (e.g. a search or indexing thread). That's exactly the bug I was thinking of if so. You must never block any of Elasticsearch's threads on a remote request like this. Either fork this work onto to a thread that you own (and don't mind blocking) or perform your HTTP request with an async API.
This happens on a search thread i believe, yes. Every search sent by our clients invokes the plugin with their security related info to determine if they have access to the document or not, then that info passed to the plugin is sent through in the httpClient you linked there to our Rest API.
Either fork this work onto to a thread that you own (and don't mind blocking) or perform your HTTP request with an async API.
A bit unfamiliar with this topic but i will do some research. Is the forking of the work something that could be done within the same custom plugin without causing any issues, as long as i use a thread pool that isnt earmarked for searching and indexing?
It looks like the plugin is being classified as a "FilterScript" plugin and is using the HttpClient in I think what's called a LeafFactory, hopefully this can help fill in missing details:
public class SecurityPlugin extends Plugin implements ScriptPlugin, ActionPlugin {
private static final Logger log = LogManager.getLogger(SecurityPlugin.class);
public List<RestHandler> getRestHandlers(final Settings settings,
final RestController restController,
final ClusterSettings clusterSettings,
final IndexScopedSettings indexScopedSettings,
final SettingsFilter settingsFilter,
final IndexNameExpressionResolver indexNameExpressionResolver,
final Supplier<DiscoveryNodes> nodesInCluster) {
return Collections.singletonList(new CacheRefreshRestAction());
}
@Override
public ScriptEngine getScriptEngine(
Settings settings,
Collection<ScriptContext<?>> contexts
) {
return new SecurityScriptEngine();
}
private static class SecurityScriptEngine implements ScriptEngine {
private AuthorizationService authorizationService;
SecurityScriptEngine() {
authorizationService = AuthorizationServiceFactory.getInstance();
}
@Override
public String getType() {
return "hcp_security";
}
@Override
public <T> T compile(
String scriptName,
String scriptSource,
ScriptContext<T> context,
Map<String, String> params
) {
log.info("compile");
if (!context.equals(FilterScript.CONTEXT)) {
throw new IllegalArgumentException(getType()
+ " scripts cannot be used for context ["
+ context.name + "]");
}
// we use the script "source" as the script identifier
if ("apply_permissions".equals(scriptSource)) {
FilterScript.Factory factory = new SecurityPermissionsFactory(**authorizationService**);
return context.factoryClazz.cast(factory);
}
throw new IllegalArgumentException("Unknown script name "
+ scriptSource);
}
@Override
public void close() {
// optionally close resources
log.info("close");
}
@Override
public Set<ScriptContext<?>> getSupportedContexts() {
Set<ScriptContext<?>> s = new HashSet<>();
s.add(FilterScript.CONTEXT);
return s;
}
private static class SecurityPermissionsFactory implements FilterScript.Factory,
ScriptFactory {
private AuthorizationService authorizationService;
SecurityPermissionsFactory(AuthorizationService authorizationService) {
this.authorizationService = authorizationService;
}
@Override
public boolean isResultDeterministic() {
// need to verify how this works
return true;
}
@Override
public FilterScript.LeafFactory newFactory(
Map<String, Object> params,
SearchLookup lookup
) {
log.debug("newFactory");
return new SecurityPermissionsLeafFactory(params, lookup, authorizationService);
}
}
private static class SecurityPermissionsLeafFactory implements FilterScript.LeafFactory {
private final Map<String, Object> params;
private final SearchLookup lookup;
private final String consumerName;
private final List<String> roles;
private final AuthorizationService authorizationService;
@SuppressWarnings("unchecked")
private SecurityPermissionsLeafFactory(
Map<String, Object> params, SearchLookup lookup, AuthorizationService authorizationService) {
if (!params.containsKey("consumer_name")) {
throw new IllegalArgumentException(
"Missing parameter [consumerName]");
}
this.params = params;
this.lookup = lookup;
consumerName = params.get("consumer_name").toString();
roles = params.get("roles") != null ? (List<String>) params.get("roles") : null;
this.authorizationService = authorizationService;
}
@Override
public FilterScript newInstance(DocReader docReader) throws IOException {
log.debug("newInstance");
/*
* the field and/or term don't exist in this segment,
* so always return 0
*/
return new FilterScript(params, lookup, docReader) {
@Override
public boolean execute() {
log.debug("execute");
Map<String, ScriptDocValues<?>> docMap = getDoc();
return authorizationService.isAuthorized(
consumerName,
roles,
getString(docMap, "security.securitySourceSystemCode"),
getStringList(docMap, "security.securityPermission.securityPermissionValue"),
getStringList(docMap, "security.securityPermissionAny.securityPermissionAnyValue")
);
}
private String getString(Map<String, ScriptDocValues<?>> docMap, String key) {
if (docMap.containsKey(key)) {
return ((ScriptDocValues.Strings) docMap.get(key)).getValue();
} else {
return null;
}
}
@SuppressWarnings("unchecked")
private List<String> getStringList(Map<String, ScriptDocValues<?>> docMap, String key) {
if (docMap.containsKey(key)) {
return ((List<String>) docMap.get(key));
} else {
return null;
}
}
};
}
}
}
}
You must not block any of Elasticsearch's threads, including (but not limited to) the search and indexing ones. The best approach is to go async (threads are not cheap and blocking them on network calls is very wasteful) but if you must block then do it on your own thread.
Got it, thank you so much for your time on this. Although its not going to be an easy lift, it helps a ton to know that you have seen similar situations with custom plugins so we know to hone in on this for the troubleshooting. We were grasping at straws looking for the root cause all the past weeks.
We fixed this issue by adding aggressive timeouts and stale connection management to the aforementioned http clients in the custom plugin. The root cause diagnosis on the blocking of threads by David does appear to be the root cause for the queue backlog.
Thanks again for all your help on this @DavidTurner. Marking this resolved.
Apache, Apache Lucene, Apache Hadoop, Hadoop, HDFS and the yellow elephant
logo are trademarks of the
Apache Software Foundation
in the United States and/or other countries.
