[data] reset cpu + gpu metrics on executor shutdown and updating task submission/block generation metrics

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

@@ -451,7 +451,7 @@
     unit="seconds",
     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}}",
         )
     ],
@@ -466,7 +466,7 @@
     unit="seconds",
     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/resource_manager.py

@@ -330,6 +330,16 @@ def get_budget(self, op: PhysicalOperator) -> Optional[ExecutionResources]:
             return None
         return self._op_resource_allocator.get_budget(op)
 
+    def clear_usages_and_budget(self):
+        """Resets the usages and budgets. This function should only be called on
+        executor shutdown.
+        """
+        assert isinstance(self.op_resource_allocator, ReservationOpResourceAllocator)
+        for op in self._op_usages:
+            self._op_usages[op] = ExecutionResources.zero()
+            self.op_resource_allocator._op_budgets[op] = ExecutionResources.zero()
+            self.op_resource_allocator._output_budgets[op] = 0
+
 
 class OpResourceAllocator(ABC):
     """An interface for dynamic operator resource allocation.

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

@@ -258,8 +258,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.clear_usages_and_budget()
+            self.update_metrics(0)
             if self._data_context.enable_auto_log_stats:
                 logger.info(stats_summary_string)
             # Close the progress bars from top to bottom to avoid them jumping