[Observability] Clustermgtd to emit heartbeat into event logs rather than explicit CW metric

CHANGELOG.md

@@ -8,7 +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 metric `ClustermgtdHeartbeat` to signal clustermgtd heartbeat. 
+- 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

@@ -41,7 +41,6 @@
 from common.utils import check_command_output, read_json, sleep_remaining_loop_time, time_is_up, wait_remaining_time
 from retrying import retry
 from slurm_plugin.capacity_block_manager import CapacityBlockManager
-from slurm_plugin.cloudwatch_utils import CloudWatchMetricsPublisher
 from slurm_plugin.cluster_event_publisher import ClusterEventPublisher
 from slurm_plugin.common import TIMESTAMP_FORMAT, ScalingStrategy, log_exception, print_with_count
 from slurm_plugin.console_logger import ConsoleLogger
@@ -61,7 +60,6 @@
 LOOP_TIME = 60
 CONSOLE_OUTPUT_WAIT_TIME = 5 * 60
 MAXIMUM_TASK_BACKLOG = 100
-CW_METRICS_HEARTBEAT = "ClustermgtdHeartbeat"
 log = logging.getLogger(__name__)
 compute_logger = log.getChild("console_output")
 event_logger = log.getChild("events")
@@ -403,7 +401,6 @@ def __init__(self, config):
         self._event_publisher = None
         self._partition_nodelist_mapping_instance = None
         self._capacity_block_manager = None
-        self._metrics_publisher = None
         self.set_config(config)
 
     def set_config(self, config: ClustermgtdConfig):
@@ -429,7 +426,6 @@ def set_config(self, config: ClustermgtdConfig):
             self._instance_manager = self._initialize_instance_manager(config)
             self._console_logger = self._initialize_console_logger(config)
             self._capacity_block_manager = self._initialize_capacity_block_manager(config)
-            self._metrics_publisher = self._initialize_metrics_publisher(config)
 
     def shutdown(self):
         if self._task_executor:
@@ -484,16 +480,6 @@ def _initialize_capacity_block_manager(config):
             region=config.region, fleet_config=config.fleet_config, boto3_config=config.boto3_config
         )
 
-    @staticmethod
-    def _initialize_metrics_publisher(config):
-        """Initialize CloudWatch metrics publisher."""
-        return CloudWatchMetricsPublisher(
-            region=config.region,
-            cluster_name=config.cluster_name,
-            instance_id=config.head_node_instance_id,
-            boto3_config=config.boto3_config,
-        )
-
     def _update_compute_fleet_status(self, status):
         log.info("Updating compute fleet status from %s to %s", self._compute_fleet_status, status)
         self._compute_fleet_status_manager.update_status(status)
@@ -588,16 +574,16 @@ def manage_cluster(self):
         # Write clustermgtd heartbeat to file
         self._write_timestamp_to_file()
 
-        # Publish heartbeat metric to CloudWatch
-        self._metrics_publisher.put_metric(metric_name=CW_METRICS_HEARTBEAT, value=1)
-
     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)