From 6ff053dfacba5e1c9c33fffaffbb4c9c2364f75d Mon Sep 17 00:00:00 2001 From: Anthony Romaniello Date: Fri, 1 Mar 2024 18:52:36 -0500 Subject: [PATCH] add NTP status to SEA metadata --- .../actions/acquire_sea_data_product.py | 5 ++++ scos_actions/hardware/utils.py | 23 ++++++++++++++++--- 2 files changed, 25 insertions(+), 3 deletions(-) diff --git a/scos_actions/actions/acquire_sea_data_product.py b/scos_actions/actions/acquire_sea_data_product.py index 337503c7..f3d368a1 100644 --- a/scos_actions/actions/acquire_sea_data_product.py +++ b/scos_actions/actions/acquire_sea_data_product.py @@ -45,6 +45,7 @@ get_current_cpu_temperature, get_disk_smart_data, get_max_cpu_temperature, + get_ntp_status, ) from scos_actions.metadata.sigmf_builder import SigMFBuilder from scos_actions.metadata.structs import ( @@ -778,6 +779,10 @@ def capture_diagnostics( cpu_diag["ssd_smart_data"] = ntia_diagnostics.SsdSmartData(**smart_data) except: logger.warning("Failed to get SSD SMART data") + try: + cpu_diag["ntp_active"], cpu_diag["ntp_sync"] = get_ntp_status() + except: + logger.warning("Failed to get NTP status") try: # Disk usage disk_usage = get_disk_usage() cpu_diag["disk_usage"] = disk_usage diff --git a/scos_actions/hardware/utils.py b/scos_actions/hardware/utils.py index 49642da7..05de2956 100644 --- a/scos_actions/hardware/utils.py +++ b/scos_actions/hardware/utils.py @@ -1,6 +1,6 @@ import logging import subprocess -from typing import Dict +from typing import Dict, Tuple, Union import psutil from its_preselector.web_relay import WebRelay @@ -66,7 +66,23 @@ def get_current_cpu_temperature(fahrenheit: bool = False) -> float: raise e -def get_disk_smart_data(disk: str) -> dict: +def get_ntp_status() -> Union[Tuple[bool, bool], str]: + """ + Get system NTP status by parsing the output of ``timedatectl``. + + :returns: A tuple of booleans: (ntp_active, ntp_synchronized). + """ + try: + status = subprocess.check_output(["timedatectl"]).decode("utf-8") + except Exception: + logger.exception(f"Unable to get NTP status from timedatectl") + return "Unavailable" + ntp_active = "NTP service: active" in status + ntp_synchronized = "System clock synchronized: yes" in status + return ntp_active, ntp_synchronized + + +def get_disk_smart_data(disk: str) -> Union[dict, str]: """ Get selected SMART data for the chosen disk. @@ -81,7 +97,8 @@ def get_disk_smart_data(disk: str) -> dict: https://nvmexpress.org/wp-content/uploads/NVM-Express-1_4-2019.06.10-Ratified.pdf :param disk: The desired disk, e.g., ``/dev/nvme0n1``. - :return: A dictionary containing the retrieved data from the SMART report. + :return: A dictionary containing the retrieved data from the SMART report, or + the string "Unavailable" if ``smartctl`` fails to run. """ try: report = subprocess.check_output(["smartctl", "-a", disk]).decode("utf-8")