Fix validation result status transitions

src/alert-manager/src/alert-parser/node_alert_monitor.py

@@ -225,16 +225,20 @@ def handle_node_status_change(self, node: str, timestamp: float, status: float,
                 logger.info(f'{len(shrinked_alerts)} alerts after shrinking for node {node} at time {timestamp} due to tolerance time')
                 if period_alerts.empty:
                     logger.info(f"Node {node} is continuously unschedulable but in {from_status} with no alerts. No action taken.")
-                elif period_alerts['alertname'].str.contains('CordonValidationFailedNodes').any():
-                    validation_alerts = period_alerts[period_alerts['alertname'].str.contains('CordonValidationFailedNodes')]
+                elif (period_alerts['alertname'] == 'CordonValidationFailedNodes').any():
+                    validation_alerts = period_alerts[period_alerts['alertname'] == 'CordonValidationFailedNodes']
                     validation_time = validation_alerts['timestamp'].max()
-                    to_status = NodeStatus.CORDONED.value
-                    reason, detail = self.alert_mapper.summary_events_into_reason_detail(shrinked_alerts)
+                    shrinked_validation_alerts = self.alert_fetcher.shrink_alerts(validation_alerts)
+                    to_status = NodeStatus.TRIAGED_UNKNOWN.value
+                    reason, detail = self.alert_mapper.summary_events_into_reason_detail(shrinked_validation_alerts)
                     self.node_updater.update_status_action(node, from_status, to_status, validation_time, reason, detail)
-                elif period_alerts['alertname'].str.contains('RecoverValidatedNodes').any():
+                elif (period_alerts['alertname'] == 'RecoverValidatedNodes').any():
+                    recovery_alerts = period_alerts[period_alerts['alertname'] == 'RecoverValidatedNodes']
+                    recovery_time = recovery_alerts['timestamp'].max()
+                    shrinked_recovery_alerts = self.alert_fetcher.shrink_alerts(recovery_alerts)
                     to_status = NodeStatus.AVAILABLE_NODATA.value
-                    reason, detail = self.alert_mapper.summary_events_into_reason_detail(shrinked_alerts)
-                    self.node_updater.update_status_action(node, from_status, to_status, timestamp, reason, detail)
+                    reason, detail = self.alert_mapper.summary_events_into_reason_detail(shrinked_recovery_alerts)
+                    self.node_updater.update_status_action(node, from_status, to_status, recovery_time, reason, detail)
                 # TODO check validation job status
                 elif len(shrinked_new_alerts) > 0 and not period_alerts['alertname'].str.contains('NodeNotReady').any() and not period_alerts['alertname'].str.contains('RecoverValidatedNodes').any():
                     to_status = NodeStatus.CORDONED.value
@@ -283,6 +287,23 @@ def process_node_changes(self, node: str, changes: Dict[float, float], end_time:
             alerts = self.alert_fetcher.get_node_alert_records(
                 end_time, f"{time_offset}s", nodes=[node], severity="error"
             )
+            if node_status.Status == NodeStatus.VALIDATING.value:
+                recovery_alerts = self.alert_fetcher.get_node_alert_records(
+                    end_time,
+                    f"{time_offset}s",
+                    nodes=[node],
+                    alertname="RecoverValidatedNodes",
+                )
+                alert_frames = [
+                    frame for frame in (alerts, recovery_alerts)
+                    if frame is not None and not frame.empty
+                ]
+                alerts = (
+                    pd.concat(alert_frames, ignore_index=True)
+                    .drop_duplicates()
+                    .sort_values("timestamp")
+                    if alert_frames else None
+                )
 
             sorted_changes = sorted(changes.items(), key=lambda x: x[0])
             for timestamp, status in sorted_changes:

src/alert-manager/src/alert-parser/tests/test_alert_monitor.py

@@ -178,7 +178,7 @@ def test_handle_node_status_change(monitor, mock_alert_fetcher, mock_alert_mappe
     mock_alert_fetcher.reset_mock()
     mock_alert_mapper.reset_mock()
 
-    # case 3: Node status change from validating to cordoned
+    # case 4: Validation failure moves the node to triaged_unknown
     status = -1 
     node_status = NodeStatusRecord(
         Timestamp=datetime.fromtimestamp(timestamp - 100, tz=timezone.utc),
@@ -187,14 +187,18 @@ def test_handle_node_status_change(monitor, mock_alert_fetcher, mock_alert_mappe
         NodeId=node,
         Endpoint='test-endpoint'
     )
+    validation_failure_time = timestamp - 20
     alerts = pd.DataFrame({
-        'alertname': ['CordonValidationFailedNodes'],
-        'timestamp': [datetime.fromtimestamp(timestamp, tz=timezone.utc)],
-        'node_name': [node],
-        'summary': [f'{node} should be cordoned']
+        'alertname': ['NodeNotReady', 'CordonValidationFailedNodes'],
+        'timestamp': [
+            datetime.fromtimestamp(timestamp - 30, tz=timezone.utc),
+            datetime.fromtimestamp(validation_failure_time, tz=timezone.utc),
+        ],
+        'node_name': [node, node],
+        'summary': [f'{node} was not ready', f'{node} failed validation']
     })
     mock_alert_fetcher.find_node_alerts.return_value = alerts
-    mock_alert_fetcher.shrink_alerts.return_value = alerts
+    mock_alert_fetcher.shrink_alerts.side_effect = lambda frame: frame
     mock_alert_mapper.summary_events_into_reason_detail.return_value = ("reason", "detail")
 
     monitor.handle_node_status_change(node, timestamp, status, alerts, node_status)
@@ -204,7 +208,94 @@ def test_handle_node_status_change(monitor, mock_alert_fetcher, mock_alert_mappe
     args = mock_node_updater.update_status_action.call_args.args
     assert args[0] == node
     assert args[1] == 'validating'
-    assert args[2] == 'cordoned'
+    assert args[2] == 'triaged_unknown'
+    assert args[3] == datetime.fromtimestamp(validation_failure_time, tz=timezone.utc)
+    reason_alerts = mock_alert_mapper.summary_events_into_reason_detail.call_args.args[0]
+    assert reason_alerts['alertname'].tolist() == ['CordonValidationFailedNodes']
+
+    # Reset mocks for next test
+    mock_node_updater.reset_mock()
+    mock_alert_fetcher.reset_mock()
+    mock_alert_mapper.reset_mock()
+
+    # case 5: Validation success moves the node to available_nodata
+    recovery_time = timestamp - 10
+    alerts = pd.DataFrame({
+        'alertname': ['NodeNotReady', 'RecoverValidatedNodes'],
+        'timestamp': [
+            datetime.fromtimestamp(timestamp - 30, tz=timezone.utc),
+            datetime.fromtimestamp(recovery_time, tz=timezone.utc),
+        ],
+        'node_name': [node, node],
+        'summary': [f'{node} was not ready', f'{node} passed validation']
+    })
+    mock_alert_fetcher.find_node_alerts.return_value = alerts
+    mock_alert_fetcher.shrink_alerts.side_effect = lambda frame: frame
+    mock_alert_mapper.summary_events_into_reason_detail.return_value = ("reason", "detail")
+
+    monitor.handle_node_status_change(node, timestamp, status, alerts, node_status)
+
+    assert mock_node_updater.update_status_action.called
+    args = mock_node_updater.update_status_action.call_args.args
+    assert args[0] == node
+    assert args[1] == 'validating'
+    assert args[2] == 'available_nodata'
+    assert args[3] == datetime.fromtimestamp(recovery_time, tz=timezone.utc)
+    reason_alerts = mock_alert_mapper.summary_events_into_reason_detail.call_args.args[0]
+    assert reason_alerts['alertname'].tolist() == ['RecoverValidatedNodes']
+
+
+def test_process_validating_node_fetches_recovery_alerts(
+    monitor, mock_alert_fetcher, mock_node_updater
+):
+    from ltp_storage.data_schema.node_status import NodeStatusRecord
+
+    node = "test-node"
+    end_time = 1100.0
+    node_status = NodeStatusRecord(
+        Timestamp=datetime.fromtimestamp(900, tz=timezone.utc),
+        HostName=node,
+        Status='validating',
+        NodeId=node,
+        Endpoint='test-endpoint'
+    )
+    error_alert = pd.DataFrame({
+        'alertname': ['NodeNotReady'],
+        'timestamp': [datetime.fromtimestamp(1000, tz=timezone.utc)],
+        'node_name': [node],
+        'summary': ['node not ready'],
+        'severity': ['error'],
+    })
+    recovery_alert = pd.DataFrame({
+        'alertname': ['RecoverValidatedNodes'],
+        'timestamp': [datetime.fromtimestamp(1050, tz=timezone.utc)],
+        'node_name': [node],
+        'summary': ['validation passed'],
+        'severity': ['info'],
+    })
+    mock_node_updater.get_node_latest_status.return_value = node_status
+    mock_alert_fetcher.get_node_alert_records.side_effect = [
+        error_alert,
+        recovery_alert,
+    ]
+
+    with patch.object(monitor, 'handle_node_status_change') as mock_handle:
+        monitor.process_node_changes(node, {end_time: -1}, end_time)
+
+    assert mock_alert_fetcher.get_node_alert_records.call_count == 2
+    assert mock_alert_fetcher.get_node_alert_records.call_args_list[0].kwargs == {
+        'nodes': [node],
+        'severity': 'error',
+    }
+    assert mock_alert_fetcher.get_node_alert_records.call_args_list[1].kwargs == {
+        'nodes': [node],
+        'alertname': 'RecoverValidatedNodes',
+    }
+    merged_alerts = mock_handle.call_args.args[3]
+    assert merged_alerts['alertname'].tolist() == [
+        'NodeNotReady',
+        'RecoverValidatedNodes',
+    ]
 
 
 def test_handle_validating_node_with_empty_alerts(monitor, mock_alert_fetcher, mock_alert_mapper, mock_node_updater):

src/alert-manager/src/alert-parser/tests/test_alert_util.py

@@ -118,6 +118,22 @@ def test_shrink_alerts(self):
         self.assertEqual(result.iloc[0]['node_name'], 'mi300-00007n')
         self.assertEqual(result.iloc[0]['alertname'], 'RecoverValidatedNodes')
 
+    def test_fetch_logs_forwards_alertname(self):
+        self.fetcher.client.query_alerts.reset_mock()
+        self.fetcher.client.query_alerts.return_value = []
+
+        self.fetcher.fetch_logs(
+            end_time_stamp=1000,
+            time_offset="5m",
+            nodes=["test-node"],
+            alertname="RecoverValidatedNodes",
+        )
+
+        kwargs = self.fetcher.client.query_alerts.call_args.kwargs
+        self.assertEqual(kwargs["nodes"], ["test-node"])
+        self.assertEqual(kwargs["alertname"], "RecoverValidatedNodes")
+        self.assertIsNone(kwargs["severity"])
+
 @pytest.fixture(scope='module')
 def mock_kusto_client():
     """Mock alert client for testing purposes"""

src/alert-manager/src/alert-parser/utils/alert_util.py

@@ -27,16 +27,19 @@ def __init__(self):
         self.endpoint = os.getenv("CLUSTER_ID")
         self.client = create_alert_client(endpoint=self.endpoint)
 
-    def fetch_logs(self, end_time_stamp, time_offset, nodes=None, severity=None):
+    def fetch_logs(self, end_time_stamp, time_offset, nodes=None, severity=None, alertname=None):
         """Fetch raw alert logs from Kusto"""
         end_time = datetime.fromtimestamp(end_time_stamp)
         time_offset_delta = parse_duration(time_offset)
         start_time = end_time - time_offset_delta
-        records = []
-        if severity is None:
-            records = self.client.query_alerts(start_time=start_time, end_time=end_time, nodes=nodes, endpoint=self.endpoint)
-        else:
-            records = self.client.query_alerts(start_time=start_time, end_time=end_time, nodes=nodes, severity=severity, endpoint=self.endpoint)
+        records = self.client.query_alerts(
+            start_time=start_time,
+            end_time=end_time,
+            nodes=nodes,
+            severity=severity,
+            alertname=alertname,
+            endpoint=self.endpoint,
+        )
         logger.info(f"Fetched {len(records)} alert logs from Kusto.")
         return records if records else None
 
@@ -87,10 +90,16 @@ def shrink_alerts(self, alerts_df):
 
         return result_df
 
-    def get_node_alert_records(self, end_time_stamp, time_offset, nodes=None, severity=None):
+    def get_node_alert_records(self, end_time_stamp, time_offset, nodes=None, severity=None, alertname=None):
         """Get processed alert records for nodes"""
         logger.info(f"Fetching alerts from Kusto for nodes: {nodes} with time offset: {time_offset} and end time: {end_time_stamp}")
-        alerts_data = self.fetch_logs(end_time_stamp, time_offset, nodes=nodes, severity=severity)
+        alerts_data = self.fetch_logs(
+            end_time_stamp,
+            time_offset,
+            nodes=nodes,
+            severity=severity,
+            alertname=alertname,
+        )
         if alerts_data is None:
             return None
 

src/alert-manager/src/job-status-change-notification/controllers/alert.js

@@ -80,9 +80,9 @@ const sendAlerts = async (alerts) => {
 };
 
 const uncordonNodes = async (nodeList) => {
-  logger.info(`Uncordoning nodes: ${nodeList.join(", ")} ...`);
+  logger.info(`Reporting successfully validated nodes: ${nodeList.join(", ")} ...`);
 
-  // create alerts for uncordon nodes
+  // create alerts for nodes that are ready for data synchronization
   const alerts = nodeList.map(node => ({
     status: "firing",
     labels: {
@@ -91,13 +91,13 @@ const uncordonNodes = async (nodeList) => {
       node_name: node,
     },
     annotations: {
-      summary: `The node ${node} has been validated and be uncordoned.`,
+      summary: `The node ${node} has passed validation and is ready for data synchronization.`,
     },
   }));
 
   await sendAlerts(alerts);
 
-  logger.info(`Successfully uncordoned nodes: ${nodeList.join(", ")}`);
+  logger.info(`Successfully reported validated nodes: ${nodeList.join(", ")}`);
 }
 
 const cordonNodes = async (nodeList) => {
@@ -109,7 +109,7 @@ const cordonNodes = async (nodeList) => {
     status: "firing",
     labels: {
       alertname: "CordonValidationFailedNodes",
-      severity: "info",
+      severity: "error",
       node_name: node.name,
     },
     annotations: {