nssuresh2007
commented
2 years ago Following are some detailed thoughts on proposed approach:
Problem Statement
Many times, a single search query which is resource intensive can guzzle a lot of resources and a bunch of such queries can degrade the performance of the cluster. Currently we do not have a mechanism to identify and terminate the problematic queries when a node is in duress. Existing mechanisms like circuit breaker, thread pool size threshold act as a blanket mechanism and does not specifically target the problematic queries alone.
Goals
Milestone 1: Identify and reject the on-going resource intensive tasks on a shard/node if they have breached limits and does not recover within a certain threshold. It only rejects the task on a particular shard and other shard tasks can still execute successfully. Milestone 2: In previous milestone, we are only tracking shard level tasks. Now, include tracking for co-ordinator task resource consumption and cancellation logic based on the threshold. Milestone 3: Build aggregated view of resource consumption for a query by rolling up the consumption stats from all the shards and aggregating them under the parent task id. This aggregated view can be used to build guard-rails to track and cancel the request which consumes lots of resources across nodes.
Non-Goals
We are not targeting to build backpressure for spikes in search request rate as a part of this task. It would be handled as a part of rejection of incoming requests (https://github.com/opensearch-project/OpenSearch/issues/1180) task.
Key Considerations
- Resource Tracking Framework would be utilized to provides stats on task resource consumption.
- Initially, we are targeting the shard search task and not the co-ordinator task.
Proposed Approach
[Following sections describe the approach for Milestone 1]
Measure the resource consumption (CPU, Heap Memory) at frequent checkpoints within query phase of shard search request. If the node is in duress (JVM MP above threshold, CPU Utilization reached threshold) and if the total heap memory occupied by search shard tasks is >= 5% of total heap, then check the following criteria for each Search Task — CPU cycles spent, heap memory occupied by the task. If the task has been exceeded CPU cycles threshold and is among the top tasks based on heap memory occupied with huge variance from average resource consumption, then we will cancel the search task.
Different checkpoints to consider in Query Phase
Approach 1 - Using multiple checkpoints to track within the same task thread: Query phase in each search shard task undergoes different sub-phases (Pre-aggregation, Search, Rescoring, Suggest, Post-aggregation) and we will be checkpointing after each phase is completed. Among these, search phase does the actual lucene search and is very intensive. Hence we cannot checkpoint only after the search is completed and we will add a cancellable callback which would periodically checkpoint during actual search execution itself.
Approach 2 - Using separate observer thread We will be using separate observer thread to monitor the tasks at a fixed frequency. We will not evaluate through different checkpoints, but track tasks at a fixed frequency.
Deciding if node is in duress
Current JVM MP on the node and CPU utilization are used as criteria to determine if the node is in duress.
Identifying top resource consuming tasks
When TaskResourceTrackingService measures the resource stats, it will also keep track of top-N tasks based on the heap memory consumption. This would be used to identify and cancel the top resource intensive tasks if the variance is considerably higher.
Why not add cancellation logic in Fetch phase also?
Every search request goes through two phases - Query and Fetch phase. Query phase is responsible for doing the actual search and get the matching document ids from each shard. Fetch phase enriches the document ids with document information. Query phase is usually very heavy and resource consumption varies depending upon the nature of the query and the workload and hence we track query phase extensively. Also, once query phase is completed, search query is about to get finished and we do not want to cancel it which would result in wastage of the resource that executed till now.
PoC Testing
Did code changes for PoC Testing which included following logic — Heap to track the top-N requests, measure resource utilization after every sub-phase, cancel the top most resource consuming query. (Did not include logic for duration of running request, variance logic)
Executed two different types of queries - Light and heavy as follows:
Light weight query:
curl "localhost:9200/_search?q=*:*"Comparatively heavy aggregation query:
curl -X GET "localhost:9200/_search?pretty" -H 'Content-Type: application/json'
-d'{"query": {"range":{"dropoff_datetime":{"from":"01/01/2015","to":"21/01/2015",
"include_lower":true,"include_upper":true,"format":"dd/MM/yyyy","boost":1.0}}},
"aggregations":{"dropoffs_over_time":{"date_histogram":{"field":"dropoff_datetime",
"calendar_interval":"day","offset":0,"order":{"_key":"asc"},"keyed":false,"min_doc_count":0}}}}'While the queries were getting executed, the top queries consuming lot of heap were getting cancelled as below, whereas the light-weight queries were always successful:
[2022-07-28T18:29:08,400][INFO ][o.o.s.q.QueryPhase ] [ip-172-31-51-111.us-west-2.compute.internal] This task 1317 is currently the highest resource consuming task and it is being cancelled
[2022-07-28T18:29:08,401][TRACE][o.o.s.SearchService ] [ip-172-31-51-111.us-west-2.compute.internal] Query phase failed
org.opensearch.tasks.TaskCancelledException: Task is cancelled due to high resource consumption
at org.opensearch.search.query.QueryPhase.measureResourceConsumption(QueryPhase.java:157) ~[opensearch-2.1.0.jar:2.1.0-SNAPSHOT]
at org.opensearch.search.query.QueryPhase.execute(QueryPhase.java:187) ~[opensearch-2.1.0.jar:2.1.0-SNAPSHOT]
at org.opensearch.search.SearchService.loadOrExecuteQueryPhase(SearchService.java:455) ~[opensearch-2.1.0.jar:2.1.0-SNAPSHOT]
at org.opensearch.search.SearchService.executeQueryPhase(SearchService.java:523) [opensearch-2.1.0.jar:2.1.0-SNAPSHOT]
at org.opensearch.search.SearchService$2.lambda$onResponse$0(SearchService.java:490) [opensearch-2.1.0.jar:2.1.0-SNAPSHOT]
at org.opensearch.action.ActionRunnable.lambda$supply$0(ActionRunnable.java:73) [opensearch-2.1.0.jar:2.1.0-SNAPSHOT]
at org.opensearch.action.ActionRunnable$2.doRun(ActionRunnable.java:88) [opensearch-2.1.0.jar:2.1.0-SNAPSHOT]
at org.opensearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:52) [opensearch-2.1.0.jar:2.1.0-SNAPSHOT]
at org.opensearch.threadpool.TaskAwareRunnable.doRun(TaskAwareRunnable.java:78) [opensearch-2.1.0.jar:2.1.0-SNAPSHOT]
at org.opensearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:52) [opensearch-2.1.0.jar:2.1.0-SNAPSHOT]
at org.opensearch.common.util.concurrent.TimedRunnable.doRun(TimedRunnable.java:59) [opensearch-2.1.0.jar:2.1.0-SNAPSHOT]
at org.opensearch.common.util.concurrent.ThreadContext$ContextPreservingAbstractRunnable.doRun(ThreadContext.java:806) [opensearch-2.1.0.jar:2.1.0-SNAPSHOT]
at org.opensearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:52) [opensearch-2.1.0.jar:2.1.0-SNAPSHOT]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) [?:?]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) [?:?]
at java.lang.Thread.run(Thread.java:833) [?:?][ec2-user@ip-172-31-51-111 opensearch]$ curl -X GET "localhost:9200/_search?pretty"
-H 'Content-Type: application/json'
-d'{"query": {"range":{"dropoff_datetime":{"from":"01/01/2015","to":"21/01/2015",
"include_lower":true,"include_upper":true,"format":"dd/MM/yyyy","boost":1.0}}},
"aggregations":{"dropoffs_over_time":{"date_histogram":{"field":"dropoff_datetime",
"calendar_interval":"day","offset":0,"order":{"_key":"asc"},"keyed":false,"min_doc_count":0}}}}'
{
"error" : {
"root_cause" : [
{
"type" : "task_cancelled_exception",
"reason" : "Task is cancelled due to high resource consumption"
}
],
"type" : "search_phase_execution_exception",
"reason" : "all shards failed",
"phase" : "query",
"grouped" : true,
"failed_shards" : [
{
"shard" : 0,
"index" : "nyc_taxis",
"node" : "MGkMkg9wREW3IVewZ7U_jw",
"reason" : {
"type" : "task_cancelled_exception",
"reason" : "Task is cancelled due to high resource consumption"
}
}
]
},
"status" : 500
}Other Approaches considered
- Measure resource consumption and add cancellation logic at end of each phase/sub-phases.
- As mentioned above, Query Phase does all the heavy-lifting and measuring at the end of the phase is too late to cancel the request.
- Cancelling based on fixed thresholds for per-request resource consumption.
- Each workload is different based on index and hardware configuration and fixed thresholds for per-request resource consumption might not be apt for all diverse workloads.
Points to Note
- [Milestone 1] If the search task is cancelled, it should be retried on another replica.
- [Milestone 1] Should the tracking of top-N requests can be moved to shard level, instead of node level? — Tracking at shard level for top-N might be over-complicating and does not add much value. Instead, we can just track the total heap memory occupied by search requests at shard level alone.
- [Milestone 3] We can also expose configuration for users to set a threshold for resource consumption beyond which the request will be cancelled.
- [Milestone 3] Cancelled queries can be added to slow logs, so that it is easier for users to debug.
Open Items
- From where can we measure the resource consumption for currently executing tasks? Should it be done on a separate observer thread? Or, can it happen on the same tasks’ thread?
- Separate observer thread: Using a separate observer thread to checkpoint resource consumption periodically at a fixed frequency. This would avoid any performance impact due to measuring resource consumption since it runs on a separate thread. It also provides improved code maintainability, since it does not require code interleaving within search logic.
- Using same task thread: We can measure resource consumption of task within the same task itself at different milestones. Additional overhead of measuring resource consumption needs to be handled by same task thread.
I will evaluate more on both the approaches and more details.
Metrics to be added
Following would be rolling window metric for every 1 minute and would be exposed through stats API.
ExpensiveTaskCancelledCount- Counter which indicates the number of expensive tasks that are cancelled.TopNTaskCPUUtilization- Absolute value of CPU utilization for all top-N requests for every minute.TopNTaskHeapMem- Absolute value of Heap Mem utilization for all top-N requests for every minute.
Additional changes
- Shard level tracking of resource consumption — Resource consumption is currently measured per task and we do not build a shard-specific view to see how much heap memory is occupied by each shard. Similar to Indexing Backpressure (https://github.com/opensearch-project/OpenSearch/issues/478), we can cancel the requests based on the pressure on a shard.
Please share your thoughts on the above.
Bukhtawar
reta
dblock
dreamer-89
elfisher
kgcreative
rohin
anasalkouz
kartg
PritLadani
DarshitChanpura
Is your feature request related to a problem? Please describe.
1042 aims to build back-pressure support for Search requests. This will help in recovering a node which is running short on system resources and the already running search requests are not finishing and making things worse.
Describe the solution you'd like
Cancelling on-going most resource intensive search requests on a shard/node, if the resource limits for that shard/node have started breaching the assigned limits (https://github.com/opensearch-project/OpenSearch/issues/1180), and there is no recovery for a certain time threshold. The back-pressure model should support identification of queries which are most resource guzzling with minimal wasteful work. These can then be cancelled for recovering a node under load and continue doing useful work.
Describe alternatives you've considered A clear and concise description of any alternative solutions or features you've considered.
Additional context https://github.com/opensearch-project/OpenSearch/issues/1180 - This issue covers rejection of incoming requests.