[SWDEV-511234] Added amdsmi_get_gpu_cper_entries & CLI implementation

Added amdsmi_get_gpu_cper_entries() in the python and C APIs

Signed-off-by: Maisam Arif <Maisam.Arif@amd.com>
Signed-off-by: Oliveira, Daniel <daniel.oliveira@amd.com>
Co-authored-by: Saeed, Oosman <Oosman.Saeed@amd.com>
Co-authored-by: AL Musaffar, Yazen <Yazen.ALMusaffar@amd.com>

amdsmi_cli/amdsmi_cli.py

@@ -96,7 +96,8 @@ if __name__ == "__main__":
                                     amd_smi_commands.monitor,
                                     amd_smi_commands.rocm_smi,
                                     amd_smi_commands.xgmi,
-                                    amd_smi_commands.partition)
+                                    amd_smi_commands.partition,
+                                    amd_smi_commands.ras)
     try:
         try:
             argcomplete.autocomplete(amd_smi_parser)
@@ -105,7 +106,7 @@ if __name__ == "__main__":
 
         valid_commands = ['version', 'list', 'static', 'firmware', 'bad-pages',
                           'metric', 'process', 'profile', 'event', 'topology', 'set',
-                          'reset', 'monitor', 'xgmi', 'partition', '--help', '-h']
+                          'reset', 'monitor', 'xgmi', 'partition', 'ras', '--help', '-h']
 
         sys.argv = [arg.lower() if arg.startswith('--') or not arg.startswith('-')
                     else arg for arg in sys.argv]
@@ -117,11 +118,12 @@ if __name__ == "__main__":
             raise amdsmi_cli_exceptions.AmdSmiInvalidSubcommandException(sys.argv[1],amd_smi_commands.logger.destination)
 
         # Handle command modifiers before subcommand execution
-        if args.json:
+            # human readable is the default output format
+        if hasattr(args, 'json') and args.json:
             amd_smi_commands.logger.format = amd_smi_commands.logger.LoggerFormat.json.value
-        if args.csv:
+        if hasattr(args, 'csv') and args.csv:
             amd_smi_commands.logger.format = amd_smi_commands.logger.LoggerFormat.csv.value
-        if args.file:
+        if hasattr(args, 'file') and args.file:
             amd_smi_commands.logger.destination = args.file
 
         # Remove previous log handlers

amdsmi_cli/amdsmi_commands.py

@@ -34,12 +34,12 @@ from amdsmi_helpers import AMDSMIHelpers
 from amdsmi_logger import AMDSMILogger
 from amdsmi import amdsmi_exception, amdsmi_interface
 
-
 class AMDSMICommands():
     """This class contains all the commands corresponding to AMDSMIParser
     Each command function will interact with AMDSMILogger to handle
     displaying the output to the specified format and destination.
     """
+    
     def __init__(self, format='human_readable', destination='stdout') -> None:
         self.helpers = AMDSMIHelpers()
         self.logger = AMDSMILogger(format=format, destination=destination)
@@ -175,6 +175,7 @@ class AMDSMICommands():
         elif self.logger.is_json_format() or self.logger.is_csv_format():
             self.logger.print_output()
 
