Support xpu timer metric collection.

dlrover/python/common/grpc.py

@@ -490,17 +490,19 @@ class NodeCheckpointState(Message):
 
 @dataclass
 class DiagnosisTrainingLog(Message):
+    type: str = ""
     timestamp: int = 0
+    log_content = ""
 
 
 @dataclass
-class DiagnosisCudaLog(Message):
-    timestamp: int = 0
-
-
-@dataclass
-class DiagnosisChipMetrics(Message):
+class DiagnosisAgentMetric(Message):
+    type: str = ""
     timestamp: int = 0
+    metric_content: str = ""
+    node_id: int = -1
+    node_type: str = ""
+    node_rank: int = -1
 
 
 @dataclass

dlrover/python/diagnosis/common/constants.py

@@ -12,12 +12,27 @@
 # limitations under the License.
 
 
+class EnvConfigKey(object):
+    XPU_TIMER_PORT = "XPU_TIMER_PORT"
+
+
 class InferenceConfigKey(object):
     LOG_FILE = "log_file"
     ERRORS = "errors"
 
 
-class DiagnoseAction(object):
+class DiagnosisConstant(object):
+    MASTER_DIAGNOSIS_OBSERVING_INTERVAL_SECS = 180
+    AGENT_PERIODICALLY_DIAGNOSIS_INTERVAL_SECS = 60
+
+
+class DiagnosisDataType(object):
+    GENERIC = "GENERIC"
+    TRAINING_LOG = "TRAINING_LOG"
+    TRAINING_HANG_DETECTION = "TRAINING_HANG_DETECTION"
+
+
+class DiagnosisAction(object):
     NO_ACTION = "no_action"
     RESTART_WORKER = "restart_worker"
     RELAUNCH_WORKER = "relaunch_worker"

dlrover/python/diagnosis/common/diagnosis_data.py

@@ -11,69 +11,117 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from abc import ABCMeta, abstractmethod
+from abc import ABCMeta
 from datetime import datetime
-from typing import List, Optional
+from typing import List
 
-
-class DiagnosisDataType:
-    CUDALOG = "cuda_log"
-    TRAININGLOG = "training_log"
-    CHIPMETRICES = "chip_metrics"
+from dlrover.python.common import env_utils
+from dlrover.python.diagnosis.common.constants import DiagnosisDataType
 
 
 class DiagnosisData(metaclass=ABCMeta):
-    def __init__(self):
-        pass
-
-    @abstractmethod
-    def get_timestamp(self) -> float:
-        pass
-
-    @abstractmethod
-    def get_type(self) -> str:
-        pass
-
-
-class CudaLog(DiagnosisData):
-    def __init__(self, timestamp: int):
+    def __init__(self, data_type, timestamp: int = 0, data_content: str = ""):
+        self._data_type = data_type
         if timestamp == 0:
-            self.timestamp = int(round(datetime.now().timestamp()))
+            self._timestamp = int(round(datetime.now().timestamp()))
         else:
-            self.timestamp = timestamp
-
-    def get_timestamp(self) -> int:
-        return self.timestamp
-
-    def get_type(self) -> str:
-        return DiagnosisDataType.CUDALOG
-
-
-class TrainingLog(DiagnosisData):
+            self._timestamp = timestamp
+        self._data_content = data_content
+
+    @property
+    def data_type(self) -> str:
+        return self._data_type
+
+    @property
+    def timestamp(self) -> int:
+        return self._timestamp
+
+    @property
+    def data_content(self) -> str:
+        return self._data_content
+
+
+class AgentMetric(DiagnosisData):
+    def __init__(
+        self,
+        timestamp: int = 0,
+        data_type: str = DiagnosisDataType.GENERIC,
+        data_content: str = "",
+        is_final_result=False,
+        need_report=False,
+        node_id: int = -1,
+        node_type: str = "",
+        node_rank: int = -1,
+    ):
+        """
+        General metric
+
+        Args:
+            data_type (str): Type of metric. Defaults to "GENERIC".
+            data_content (str): Content of the metric. Defaults to "".
+            is_final_result (bool, optional): Whether the metric is final
+                result or not. Defaults to False.
+            need_report (bool, optional): Whether the metric needs
+                report(to Brain). Defaults to False.
+            node_id (int): Node ID. Defaults to -1.
+            node_type (str): Node type. Defaults to "".
+            node_rank (int): Node rank. Defaults to -1.
+        """
+
+        super().__init__(data_type, timestamp, data_content)
+        self._is_final_result = is_final_result
+        self._need_report = need_report
+        self._node_id = node_id
+        self._node_type = node_type
+        self._node_rank = node_rank
+
+    @property
+    def is_final_result(self):
+        return self._is_final_result
+
+    @property
+    def need_report(self):
+        return self._need_report
+
+    @property
+    def node_id(self):
+        return self._node_id
+
+    @property
+    def node_type(self):
+        return self._node_type
+
+    @property
+    def node_rank(self):
+        return self._node_rank
+
+    def is_resolvable(self):
+        if self.data_type == DiagnosisDataType.TRAINING_HANG_DETECTION:
+            return True
+        # TODO: add more resolvable metric type later
+        return False
+
+
+class TrainingLog(AgentMetric):
     def __init__(self, timestamp: int = 0, logs: List[str] = None):
