[df] Export more metrics from query engine#4983
Conversation
This PR introduces ratelimiting knobs for datafusion queries as an emergency mechanism to control the rate of DF queries if they end up putting the system under pressure. By default, this rate limiting is disabled. Along the way, the PR also translates some client side DF errors to be 400s (bad request) instead of 500s.
This PR introduces two main metrics for datafusion. 1. Latency tracking for the `/query` endpoint. Although, the exported histograms will be influnced by the query complexity submitted by the user, I found that if I'm adding num /query calls and a metric for slow durations, I might as well just track the whole endpoint latency which will give us both and a couple more. We also typically have baseline query load (for exporting metrics) which will at least help set the baseline for such histograms. 2. I've also added tracking for the number of active local scanner that we have (guage and counters) that might help track if we're for example leaking scanner (due to aborted original request). It might be useful to export the number of scanner per table, put I would like more opinions on whether this is worth the extra metrics dimensions or not. Now, for the things that I didn't add: 1. I was planning to report the active time that the scanner spent actively reading. However, this turned out to be tricky because scanners are long lived throughout the query's lifetime. So capturing their lifetime will also capture the time they're not polled (e.g. between batches RPCs from the remote scanner). I considered instrumenting the handling for the NextBatch RPC from the remote scanner manager, but that wouldn't cover local scans. I think our rocksb metrics in background iterator might be good enough to capture the cost of the scans themseleves? 2. I was planning to log some metrics from the query coordinator about the number of batches/rows and latency for remote scanner, etc. But it turned out that you can get those on demand from DF with `explin analyze $query` so ditched those as well.
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 0804709e18
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| options.rate_limiting.as_ref().map(|limit| { | ||
| RateLimiter::new(gardal::Limit::from(limit.clone()), gardal::TokioClock) | ||
| }), |
There was a problem hiding this comment.
When rate limiting is configured, this allocates a fresh bucket for every QueryContext::create call. The admin REST context is attached to AdminService from crates/node/src/roles/admin.rs:143-156, while the cluster-controller builds a separate context from the same options.admin.query_engine in crates/admin/src/cluster_controller/service.rs:130-131; with both services enabled, a rate = "1/s" limit permits one REST query plus one gRPC query per second on the same node instead of a single per-node budget.
Useful? React with 👍 / 👎.
| if let Some(limiter) = self.rate_limiter.as_ref() { | ||
| limiter.try_consume_one()?; | ||
| } |
There was a problem hiding this comment.
Because the token is consumed inside QueryContext::execute, the background storage accounting task also spends the same bucket: AdminRole clones this context into the REST service and then moves it into StorageAccountingTask (crates/node/src/roles/admin.rs:156 and :179-182), and the task calls .execute(STORAGE_QUERY) in crates/admin/src/storage_accounting.rs:84-87. With accounting enabled and a low query limit, the periodic metrics query can either be rate-limited itself or consume the token that should be reserved for user /query requests.
Useful? React with 👍 / 👎.
muhamadazmy
left a comment
muhamadazmy
left a comment
There was a problem hiding this comment.
Thank you so much @MohamedBassem for this PR. The changes looks good to me but I left couple of nits and requests.
| let duration_guard = QueryDurationGuard { | ||
| start: Instant::now(), | ||
| }; |
There was a problem hiding this comment.
nit: Maybe a QueryDurationGuard::new() is nicer ?
|
|
||
| /// Records the elapsed time since `start` into the query-duration histogram when dropped. | ||
| struct QueryDurationGuard { | ||
| start: Instant, |
There was a problem hiding this comment.
Use our MillisSinceEpoch instead
|
|
||
| impl Drop for QueryDurationGuard { | ||
| fn drop(&mut self) { | ||
| histogram!(QUERY_ENGINE_QUERY_DURATION_SECONDS).record(self.start.elapsed().as_secs_f64()); |
There was a problem hiding this comment.
I am still not entirely sure about this metric. You already mentioned in the commit message that this can/will be skewed by the complexity of the query. But will also be skewed by slow clients, network speed, or cancelled queries (users dropping the stream before draining it completely).
What I am trying to say it's heavily affected by user behaviour, and worried it can be misleading in some situations.
What might be useful imho is a metric that measure latency of query_context.execute then time of first record or TTFB.
@AhmedSoliman what do you hink?
There was a problem hiding this comment.
Ah, I only considered the complexity of the query, but didn't consider the client's speed in draining the stream. That's a good point. We can definitely do TTFB, but then it wouldn't say much about how expensive the query is. I assume that maybe I can try to track the time we spent blocked on the record batch stream, but I don't know if this is going to be worth the complexity.
| mut address_book: AddressBook, | ||
| ) -> Result<Self, BuildError> { | ||
| metric_definitions::describe_metrics(); | ||
| restate_storage_query_datafusion::describe_metrics(); |
There was a problem hiding this comment.
I would avoid calling this from node crate tbh. describe_metric() should be private and internal to the crate.
I understand that datafusion has no singleton object to call this one on but it's idempotent and maybe we can just call it when creating a context (we only have 2 of these)
|
|
||
| impl InstrumentedRecordBatchStream { | ||
| fn new(inner: SendableRecordBatchStream) -> Self { | ||
| gauge!(DATAFUSION_NUM_ACTIVE_SCANNERS).increment(1); |
There was a problem hiding this comment.
Gauges are not very useful because they can be faster than the scrape time. A common pattern that we use is to introduce 2 counters, one for "creation" and the other for "drop" (you already have the creation one DATAFUSION_NUM_SCANNERS)
With the 2 counters, we can always see how many alive scanners at any point in time (created - dropped) and rate of new scanners.
Also with the InstrumentedRecordBatchStream we can also have histogram for the scanner age (not sure if it's useful though, up to you)
There was a problem hiding this comment.
I like the two counters idea, that's brilliant :)
As for adding a histogram for the batch stream, the problem is that their lifetime is tied to the lifetime of the overall parent query rather than the active time they spent scanning. Every remote scanner opens a scanner stream when it gets the Open RPC, and keeps it open over the duration of the multiple NextBatch and up until the End RPC. So I though that wouldn't be very useful.
2 participants
This PR introduces two main metrics for datafusion.
Latency tracking for the
/queryendpoint. Although, the exported histograms will be influnced by the query complexity submitted by the user, I found that if I'm adding num /query calls and a metric for slow durations,I might as well just track the whole endpoint latency which will give us both and a couple more. We also typically have baseline query load (for exporting metrics) which will at least help set the baseline for such histograms.
I've also added tracking for the number of active local scanner that we have (guage and counters) that might help track if we're for example leaking scanner (due to aborted original request). It might be useful to export
the number of scanner per table, put I would like more opinions on whether this is worth the extra metrics dimensions or not.
Now, for the things that I didn't add:
capture the time they're not polled (e.g. between batches RPCs from the remote scanner). I considered instrumenting the handling for the NextBatch RPC from the remote scanner manager, but that wouldn't cover local scans. I think
our rocksb metrics in background iterator might be good enough to capture the cost of the scans themseleves?
explin analyze $queryso ditchedthose as well.
Stack created with Sapling. Best reviewed with ReviewStack.