+
     def list(self, args, multiple_devices=False, gpu=None):
         """List information for target gpu
 
@@ -6160,6 +6161,108 @@ class AMDSMICommands():
                 with self.logger.destination.open('a', encoding="utf-8") as output_file:
                     output_file.write(legend_output + '\n')
 
+
+    def ras(self, args, multiple_devices=False, gpu=None, cper=None,
+            severity=None, folder=None, file_limit=None, follow=None):
+        """
+        Retrieve and process CPER (RAS) entries for a target GPU.
+
+        Expected command (all options only):
+        amd-smi ras --cper --severity=nonfatal-uncorrected,fatal --folder <folder_name> --file_limit=1000 --follow
+
+        Since no timestamp is provided on the command line, the function starts from a default cursor of 0.
+        The output file name is auto-generated using the timestamp from the CPER header data (converted from
+        the header’s "YYYY/MM/DD HH:MM:SS" format), along with the GPU/platform ID and error severity.
+        """
+        # GPU handle logic.
+        if gpu:
+            args.gpu = gpu
+        if cper:
+            args.cper = cper
+        if severity:
+            args.severity = severity
+        if folder:
+            args.folder = folder
+        if file_limit:
+            args.file_limit = file_limit
+        if follow:
+            args.follow = follow
+
+        if args.gpu == None:
+            args.gpu = self.device_handles
+
+        self.helpers.check_required_groups()
+        handled_multiple_gpus, device_handle = self.helpers.handle_gpus(args, self.logger, self.ras)
+        if handled_multiple_gpus:
+            return
+
+        args.gpu = device_handle
+
+        # Parse severity mask dynamically from the --severity option.
+        severity_mask = 0
+        # drop duplicates of args
+        logging.debug(args)
+        for sev in list(set(args.severity)):
+            if sev == "all":
+                # Set bits for NON_FATAL_UNCORRECTED (0), FATAL (1), and NON_FATAL_CORRECTED (2)
+                severity_mask |= ((1 << 0) | (1 << 1) | (1 << 2))
+            elif sev == "fatal":
+                # Set bit corresponding to AMDSMI_CPER_SEV_FATAL (which is 1)
+                severity_mask |= (1 << 1)
+            elif sev in ("nonfatal", "nonfatal-uncorrected"):
+                # Set bit corresponding to AMDSMI_CPER_SEV_NON_FATAL_UNCORRECTED (which is 0)
+                severity_mask |= (1 << 0)
+            elif sev in ("nonfatal-corrected", "corrected"):
+                # Set bit corresponding to AMDSMI_CPER_SEV_NON_FATAL_CORRECTED (which is 2)
+                severity_mask |= (1 << 2)
+
+        if args.cper:
+            # Start from cursor 0 (no timestamp argument provided).
+            cursor = 0
+            buffer_size = 1048576
+            file_limit = int(args.file_limit) if args.file_limit else 1000
+
+            # Print exit message only once and only when follow is set
+            if self.logger.cper_exit_message() and args.follow:
+                print('Press q and hit ENTER when you want to stop.')
+                self.logger.set_cper_exit_message(False)
+
+            # Main loop: continuously retrieve CPER entries if --follow is set.
+            gpu_id = self.helpers.get_gpu_id_from_device_handle(args.gpu)
+            if args.folder:
+                print(f'Dumping CPER file header entries for GPU {gpu_id} in folder {args.folder}\n')
+            else:
+                print(f'Dumping CPER file header entries for GPU {gpu_id}:\n')
+
+            self.stop = False
+            while True:
+                try:
+                    entries, new_cursor, cper_data = amdsmi_interface.amdsmi_get_gpu_cper_entries(
+                        args.gpu, severity_mask, buffer_size, cursor)
+                    logging.debug(f"cper_entries | entries: {entries}")
+                except amdsmi_exception.AmdSmiLibraryException as e:
+                    if e.get_error_code() == amdsmi_interface.amdsmi_wrapper.AMDSMI_STATUS_NO_PERM:
+                        raise PermissionError('Error opening CPER file. This command requires elevation') from e
+                    if e.get_error_code() == amdsmi_interface.amdsmi_wrapper.AMDSMI_STATUS_FILE_NOT_FOUND:
+                        raise FileNotFoundError('Error opening CPER file. This command requires a CPER to be enabled.') from e
+                    if e.get_error_code() == amdsmi_interface.amdsmi_wrapper.AMDSMI_STATUS_FILE_ERROR:
+                        raise FileExistsError('Error opening CPER file. Unable to read CPER File') from e
+                    else:
+                        logging.debug(f"Error retrieving CPER entries: {e}")
+                        break
+                if entries:
+                    self.helpers.dump_entries(args.folder, entries, cper_data)
+                if len(entries) == 0 or not args.follow:
+                    break
+                cursor = new_cursor
+                time.sleep(5)
+                user_input = input()
+                if user_input == 'q':
+                    print("Escape Sequence Detected; Exiting")
+                    self.stop = True
+                    break
+
+
     def _event_thread(self, commands, i):
         devices = commands.device_handles
         if len(devices) == 0:

amdsmi_cli/amdsmi_helpers.py

@@ -19,18 +19,19 @@
 # IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
 # CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 
+import grp
+import json
 import logging
 import math
+import multiprocessing
 import os
-import grp
 import platform
+import re
 import sys
 import time
-import re
-import multiprocessing
-import json
 
 from enum import Enum
+from pathlib import Path
 from typing import List, Set, Union
 
 from amdsmi_init import *
@@ -55,7 +56,11 @@ class AMDSMIHelpers():
 
         self._is_linux = False
         self._is_windows = False
+
+        # Counts and Tracking variables
         self._count_of_sets_called = 0
+        self._count_of_cper_files = 0
+
 
         # Check if the system is a virtual OS
         if self.operating_system.startswith("Linux"):
@@ -95,6 +100,7 @@ class AMDSMIHelpers():
             except amdsmi_exception.AmdSmiLibraryException as e:
                 logging.debug("Unable to determine virtualization status: " + str(e.get_error_code()))
 
+
     def increment_set_count(self):
         self._count_of_sets_called += 1
 
@@ -103,6 +109,14 @@ class AMDSMIHelpers():
         return self._count_of_sets_called
 
 
+    def increment_cper_count(self):
+        self._count_of_cper_files += 1
+
+
+    def get_cper_count(self):
+        return self._count_of_cper_files
+
+
     def is_virtual_os(self):
         return self._is_virtual_os
 
@@ -116,6 +130,7 @@ class AMDSMIHelpers():
         # Returns True if system is baremetal, if system is hypervisor this should return False
         return self._is_baremetal
 
+
     def is_passthrough(self):
         return self._is_passthrough
 
@@ -197,7 +212,7 @@ class AMDSMIHelpers():
         """
         cpu_choices = {}
         cpu_choices_str = ""
