Fix event action consuming.

dlrover/python/diagnosis/common/diagnosis_action.py

@@ -123,6 +123,7 @@ def __init__(
     ):
         super().__init__(
             DiagnosisActionType.EVENT,
+            instance=DiagnosisConstant.MASTER_INSTANCE,
             timestamp=timestamp,
             expired_time_period=expired_time_period,
         )
@@ -203,7 +204,7 @@ def add_action(self, new_action: DiagnosisAction):
             try:
                 if new_action.is_needed():
                     ins_actions.put(new_action, timeout=3)
-                    logger.info(f"New diagnosis action {new_action}")
+                    logger.info(f"New diagnosis action: {new_action}")
             except queue.Full:
                 logger.warning(
                     f"Diagnosis actions for {instance} is full, "

dlrover/python/diagnosis/inferencechain/inference_chain.py

@@ -36,7 +36,7 @@ def __init__(
         self.operators = operators
 
     def infer(self) -> List[Inference]:
-        logger.info(f"Infer {self.inferences}")
+        logger.debug(f"Infer {self.inferences}")
         inferences = self.inferences
         while True:
             has_new_inference = False
@@ -66,5 +66,5 @@ def get_operator(self, inference: Inference) -> InferenceOperator:
         for operator in self.operators:
             if operator.is_compatible(inference):
                 return operator
-        logger.info(f"No operator for inference: {inference.__dict__}")
+        logger.debug(f"No operator for inference: {inference.__dict__}")
         return None  # type: ignore

dlrover/python/master/elastic_training/net_topology.py

@@ -10,14 +10,17 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+
 from abc import ABCMeta, abstractmethod
 from collections import OrderedDict
 from dataclasses import dataclass
 from typing import Dict, List, Tuple
 
+from dlrover.python.common.serialize import JsonSerializable
+
 
 @dataclass
-class NodeTopologyMeta(object):
+class NodeTopologyMeta(JsonSerializable):
     node_id: int = 0
     node_rank: int = 0
     process_num: int = 0

dlrover/python/master/node/dist_job_manager.py

@@ -36,12 +36,10 @@
 from dlrover.python.common.grpc import ParallelConfig
 from dlrover.python.common.log import default_logger as logger
 from dlrover.python.common.node import Node, NodeGroupResource
-from dlrover.python.diagnosis.common.constants import (
-    DiagnosisActionType,
-    DiagnosisConstant,
-)
+from dlrover.python.diagnosis.common.constants import DiagnosisConstant
 from dlrover.python.diagnosis.common.diagnosis_action import (
     DiagnosisAction,
+    EventAction,
     NoAction,
 )
 from dlrover.python.master.monitor.error_monitor import K8sJobErrorMonitor
@@ -670,10 +668,20 @@ def _get_pod_unique_labels(self, node: Node):
         }
 
     def _process_diagnosis_action(self, action: DiagnosisAction):
-        if not action or action.action_type == DiagnosisActionType.NONE:
+        if not action or isinstance(action, NoAction):
             return
 
-        # TODO
+        if isinstance(action, EventAction):
+            self._report_event(
+                action.event_type,
+                action.event_instance,
+                action.event_action,
+                action.event_msg,
+                action.event_labels,
+            )
+        else:
+            # TODO: deal with other action
+            pass
 
     def _process_event(self, event: NodeEvent):
         node_type = event.node.type

dlrover/python/tests/test_diagnosis.py

@@ -47,7 +47,7 @@ def test_action_basic(self):
         )
         self.assertEqual(event_action.action_type, DiagnosisActionType.EVENT)
         self.assertEqual(
-            event_action._instance, DiagnosisConstant.LOCAL_INSTANCE
+            event_action._instance, DiagnosisConstant.MASTER_INSTANCE
         )
         self.assertEqual(event_action.event_type, "info")
         self.assertEqual(event_action.event_instance, "job")
@@ -101,15 +101,15 @@ def test_action_queue(self):
         )
         self.assertEqual(
             action_queue.next_action(
-                instance=DiagnosisConstant.LOCAL_INSTANCE
+                instance=DiagnosisConstant.MASTER_INSTANCE
             ).action_type,
             DiagnosisActionType.EVENT,
         )
         self.assertEqual(
             action_queue.next_action(
                 instance=DiagnosisConstant.LOCAL_INSTANCE
             ).action_type,
-            DiagnosisActionType.EVENT,
+            DiagnosisActionType.NONE,
         )
         self.assertEqual(
             action_queue.next_action(instance=1).action_type,

dlrover/python/tests/test_job_manager.py

@@ -39,6 +39,10 @@
     ParallelConfig,
 )
 from dlrover.python.common.node import NodeGroupResource, NodeResource
+from dlrover.python.diagnosis.common.diagnosis_action import (
+    EventAction,
+    NoAction,
+)
 from dlrover.python.master.dist_master import DistributedJobMaster
 from dlrover.python.master.monitor.error_monitor import SimpleErrorMonitor
 from dlrover.python.master.monitor.speed_monitor import SpeedMonitor
@@ -869,6 +873,21 @@ def test_get_pending_timeout(self):
         # reset
         _dlrover_context.seconds_to_wait_pending_pod = 900
 
+    @patch.object(DistributedJobManager, "_report_event")
+    def test_process_diagnosis_action(self, mock_method):
+        params = MockK8sPSJobArgs()
+        params.initilize()
+        manager = create_job_manager(params, SpeedMonitor())
+
+        manager._process_diagnosis_action(None)
+        self.assertEqual(mock_method.call_count, 0)
+
+        manager._process_diagnosis_action(NoAction)
+        self.assertEqual(mock_method.call_count, 0)
+
+        manager._process_diagnosis_action(EventAction())
+        self.assertEqual(mock_method.call_count, 1)
+
 
 class LocalJobManagerTest(unittest.TestCase):
     def test_local_job_manager(self):