Merge pull request #72 from IBM/hotfix_processStats

Re-Introducing TOP to fix #71

Author: baslerj

python/influx/definitions.py

@@ -1011,6 +1011,7 @@ def add_table_definitions(cls, database: Database):
                 'PID',
                 'USER',
                 'hostName',
+                'collectionType',
                 'ssh_type'
             ],# time key is capture time
             retention_policy=cls._RP_DAYS_14(),

python/sppConnection/ssh_client.py

@@ -210,7 +210,7 @@ def execute_commands(self, commands: List[SshCommand], verbose: bool = False) ->
         if(verbose):
             LOGGER.info("> connection successfull")
 
-        new_command_list = []
+        new_command_list: List[SshCommand] = []
         for ssh_command in commands:
 
             try:

python/sppmon.py

@@ -56,6 +56,7 @@
  03/29/2021 version 0.13.2 Fixes to typing, reducing error messages and tracking code for NaN bug
  06/07/2021 version 0.13.3 Hotfixing version endpoint for SPP 10.1.8.1
  06/09/2021 version 0.13.4 Hotfixing storage execption, chaning top-level execption handling to reduce the need of further hotfixes
+ 08/06/2021 version 0.13.5 Fixing PS having unituitive CPU-recording, reintroducing TOP to collect CPU informations only
 """
 from __future__ import annotations
 import functools
@@ -83,7 +84,7 @@
 from utils.spp_utils import SppUtils
 
 # Version:
-VERSION = "0.13.4  (2021/06/09)"
+VERSION = "0.13.5  (2021/08/06)"
 
 # ----------------------------------------------------------------------------
 # command line parameter parsing

python/sppmonMethods/ssh.py

@@ -149,15 +149,24 @@ def __init__(self, influx_client: InfluxClient, config_file: Dict[str, Any], ver
 
         # SERVER
         # add server later due multiple processes
-        self.__ps_grep_list = ["mongod", "beam.smp", "java"] # be aware this is double declared below
-        for grep_name in self.__ps_grep_list:
+        self.__process_grep_list = ["mongod", "beam.smp", "java"] # be aware this is double declared below
+        for grep_name in self.__process_grep_list:
             self.__client_commands[SshTypes.SERVER].append(
                 SshCommand(
                     command=f"ps -o \"%cpu,%mem,comm,rss,vsz,user,pid,etimes\" -p $(pgrep -d',' -f {grep_name}) S -ww",
                     parse_function=self._parse_ps_cmd,
                     table_name="processStats"
                 )
             )
+        # Top commands for CPU Only
+        for grep_name in self.__process_grep_list:
+            self.__client_commands[SshTypes.SERVER].append(
+                SshCommand(
+                    command=f"top -bs -w 512 -n1 -p $(pgrep -d',' -f {grep_name})",
+                    parse_function=self._parse_top_cmd,
+                    table_name="processStats"
+                )
+            )
 
         # ################ END OF SSH COMMAND LIST GROUPS ############################
 
@@ -221,6 +230,74 @@ def ssh(self) -> None:
                 ExceptionUtils.exception_info(
                     error=error, extra_message=f"Top-level-error when excecuting {ssh_type.value} ssh commands, skipping them all")
 
+    def _parse_top_cmd(self, ssh_command: SshCommand, ssh_type: SshTypes) -> Tuple[str, List[Dict[str, Any]]]:
+        """Parses the result of the `top` command, splitting it into its parts.
+
+        Arguments:
+            ssh_command {SshCommand} -- command with saved result
+            ssh_type {SshTypes} -- type of the client
+        Raises:
+            ValueError: no command given or no result saved
+            ValueError: no ssh type given
+        Returns:
+            Tuple[str, List[Dict[str, Any]]] -- Tuple of the tablename and a insert list
+        """
+
+        if(not ssh_command or not ssh_command.result):
+            raise ValueError("no command given or empty result")
+        if(not ssh_type):
+            raise ValueError("no sshtype given")
+        if(not ssh_command.table_name):
+            raise ValueError("need table name to insert parsed value")
+
+        result_lines = ssh_command.result.splitlines()
+        header = result_lines[6].split()
+
+        values: List[Dict[str, Any]] = list(
+            map(lambda row: dict(zip(header, row.split())), result_lines[7:])) # type: ignore
+
+        # All lines above (header) 5 are pruned, not used anymore. This data is tracked via ps (see Issue #71)
+
+        time_pattern = re.compile(r"(\d+):(\d{2})(?:\.(\d{2}))?")
+
+        # remove `top` from commands, it is also tracked
+        values = list(filter(lambda row: row["COMMAND"] in self.__process_grep_list, values))
+
+        for row in values:
+            # Delete Memory, this is tracked by ps command (See Issue #71)
+            row.pop("VIRT", None)
+            row.pop("RES", None)
+            row.pop("SHR", None)
+            row.pop("%MEM", None)
+            # Add information
+            row["collectionType"] = "TOP"
+
+            # unused information
+            row.pop("PR", None)
+            row.pop("NI", None)
+            row.pop("S", None)
+
+
+            # set default needed fields
+            row['hostName'] = ssh_command.host_name
+            row['ssh_type'] = ssh_type.name
+            (time_key, time_value) = SppUtils.get_capture_timestamp_sec()
+            row[time_key] = time_value
+
+            # split time into seconds
+            match = re.match(time_pattern, row['TIME+'])
+            if(match):
+                time_list = match.groups()
+                (hours, minutes, seconds) = time_list
+                if(seconds is None):
+                    seconds = 0
+                time = int(hours)*pow(60, 2) + int(minutes)*pow(60, 1) + int(seconds)*pow(60, 0)
+            else:
+                time = None
+            row['TIME+'] = time
+
+        return (ssh_command.table_name, values)
+
     def _parse_ps_cmd(self, ssh_command: SshCommand, ssh_type: SshTypes) -> Tuple[str, List[Dict[str, Any]]]:
         """Parses the result of the `df` command, splitting it into its parts.
 