-        #import pdb;pdb.set_trace()
+
         try:
             cpu_handles = []
             # amdsmi_get_cpusocket_handles() returns the cpu socket handles stored for cpu_id
@@ -230,6 +245,7 @@ class AMDSMIHelpers():
 
         return (cpu_choices, cpu_choices_str)
 
+
     def get_core_choices(self):
         """Return dictionary of possible Core choices and string of the output:
             Dictionary will be in format: coress[ID]: Device Handle)
@@ -705,11 +721,13 @@ class AMDSMIHelpers():
         except:
             return False
 
+
     def get_perf_levels(self):
         perf_levels_str = [clock.name for clock in amdsmi_interface.AmdSmiDevPerfLevel]
         perf_levels_int = list(set(clock.value for clock in amdsmi_interface.AmdSmiDevPerfLevel))
         return perf_levels_str, perf_levels_int
 
+
     def get_accelerator_partition_profile_config(self):
         device_handles = amdsmi_interface.amdsmi_get_processor_handles()
         accelerator_partition_profiles = {'profile_indices':[], 'profile_types':[], 'memory_caps': []}
@@ -726,6 +744,7 @@ class AMDSMIHelpers():
                 break
         return accelerator_partition_profiles
 
+
     def get_accelerator_choices_types_indices(self):
         return_val = ("N/A", {'profile_indices':[], 'profile_types':[]})
         accelerator_partition_profiles = self.get_accelerator_partition_profile_config()
@@ -735,6 +754,7 @@ class AMDSMIHelpers():
             return_val = (accelerator_choices, accelerator_partition_profiles)
         return return_val
 
+
     def get_memory_partition_types(self):
         memory_partitions_str = [partition.name for partition in amdsmi_interface.AmdSmiMemoryPartitionType]
         if 'UNKNOWN' in memory_partitions_str:
@@ -854,6 +874,7 @@ class AMDSMIHelpers():
         else:
             sys.exit('Confirmation not given. Exiting without setting value')
 
+
     def confirm_changing_memory_partition_gpu_reload_warning(self, auto_respond=False):
         """ Print the warning for running outside of specification and prompt user to accept the terms.
 
@@ -879,6 +900,7 @@ class AMDSMIHelpers():
             print('Confirmation not given. Exiting without setting value')
             sys.exit(1)
 
