[data] reset cpu + gpu metrics on executor shutdown and updating task submission/block generation metrics (ray-project#57246)

<!-- Thank you for your contribution! Please review
https://github.com/ray-project/ray/blob/master/CONTRIBUTING.rst before
opening a pull request. -->

<!-- Please add a reviewer to the assignee section when you create a PR.
If you don't have the access to it, we will shortly find a reviewer and
assign them to your PR. -->

## Why are these changes needed?
On executor shutdown, the metrics persist even after execution. The plan
is to reset on streaming_executor.shutdown. This PR also includes 2
potential drive-by fixes for metric calculation
<!-- Please give a short summary of the change and the problem this
solves. -->

## Related issue number

<!-- For example: "Closes ray-project#1234" -->

## Checks

- [ ] I've signed off every commit(by using the -s flag, i.e., `git
commit -s`) in this PR.
- [ ] I've run pre-commit jobs to lint the changes in this PR.
([pre-commit
setup](https://docs.ray.io/en/latest/ray-contribute/getting-involved.html#lint-and-formatting))
- [ ] I've included any doc changes needed for
https://docs.ray.io/en/master/.
- [ ] I've added any new APIs to the API Reference. For example, if I
added a
method in Tune, I've added it in `doc/source/tune/api/` under the
           corresponding `.rst` file.
- [ ] I've made sure the tests are passing. Note that there might be a
few flaky tests, see the recent failures at https://flakey-tests.ray.io/
- Testing Strategy
   - [ ] Unit tests
   - [ ] Release tests
   - [ ] This PR is not tested :(

---------

Signed-off-by: iamjustinhsu <[email protected]>
Signed-off-by: Future-Outlier <[email protected]>

Author: iamjustinhsu

python/ray/dashboard/modules/metrics/dashboards/data_dashboard_panels.py

@@ -453,7 +453,7 @@
     unit="s",
     targets=[
         Target(
-            expr='sum(ray_data_block_generation_time{{{global_filters}, operator=~"$Operator"}}) by (dataset, operator)',
+            expr='increase(ray_data_block_generation_time{{{global_filters}, operator=~"$Operator"}}[5m]) / increase(ray_data_num_task_outputs_generated{{{global_filters}, operator=~"$Operator"}}[5m])',
             legend="Block Generation Time: {{dataset}}, {{operator}}",
         )
     ],
@@ -468,7 +468,7 @@
     unit="s",
     targets=[
         Target(
-            expr='sum(ray_data_task_submission_backpressure_time{{{global_filters}, operator=~"$Operator"}}) by (dataset, operator)',
+            expr='increase(ray_data_task_submission_backpressure_time{{{global_filters}, operator=~"$Operator"}}[5m]) / increase(ray_data_num_tasks_submitted{{{global_filters}, operator=~"$Operator"}}[5m])',
             legend="Backpressure Time: {{dataset}}, {{operator}}",
         )
     ],

python/ray/data/_internal/execution/streaming_executor.py

@@ -277,8 +277,9 @@ def shutdown(self, force: bool, exception: Optional[Exception] = None):
             stats_summary_string = self._final_stats.to_summary().to_string(
                 include_parent=False
             )
-            # Reset the scheduling loop duration gauge.
-            self._sched_loop_duration_s.set(0, tags={"dataset": self._dataset_id})
+            # Reset the scheduling loop duration gauge + resource manager budgets/usages.
+            self._resource_manager.update_usages()
+            self.update_metrics(0)
             if self._data_context.enable_auto_log_stats:
                 logger.info(stats_summary_string)
             # Close the progress manager with a finishing message.
@@ -387,7 +388,12 @@ def update_metrics(self, sched_loop_duration: int):
 
     def _update_budget_metrics(self, op: PhysicalOperator, tags: Dict[str, str]):
         budget = self._resource_manager.get_budget(op)
-        if budget is not None:
+        if budget is None:
+            cpu_budget = 0
+            gpu_budget = 0
+            memory_budget = 0
+            object_store_memory_budget = 0
+        else:
             # Convert inf to -1 to represent unlimited budget in metrics
             cpu_budget = -1 if math.isinf(budget.cpu) else budget.cpu
             gpu_budget = -1 if math.isinf(budget.gpu) else budget.gpu
@@ -397,10 +403,11 @@ def _update_budget_metrics(self, op: PhysicalOperator, tags: Dict[str, str]):
                 if math.isinf(budget.object_store_memory)
                 else budget.object_store_memory
             )
-            self._cpu_budget_gauge.set(cpu_budget, tags=tags)
-            self._gpu_budget_gauge.set(gpu_budget, tags=tags)
-            self._memory_budget_gauge.set(memory_budget, tags=tags)
-            self._osm_budget_gauge.set(object_store_memory_budget, tags=tags)
+
+        self._cpu_budget_gauge.set(cpu_budget, tags=tags)
+        self._gpu_budget_gauge.set(gpu_budget, tags=tags)
+        self._memory_budget_gauge.set(memory_budget, tags=tags)
+        self._osm_budget_gauge.set(object_store_memory_budget, tags=tags)
 
     def _update_max_bytes_to_read_metric(
         self, op: PhysicalOperator, tags: Dict[str, str]