[Observability] Clustermgtd to emit heartbeat event into its events log.

Author: gmarciani

CHANGELOG.md

@@ -8,6 +8,7 @@ This file is used to list changes made in each version of the aws-parallelcluste
 
 **CHANGES**
 - Direct users to slurm_resume log to see EC2 error codes if no instances are launched.
+- Emit clustermgtd heartbeat as a structured event to CloudWatch Logs for metric filter-based monitoring.
 
 3.14.1
 ------

src/slurm_plugin/cluster_event_publisher.py

@@ -104,6 +104,10 @@
     "message": "Count of EC2 instances backing the cluster compute fleet",
     "event_type": "cluster-instance-count",
 }
+CLUSTERMGTD_HEARTBEAT = {
+    "message": "Clustermgtd heartbeat",
+    "event_type": "clustermgtd-heartbeat",
+}
 
 
 class ClusterEventPublisher:
@@ -474,6 +478,37 @@ def publish_compute_node_events(self, compute_nodes: List[SlurmNode], cluster_in
             event_supplier=({"detail": self._describe_node(node)} for node in compute_nodes),
         )
 
+    # Example event generated by this function:
+    # {
+    #     "datetime": "2023-04-03T18:10:13.089+00:00",
+    #     "version": 0,
+    #     "scheduler": "slurm",
+    #     "cluster-name": "integ-tests-9hbu2op3liukbqqz-develop",
+    #     "node-role": "HeadNode",
+    #     "component": "clustermgtd",
+    #     "level": "INFO",
+    #     "instance-id": "i-04886f8b56e5967ee",
+    #     "event-type": "clustermgtd-heartbeat",
+    #     "message": "Clustermgtd heartbeat",
+    #     "detail": {
+    #         "heartbeat-timestamp": "2023-04-03T18:10:13.089+00:00"
+    #     }
+    # }
+    @log_exception(logger, "publish_heartbeat_event", catch_exception=Exception, raise_on_error=False)
+    def publish_heartbeat_event(self, heartbeat_timestamp: datetime):
+        """Publish heartbeat event."""
+        timestamp = ClusterEventPublisher.timestamp()
+        self.publish_event(
+            logging.INFO,
+            **CLUSTERMGTD_HEARTBEAT,
+            timestamp=timestamp,
+            detail={
+                "heartbeat-timestamp": (
+                    heartbeat_timestamp.isoformat(timespec="milliseconds") if heartbeat_timestamp else None
+                ),
+            },
+        )
+
     # Slurm Resume Events
     # Note: This uses the same schema as `publish_unhealthy_static_node_events` from clustermgtd
     # Example event generated by this function:

src/slurm_plugin/clustermgtd.py

@@ -577,10 +577,13 @@ def manage_cluster(self):
     def _write_timestamp_to_file(self):
         """Write timestamp into shared file so compute nodes can determine if head node is online."""
         # Make clustermgtd heartbeat readable to all users
+        heartbeat_time = datetime.now(tz=timezone.utc)
         with open(os.open(self._config.heartbeat_file_path, os.O_WRONLY | os.O_CREAT, 0o644), "w") as timestamp_file:
             # Note: heartbeat must be written with datetime.strftime to convert localized datetime into str
             # datetime.strptime will not work with str(datetime)
-            timestamp_file.write(datetime.now(tz=timezone.utc).strftime(TIMESTAMP_FORMAT))
+            timestamp_file.write(heartbeat_time.strftime(TIMESTAMP_FORMAT))
+        # Publish heartbeat event to events log
+        self._event_publisher.publish_heartbeat_event(heartbeat_time)
 
     @staticmethod
     @retry(stop_max_attempt_number=2, wait_fixed=1000)

tests/slurm_plugin/test_cluster_event_publisher.py