+
     def is_valid_profile(self, profile):
         profile_presets = amdsmi_interface.amdsmi_wrapper.amdsmi_power_profile_preset_masks_t__enumvalues
         if profile in profile_presets:
@@ -924,6 +946,7 @@ class AMDSMIHelpers():
             return f"{value} {unit}".rstrip()
         return f"{value}"
 
+
     class SI_Unit(float, Enum):
         GIGA = 1000000000  # 10^9
         MEGA = 1000000     # 10^6
@@ -937,6 +960,7 @@ class AMDSMIHelpers():
         MICRO = 0.000001   # 10^-6
         NANO = 0.000000001 # 10^-9
 
+
     def convert_SI_unit(self, val: Union[int, float], unit_in: SI_Unit, unit_out = SI_Unit.BASE) -> Union[int, float]:
         """This function will convert a value into another
          scientific (SI) unit. Defaults unit_out to SI_Unit.BASE
@@ -956,6 +980,7 @@ class AMDSMIHelpers():
         else:
             raise TypeError("val must be an int or float")
 
+
     def get_pci_device_ids(self) -> Set[str]:
         pci_devices_path = "/sys/bus/pci/devices"
         pci_devices: set[str] = set()
@@ -969,6 +994,7 @@ class AMDSMIHelpers():
                 continue
         return pci_devices
 
+
     def progressbar(self, it, prefix="", size=60, out=sys.stdout, add_newline=False):
         count = len(it)
         if (add_newline):
@@ -985,12 +1011,14 @@ class AMDSMIHelpers():
             show(i+1)
         print("\n\n", end='\r', flush=True, file=out)
 
+
     def showProgressbar(self, title="", timeInSeconds=13, add_newline=False):
         if title != "":
             title += " "
         for i in self.progressbar(range(timeInSeconds), title, 40, add_newline=add_newline):
             time.sleep(1)
 
+
     def check_required_groups(self):
         """
         Check if the current user is a member of the required groups.
@@ -1016,3 +1044,60 @@ class AMDSMIHelpers():
             ) % ", ".join(sorted(missing_groups))
             print(msg)
             logging.warning(msg)
+
+    def hexdump(self, data, size, filepath):
+        """
+        Converts binary data to a hex dump string, similar to the hexdump utility.
+        """
+        def to_printable_ascii(byte):
+            return chr(byte) if 32 <= byte <= 126 else "."
+
+        with open(filepath, 'w') as f:
+            offset = 0
+            while offset < size:
+                chunk = data[offset:offset + 16]
+                hex_values = " ".join(f"{byte:02x}" for byte in chunk)
+                ascii_values = "".join(to_printable_ascii(byte) for byte in chunk)
+                print(f"{offset:08x} {hex_values:<48} |{ascii_values}|", file=f)
+                offset += 16
+
+    def dump_entries(self, folder, entries, cper_data):
+        if folder:
+            folder = Path(folder)
+            folder.mkdir(parents=True, exist_ok=True)  # Ensure folder exists
+
+            # Loop through all entries in the dictionary.
+            for entry_index, entry in enumerate(entries.values()):
+                # Assume 'entry' is a dictionary with keys: "error_severity" and "notify_type".
+                error_severity = entry.get("error_severity", "Unknown")
+                notify_type = entry.get("notify_type", "Unknown")
+
+                if error_severity == "non_fatal_uncorrected":
+                    prefix = "uncorrected"
+                elif error_severity == "non_fatal_corrected":
+                    prefix = "corrected"
+                elif error_severity == "fatal":
+                    prefix = "fatal"
+                    if notify_type == "BOOT":
+                        prefix = "boot"
+
+                # Construct a unique file name using the key to avoid overwriting
+                entry_file = f"{prefix}_{self.get_cper_count()}.json"
+                output_path = folder / entry_file
+
+                cper_data_file = f"{prefix}_{self.get_cper_count()}.cper"
+                cper_data_file_path = folder / cper_data_file
+                self.hexdump(cper_data[entry_index]["bytes"], cper_data[entry_index]["size"], cper_data_file_path)
+
+                try:
+                    with output_path.open("w") as f:
+                        logging.debug(f"Writing entry {self.get_cper_count()}: {entry} to {output_path}")
+                        # Dump the single entry as JSON, handling bytes via the lambda.
+                        f.write(json.dumps(entry, indent=2,
+                                           default=lambda o: o.decode('utf-8') if isinstance(o, bytes) else o))
+                except Exception as e:
+                    logging.error(f"Failed to write entry {self.get_cper_count()} to {output_path}: {e}")
+        else:
+            print(json.dumps(entries, indent=2,
+                             default=lambda o: o.decode('utf-8') if isinstance(o, bytes) else o))
+        self.increment_cper_count()

