| #!/usr/bin/env python3 |
| |
| r""" |
| See class prolog below for details. |
| """ |
| |
| import json |
| import logging |
| import os |
| import platform |
| import re |
| import subprocess |
| import sys |
| import time |
| from errno import EACCES, EPERM |
| |
| import yaml |
| |
| sys.dont_write_bytecode = True |
| |
| |
| script_dir = os.path.dirname(os.path.abspath(__file__)) |
| sys.path.append(script_dir) |
| # Walk path and append to sys.path |
| for root, dirs, files in os.walk(script_dir): |
| for dir in dirs: |
| sys.path.append(os.path.join(root, dir)) |
| |
| from ssh_utility import SSHRemoteclient # NOQA |
| from telnet_utility import TelnetRemoteclient # NOQA |
| |
| r""" |
| User define plugins python functions. |
| |
| It will imports files from directory plugins |
| |
| plugins |
| ├── file1.py |
| └── file2.py |
| |
| Example how to define in YAML: |
| - plugin: |
| - plugin_name: plugin.foo_func.foo_func_yaml |
| - plugin_args: |
| - arg1 |
| - arg2 |
| """ |
| plugin_dir = os.path.join(os.path.dirname(__file__), "plugins") |
| sys.path.append(plugin_dir) |
| |
| for module in os.listdir(plugin_dir): |
| if module == "__init__.py" or not module.endswith(".py"): |
| continue |
| |
| plugin_module = f"plugins.{module[:-3]}" |
| try: |
| plugin = __import__(plugin_module, globals(), locals(), [], 0) |
| except Exception as e: |
| print(f"PLUGIN: Exception: {e}") |
| print(f"PLUGIN: Module import failed: {module}") |
| continue |
| |
| r""" |
| This is for plugin functions returning data or responses to the caller |
| in YAML plugin setup. |
| |
| Example: |
| |
| - plugin: |
| - plugin_name: version = plugin.ssh_execution.ssh_execute_cmd |
| - plugin_args: |
| - ${hostname} |
| - ${username} |
| - ${password} |
| - "cat /etc/os-release | grep VERSION_ID | awk -F'=' '{print $2}'" |
| - plugin: |
| - plugin_name: plugin.print_vars.print_vars |
| - plugin_args: |
| - version |
| |
| where first plugin "version" var is used by another plugin in the YAML |
| block or plugin |
| |
| """ |
| global global_log_store_path |
| global global_plugin_dict |
| global global_plugin_list |
| |
| # Hold the plugin return values in dict and plugin return vars in list. |
| # Dict is to reference and update vars processing in parser where as |
| # list is for current vars from the plugin block which needs processing. |
| global_plugin_dict = {} |
| global_plugin_list = [] |
| |
| # Hold the plugin return named declared if function returned values are |
| # list,dict. |
| # Refer this name list to look up the plugin dict for eval() args function |
| # Example ['version'] |
| global_plugin_type_list = [] |
| |
| # Path where logs are to be stored or written. |
| global_log_store_path = "" |
| |
| # Plugin error state defaults. |
| plugin_error_dict = { |
| "exit_on_error": False, |
| "continue_on_error": False, |
| } |
| |
| |
| class ffdc_collector: |
| r""" |
| Execute commands from configuration file to collect log files. |
| Fetch and store generated files at the specified location. |
| |
| """ |
| |
| def __init__( |
| self, |
| hostname, |
| username, |
| password, |
| port_ssh, |
| port_https, |
| port_ipmi, |
| ffdc_config, |
| location, |
| remote_type, |
| remote_protocol, |
| env_vars, |
| econfig, |
| log_level, |
| ): |
| r""" |
| Initialize the FFDCCollector object with the provided parameters. |
| |
| This method initializes an FFDCCollector object with the given |
| attributes. The attributes represent the configuration for connecting |
| to a remote system, collecting log data, and storing the collected |
| data. |
| |
| Parameters: |
| hostname (str): Name or IP address of the targeted |
| (remote) system. |
| username (str): User on the targeted system with access |
| to log files. |
| password (str): Password for the user on the targeted |
| system. |
| port_ssh (int, optional): SSH port value. Defaults to 22. |
| port_https (int, optional): HTTPS port value. Defaults to 443. |
| port_ipmi (int, optional): IPMI port value. Defaults to 623. |
| ffdc_config (str): Configuration file listing commands |
| and files for FFDC. |
| location (str): Where to store collected log data. |
| remote_type (str): Block YAML type name of the remote |
| host. |
| remote_protocol (str): Protocol to use to collect data. |
| env_vars (dict, optional): User-defined CLI environment variables. |
| Defaults to None. |
| econfig (str, optional): User-defined environment variables |
| YAML file. Defaults to None. |
| log_level (str, optional): Log level for the collector. |
| Defaults to "INFO". |
| """ |
| |
| self.hostname = hostname |
| self.username = username |
| self.password = password |
| self.port_ssh = str(port_ssh) |
| self.port_https = str(port_https) |
| self.port_ipmi = str(port_ipmi) |
| self.ffdc_config = ffdc_config |
| self.location = location + "/" + remote_type.upper() |
| self.ssh_remoteclient = None |
| self.telnet_remoteclient = None |
| self.ffdc_dir_path = "" |
| self.ffdc_prefix = "" |
| self.target_type = remote_type.upper() |
| self.remote_protocol = remote_protocol.upper() |
| self.env_vars = env_vars if env_vars else {} |
| self.econfig = econfig if econfig else {} |
| self.start_time = 0 |
| self.elapsed_time = "" |
| self.env_dict = {} |
| self.logger = None |
| |
| """ |
| Set prefix values for SCP files and directories. |
| Since the time stamp is at second granularity, these values are set |
| here to be sure that all files for this run will have the same |
| timestamps and be saved in the same directory. |
| self.location == local system for now |
| """ |
| self.set_ffdc_default_store_path() |
| |
| # Logger for this run. Need to be after set_ffdc_default_store_path() |
| self.script_logging(getattr(logging, log_level.upper())) |
| |
| # Verify top level directory exists for storage |
| self.validate_local_store(self.location) |
| |
| if self.verify_script_env(): |
| try: |
| with open(self.ffdc_config, "r") as file: |
| self.ffdc_actions = yaml.safe_load(file) |
| except yaml.YAMLError as e: |
| self.logger.error(e) |
| sys.exit(-1) |
| |
| if self.target_type not in self.ffdc_actions: |
| self.logger.error( |
| "\n\tERROR: %s is not listed in %s.\n\n" |
| % (self.target_type, self.ffdc_config) |
| ) |
| sys.exit(-1) |
| |
| self.logger.info("\n\tENV: User define input YAML variables") |
| self.env_dict = self.load_env() |
| else: |
| sys.exit(-1) |
| |
| def verify_script_env(self): |
| r""" |
| Verify that all required environment variables are set. |
| |
| This method checks if all required environment variables are set. |
| If any required variable is missing, the method returns False. |
| Otherwise, it returns True. |
| |
| Returns: |
| bool: True if all required environment variables are set, |
| False otherwise. |
| """ |
| # Import to log version |
| import click |
| import paramiko |
| |
| run_env_ok = True |
| |
| try: |
| redfishtool_version = ( |
| self.run_tool_cmd("redfishtool -V").split(" ")[2].strip("\n") |
| ) |
| except Exception as e: |
| self.logger.error("\tEXCEPTION redfishtool: %s", e) |
| redfishtool_version = "Not Installed (optional)" |
| |
| try: |
| ipmitool_version = self.run_tool_cmd("ipmitool -V").split(" ")[2] |
| except Exception as e: |
| self.logger.error("\tEXCEPTION ipmitool: %s", e) |
| ipmitool_version = "Not Installed (optional)" |
| |
| self.logger.info("\n\t---- Script host environment ----") |
| self.logger.info( |
| "\t{:<10} {:<10}".format("Script hostname", os.uname()[1]) |
| ) |
| self.logger.info( |
| "\t{:<10} {:<10}".format("Script host os", platform.platform()) |
| ) |
| self.logger.info( |
| "\t{:<10} {:>10}".format("Python", platform.python_version()) |
| ) |
| self.logger.info("\t{:<10} {:>10}".format("PyYAML", yaml.__version__)) |
| self.logger.info("\t{:<10} {:>10}".format("click", click.__version__)) |
| self.logger.info( |
| "\t{:<10} {:>10}".format("paramiko", paramiko.__version__) |
| ) |
| self.logger.info( |
| "\t{:<10} {:>9}".format("redfishtool", redfishtool_version) |
| ) |
| self.logger.info( |
| "\t{:<10} {:>12}".format("ipmitool", ipmitool_version) |
| ) |
| |
| if eval(yaml.__version__.replace(".", ",")) < (5, 3, 0): |
| self.logger.error( |
| "\n\tERROR: Python or python packages do not meet minimum" |
| " version requirement." |
| ) |
| self.logger.error( |
| "\tERROR: PyYAML version 5.3.0 or higher is needed.\n" |
| ) |
| run_env_ok = False |
| |
| self.logger.info("\t---- End script host environment ----") |
| return run_env_ok |
| |
| def script_logging(self, log_level_attr): |
| """ |
| Create a logger for the script with the specified log level. |
| |
| This method creates a logger for the script with the specified |
| log level. The logger is configured to write log messages to a file |
| and the console. |
| |
| self.logger = logging.getLogger(__name__) |
| |
| Setting logger with __name__ will add the trace |
| Example: |
| |
| INFO:ffdc_collector: System Type: OPENBMC |
| |
| Currently, set to empty purposely to log as |
| System Type: OPENBMC |
| |
| Parameters: |
| log_level_attr (str): The log level for the logger |
| (e.g., "DEBUG", "INFO", "WARNING", |
| "ERROR", "CRITICAL"). |
| |
| Returns: |
| None |
| """ |
| self.logger = logging.getLogger() |
| self.logger.setLevel(log_level_attr) |
| |
| log_file_handler = logging.FileHandler( |
| self.ffdc_dir_path + "collector.log" |
| ) |
| stdout_handler = logging.StreamHandler(sys.stdout) |
| |
| self.logger.addHandler(log_file_handler) |
| self.logger.addHandler(stdout_handler) |
| |
| # Turn off paramiko INFO logging |
| logging.getLogger("paramiko").setLevel(logging.WARNING) |
| |
| def target_is_pingable(self): |
| r""" |
| Check if the target system is ping-able. |
| |
| This method checks if the target system is reachable by sending an |
| ICMP echo request (ping). If the target system responds to the ping, |
| the method returns True. Otherwise, it returns False. |
| |
| Returns: |
| bool: True if the target system is ping-able, False otherwise. |
| """ |
| response = os.system("ping -c 2 %s 2>&1 >/dev/null" % self.hostname) |
| if response == 0: |
| self.logger.info( |
| "\n\t[Check] %s is ping-able.\t\t [OK]" % self.hostname |
| ) |
| return True |
| else: |
| self.logger.error( |
| "\n\tERROR: %s is not ping-able. FFDC collection aborted.\n" |
| % self.hostname |
| ) |
| sys.exit(-1) |
| return False |
| |
| def collect_ffdc(self): |
| r""" |
| Initiate FFDC collection based on the requested protocol. |
| |
| This method initiates FFDC (First Failure Data Capture) collection |
| based on the requested protocol (SSH,SCP, TELNET, REDFISH, IPMI). |
| The method establishes a connection to the target system using the |
| specified protocol and collects the required FFDC data. |
| |
| Returns: |
| None |
| """ |
| self.logger.info( |
| "\n\t---- Start communicating with %s ----" % self.hostname |
| ) |
| self.start_time = time.time() |
| |
| # Find the list of target and protocol supported. |
| check_protocol_list = [] |
| config_dict = self.ffdc_actions |
| |
| for target_type in config_dict.keys(): |
| if self.target_type != target_type: |
| continue |
| |
| for k, v in config_dict[target_type].items(): |
| if v["PROTOCOL"][0] not in check_protocol_list: |
| check_protocol_list.append(v["PROTOCOL"][0]) |
| |
| self.logger.info( |
| "\n\t %s protocol type: %s" |
| % (self.target_type, check_protocol_list) |
| ) |
| |
| verified_working_protocol = self.verify_protocol(check_protocol_list) |
| |
| if verified_working_protocol: |
| self.logger.info( |
| "\n\t---- Completed protocol pre-requisite check ----\n" |
| ) |
| |
| # Verify top level directory exists for storage |
| self.validate_local_store(self.location) |
| |
| if (self.remote_protocol not in verified_working_protocol) and ( |
| self.remote_protocol != "ALL" |
| ): |
| self.logger.info( |
| "\n\tWorking protocol list: %s" % verified_working_protocol |
| ) |
| self.logger.error( |
| "\tERROR: Requested protocol %s is not in working protocol" |
| " list.\n" % self.remote_protocol |
| ) |
| sys.exit(-1) |
| else: |
| self.generate_ffdc(verified_working_protocol) |
| |
| def ssh_to_target_system(self): |
| r""" |
| Establish an SSH connection to the target system. |
| |
| This method establishes an SSH connection to the target system using |
| the provided hostname, username, password, and SSH port. If the |
| connection is successful, the method returns True. Otherwise, it logs |
| an error message and returns False. |
| |
| Returns: |
| bool: True if the connection is successful, False otherwise. |
| """ |
| |
| self.ssh_remoteclient = SSHRemoteclient( |
| self.hostname, self.username, self.password, self.port_ssh |
| ) |
| |
| if self.ssh_remoteclient.ssh_remoteclient_login(): |
| self.logger.info( |
| "\n\t[Check] %s SSH connection established.\t [OK]" |
| % self.hostname |
| ) |
| |
| # Check scp connection. |
| # If scp connection fails, |
| # continue with FFDC generation but skip scp files to local host. |
| self.ssh_remoteclient.scp_connection() |
| return True |
| else: |
| self.logger.info( |
| "\n\t[Check] %s SSH connection.\t [NOT AVAILABLE]" |
| % self.hostname |
| ) |
| return False |
| |
| def telnet_to_target_system(self): |
| r""" |
| Establish a Telnet connection to the target system. |
| |
| This method establishes a Telnet connection to the target system using |
| the provided hostname, username, and Telnet port. If the connection is |
| successful, the method returns True. Otherwise, it logs an error |
| message and returns False. |
| |
| Returns: |
| bool: True if the connection is successful, False otherwise. |
| """ |
| self.telnet_remoteclient = TelnetRemoteclient( |
| self.hostname, self.username, self.password |
| ) |
| if self.telnet_remoteclient.tn_remoteclient_login(): |
| self.logger.info( |
| "\n\t[Check] %s Telnet connection established.\t [OK]" |
| % self.hostname |
| ) |
| return True |
| else: |
| self.logger.info( |
| "\n\t[Check] %s Telnet connection.\t [NOT AVAILABLE]" |
| % self.hostname |
| ) |
| return False |
| |
| def generate_ffdc(self, working_protocol_list): |
| r""" |
| Generate FFDC (First Failure Data Capture) based on the remote host |
| type and working protocols. |
| |
| This method determines the actions to be performed for generating FFDC |
| based on the remote host type and the list of confirmed working |
| protocols. The method iterates through the available actions for the |
| remote host type and checks if any of the working protocols are |
| supported. If a supported protocol is found, the method executes the |
| corresponding FFDC generation action. |
| |
| Parameters: |
| working_protocol_list (list): A list of confirmed working |
| protocols to connect to the |
| remote host. |
| |
| Returns: |
| None |
| """ |
| self.logger.info( |
| "\n\t---- Executing commands on " + self.hostname + " ----" |
| ) |
| self.logger.info( |
| "\n\tWorking protocol list: %s" % working_protocol_list |
| ) |
| |
| config_dict = self.ffdc_actions |
| for target_type in config_dict.keys(): |
| if self.target_type != target_type: |
| continue |
| |
| self.logger.info("\n\tFFDC Path: %s " % self.ffdc_dir_path) |
| global_plugin_dict["global_log_store_path"] = self.ffdc_dir_path |
| self.logger.info("\tSystem Type: %s" % target_type) |
| for k, v in config_dict[target_type].items(): |
| protocol = v["PROTOCOL"][0] |
| |
| if ( |
| self.remote_protocol not in working_protocol_list |
| and self.remote_protocol != "ALL" |
| ) or protocol not in working_protocol_list: |
| continue |
| |
| if protocol in working_protocol_list: |
| if protocol in ["SSH", "SCP"]: |
| self.protocol_ssh(protocol, target_type, k) |
| elif protocol == "TELNET": |
| self.protocol_telnet(target_type, k) |
| elif protocol in ["REDFISH", "IPMI", "SHELL"]: |
| self.protocol_service_execute(protocol, target_type, k) |
| else: |
| self.logger.error( |
| "\n\tERROR: %s is not available for %s." |
| % (protocol, self.hostname) |
| ) |
| |
| # Close network connection after collecting all files |
| self.elapsed_time = time.strftime( |
| "%H:%M:%S", time.gmtime(time.time() - self.start_time) |
| ) |
| self.logger.info("\n\tTotal time taken: %s" % self.elapsed_time) |
| if self.ssh_remoteclient: |
| self.ssh_remoteclient.ssh_remoteclient_disconnect() |
| if self.telnet_remoteclient: |
| self.telnet_remoteclient.tn_remoteclient_disconnect() |
| |
| def protocol_ssh(self, protocol, target_type, sub_type): |
| r""" |
| Perform actions using SSH and SCP protocols. |
| |
| This method executes a set of commands using the SSH protocol to |
| connect to the target system and collect FFDC data. The method takes |
| the protocol, target type, and sub-type as arguments and performs the |
| corresponding actions based on the provided parameters. |
| |
| Parameters: |
| protocol (str): The protocol to execute (SSH or SCP). |
| target_type (str): The type group of the remote host. |
| sub_type (str): The group type of commands to execute. |
| |
| Returns: |
| None |
| """ |
| if protocol == "SCP": |
| self.group_copy(self.ffdc_actions[target_type][sub_type]) |
| else: |
| self.collect_and_copy_ffdc( |
| self.ffdc_actions[target_type][sub_type] |
| ) |
| |
| def protocol_telnet(self, target_type, sub_type): |
| r""" |
| Perform actions using the Telnet protocol. |
| |
| This method executes a set of commands using the Telnet protocol to |
| connect to the target system and collect FFDC data. The method takes |
| the target type and sub-type as arguments and performs the |
| corresponding actions based on the provided parameters. |
| |
| Parameters: |
| target_type (str): The type group of the remote host. |
| sub_type (str): The group type of commands to execute. |
| |
| Returns: |
| None |
| """ |
| self.logger.info( |
| "\n\t[Run] Executing commands on %s using %s" |
| % (self.hostname, "TELNET") |
| ) |
| telnet_files_saved = [] |
| progress_counter = 0 |
| list_of_commands = self.ffdc_actions[target_type][sub_type]["COMMANDS"] |
| for index, each_cmd in enumerate(list_of_commands, start=0): |
| command_txt, command_timeout = self.unpack_command(each_cmd) |
| result = self.telnet_remoteclient.execute_command( |
| command_txt, command_timeout |
| ) |
| if result: |
| try: |
| targ_file = self.ffdc_actions[target_type][sub_type][ |
| "FILES" |
| ][index] |
| except IndexError: |
| targ_file = command_txt |
| self.logger.warning( |
| "\n\t[WARN] Missing filename to store data from" |
| " telnet %s." % each_cmd |
| ) |
| self.logger.warning( |
| "\t[WARN] Data will be stored in %s." % targ_file |
| ) |
| targ_file_with_path = ( |
| self.ffdc_dir_path + self.ffdc_prefix + targ_file |
| ) |
| # Creates a new file |
| with open(targ_file_with_path, "w") as fp: |
| fp.write(result) |
| fp.close |
| telnet_files_saved.append(targ_file) |
| progress_counter += 1 |
| self.print_progress(progress_counter) |
| self.logger.info("\n\t[Run] Commands execution completed.\t\t [OK]") |
| for file in telnet_files_saved: |
| self.logger.info("\n\t\tSuccessfully save file " + file + ".") |
| |
| def protocol_service_execute(self, protocol, target_type, sub_type): |
| r""" |
| Perform actions for a given protocol. |
| |
| This method executes a set of commands using the specified protocol to |
| connect to the target system and collect FFDC data. The method takes |
| the protocol, target type, and sub-type as arguments and performs the |
| corresponding actions based on the provided parameters. |
| |
| Parameters: |
| protocol (str): The protocol to execute |
| (REDFISH, IPMI, or SHELL). |
| target_type (str): The type group of the remote host. |
| sub_type (str): The group type of commands to execute. |
| |
| Returns: |
| None |
| """ |
| self.logger.info( |
| "\n\t[Run] Executing commands to %s using %s" |
| % (self.hostname, protocol) |
| ) |
| executed_files_saved = [] |
| progress_counter = 0 |
| list_of_cmd = self.get_command_list( |
| self.ffdc_actions[target_type][sub_type] |
| ) |
| for index, each_cmd in enumerate(list_of_cmd, start=0): |
| plugin_call = False |
| if isinstance(each_cmd, dict): |
| if "plugin" in each_cmd: |
| # If the error is set and plugin explicitly |
| # requested to skip execution on error.. |
| if plugin_error_dict[ |
| "exit_on_error" |
| ] and self.plugin_error_check(each_cmd["plugin"]): |
| self.logger.info( |
| "\n\t[PLUGIN-ERROR] exit_on_error: %s" |
| % plugin_error_dict["exit_on_error"] |
| ) |
| self.logger.info( |
| "\t[PLUGIN-SKIP] %s" % each_cmd["plugin"][0] |
| ) |
| continue |
| plugin_call = True |
| # call the plugin |
| self.logger.info("\n\t[PLUGIN-START]") |
| result = self.execute_plugin_block(each_cmd["plugin"]) |
| self.logger.info("\t[PLUGIN-END]\n") |
| else: |
| each_cmd = self.yaml_env_and_plugin_vars_populate(each_cmd) |
| |
| if not plugin_call: |
| result = self.run_tool_cmd(each_cmd) |
| if result: |
| try: |
| file_name = self.get_file_list( |
| self.ffdc_actions[target_type][sub_type] |
| )[index] |
| # If file is specified as None. |
| if file_name == "None": |
| continue |
| targ_file = self.yaml_env_and_plugin_vars_populate( |
| file_name |
| ) |
| except IndexError: |
| targ_file = each_cmd.split("/")[-1] |
| self.logger.warning( |
| "\n\t[WARN] Missing filename to store data from %s." |
| % each_cmd |
| ) |
| self.logger.warning( |
| "\t[WARN] Data will be stored in %s." % targ_file |
| ) |
| |
| targ_file_with_path = ( |
| self.ffdc_dir_path + self.ffdc_prefix + targ_file |
| ) |
| |
| # Creates a new file |
| with open(targ_file_with_path, "w") as fp: |
| if isinstance(result, dict): |
| fp.write(json.dumps(result)) |
| else: |
| fp.write(result) |
| fp.close |
| executed_files_saved.append(targ_file) |
| |
| progress_counter += 1 |
| self.print_progress(progress_counter) |
| |
| self.logger.info("\n\t[Run] Commands execution completed.\t\t [OK]") |
| |
| for file in executed_files_saved: |
| self.logger.info("\n\t\tSuccessfully save file " + file + ".") |
| |
| def collect_and_copy_ffdc( |
| self, ffdc_actions_for_target_type, form_filename=False |
| ): |
| r""" |
| Send commands and collect FFDC data from the targeted system. |
| |
| This method sends a set of commands and collects FFDC data from the |
| targeted system based on the provided ffdc_actions_for_target_type |
| dictionary. The method also has an optional form_filename parameter, |
| which, if set to True, prepends the target type to the output file |
| name. |
| |
| Parameters: |
| ffdc_actions_for_target_type (dict): A dictionary containing |
| commands and files for the |
| selected remote host type. |
| form_filename (bool, optional): If True, prepends the target |
| type to the output file name. |
| Defaults to False. |
| |
| Returns: |
| None |
| """ |
| # Executing commands, if any |
| self.ssh_execute_ffdc_commands( |
| ffdc_actions_for_target_type, form_filename |
| ) |
| |
| # Copying files |
| if self.ssh_remoteclient.scpclient: |
| self.logger.info( |
| "\n\n\tCopying FFDC files from remote system %s.\n" |
| % self.hostname |
| ) |
| |
| # Retrieving files from target system |
| list_of_files = self.get_file_list(ffdc_actions_for_target_type) |
| self.scp_ffdc( |
| self.ffdc_dir_path, |
| self.ffdc_prefix, |
| form_filename, |
| list_of_files, |
| ) |
| else: |
| self.logger.info( |
| "\n\n\tSkip copying FFDC files from remote system %s.\n" |
| % self.hostname |
| ) |
| |
| def get_command_list(self, ffdc_actions_for_target_type): |
| r""" |
| Fetch a list of commands from the configuration file. |
| |
| This method retrieves a list of commands from the |
| ffdc_actions_for_target_type dictionary, which contains commands and |
| files for the selected remote host type. The method returns the list |
| of commands. |
| |
| Parameters: |
| ffdc_actions_for_target_type (dict): A dictionary containing |
| commands and files for the |
| selected remote host type. |
| |
| Returns: |
| list: A list of commands. |
| """ |
| try: |
| list_of_commands = ffdc_actions_for_target_type["COMMANDS"] |
| except KeyError: |
| list_of_commands = [] |
| return list_of_commands |
| |
| def get_file_list(self, ffdc_actions_for_target_type): |
| r""" |
| Fetch a list of files from the configuration file. |
| |
| This method retrieves a list of files from the |
| ffdc_actions_for_target_type dictionary, which contains commands and |
| files for the selected remote host type. The method returns the list |
| of files. |
| |
| Parameters: |
| ffdc_actions_for_target_type (dict): A dictionary containing |
| commands and files for the |
| selected remote host type. |
| |
| Returns: |
| list: A list of files. |
| """ |
| try: |
| list_of_files = ffdc_actions_for_target_type["FILES"] |
| except KeyError: |
| list_of_files = [] |
| return list_of_files |
| |
| def unpack_command(self, command): |
| r""" |
| Unpack a command from the configuration file, handling both dictionary |
| and string inputs. |
| |
| This method takes a command from the configuration file, which can be |
| either a dictionary or a string. If the input is a dictionary, the |
| method extracts the command text and timeout from the dictionary. |
| If the input is a string, the method assumes a default timeout of |
| 60 seconds. |
| The method returns a tuple containing the command text and timeout. |
| |
| Parameters: |
| command (dict or str): A command from the configuration file, |
| which can be either a dictionary or a |
| string. |
| |
| Returns: |
| tuple: A tuple containing the command text and timeout. |
| """ |
| if isinstance(command, dict): |
| command_txt = next(iter(command)) |
| command_timeout = next(iter(command.values())) |
| elif isinstance(command, str): |
| command_txt = command |
| # Default command timeout 60 seconds |
| command_timeout = 60 |
| |
| return command_txt, command_timeout |
| |
| def ssh_execute_ffdc_commands( |
| self, ffdc_actions_for_target_type, form_filename=False |
| ): |
| r""" |
| Send commands in the ffdc_config file to the targeted system using SSH. |
| |
| This method sends a set of commands and collects FFDC data from the |
| targeted system using the SSH protocol. The method takes the |
| ffdc_actions_for_target_type dictionary and an optional |
| form_filename parameter as arguments. |
| |
| If form_filename is set to True, the method prepends the target type |
| to the output file name. The method returns the output of the executed |
| commands. |
| |
| It also prints the progress counter string + on the console. |
| |
| Parameters: |
| ffdc_actions_for_target_type (dict): A dictionary containing |
| commands and files for the |
| selected remote host type. |
| form_filename (bool, optional): If True, prepends the target |
| type to the output file name. |
| Defaults to False. |
| |
| Returns: |
| None |
| """ |
| self.logger.info( |
| "\n\t[Run] Executing commands on %s using %s" |
| % (self.hostname, ffdc_actions_for_target_type["PROTOCOL"][0]) |
| ) |
| |
| list_of_commands = self.get_command_list(ffdc_actions_for_target_type) |
| # If command list is empty, returns |
| if not list_of_commands: |
| return |
| |
| progress_counter = 0 |
| for command in list_of_commands: |
| command_txt, command_timeout = self.unpack_command(command) |
| |
| if form_filename: |
| command_txt = str(command_txt % self.target_type) |
| |
| ( |
| cmd_exit_code, |
| err, |
| response, |
| ) = self.ssh_remoteclient.execute_command( |
| command_txt, command_timeout |
| ) |
| |
| if cmd_exit_code: |
| self.logger.warning( |
| "\n\t\t[WARN] %s exits with code %s." |
| % (command_txt, str(cmd_exit_code)) |
| ) |
| self.logger.warning("\t\t[WARN] %s " % err) |
| |
| progress_counter += 1 |
| self.print_progress(progress_counter) |
| |
| self.logger.info("\n\t[Run] Commands execution completed.\t\t [OK]") |
| |
| def group_copy(self, ffdc_actions_for_target_type): |
| r""" |
| SCP a group of files (wildcard) from the remote host. |
| |
| This method copies a group of files from the remote host using the SCP |
| protocol. The method takes the fdc_actions_for_target_type dictionary |
| as an argument, which contains commands and files for the selected |
| remote host type. |
| |
| Parameters: |
| fdc_actions_for_target_type (dict): A dictionary containing |
| commands and files for the |
| selected remote host type. |
| |
| Returns: |
| None |
| """ |
| if self.ssh_remoteclient.scpclient: |
| self.logger.info( |
| "\n\tCopying files from remote system %s via SCP.\n" |
| % self.hostname |
| ) |
| |
| list_of_commands = self.get_command_list( |
| ffdc_actions_for_target_type |
| ) |
| # If command list is empty, returns |
| if not list_of_commands: |
| return |
| |
| for command in list_of_commands: |
| try: |
| command = self.yaml_env_and_plugin_vars_populate(command) |
| except IndexError: |
| self.logger.error("\t\tInvalid command %s" % command) |
| continue |
| |
| ( |
| cmd_exit_code, |
| err, |
| response, |
| ) = self.ssh_remoteclient.execute_command(command) |
| |
| # If file does not exist, code take no action. |
| # cmd_exit_code is ignored for this scenario. |
| if response: |
| scp_result = self.ssh_remoteclient.scp_file_from_remote( |
| response.split("\n"), self.ffdc_dir_path |
| ) |
| if scp_result: |
| self.logger.info( |
| "\t\tSuccessfully copied from " |
| + self.hostname |
| + ":" |
| + command |
| ) |
| else: |
| self.logger.info("\t\t%s has no result" % command) |
| |
| else: |
| self.logger.info( |
| "\n\n\tSkip copying files from remote system %s.\n" |
| % self.hostname |
| ) |
| |
| def scp_ffdc( |
| self, |
| targ_dir_path, |
| targ_file_prefix, |
| form_filename, |
| file_list=None, |
| quiet=None, |
| ): |
| r""" |
| SCP all files in the file_dict to the indicated directory on the local |
| system. |
| |
| This method copies all files specified in the file_dict dictionary |
| from the targeted system to the local system using the SCP protocol. |
| The method takes the target directory path, target file prefix, and a |
| boolean flag form_filename as required arguments. |
| |
| The file_dict argument is optional and contains the files to be copied. |
| The quiet argument is also optional and, if set to True, suppresses |
| the output of the SCP operation. |
| |
| Parameters: |
| targ_dir_path (str): The path of the directory to receive |
| the files on the local system. |
| targ_file_prefix (str): Prefix which will be prepended to each |
| target file's name. |
| form_filename (bool): If True, prepends the target type to |
| the file names. |
| file_dict (dict, optional): A dictionary containing the files to |
| be copied. Defaults to None. |
| quiet (bool, optional): If True, suppresses the output of the |
| SCP operation. Defaults to None. |
| |
| Returns: |
| None |
| """ |
| progress_counter = 0 |
| for filename in file_list: |
| if form_filename: |
| filename = str(filename % self.target_type) |
| source_file_path = filename |
| targ_file_path = ( |
| targ_dir_path + targ_file_prefix + filename.split("/")[-1] |
| ) |
| |
| # If source file name contains wild card, copy filename as is. |
| if "*" in source_file_path: |
| scp_result = self.ssh_remoteclient.scp_file_from_remote( |
| source_file_path, self.ffdc_dir_path |
| ) |
| else: |
| scp_result = self.ssh_remoteclient.scp_file_from_remote( |
| source_file_path, targ_file_path |
| ) |
| |
| if not quiet: |
| if scp_result: |
| self.logger.info( |
| "\t\tSuccessfully copied from " |
| + self.hostname |
| + ":" |
| + source_file_path |
| + ".\n" |
| ) |
| else: |
| self.logger.info( |
| "\t\tFail to copy from " |
| + self.hostname |
| + ":" |
| + source_file_path |
| + ".\n" |
| ) |
| else: |
| progress_counter += 1 |
| self.print_progress(progress_counter) |
| |
| def set_ffdc_default_store_path(self): |
| r""" |
| Set default values for self.ffdc_dir_path and self.ffdc_prefix. |
| |
| This method sets default values for the self.ffdc_dir_path and |
| self.ffdc_prefix class variables. |
| |
| The collected FFDC files will be stored in the directory |
| /self.location/hostname_timestr/, with individual files having the |
| format timestr_filename where timestr is in %Y%m%d-%H%M%S. |
| |
| Returns: |
| None |
| """ |
| timestr = time.strftime("%Y%m%d-%H%M%S") |
| self.ffdc_dir_path = ( |
| self.location + "/" + self.hostname + "_" + timestr + "/" |
| ) |
| self.ffdc_prefix = timestr + "_" |
| self.validate_local_store(self.ffdc_dir_path) |
| |
| # Need to verify local store path exists prior to instantiate this class. |
| # This class method to validate log path before referencing this class. |
| @classmethod |
| def validate_local_store(cls, dir_path): |
| r""" |
| Ensure the specified directory exists to store FFDC files locally. |
| |
| This method checks if the provided dir_path exists. If the directory |
| does not exist, the method creates it. The method does not return any |
| value. |
| |
| Parameters: |
| dir_path (str): The directory path where collected FFDC data files |
| will be stored. |
| |
| Returns: |
| None |
| """ |
| if not os.path.exists(dir_path): |
| try: |
| os.makedirs(dir_path, 0o755) |
| except (IOError, OSError) as e: |
| # PermissionError |
| if e.errno == EPERM or e.errno == EACCES: |
| print( |
| "\tERROR: os.makedirs %s failed with" |
| " PermissionError.\n" % dir_path |
| ) |
| else: |
| print( |
| "\tERROR: os.makedirs %s failed with %s.\n" |
| % (dir_path, e.strerror) |
| ) |
| sys.exit(-1) |
| |
| def print_progress(self, progress): |
| r""" |
| Print the current activity progress. |
| |
| This method prints the current activity progress using the provided |
| progress counter. The method does not return any value. |
| |
| Parameters: |
| progress (int): The current activity progress counter. |
| |
| Returns: |
| None |
| """ |
| sys.stdout.write("\r\t" + "+" * progress) |
| sys.stdout.flush() |
| time.sleep(0.1) |
| |
| def verify_redfish(self): |
| r""" |
| Verify if the remote host has the Redfish service active. |
| |
| This method checks if the remote host has the Redfish service active |
| by sending a GET request to the Redfish base URL /redfish/v1/. |
| If the request is successful (status code 200), the method returns |
| stdout output of the run else error message. |
| |
| Returns: |
| str: Redfish service executed output. |
| """ |
| redfish_parm = ( |
| "redfishtool -r " |
| + self.hostname |
| + ":" |
| + self.port_https |
| + " -S Always raw GET /redfish/v1/" |
| ) |
| return self.run_tool_cmd(redfish_parm, True) |
| |
| def verify_ipmi(self): |
| r""" |
| Verify if the remote host has the IPMI LAN service active. |
| |
| This method checks if the remote host has the IPMI LAN service active |
| by sending an IPMI "power status" command. |
| |
| If the command is successful (returns a non-empty response), |
| else error message. |
| |
| Returns: |
| str: IPMI LAN service executed output. |
| """ |
| if self.target_type == "OPENBMC": |
| ipmi_parm = ( |
| "ipmitool -I lanplus -C 17 -U " |
| + self.username |
| + " -P " |
| + self.password |
| + " -H " |
| + self.hostname |
| + " -p " |
| + str(self.port_ipmi) |
| + " power status" |
| ) |
| else: |
| ipmi_parm = ( |
| "ipmitool -I lanplus -P " |
| + self.password |
| + " -H " |
| + self.hostname |
| + " -p " |
| + str(self.port_ipmi) |
| + " power status" |
| ) |
| |
| return self.run_tool_cmd(ipmi_parm, True) |
| |
| def run_tool_cmd(self, parms_string, quiet=False): |
| r""" |
| Run a CLI standard tool or script with the provided command options. |
| |
| This method runs a CLI standard tool or script with the provided |
| parms_string command options. If the quiet parameter is set to True, |
| the method suppresses the output of the command. |
| The method returns the output of the command as a string. |
| |
| Parameters: |
| parms_string (str): The command options for the CLI tool or |
| script. |
| quiet (bool, optional): If True, suppresses the output of the |
| command. Defaults to False. |
| |
| Returns: |
| str: The output of the command as a string. |
| """ |
| |
| result = subprocess.run( |
| [parms_string], |
| stdout=subprocess.PIPE, |
| stderr=subprocess.PIPE, |
| shell=True, |
| universal_newlines=True, |
| ) |
| |
| if result.stderr and not quiet: |
| if self.password in parms_string: |
| parms_string = parms_string.replace(self.password, "********") |
| self.logger.error("\n\t\tERROR with %s " % parms_string) |
| self.logger.error("\t\t" + result.stderr) |
| |
| return result.stdout |
| |
| def verify_protocol(self, protocol_list): |
| r""" |
| Perform a working check for the provided list of protocols. |
| |
| This method checks if the specified protocols are available on the |
| remote host. The method iterates through the protocol_list and |
| attempts to establish a connection using each protocol. |
| |
| If a connection is successfully established, the method append to the |
| list and if any protocol fails to connect, the method ignores it. |
| |
| Parameters: |
| protocol_list (list): A list of protocols to check. |
| |
| Returns: |
| list: All protocols are available list. |
| """ |
| |
| tmp_list = [] |
| if self.target_is_pingable(): |
| tmp_list.append("SHELL") |
| |
| for protocol in protocol_list: |
| if self.remote_protocol != "ALL": |
| if self.remote_protocol != protocol: |
| continue |
| |
| # Only check SSH/SCP once for both protocols |
| if ( |
| protocol == "SSH" |
| or protocol == "SCP" |
| and protocol not in tmp_list |
| ): |
| if self.ssh_to_target_system(): |
| # Add only what user asked. |
| if self.remote_protocol != "ALL": |
| tmp_list.append(self.remote_protocol) |
| else: |
| tmp_list.append("SSH") |
| tmp_list.append("SCP") |
| |
| if protocol == "TELNET": |
| if self.telnet_to_target_system(): |
| tmp_list.append(protocol) |
| |
| if protocol == "REDFISH": |
| if self.verify_redfish(): |
| tmp_list.append(protocol) |
| self.logger.info( |
| "\n\t[Check] %s Redfish Service.\t\t [OK]" |
| % self.hostname |
| ) |
| else: |
| self.logger.info( |
| "\n\t[Check] %s Redfish Service.\t\t [NOT AVAILABLE]" |
| % self.hostname |
| ) |
| |
| if protocol == "IPMI": |
| if self.verify_ipmi(): |
| tmp_list.append(protocol) |
| self.logger.info( |
| "\n\t[Check] %s IPMI LAN Service.\t\t [OK]" |
| % self.hostname |
| ) |
| else: |
| self.logger.info( |
| "\n\t[Check] %s IPMI LAN Service.\t\t [NOT AVAILABLE]" |
| % self.hostname |
| ) |
| |
| return tmp_list |
| |
| def load_env(self): |
| r""" |
| Load the user environment variables from a YAML file. |
| |
| This method reads the environment variables from a YAML file specified |
| in the ENV_FILE environment variable. If the file is not found or |
| there is an error reading the file, an exception is raised. |
| |
| The YAML file should have the following format: |
| |
| .. code-block:: yaml |
| |
| VAR_NAME: VAR_VALUE |
| |
| Where VAR_NAME is the name of the environment variable, and |
| VAR_VALUE is its value. |
| |
| After loading the environment variables, they are stored in the |
| self.env attribute for later use. |
| """ |
| |
| os.environ["hostname"] = self.hostname |
| os.environ["username"] = self.username |
| os.environ["password"] = self.password |
| os.environ["port_ssh"] = self.port_ssh |
| os.environ["port_https"] = self.port_https |
| os.environ["port_ipmi"] = self.port_ipmi |
| |
| # Append default Env. |
| self.env_dict["hostname"] = self.hostname |
| self.env_dict["username"] = self.username |
| self.env_dict["password"] = self.password |
| self.env_dict["port_ssh"] = self.port_ssh |
| self.env_dict["port_https"] = self.port_https |
| self.env_dict["port_ipmi"] = self.port_ipmi |
| |
| try: |
| tmp_env_dict = {} |
| if self.env_vars: |
| tmp_env_dict = json.loads(self.env_vars) |
| # Export ENV vars default. |
| for key, value in tmp_env_dict.items(): |
| os.environ[key] = value |
| self.env_dict[key] = str(value) |
| |
| # Load user specified ENV config YAML. |
| if self.econfig: |
| with open(self.econfig, "r") as file: |
| try: |
| tmp_env_dict = yaml.load(file, Loader=yaml.SafeLoader) |
| except yaml.YAMLError as e: |
| self.logger.error(e) |
| sys.exit(-1) |
| # Export ENV vars. |
| for key, value in tmp_env_dict["env_params"].items(): |
| os.environ[key] = str(value) |
| self.env_dict[key] = str(value) |
| except json.decoder.JSONDecodeError as e: |
| self.logger.error("\n\tERROR: %s " % e) |
| sys.exit(-1) |
| except FileNotFoundError as e: |
| self.logger.error("\n\tERROR: %s " % e) |
| sys.exit(-1) |
| |
| # This to mask the password from displaying on the console. |
| mask_dict = self.env_dict.copy() |
| for k, v in mask_dict.items(): |
| if k.lower().find("password") != -1: |
| hidden_text = [] |
| hidden_text.append(v) |
| password_regex = ( |
| "(" + "|".join([re.escape(x) for x in hidden_text]) + ")" |
| ) |
| mask_dict[k] = re.sub(password_regex, "********", v) |
| |
| self.logger.info(json.dumps(mask_dict, indent=8, sort_keys=False)) |
| |
| def execute_python_eval(self, eval_string): |
| r""" |
| Execute a qualified Python function string using the eval() function. |
| |
| This method executes a provided Python function string using the |
| eval() function. |
| |
| The method takes the eval_string as an argument, which is expected to |
| be a valid Python function call. |
| |
| The method returns the result of the executed function. |
| |
| Example: |
| eval(plugin.foo_func.foo_func(10)) |
| |
| Parameters: |
| eval_string (str): A valid Python function call string. |
| |
| Returns: |
| str: The result of the executed function and on failure return |
| PLUGIN_EVAL_ERROR. |
| """ |
| try: |
| self.logger.info("\tExecuting plugin func()") |
| self.logger.debug("\tCall func: %s" % eval_string) |
| result = eval(eval_string) |
| self.logger.info("\treturn: %s" % str(result)) |
| except ( |
| ValueError, |
| SyntaxError, |
| NameError, |
| AttributeError, |
| TypeError, |
| ) as e: |
| self.logger.error("\tERROR: execute_python_eval: %s" % e) |
| # Set the plugin error state. |
| plugin_error_dict["exit_on_error"] = True |
| self.logger.info("\treturn: PLUGIN_EVAL_ERROR") |
| return "PLUGIN_EVAL_ERROR" |
| |
| return result |
| |
| def execute_plugin_block(self, plugin_cmd_list): |
| r""" |
| Pack the plugin commands into qualified Python string objects. |
| |
| This method processes the plugin_cmd_list argument, which is expected |
| to contain a list of plugin commands read from a YAML file. The method |
| iterates through the list, constructs a qualified Python string object |
| for each plugin command, and returns a list of these string objects. |
| |
| Parameters: |
| plugin_cmd_list (list): A list of plugin commands containing |
| plugin names and arguments. |
| Plugin block read from YAML |
| [ |
| {'plugin_name':'plugin.foo_func.my_func'}, |
| {'plugin_args':[10]}, |
| ] |
| |
| Example: |
| Execute and no return response |
| - plugin: |
| - plugin_name: plugin.foo_func.my_func |
| - plugin_args: |
| - arg1 |
| - arg2 |
| |
| Execute and return a response |
| - plugin: |
| - plugin_name: result = plugin.foo_func.my_func |
| - plugin_args: |
| - arg1 |
| - arg2 |
| |
| Execute and return multiple values response |
| - plugin: |
| - plugin_name: result1,result2 = plugin.foo_func.my_func |
| - plugin_args: |
| - arg1 |
| - arg2 |
| |
| Returns: |
| str: Execute and not response or a string value(s) responses, |
| |
| """ |
| try: |
| idx = self.key_index_list_dict("plugin_name", plugin_cmd_list) |
| plugin_name = plugin_cmd_list[idx]["plugin_name"] |
| # Equal separator means plugin function returns result. |
| if " = " in plugin_name: |
| # Ex. ['result', 'plugin.foo_func.my_func'] |
| plugin_name_args = plugin_name.split(" = ") |
| # plugin func return data. |
| for arg in plugin_name_args: |
| if arg == plugin_name_args[-1]: |
| plugin_name = arg |
| else: |
| plugin_resp = arg.split(",") |
| # ['result1','result2'] |
| for x in plugin_resp: |
| global_plugin_list.append(x) |
| global_plugin_dict[x] = "" |
| |
| # Walk the plugin args ['arg1,'arg2'] |
| # If the YAML plugin statement 'plugin_args' is not declared. |
| plugin_args = [] |
| if any("plugin_args" in d for d in plugin_cmd_list): |
| idx = self.key_index_list_dict("plugin_args", plugin_cmd_list) |
| if idx is not None: |
| plugin_args = plugin_cmd_list[idx].get("plugin_args", []) |
| plugin_args = self.yaml_args_populate(plugin_args) |
| else: |
| plugin_args = self.yaml_args_populate([]) |
| |
| # Pack the args list to string parameters for plugin function. |
| parm_args_str = self.yaml_args_string(plugin_args) |
| |
| """ |
| Example of plugin_func: |
| plugin.redfish.enumerate_request( |
| "xx.xx.xx.xx:443", |
| "root", |
| "********", |
| "/redfish/v1/", |
| "json") |
| """ |
| if parm_args_str: |
| plugin_func = f"{plugin_name}({parm_args_str})" |
| else: |
| plugin_func = f"{plugin_name}()" |
| |
| # Execute plugin function. |
| if global_plugin_dict: |
| resp = self.execute_python_eval(plugin_func) |
| # Update plugin vars dict if there is any. |
| if resp != "PLUGIN_EVAL_ERROR": |
| self.response_args_data(resp) |
| else: |
| resp = self.execute_python_eval(plugin_func) |
| except Exception as e: |
| # Set the plugin error state. |
| plugin_error_dict["exit_on_error"] = True |
| self.logger.error("\tERROR: execute_plugin_block: %s" % e) |
| pass |
| |
| # There is a real error executing the plugin function. |
| if resp == "PLUGIN_EVAL_ERROR": |
| return resp |
| |
| # Check if plugin_expects_return (int, string, list,dict etc) |
| if any("plugin_expects_return" in d for d in plugin_cmd_list): |
| idx = self.key_index_list_dict( |
| "plugin_expects_return", plugin_cmd_list |
| ) |
| plugin_expects = plugin_cmd_list[idx]["plugin_expects_return"] |
| if plugin_expects: |
| if resp: |
| if ( |
| self.plugin_expect_type(plugin_expects, resp) |
| == "INVALID" |
| ): |
| self.logger.error("\tWARN: Plugin error check skipped") |
| elif not self.plugin_expect_type(plugin_expects, resp): |
| self.logger.error( |
| "\tERROR: Plugin expects return data: %s" |
| % plugin_expects |
| ) |
| plugin_error_dict["exit_on_error"] = True |
| elif not resp: |
| self.logger.error( |
| "\tERROR: Plugin func failed to return data" |
| ) |
| plugin_error_dict["exit_on_error"] = True |
| |
| return resp |
| |
| def response_args_data(self, plugin_resp): |
| r""" |
| Parse the plugin function response and update plugin return variable. |
| |
| plugin_resp Response data from plugin function. |
| """ |
| resp_list = [] |
| resp_data = "" |
| |
| # There is nothing to update the plugin response. |
| if len(global_plugin_list) == 0 or plugin_resp == "None": |
| return |
| |
| if isinstance(plugin_resp, str): |
| resp_data = plugin_resp.strip("\r\n\t") |
| resp_list.append(resp_data) |
| elif isinstance(plugin_resp, bytes): |
| resp_data = str(plugin_resp, "UTF-8").strip("\r\n\t") |
| resp_list.append(resp_data) |
| elif isinstance(plugin_resp, tuple): |
| if len(global_plugin_list) == 1: |
| resp_list.append(plugin_resp) |
| else: |
| resp_list = list(plugin_resp) |
| resp_list = [x.strip("\r\n\t") for x in resp_list] |
| elif isinstance(plugin_resp, list): |
| if len(global_plugin_list) == 1: |
| resp_list.append([x.strip("\r\n\t") for x in plugin_resp]) |
| else: |
| resp_list = [x.strip("\r\n\t") for x in plugin_resp] |
| elif isinstance(plugin_resp, int) or isinstance(plugin_resp, float): |
| resp_list.append(plugin_resp) |
| |
| # Iterate if there is a list of plugin return vars to update. |
| for idx, item in enumerate(resp_list, start=0): |
| # Exit loop, done required loop. |
| if idx >= len(global_plugin_list): |
| break |
| # Find the index of the return func in the list and |
| # update the global func return dictionary. |
| try: |
| dict_idx = global_plugin_list[idx] |
| global_plugin_dict[dict_idx] = item |
| except (IndexError, ValueError) as e: |
| self.logger.warn("\tWARN: response_args_data: %s" % e) |
| pass |
| |
| # Done updating plugin dict irrespective of pass or failed, |
| # clear all the list element for next plugin block execute. |
| global_plugin_list.clear() |
| |
| def yaml_args_string(self, plugin_args): |
| r""" |
| Pack the arguments into a string representation. |
| |
| This method processes the plugin_arg argument, which is expected to |
| contain a list of arguments. The method iterates through the list, |
| converts each argument to a string, and concatenates them into a |
| single string. Special handling is applied for integer, float, and |
| predefined plugin variable types. |
| |
| Ecample: |
| From |
| ['xx.xx.xx.xx:443', 'root', '********', '/redfish/v1/', 'json'] |
| to |
| "xx.xx.xx.xx:443","root","********","/redfish/v1/","json" |
| |
| Parameters: |
| plugin_args (list): A list of arguments to be packed into |
| a string. |
| |
| Returns: |
| str: A string representation of the arguments. |
| """ |
| args_str = "" |
| |
| for i, arg in enumerate(plugin_args): |
| if arg: |
| if isinstance(arg, (int, float)): |
| args_str += str(arg) |
| elif arg in global_plugin_type_list: |
| args_str += str(global_plugin_dict[arg]) |
| else: |
| args_str += f'"{arg.strip("\r\n\t")}"' |
| |
| # Skip last list element. |
| if i != len(plugin_args) - 1: |
| args_str += "," |
| |
| return args_str |
| |
| def yaml_args_populate(self, yaml_arg_list): |
| r""" |
| Decode environment and plugin variables and populate the argument list. |
| |
| This method processes the yaml_arg_list argument, which is expected to |
| contain a list of arguments read from a YAML file. The method iterates |
| through the list, decodes environment and plugin variables, and |
| returns a populated list of arguments. |
| |
| .. code-block:: yaml |
| |
| - plugin_args: |
| - arg1 |
| - arg2 |
| |
| ['${hostname}:${port_https}', '${username}', '/redfish/v1/', 'json'] |
| |
| Returns the populated plugin list |
| ['xx.xx.xx.xx:443', 'root', '/redfish/v1/', 'json'] |
| |
| Parameters: |
| yaml_arg_list (list): A list of arguments containing environment |
| and plugin variables. |
| |
| Returns: |
| list: A populated list of arguments with decoded environment and |
| plugin variables. |
| """ |
| if isinstance(yaml_arg_list, list): |
| populated_list = [] |
| for arg in yaml_arg_list: |
| if isinstance(arg, (int, float)): |
| populated_list.append(arg) |
| elif isinstance(arg, str): |
| arg_str = self.yaml_env_and_plugin_vars_populate(str(arg)) |
| populated_list.append(arg_str) |
| else: |
| populated_list.append(arg) |
| |
| return populated_list |
| |
| def yaml_env_and_plugin_vars_populate(self, yaml_arg_str): |
| r""" |
| Update environment variables and plugin variables based on the |
| provided YAML argument string. |
| |
| This method processes the yaml_arg_str argument, which is expected |
| to contain a string representing environment variables and plugin |
| variables in the format: |
| |
| .. code-block:: yaml |
| |
| - cat ${MY_VAR} |
| - ls -AX my_plugin_var |
| |
| The method parses the string, extracts the variable names, and updates |
| the corresponding environment variables and plugin variables. |
| |
| Parameters: |
| yaml_arg_str (str): A string containing environment and plugin |
| variable definitions in YAML format. |
| |
| Returns: |
| str: The updated YAML argument string with plugin variables |
| replaced. |
| """ |
| |
| # Parse and convert the Plugin YAML vars string to python vars |
| # Example: |
| # ${my_hostname}:${port_https} -> ['my_hostname', 'port_https'] |
| try: |
| # Example, list of matching |
| # env vars ['username', 'password', 'hostname'] |
| # Extra escape \ for special symbols. '\$\{([^\}]+)\}' works good. |
| env_var_regex = r"\$\{([^\}]+)\}" |
| env_var_names_list = re.findall(env_var_regex, yaml_arg_str) |
| |
| for var in env_var_names_list: |
| env_var = os.environ.get(var) |
| if env_var: |
| env_replace = "${" + var + "}" |
| yaml_arg_str = yaml_arg_str.replace(env_replace, env_var) |
| except Exception as e: |
| self.logger.error("\tERROR:yaml_env_vars_populate: %s" % e) |
| pass |
| |
| """ |
| Parse the string for plugin vars. |
| Implement the logic to update environment variables based on the |
| extracted variable names. |
| """ |
| try: |
| # Example, list of plugin vars env_var_names_list |
| # ['my_hostname', 'port_https'] |
| global_plugin_dict_keys = set(global_plugin_dict.keys()) |
| # Skip env var list already populated above code block list. |
| plugin_var_name_list = [ |
| var |
| for var in global_plugin_dict_keys |
| if var not in env_var_names_list |
| ] |
| |
| for var in plugin_var_name_list: |
| plugin_var_value = global_plugin_dict[var] |
| if yaml_arg_str in global_plugin_dict: |
| """ |
| If this plugin var exist but empty in dict, don't replace. |
| his is either a YAML plugin statement incorrectly used or |
| user added a plugin var which is not going to be populated. |
| """ |
| if isinstance(plugin_var_value, (list, dict)): |
| """ |
| List data type or dict can't be replaced, use |
| directly in eval function call. |
| """ |
| global_plugin_type_list.append(var) |
| else: |
| yaml_arg_str = yaml_arg_str.replace( |
| str(var), str(plugin_var_value) |
| ) |
| except (IndexError, ValueError) as e: |
| self.logger.error("\tERROR: yaml_plugin_vars_populate: %s" % e) |
| pass |
| |
| # From ${my_hostname}:${port_https} -> ['my_hostname', 'port_https'] |
| # to populated values string as |
| # Example: xx.xx.xx.xx:443 and return the string |
| return yaml_arg_str |
| |
| def plugin_error_check(self, plugin_dict): |
| r""" |
| Process plugin error dictionary and return the corresponding error |
| message. |
| |
| This method checks if any dictionary in the plugin_dict list contains |
| a "plugin_error" key. If such a dictionary is found, it retrieves the |
| value associated with the "plugin_error" key and returns the |
| corresponding error message from the plugin_error_dict attribute. |
| |
| Parameters: |
| plugin_dict (list of dict): A list of dictionaries containing |
| plugin error information. |
| |
| Returns: |
| str: The error message corresponding to the "plugin_error" value, |
| or None if no error is found. |
| """ |
| if any("plugin_error" in d for d in plugin_dict): |
| for d in plugin_dict: |
| if "plugin_error" in d: |
| value = d["plugin_error"] |
| return self.plugin_error_dict.get(value, None) |
| return None |
| |
| def key_index_list_dict(self, key, list_dict): |
| r""" |
| Find the index of the first dictionary in the list that contains |
| the specified key. |
| |
| Parameters: |
| key (str): The key to search for in the |
| dictionaries. |
| list_dict (list of dict): A list of dictionaries to search |
| through. |
| |
| Returns: |
| int: The index of the first dictionary containing the key, or -1 |
| if no match is found. |
| """ |
| for i, d in enumerate(list_dict): |
| if key in d: |
| return i |
| return -1 |
| |
| def plugin_expect_type(self, type, data): |
| r""" |
| Check if the provided data matches the expected type. |
| |
| This method checks if the data argument matches the specified type. |
| It supports the following types: "int", "float", "str", "list", "dict", |
| and "tuple". |
| |
| If the type is not recognized, it logs an info message and returns |
| "INVALID". |
| |
| Parameters: |
| type (str): The expected data type. |
| data: The data to check against the expected type. |
| |
| Returns: |
| bool or str: True if the data matches the expected type, False if |
| not, or "INVALID" if the type is not recognized. |
| """ |
| if type == "int": |
| return isinstance(data, int) |
| elif type == "float": |
| return isinstance(data, float) |
| elif type == "str": |
| return isinstance(data, str) |
| elif type == "list": |
| return isinstance(data, list) |
| elif type == "dict": |
| return isinstance(data, dict) |
| elif type == "tuple": |
| return isinstance(data, tuple) |
| else: |
| self.logger.info("\tInvalid data type requested: %s" % type) |
| return "INVALID" |