@@ -2958,3 +2958,31 @@ def test_publish_compute_node_events(compute_nodes, expected_details, level_filt
     assert_that(received_events).is_length(len(expected_details))
     for received_event, expected_detail in zip(received_events, expected_details):
         assert_that(received_event).is_equal_to(expected_detail)
+
+
+@pytest.mark.parametrize(
+    "heartbeat_timestamp, expected_detail",
+    [
+        pytest.param(
+            datetime(year=2023, month=4, day=3, hour=18, minute=10, second=13, microsecond=89000, tzinfo=timezone.utc),
+            {"heartbeat-timestamp": "2023-04-03T18:10:13.089+00:00"},
+            id="valid timestamp",
+        ),
+        pytest.param(
+            None,
+            {"heartbeat-timestamp": None},
+            id="No timestamp",
+        ),
+    ],
+)
+def test_publish_heartbeat_event(heartbeat_timestamp: datetime, expected_detail: dict):
+    """Test that publish_heartbeat_event publishes the correct event."""
+    received_events = []
+    event_publisher = ClusterEventPublisher(event_handler(received_events, level_filter=["INFO"]))
+
+    event_publisher.publish_heartbeat_event(heartbeat_timestamp)
+
+    # Verify the event was published with the correct details
+    assert_that(received_events).is_length(1)
+    received_event = received_events[0]
+    assert_that(received_event).is_equal_to({"clustermgtd-heartbeat": expected_detail})

tests/slurm_plugin/test_clustermgtd.py

@@ -4112,3 +4112,49 @@ def test_find_active_nodes(partitions_name_map, expected_nodelist):
     """Unit test for the `ClusterManager._find_active_nodes()` method."""
     result_nodelist = ClusterManager._find_active_nodes(partitions_name_map)
     assert_that(result_nodelist).is_equal_to(expected_nodelist)
+
+
+@pytest.mark.parametrize(
+    "write_heartbeat_file_succeeds, expect_heartbeat_event_published",
+    [
+        pytest.param(True, True, id="When heartbeat file is written, heartbeat event is written"),
+        pytest.param(False, False, id="When heartbeat file is not written, heartbeat event is not written"),
+    ],
+)
+@pytest.mark.usefixtures(
+    "initialize_instance_manager_mock", "initialize_executor_mock", "initialize_console_logger_mock"
+)
+def test_heartbeat_reporting(write_heartbeat_file_succeeds, expect_heartbeat_event_published, mocker, tmp_path):
+    """Test that heartbeat event is published only when file write succeeds."""
+    from slurm_plugin.common import TIMESTAMP_FORMAT
+
+    heartbeat_file_path = str(tmp_path / "heartbeat")
+    mock_sync_config = SimpleNamespace(
+        heartbeat_file_path=heartbeat_file_path,
+        insufficient_capacity_timeout=600,
+        cluster_name="test-cluster",
+        head_node_instance_id="i-instance-id",
+        region="us-east-2",
+        boto3_config=None,
+        fleet_config={},
+    )
+
+    cluster_manager = ClusterManager(mock_sync_config)
+    mock_event_publisher = mocker.patch.object(cluster_manager, "_event_publisher")
+
+    if write_heartbeat_file_succeeds:
+        cluster_manager._write_timestamp_to_file()
+        heartbeat_file = tmp_path / "heartbeat"
+        assert_that(heartbeat_file.exists()).is_true()
+    else:
+        mocker.patch("os.open", side_effect=OSError("Mocked write failure"))
+        with pytest.raises(OSError):
+            cluster_manager._write_timestamp_to_file()
+
+    if expect_heartbeat_event_published:
+        mock_event_publisher.publish_heartbeat_event.assert_called_once()
+        event_timestamp = mock_event_publisher.publish_heartbeat_event.call_args[0][0]
+        assert_that(event_timestamp).is_instance_of(datetime)
+        assert_that(event_timestamp.strftime(TIMESTAMP_FORMAT)).is_equal_to(heartbeat_file.read_text())
+    else:
+        mock_event_publisher.publish_heartbeat_event.assert_not_called()