amdsmi_cli/amdsmi_logger.py

@@ -42,6 +42,7 @@ class AMDSMILogger():
         self.secondary_table_header = ""
         self.warning_message = ""
         self.helpers = AMDSMIHelpers()
+        self._cper_exit_message = True
 
 
     class LoggerFormat(Enum):
@@ -78,6 +79,26 @@ class AMDSMILogger():
         self.multiple_device_output.clear()
 
 
+    def cper_exit_message(self):
+        """ Store the cper exit message
+            params:
+                message (str) - message to store
+            return:
+                cper_exit_message (bool) - True if cper exit message is set
+        """
+        return self._cper_exit_message
+
+
+    def set_cper_exit_message(self, flag:bool):
+        """ Set the cper exit message
+            params:
+                flag (bool) - True if cper exit message is set
+            return:
+                Nothing
+        """
+        self._cper_exit_message = flag
+
+
     def _capitalize_keys(self, input_dict):
         output_dict = {}
         for key in input_dict.keys():

amdsmi_cli/amdsmi_parser.py

@@ -69,7 +69,7 @@ class AMDSMIParser(argparse.ArgumentParser):
     """
     def __init__(self, version, list, static, firmware, bad_pages, metric,
                  process, profile, event, topology, set_value, reset, monitor,
-                 rocmsmi, xgmi, partition):
+                 rocmsmi, xgmi, partition, ras):
 
         # Helper variables
         self.helpers = AMDSMIHelpers()
@@ -115,7 +115,7 @@ class AMDSMIParser(argparse.ArgumentParser):
         # Store possible subcommands & aliases for later errors
         self.possible_commands = ['version', 'list', 'static', 'firmware', 'ucode', 'bad-pages',
                                   'metric', 'process', 'profile', 'event', 'topology', 'set',
-                                  'reset', 'monitor', 'dmon', 'xgmi', 'partition']
+                                  'reset', 'monitor', 'dmon', 'xgmi', 'partition', 'ras']
 
         # Add all subparsers
         self._add_version_parser(self.subparsers, version)
@@ -134,6 +134,7 @@ class AMDSMIParser(argparse.ArgumentParser):
         self._add_rocm_smi_parser(self.subparsers, rocmsmi)
         self._add_xgmi_parser(self.subparsers, xgmi)
         self._add_partition_parser(self.subparsers, partition)
+        self._add_ras_parser(self.subparsers, ras)
 
 
     def _not_negative_int(self, int_value, sub_arg=None):
@@ -241,6 +242,24 @@ class AMDSMIParser(argparse.ArgumentParser):
         return AMDSMIFreqArgs
 
 
+    def _check_folder_path(self):
+        """ Argument action validator:
+            Returns a path to folder from the folder path provided.
+            If the path doesn't exist create it.
+        """
+        class CheckOutputFilePath(argparse.Action):
+            outputformat = self.helpers.get_output_format()
+            # Checks the values
+            def __call__(self, parser, args, values, option_string=None):
+                path = Path(values)
+                path.mkdir(parents=True, exist_ok=True)
+                if not path.exists():
+                    raise amdsmi_cli_exceptions.AmdSmiInvalidFilePathException(path, CheckOutputFilePath.outputformat)
+                elif path.is_dir():
+                    setattr(args, self.dest, path)
+        return CheckOutputFilePath
+
+
     def _check_output_file_path(self):
         """ Argument action validator:
             Returns a path to a file from the output file path provided.
@@ -408,7 +427,7 @@ class AMDSMIParser(argparse.ArgumentParser):
         return _CoreSelectAction
 
 
