feat: Add observability for query state transitions in Presto #26418
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
… Presto Summary: Adds comprehensive monitoring for query state transitions to detect and diagnose queries spending abnormally long time in DISPATCHING and FINISHING states. This addresses issues observed on DKL cluster where queries were getting stuck during dispatch and commit phases (ref: S561047). Differential Revision: D85353849
Reviewer's GuideThis PR introduces a QueryStateTransitionMonitor to record query state transition durations and detect anomalies, integrates monitoring hooks into QueryStateMachine, binds and exports the monitor metrics in the server module, and adds comprehensive unit tests for all monitoring scenarios. Sequence diagram for query state transition monitoring integrationsequenceDiagram
participant QSM as QueryStateMachine
participant Monitor as QueryStateTransitionMonitor
participant Tracker as QueryTransitionTracker
QSM->>Monitor: setStateTransitionMonitor(monitor)
Monitor->>QSM: registerQuery(queryId)
QSM->>QSM: addStateChangeListener(trackStateTransition)
QSM->>QSM: trackStateTransition(newState)
QSM->>Monitor: recordStateTransition(queryId, prevState, newState, durationMillis)
Monitor->>Tracker: recordTransition(prevState, newState, durationMillis)
alt Anomaly detected
Monitor->>Monitor: checkAndLogAnomaly(...)
end
alt Query done
Monitor->>Monitor: activeQueries.remove(queryId)
end
Class diagram for QueryStateTransitionMonitor and integration with QueryStateMachineclassDiagram
class QueryStateTransitionMonitor {
+registerQuery(queryId)
+recordStateTransition(queryId, fromState, toState, durationMillis)
+getTotalQueriesTracked()
+getActiveQueriesCount()
+getDispatchingTimeStats()
+getFinishingTimeStats()
+getPlanningTimeStats()
+getRunningTimeStats()
+getAnomalousDispatchingCount()
+getAnomalousFinishingCount()
+getAnomalousPlanningCount()
+getAnomalousRunningCount()
+getStatsSummary()
-activeQueries: Map<QueryId, QueryTransitionTracker>
-dispatchingTimeStats: TimeStat
-finishingTimeStats: TimeStat
-planningTimeStats: TimeStat
-runningTimeStats: TimeStat
-anomalousDispatchingCount: CounterStat
-anomalousFinishingCount: CounterStat
-anomalousPlanningCount: CounterStat
-anomalousRunningCount: CounterStat
-totalQueriesTracked: AtomicLong
}
class QueryTransitionTracker {
-queryId: QueryId
-lastState: QueryState
-lastTransitionTimeMillis: long
+recordTransition(fromState, toState, durationMillis)
+getLastState()
+getLastTransitionTimeMillis()
}
class QueryStateMachine {
-stateTransitionMonitor: AtomicReference<QueryStateTransitionMonitor>
-previousState: AtomicReference<QueryState>
-previousStateTimestamp: AtomicLong
+setStateTransitionMonitor(monitor)
+trackStateTransition(newState)
}
QueryStateTransitionMonitor "1" *-- "many" QueryTransitionTracker
QueryStateMachine "1" o-- "1" QueryStateTransitionMonitor
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey there - I've reviewed your changes and they look great!
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location> `presto-main-base/src/main/java/com/facebook/presto/execution/QueryStateMachine.java:1013-1023` </location>
<code_context>
+
+ QueryState prevState = previousState.get();
+ long prevTimestamp = previousStateTimestamp.get();
+ long currentTimestamp = System.currentTimeMillis();
+ long durationMillis = currentTimestamp - prevTimestamp;
+
+ if (prevState != null) {
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Using System.currentTimeMillis() for duration may be affected by system clock changes.
System.nanoTime() is preferred for measuring elapsed time, as it is not affected by system clock changes.
```suggestion
QueryState prevState = previousState.get();
long prevTimestampNanos = previousStateTimestamp.get();
long currentTimestampNanos = System.nanoTime();
long durationMillis = (currentTimestampNanos - prevTimestampNanos) / 1_000_000;
if (prevState != null) {
monitor.recordStateTransition(queryId, prevState, newState, durationMillis);
}
previousState.set(newState);
previousStateTimestamp.set(currentTimestampNanos);
```
</issue_to_address>
### Comment 2
<location> `presto-main-base/src/test/java/com/facebook/presto/execution/TestQueryStateTransitionMonitor.java:50-59` </location>
<code_context>
+ assertNotNull(monitor.getStatsSummary());
+ }
+
+ @Test
+ public void testConcurrentQueryRegistration()
+ {
+ int numQueries = 100;
+ for (int i = 0; i < numQueries; i++) {
+ QueryId queryId = new QueryId("concurrent_query_" + i);
+ monitor.registerQuery(queryId);
+ monitor.recordStateTransition(queryId, DISPATCHING, PLANNING, 1000L + i);
+ }
+
+ assertEquals(monitor.getTotalQueriesTracked(), numQueries);
+ }
+
+ @Test
+ public void testStateTransitionWithZeroDuration()
+ {
+ monitor.registerQuery(testQueryId);
</code_context>
<issue_to_address>
**suggestion (testing):** Edge case for negative duration is not tested.
Please add a test for negative duration values in state transitions to verify the monitor's behavior in these cases.
Suggested implementation:
```java
@Test
public void testStateTransitionWithNegativeDuration()
{
monitor.registerQuery(testQueryId);
// Attempt to record a state transition with a negative duration
monitor.recordStateTransition(testQueryId, DISPATCHING, PLANNING, -500L);
// Assert that the query is still tracked and no exception is thrown
assertEquals(monitor.getTotalQueriesTracked(), 1);
assertEquals(monitor.getActiveQueriesCount(), 1);
// Optionally, check stats summary if available
assertNotNull(monitor.getStatsSummary());
}
@Test
public void testStateTransitionWithZeroDuration()
{
monitor.registerQuery(testQueryId);
```
If the monitor is expected to handle negative durations in a specific way (e.g., clamp to zero, ignore, or log a warning), you may want to add more specific assertions based on the implementation of `recordStateTransition` and `getStatsSummary()`. Adjust the assertions as needed to match the expected behavior.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
+1013
to
+1023
| QueryState prevState = previousState.get(); | ||
| long prevTimestamp = previousStateTimestamp.get(); | ||
| long currentTimestamp = System.currentTimeMillis(); | ||
| long durationMillis = currentTimestamp - prevTimestamp; | ||
|
|
||
| if (prevState != null) { | ||
| monitor.recordStateTransition(queryId, prevState, newState, durationMillis); | ||
| } | ||
|
|
||
| previousState.set(newState); | ||
| previousStateTimestamp.set(currentTimestamp); |
There was a problem hiding this comment.
suggestion (bug_risk): Using System.currentTimeMillis() for duration may be affected by system clock changes.
System.nanoTime() is preferred for measuring elapsed time, as it is not affected by system clock changes.
Suggested change
| QueryState prevState = previousState.get(); | |
| long prevTimestamp = previousStateTimestamp.get(); | |
| long currentTimestamp = System.currentTimeMillis(); | |
| long durationMillis = currentTimestamp - prevTimestamp; | |
| if (prevState != null) { | |
| monitor.recordStateTransition(queryId, prevState, newState, durationMillis); | |
| } | |
| previousState.set(newState); | |
| previousStateTimestamp.set(currentTimestamp); | |
| QueryState prevState = previousState.get(); | |
| long prevTimestampNanos = previousStateTimestamp.get(); | |
| long currentTimestampNanos = System.nanoTime(); | |
| long durationMillis = (currentTimestampNanos - prevTimestampNanos) / 1_000_000; | |
| if (prevState != null) { | |
| monitor.recordStateTransition(queryId, prevState, newState, durationMillis); | |
| } | |
| previousState.set(newState); | |
| previousStateTimestamp.set(currentTimestampNanos); |
Dependency Review✅ No vulnerabilities or license issues or OpenSSF Scorecard issues found.Scanned FilesNone |
abhinavmuk04
changed the title
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Summary:
Adds comprehensive monitoring for query state transitions to detect and diagnose
queries spending abnormally long time in DISPATCHING and FINISHING states. This
addresses issues observed on DKL cluster where queries were getting stuck during
dispatch and commit phases (ref: S561047).
Differential Revision: D85353849