@@ -246,10 +323,15 @@ def _parse_ps_cmd(self, ssh_command: SshCommand, ssh_type: SshTypes) -> Tuple[st
         values: List[Dict[str, Any]] = list(
             map(lambda row: dict(zip(header, row.split())), result_lines[1:])) # type: ignore
 
-        # remove top statistic itself to avoid spam with useless information
-        values = list(filter(lambda row: row["COMMAND"] in self.__ps_grep_list, values))
+        # remove `ps` from commands, it is also tracked
+        values = list(filter(lambda row: row["COMMAND"] in self.__process_grep_list, values))
 
         for row in values:
+            # Remove CPU, it is tracked by TOP-Command (see Issue #71)
+            row.pop("%CPU", None)
+            # Add information
+            row["collectionType"] = "PS"
+
             # set default needed fields
             row['hostName'] = ssh_command.host_name
             row['ssh_type'] = ssh_type.name

python/utils/methods_utils.py

@@ -59,7 +59,9 @@ def ssh_execute_commands(cls, ssh_clients: List[SshClient], ssh_type: SshTypes,
                 LOGGER.info(f"No {ssh_type.name} ssh client present. Aborting command")
             return []
 
-        ssh_cmd_response_list = []
+        # List to persist ssh-result stats over each client
+        ssh_cmd_response_list: List[Dict[str,Union[str, int, None]]] = []
+        # list to insert into influx, tuple of table and its result-lists
         result_list: List[Tuple[str, List[Dict[str, Any]]]] = []
         for client in client_list:
 
@@ -77,7 +79,8 @@ def ssh_execute_commands(cls, ssh_clients: List[SshClient], ssh_type: SshTypes,
                 continue
 
             for ssh_command in result_commands:
-                insert_dict = {}
+                # generate stats for the ssh-stats list
+                insert_dict: Dict[str, Union[str, int, None]] = {}
                 insert_dict["host"] = ssh_command.host_name
                 insert_dict["command"] = ssh_command.cmd
                 insert_dict["output"] = json.dumps(ssh_command.result)
@@ -87,13 +90,16 @@ def ssh_execute_commands(cls, ssh_clients: List[SshClient], ssh_type: SshTypes,
 
                 ssh_cmd_response_list.append(insert_dict)
 
+                # execute the command
                 try:
                     table_result_tuple = ssh_command.parse_result(ssh_type=ssh_type)
                     if(table_result_tuple):
+                        # save the command into the result set wit its table
                         result_list.append(table_result_tuple)
                 except ValueError as error:
                     ExceptionUtils.exception_info(error=error, extra_message="Error when parsing result, skipping parsing of this result")
 
+        # append the ssh command once, cause each client already added into the ssh_command list
         result_list.append(("sshCmdResponse", ssh_cmd_response_list))
         return result_list