-        super().__init__()
-        if timestamp == 0:
-            self.timestamp = int(round(datetime.now().timestamp()))
+        if logs is None:
+            data_content = ""
         else:
-            self.timestamp = timestamp
-        self.logs: Optional[List[str]] = logs
-
-    def get_timestamp(self) -> int:
-        return self.timestamp
-
-    def get_type(self) -> str:
-        return DiagnosisDataType.TRAININGLOG
-
-
-class ChipMetrics(DiagnosisData):
-    def __init__(self, timestamp: int):
-        if timestamp == 0:
-            self.timestamp = int(round(datetime.now().timestamp()))
-        else:
-            self.timestamp = timestamp
-
-    def get_timestamp(self) -> int:
-        return self.timestamp
-
-    def get_type(self) -> str:
-        return DiagnosisDataType.CHIPMETRICES
+            data_content = "\n".join(logs)
+
+        super().__init__(
+            timestamp,
+            DiagnosisDataType.TRAINING_LOG,
+            data_content,
+            True,
+            False,
+            env_utils.get_node_id(),
+            env_utils.get_node_type(),
+            env_utils.get_node_rank(),
+        )
+
+    @property
+    def logs(self) -> List[str]:
+        if not self.data_content:
+            return []
+        return [line for line in self.data_content.splitlines()]

dlrover/python/diagnosis/datacollector/data_collector.py

@@ -31,8 +31,11 @@ def __init__(self):
 
     @abstractmethod
     def collect_data(self) -> object:
+        """The implementation of data collector."""
         pass
 
     @abstractmethod
-    def to_collect_data(self) -> bool:
-        pass
+    def is_enabled(self) -> bool:
+        """Whether the collector is enabled."""
+
+        return True

dlrover/python/diagnosis/datacollector/metrics_collector.py

@@ -11,20 +11,18 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from dlrover.python.diagnosis.common.diagnosis_data import ChipMetrics
 from dlrover.python.diagnosis.datacollector.data_collector import DataCollector
 
 
 class MetricsCollector(DataCollector):
-    def __init__(self, *args, **kwargs):
+    def __init__(self):
         """
-        MetricsCollector collects GPU metrics
+        MetricsCollector collects XPU metrics
         """
-        pass
+        super().__init__()
 
     def collect_data(self) -> object:
-        chip_metrics = ChipMetrics(0)
-        return chip_metrics
+        pass
 
-    def to_collect_data(self) -> bool:
+    def is_enabled(self) -> bool:
         return True

dlrover/python/diagnosis/datacollector/training_log_collector.py

@@ -44,5 +44,5 @@ def collect_data(self) -> TrainingLog:
         training_log = TrainingLog(logs=logs)
         return training_log
 
-    def to_collect_data(self) -> bool:
+    def is_enabled(self) -> bool:
         return True

dlrover/python/diagnosis/datacollector/xpu_timer_metric_collector.py

@@ -0,0 +1,68 @@
+# Copyright 2024 The DLRover Authors. All rights reserved.
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# 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.
+
+import requests
+
+from dlrover.python.common import env_utils
+from dlrover.python.common.log import default_logger as logger
+from dlrover.python.diagnosis.common.constants import EnvConfigKey
+from dlrover.python.diagnosis.datacollector.metrics_collector import (
+    MetricsCollector,
+)
+
+
+class XpuTimerMetricsCollector(MetricsCollector):
+    def __init__(self):
+        """
+        MetricsCollector collects GPU metrics from xpu-timer.
+        """
+        super().__init__()
+        self._metric_port = env_utils.get_env(EnvConfigKey.XPU_TIMER_PORT)
+        if self._metric_port:
+            self._metric_endpoint = (
+                "http://127.0.0.1:" + self._metric_port + "/metrics"
+            )
+        else:
+            self._metric_endpoint = None
+
+    def collect_data(self) -> str:
+        if not self.is_enabled():
+            return ""
+
+        try:
+            response = requests.get(self._metric_endpoint)
+            response.raise_for_status()
+
+            # data preprocessing
+            return self._preprocess_metrics(response.text)
+        except requests.exceptions.RequestException as e:
+            logger.warning(
+                "Error fetching metrics from "
+                f"xpu-timer: {self._metric_endpoint}, error: {e}"
+            )
+            return ""
+
+    def _preprocess_metrics(self, metric_str):
+        try:
+            metric_list = [
+                line
+                for line in metric_str.splitlines()
+                if not line.startswith("#") and not line.startswith("exposer")
+            ]
+            return "\n".join(metric_list)
+        except Exception as e:
+            logger.warning(f"Error preprocessing metrics from xpu-timer: {e}")
+            return ""
+
+    def is_enabled(self) -> bool:
+        return self._metric_endpoint is not None