[Data] Warn when configured memory is below 1.25x max USS - #65512
[Data] Warn when configured memory is below 1.25x max USS#65512viiccwen wants to merge 2 commits into
Conversation
Signed-off-by: viiccwen <vicwen@apache.org>
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a mechanism to run issue detectors at the end of successful execution using final metrics, specifically implementing a high-memory detector that warns if an operator's memory usage exceeded its configured logical memory. The feedback suggests two improvements: first, saving and restoring hanging metrics in invoke_detectors_on_execution_end to prevent them from being wiped out by _report_issues; second, using getattr defensively when accessing op.metrics.max_uss_bytes to avoid potential AttributeErrors in mock or test environments.
Signed-off-by: viiccwen <vicwen@apache.org>
|
cc @bveeramani, PTAL. Thanks! 🙌 |
ray-gardener
Bot
added
data
bveeramani
left a comment
bveeramani
left a comment
There was a problem hiding this comment.
Left some high-level thoughts.
Would you mind adding an example warning to the PR description?
| # Final detectors don't re-evaluate hanging issues, so preserve their state. | ||
| hanging_metrics = { | ||
| operator: operator.metrics._issue_detector_hanging | ||
| for operator in self.executor._topology | ||
| } | ||
| try: | ||
| self._report_issues(issues) | ||
| finally: | ||
| for operator, value in hanging_metrics.items(): | ||
| operator.metrics._issue_detector_hanging = value | ||
|
|
There was a problem hiding this comment.
What breaks if we remove this?
This code breaks abstractions barriers. We're adding code to the generic manager class that's specific to a particular issue detector implementation. In general, I think we should avoid this unless there's an extremely compelling reason
| def detect(self) -> List[Issue]: | ||
| pass | ||
|
|
||
| def detect_on_execution_end(self) -> List[Issue]: |
There was a problem hiding this comment.
The IssueDetector class is starting to look like a really shallow abstraction on top of ExecutionCallback, and I'm wondering if there's any reason we still need the IssueDetector abstraction at all (as opposed to using ExecutionCallback + logs directly).
In the meantime, I think we can architect this in a way that avoids widening the IssueDetector interface.
Rather than adding a method to the IssueDetector interface, I'm thinking we just extend the implementation for the HighMemoryDetector. Here's the pseudocode I have in mind:
Class HighMemoryDetector:
def detect(...)
For op in ops:
If op.completed() and not self._has_emitted_completion_issue[op]:
# Handle final case. Should only emit once for completed op.
elif not op.completed()
# Handle regular case
And then on the IssueDetectorManager side, you would just want to call detect() on execution success or fail.
Do you forsee and issues with this approach?
This would emit the warning when an operator completes rather than when execution completes, but I think that's okay for a simpler interface
2 participants
Description
Add an execution-end hook to the issue detector lifecycle. After a successful execution, the
high_memory_detectorwarns for each map operator with an observed maximum USS when logical memory isn't configured or is less than 125% of that maximum.The warning shows the maximum worker memory, whether memory was configured, and an exact
memory=<bytes>recommendation.Related issues
closes #65511.
Additional information
The 125% comparison and recommendation use integer arithmetic, so I round up to a whole number of bytes. : )