-    def _add_command_modifiers(self, subcommand_parser: argparse.ArgumentParser):
+    def _add_command_modifiers(self, subcommand_parser: argparse.ArgumentParser, logging_only=False):
         json_help = "Displays output in JSON format (human readable by default)."
         csv_help = "Displays output in CSV format (human readable by default)."
         file_help = "Saves output into a file on the provided path (stdout by default)."
@@ -418,12 +437,14 @@ class AMDSMIParser(argparse.ArgumentParser):
 
         command_modifier_group = subcommand_parser.add_argument_group('Command Modifiers')
 
-        # Output Format options
-        logging_args = command_modifier_group.add_mutually_exclusive_group()
-        logging_args.add_argument('--json', action='store_true', required=False, help=json_help)
-        logging_args.add_argument('--csv', action='store_true', required=False, help=csv_help)
+        if not logging_only:
+            # Output Format options
+            logging_args = command_modifier_group.add_mutually_exclusive_group()
+            logging_args.add_argument('--json', action='store_true', required=False, help=json_help)
+            logging_args.add_argument('--csv', action='store_true', required=False, help=csv_help)
+
+            command_modifier_group.add_argument('--file', action=self._check_output_file_path(), type=str, required=False, help=file_help)
 
-        command_modifier_group.add_argument('--file', action=self._check_output_file_path(), type=str, required=False, help=file_help)
         # Placing loglevel outside the subcommands so it can be used with any subcommand
         command_modifier_group.add_argument('--loglevel', action='store', type=str.upper, required=False, help=loglevel_help, default='ERROR', metavar='LEVEL',
                                             choices=loglevel_choices)
@@ -1398,6 +1419,49 @@ class AMDSMIParser(argparse.ArgumentParser):
         self._add_command_modifiers(partition_parser)
 
 
+    def _add_ras_parser(self, subparsers: argparse._SubParsersAction, func):
+        """
+        Adds the 'ras' subcommand.
+
+        Expected command:
+            amd-smi ras --cper --severity=nonfatal-uncorrected,fatal --folder <folder_name> --file_limit=1000 --follow
+
+        All parameters are provided via options; no positional arguments or optional --file/--gpu are used.
+        """
+        # Subparser help text
+        ras_help = "Retrieve CPER (RAS) entries from the driver"
+        ras_description = (
+            "Retrieve and decode CPER (RAS) entries from the kernel driver.\n"
+            "Supports filtering by severity, exporting to different formats, and continuous monitoring.\n"
+            "This command accepts options only; no positional arguments are required."
+        )
+
+        # Help text for RAS arguments
+        cper_help = "Trigger CPER data retrieval"
+
+        severity_choices = ["nonfatal-uncorrected", "fatal", "nonfatal-corrected", "all"]
+        severity_choices_str = ", ".join(severity_choices)
+        severity_help = f"Set the SEVERITY filters from the following:\n    {severity_choices_str}"
+        folder_help = "Folder to dump CPER report files"
+        file_limit_help = "Maximum number of entries per output file"
+        follow_help = "Continuously monitor for new entries"
+
+        ras_parser = subparsers.add_parser("ras", help=ras_help, description=ras_description)
+        ras_parser.formatter_class = lambda prog: AMDSMISubparserHelpFormatter(prog)
+        ras_parser.set_defaults(func=func)
+
+        # Required flags and arguments:
+        ras_parser.add_argument("--cper", action="store_true", required=True, help=cper_help)
+        ras_parser.add_argument("--severity", type=str.lower, nargs='+', default=['all'], help=severity_help, choices=severity_choices, metavar='SEVERITY')
+        ras_parser.add_argument("--folder", type=str, action=self._check_folder_path(), default=False, help=folder_help)
+        ras_parser.add_argument("--file_limit", type=self._positive_int, action='store', default=1000, help=file_limit_help)
+        ras_parser.add_argument("--follow", action="store_true", default=False, help=follow_help)
+
+        # Add common modifiers and device selection arguments.
+        self._add_device_arguments(ras_parser, required=False)
+        self._add_command_modifiers(ras_parser, logging_only=True)
+
+
     def error(self, message):
         outputformat = self.helpers.get_output_format()