Repository: Azure/WALinuxAgent
Pull request #2117 (2020-12-17 17:46:02)

Use PIDs instead of command lines to identify processes in the agent's cgroup
Using command lines to identify those processes was unreliable and hard to maintain. PIDs should address those issues.

codecov[bot]: # [Codecov](https://codecov.io/gh/Azure/WALinuxAgent/pull/2117?src=pr&el=h1) Report
> Merging [#2117](https://codecov.io/gh/Azure/WALinuxAgent/pull/2117?src=pr&el=desc) (3c2b59d) into [develop](https://codecov.io/gh/Azure/WALinuxAgent/commit/fd15961b13322a6d6b86d28e1a1c8be26ca6e078?el=desc) (fd15961) will **decrease** coverage by `0.00%`.
> The diff coverage is `78.20%`.

[![Impacted file tree graph](https://codecov.io/gh/Azure/WALinuxAgent/pull/2117/graphs/tree.svg?width=650&height=150&src=pr&token=ae90E2KYCw)](https://codecov.io/gh/Azure/WALinuxAgent/pull/2117?src=pr&el=tree)

```diff
@@             Coverage Diff             @@
##           develop    #2117      +/-   ##
===========================================
- Coverage    70.69%   70.68%   -0.01%     
===========================================
  Files           95       95              
  Lines        13076    13096      +20     
  Branches      1854     1855       +1     
===========================================
+ Hits          9244     9257      +13     
- Misses        3423     3425       +2     
- Partials       409      414       +5     
```


| [Impacted Files](https://codecov.io/gh/Azure/WALinuxAgent/pull/2117?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [azurelinuxagent/common/utils/shellutil.py](https://codecov.io/gh/Azure/WALinuxAgent/pull/2117/diff?src=pr&el=tree#diff-YXp1cmVsaW51eGFnZW50L2NvbW1vbi91dGlscy9zaGVsbHV0aWwucHk=) | `80.00% <50.00%> (-1.53%)` | :arrow_down: |
| [azurelinuxagent/common/cgroupapi.py](https://codecov.io/gh/Azure/WALinuxAgent/pull/2117/diff?src=pr&el=tree#diff-YXp1cmVsaW51eGFnZW50L2NvbW1vbi9jZ3JvdXBhcGkucHk=) | `78.71% <72.72%> (-0.60%)` | :arrow_down: |
| [azurelinuxagent/common/cgroupconfigurator.py](https://codecov.io/gh/Azure/WALinuxAgent/pull/2117/diff?src=pr&el=tree#diff-YXp1cmVsaW51eGFnZW50L2NvbW1vbi9jZ3JvdXBjb25maWd1cmF0b3IucHk=) | `78.49% <79.59%> (+2.54%)` | :arrow_up: |
| [azurelinuxagent/ga/monitor.py](https://codecov.io/gh/Azure/WALinuxAgent/pull/2117/diff?src=pr&el=tree#diff-YXp1cmVsaW51eGFnZW50L2dhL21vbml0b3IucHk=) | `77.84% <90.90%> (+<0.01%)` | :arrow_up: |
| [azurelinuxagent/ga/collect\_logs.py](https://codecov.io/gh/Azure/WALinuxAgent/pull/2117/diff?src=pr&el=tree#diff-YXp1cmVsaW51eGFnZW50L2dhL2NvbGxlY3RfbG9ncy5weQ==) | `89.43% <100.00%> (ø)` | |
| [azurelinuxagent/ga/collect\_telemetry\_events.py](https://codecov.io/gh/Azure/WALinuxAgent/pull/2117/diff?src=pr&el=tree#diff-YXp1cmVsaW51eGFnZW50L2dhL2NvbGxlY3RfdGVsZW1ldHJ5X2V2ZW50cy5weQ==) | `91.37% <0.00%> (-0.69%)` | :arrow_down: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/Azure/WALinuxAgent/pull/2117?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/Azure/WALinuxAgent/pull/2117?src=pr&el=footer). Last update [fd15961...3c2b59d](https://codecov.io/gh/Azure/WALinuxAgent/pull/2117?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).


Patch:
diff --git a/azurelinuxagent/common/cgroupapi.py b/azurelinuxagent/common/cgroupapi.py
index 18c65cd7..90c92d16 100644
--- a/azurelinuxagent/common/cgroupapi.py
+++ b/azurelinuxagent/common/cgroupapi.py
@@ -81,9 +81,9 @@ class CGroupsApi(object):
         try:
             for cgroup in extension_cgroups:
                 CGroupsTelemetry.track_cgroup(cgroup)
-        except Exception as e:  # pylint: disable=C0103
+        except Exception as exception:
             logger.warn("Cannot add cgroup '{0}' to tracking list; resource usage will not be tracked. "
-                        "Error: {1}".format(cgroup.path, ustr(e)))
+                        "Error: {1}".format(cgroup.path, ustr(exception)))
 
     @staticmethod
     def _get_extension_cgroup_name(extension_name):
@@ -105,6 +105,11 @@ class CGroupsApi(object):
         """
         return os.path.exists(SYSTEMD_RUN_PATH)
 
+    @staticmethod
+    def get_processes_in_cgroup(cgroup_path):
+        with open(os.path.join(cgroup_path, "cgroup.procs"), "r") as cgroup_procs:
+            return [int(pid) for pid in cgroup_procs.read().split()]
+
     @staticmethod
     def _foreach_controller(operation, message):
         """
@@ -150,9 +155,9 @@ class CGroupsApi(object):
 
                 if os.path.exists(procs_file):
                     procs_file_contents = fileutil.read_file(procs_file).strip()
-                    daemon_pid = fileutil.read_file(get_agent_pid_file_path()).strip()
+                    daemon_pid = CGroupsApi.get_daemon_pid()
 
-                    if daemon_pid in procs_file_contents:
+                    if ustr(daemon_pid) in procs_file_contents:
                         operation(controller, daemon_pid)
         finally:
             for _, cgroup in legacy_cgroups:
@@ -160,6 +165,10 @@ class CGroupsApi(object):
                 shutil.rmtree(cgroup, ignore_errors=True)
         return len(legacy_cgroups)
 
+    @staticmethod
+    def get_daemon_pid():
+        return int(fileutil.read_file(get_agent_pid_file_path()).strip())
+
 
 class FileSystemCgroupsApi(CGroupsApi):
     """
@@ -276,7 +285,7 @@ class FileSystemCgroupsApi(CGroupsApi):
         def move_daemon_pid(controller, daemon_pid):
             new_path = FileSystemCgroupsApi._get_agent_cgroup_path(controller)
             logger.info("Writing daemon's PID ({0}) to {1}", daemon_pid, new_path)
-            fileutil.append_file(os.path.join(new_path, "cgroup.procs"), daemon_pid)
+            fileutil.append_file(os.path.join(new_path, "cgroup.procs"), ustr(daemon_pid))
             msg = "Moved daemon's PID from legacy cgroup to {0}".format(new_path)
             add_event(op=WALAEventOperation.CGroupsCleanUp, is_success=True, message=msg)
 
@@ -629,30 +638,6 @@ After=system-{1}.slice""".format(extension_name, EXTENSIONS_ROOT_CGROUP_NAME)
             self._agent_unit_name = get_osutil().get_service_name() + ".service"
         return self._agent_unit_name
 
-    @staticmethod
-    def get_processes_in_cgroup(cgroup_path):
-        """
-        Returns an array of tuples with the PID and command line of the processes that are currently
-        within the cgroup for the given path (which must be within the cgroup filesystem).
-        """
-        #
-        # The output of the command is similar to
-        #
-        #     Directory /sys/fs/cgroup/cpu/system.slice/walinuxagent.service:
-        #     ├─27519 /usr/bin/python3 -u /usr/sbin/waagent -daemon
-        #     └─27547 python3 -u bin/WALinuxAgent-2.2.48.1-py2.7.egg -run-exthandlers
-        #
-        output = shellutil.run_command(['systemd-cgls', cgroup_path])
-
-        processes = []
-
-        for line in output.splitlines():
-            match = re.match('[^\d]*(?P<pid>\d+)\s+(?P<command>.+)', line)  # pylint: disable=W1401
-            if match is not None:
-                processes.append((match.group('pid'), match.group('command')))
-
-        return processes
-
     @staticmethod
     def _is_systemd_failure(scope_name, stderr):
         stderr.seek(0)
diff --git a/azurelinuxagent/common/cgroupconfigurator.py b/azurelinuxagent/common/cgroupconfigurator.py
index 694cd8a9..45205174 100644
--- a/azurelinuxagent/common/cgroupconfigurator.py
+++ b/azurelinuxagent/common/cgroupconfigurator.py
@@ -15,7 +15,6 @@
 # Requires Python 2.6+ and Openssl 1.0+
 
 import os
-import re
 import subprocess
 
 from azurelinuxagent.common import logger
@@ -25,15 +24,28 @@ from azurelinuxagent.common.cgroupstelemetry import CGroupsTelemetry
 from azurelinuxagent.common.exception import ExtensionErrorCodes, CGroupsException
 from azurelinuxagent.common.future import ustr
 from azurelinuxagent.common.version import get_distro
+from azurelinuxagent.common.utils import shellutil
 from azurelinuxagent.common.utils.extensionprocessutil import handle_process_completion
 from azurelinuxagent.common.event import add_event, WALAEventOperation
 
 
+class UnexpectedProcessesInCGroupException(CGroupsException):
+    """
+    Raised by CGroupConfigurator.check_processes_in_agent_cgroup() when the agent's cgroup includes processes
+    that should not belong to it.
+    The 'unexpected' property is a list of the processes (strings) that should not belong to the agent's cgroup
+    """
+    def __init__(self, unexpected):
+        super(UnexpectedProcessesInCGroupException, self).__init__("Unexpected processes in agent's cgroup")
+        self.unexpected = unexpected
+
+
 class CGroupConfigurator(object):
     """
     This class implements the high-level operations on CGroups (e.g. initialization, creation, etc)
 
-    NOTE: with the exception of start_extension_command, none of the methods in this class raise exceptions (cgroup operations should not block extensions)
+    NOTE: with the exception of start_extension_command and check_processes_in_agent_cgroup, none of the methods in this class
+    raise exceptions (cgroup operations should not block extensions)
     """
     # too-many-instance-attributes<R0902> Disabled: class complexity is OK
     # invalid-name<C0103> Disabled: class is private, so name starts with __
@@ -225,29 +237,70 @@ class CGroupConfigurator(object):
 
             self._invoke_cgroup_operation(__impl, "Failed to delete cgroups for extension '{0}'.".format(name))
 
-        def get_processes_in_agent_cgroup(self):
+        def check_processes_in_agent_cgroup(self):
             """
-            Returns an array of tuples with the PID and command line of the processes that are currently within the cgroup for the given unit.
-
-            The return value can be None if cgroups are not enabled or if an error occurs during the operation.
+            Verifies that the agent's cgroup includes only the current process, its parent and commands started using shellutil (i.e. the extension handler,
+            the daemon, and the commands started by the extension handler, respectively).
+            Other processes started by the agent (e.g. extensions) and processes not started by the agent (e.g. services installed by extensions) should
+            belong to their own cgroup.
+            The function raises an UnexpectedProcessesInCGroupException if the check fails.
             """
-            def __impl():
-                if self._agent_cpu_cgroup_path is None:
-                    return []
-                return self._cgroups_api.get_processes_in_cgroup(self._agent_cpu_cgroup_path)  # pylint: disable=E1101
-
-            def __on_error(exception):
-                #
-                # Send telemetry for a small sample of errors (if any)
-                #
-                self._get_processes_in_agent_cgroup_error_count = self._get_processes_in_agent_cgroup_error_count + 1
-                if self._get_processes_in_agent_cgroup_error_count <= 5:
-                    message = "Failed to list the processes in the agent's cgroup: {0}", ustr(exception)
-                    if message != self._get_processes_in_agent_cgroup_last_error:
-                        add_event(op=WALAEventOperation.CGroupsDebug, message=message)
-                    self._get_processes_in_agent_cgroup_last_error = message
-
-            return self._invoke_cgroup_operation(__impl, "Failed to list the processes in the agent's cgroup.", on_error=__on_error)
+            if not self.enabled():
+                return
+
+            daemon = os.getppid()
+            extension_handler = os.getpid()
+            agent_commands = set()
+            agent_commands.update(shellutil.get_running_commands())
+            agent_cgroup = CGroupsApi.get_processes_in_cgroup(self._agent_cpu_cgroup_path)
+            # get the running commands again in case a new command was started while we were fetching the processes in the cgroup;
+            agent_commands.update(shellutil.get_running_commands())
+
+            unexpected = []
+            for process in agent_cgroup:
+                if process in (daemon, extension_handler):
+                    continue
+                # check if the process is a command started by the agent or a descendant of one of those commands
+                current = process
+                while current != 0 and current not in agent_commands:
+                    current = self._get_parent(current)
+                if current == 0:
+                    unexpected.append(process)
+                    if len(unexpected) >= 5:  # collect just a small sample
+                        break
+            if unexpected:
+                raise UnexpectedProcessesInCGroupException(unexpected=self._format_processes(unexpected))
+
+        @staticmethod
+        def _format_processes(pid_list):
+            """
+            Formats the given PIDs as a sequence of strings containing the PIDs and their corresponding command line (truncated to 40 chars)
+            """
+            def get_command_line(pid):
+                try:
+                    cmdline = '/proc/{0}/cmdline'.format(pid)
+                    if os.path.exists(cmdline):
+                        with open(cmdline, "r") as cmdline_file:
+                            return "[PID: {0}] {1:40.40}".format(pid, cmdline_file.read())
+                except Exception:
+                    pass
+                return "[PID: {0}] UNKNOWN".format(pid)
+
+            return [get_command_line(pid) for pid in pid_list]
+
+        @staticmethod
+        def _get_parent(pid):
+            """
+            Returns the parent of the given process. If the parent cannot be determined returns 0 (which is the PID for the scheduler)
+            """
+            try:
+                stat = '/proc/{0}/stat'.format(pid)
+                if os.path.exists(stat):
+                    with open(stat, "r") as stat_file:
+                        return int(stat_file.read().split()[3])
+            except Exception:
+                pass
+            return 0
 
         # too-many-arguments<R0913> Disabled: argument list mimics Popen's
         def start_extension_command(self, extension_name, command, timeout, shell, cwd, env, stdout, stderr, error_code=ExtensionErrorCodes.PluginUnknownFailure):  # pylint: disable=R0913
@@ -300,41 +353,3 @@ class CGroupConfigurator(object):
         if CGroupConfigurator._instance is None:
             CGroupConfigurator._instance = CGroupConfigurator.__Impl()
         return CGroupConfigurator._instance
-
-    @staticmethod
-    def is_agent_process(command_line):
-        """
-        Returns true if the given command line corresponds to a process started by the agent.
-
-        NOTE: The function uses pattern matching to determine whether the process was spawned by the agent; this is more of a heuristic
-        than an exact check.
-        """
-        patterns = [
-            r".*waagent -daemon.*",
-            r".*(WALinuxAgent-.+\.egg|waagent) -run-exthandlers",
-            # The processes in the agent's cgroup are listed using systemd-cgls
-            r"^systemd-cgls.*walinuxagent.*$",
-            # Extensions are started using systemd-run
-            r"^systemd-run --unit=.+ --scope ",
-            #
-            # The rest of the commands are started by the environment thread; many of them are distro-specific so this list may need
-            # additions as we add support for more distros.
-            #
-            # *** Monitor DHCP client restart
-            #
-            r"^pidof (dhclient|dhclient3|systemd-networkd)",
-            r"^ip route (show|add)",
-            #
-            # *** Enable firewall
-            #
-            r"^iptables --version$",
-            r"^iptables .+ -t security",
-            #
-            # *** Monitor host name changes
-            #
-            r"^ifdown .+ && ifup .+",
-        ]
-        for p in patterns:  # pylint: disable=C0103
-            if re.match(p, command_line) is not None:
-                return True
-        return False
diff --git a/azurelinuxagent/common/utils/shellutil.py b/azurelinuxagent/common/utils/shellutil.py
index ba1c28e3..aa5f9fc2 100644
--- a/azurelinuxagent/common/utils/shellutil.py
+++ b/azurelinuxagent/common/utils/shellutil.py
@@ -207,7 +207,7 @@ def __run_command(command_action, command, log_error, encode_output):
 
 # W0622: Redefining built-in 'input'  -- disabled: the parameter name mimics subprocess.communicate()
 # R0913: Too many arguments (8/5) -- disabled: the parameter list mimics subprocess.Popen()/communicate()
-def run_command(command, input=None, stdin=None, stdout=subprocess.PIPE, stderr=subprocess.PIPE, log_error=False, encode_input=True, encode_output=True):  # pylint:disable=W0622,R0913
+def run_command(command, input=None, stdin=None, stdout=subprocess.PIPE, stderr=subprocess.PIPE, log_error=False, encode_input=True, encode_output=True, track_process=True):  # pylint:disable=W0622,R0913
     """
         Executes the given command and returns its stdout.
 
@@ -216,6 +216,8 @@ def run_command(command, input=None, stdin=None, stdout=subprocess.PIPE, stderr=
 
         If encode_output is True the stdout is returned as a string, otherwise it is returned as a bytes object.
 
+        If track_process is False the command is not added to list of running commands
+
         This function is a thin wrapper around Popen/communicate in the subprocess module:
            * The 'input' parameter corresponds to the same parameter in communicate
            * The 'stdin' parameter corresponds to the same parameters in Popen
@@ -240,10 +242,14 @@ def run_command(command, input=None, stdin=None, stdout=subprocess.PIPE, stderr=
             popen_stdin = stdin
             communicate_input = None
 
-        process = _popen(command, stdin=popen_stdin, stdout=stdout, stderr=stderr, shell=False)
+        if track_process:
+            process = _popen(command, stdin=popen_stdin, stdout=stdout, stderr=stderr, shell=False)
+        else:
+            process = subprocess.Popen(command, stdin=popen_stdin, stdout=stdout, stderr=stderr, shell=False)
 
         command_stdout, command_stderr = process.communicate(input=communicate_input)
-        _on_command_completed(process.pid)
+        if track_process:
+            _on_command_completed(process.pid)
 
         return process.returncode, command_stdout, command_stderr
 
diff --git a/azurelinuxagent/ga/collect_logs.py b/azurelinuxagent/ga/collect_logs.py
index 00a403ae..13c86790 100644
--- a/azurelinuxagent/ga/collect_logs.py
+++ b/azurelinuxagent/ga/collect_logs.py
@@ -170,7 +170,8 @@ class CollectLogsHandler(ThreadHandlerInterface):
         success = False
         msg = None
         try:
-            shellutil.run_command(final_command, log_error=True)
+            # TODO: Remove track_process (and its implementation) when the log collector is moved to the agent's cgroup
+            shellutil.run_command(final_command, log_error=True, track_process=False)
             duration = elapsed_milliseconds(start_time)
             archive_size = os.path.getsize(COMPRESSED_ARCHIVE_PATH)
 
diff --git a/azurelinuxagent/ga/monitor.py b/azurelinuxagent/ga/monitor.py
index ad4cc6dd..6ac4f6d9 100644
--- a/azurelinuxagent/ga/monitor.py
+++ b/azurelinuxagent/ga/monitor.py
@@ -21,7 +21,7 @@ import uuid
 
 import azurelinuxagent.common.logger as logger
 import azurelinuxagent.common.utils.networkutil as networkutil
-from azurelinuxagent.common.cgroupconfigurator import CGroupConfigurator
+from azurelinuxagent.common.cgroupconfigurator import CGroupConfigurator, UnexpectedProcessesInCGroupException
 from azurelinuxagent.common.cgroupstelemetry import CGroupsTelemetry
 from azurelinuxagent.common.errorstate import ErrorState
 from azurelinuxagent.common.event import add_event, WALAEventOperation, report_metric
@@ -56,39 +56,20 @@ class PollResourceUsageOperation(PeriodicOperation):
         self._error_count = 0
 
     def _operation_impl(self):
-        #
-        # Check the processes in the agent cgroup
-        #
-        processes_check_error = None
         try:
-            processes = CGroupConfigurator.get_instance().get_processes_in_agent_cgroup()
-
-            if processes is not None:
-                unexpected_processes = []
-
-                for (_, command_line) in processes:
-                    if not CGroupConfigurator.is_agent_process(command_line):
-                        unexpected_processes.append(command_line)
-
-                if unexpected_processes:
-                    unexpected_processes.sort()
-                    processes_check_error = "The agent's cgroup includes unexpected processes: {0}".format(ustr(unexpected_processes))
-        except Exception as exception:
-            processes_check_error = "Failed to check the processes in the agent's cgroup: {0}".format(ustr(exception))
-
-        # Report a small sample of errors
-        if processes_check_error != self._last_error and self._error_count < 5:
-            self._error_count += 1
-            self._last_error = processes_check_error
-            logger.info(processes_check_error)
-            add_event(op=WALAEventOperation.CGroupsDebug, message=processes_check_error)
-
-        #
-        # Report metrics
-        #
-        metrics = CGroupsTelemetry.poll_all_tracked()
-
-        for metric in metrics:
+            CGroupConfigurator.get_instance().check_processes_in_agent_cgroup()
+        except UnexpectedProcessesInCGroupException as exception:
+            exception.unexpected.sort()
+            message = "The agent's cgroup includes unexpected processes: {0}".format(exception.unexpected)
+
+            # Report a small sample of errors
+            if message != self._last_error and self._error_count < 5:
+                self._error_count += 1
+                self._last_error = message
+                logger.info(message)
+                add_event(op=WALAEventOperation.CGroupsDebug, message=message)
+
+        for metric in CGroupsTelemetry.poll_all_tracked():
             report_metric(metric.category, metric.counter, metric.instance, metric.value)
 
 


Test patch:
diff --git a/tests/common/mock_cgroup_commands.py b/tests/common/mock_cgroup_commands.py
index 3cf49ac3..09b9334b 100644
--- a/tests/common/mock_cgroup_commands.py
+++ b/tests/common/mock_cgroup_commands.py
@@ -66,12 +66,6 @@ Running scope as unit: TEST_UNIT.scope
 Thu 28 May 2020 07:25:55 AM PDT
 '''),
 
-    (r"^systemd-cgls.+/walinuxagent.service$",
-'''
-Directory /sys/fs/cgroup/cpu/system.slice/walinuxagent.service:
-├─27519 /usr/bin/python3 -u /usr/sbin/waagent -daemon
-└─27547 python3 -u bin/WALinuxAgent-2.2.48.1-py2.7.egg -run-exthandlers
-'''),
 ]
 
 __DEFAULT_FILES = [
diff --git a/tests/common/test_cgroupapi.py b/tests/common/test_cgroupapi.py
index 4be8d26b..1b81f012 100644
--- a/tests/common/test_cgroupapi.py
+++ b/tests/common/test_cgroupapi.py
@@ -527,19 +527,6 @@ class SystemdCgroupsApiTestCase(AgentTestCase):
         os.remove("/etc/systemd/system/{0}".format(unit_name))
         shellutil.run_command(["systemctl", "daemon-reload"])
 
-    def test_get_processes_in_cgroup_should_return_the_processes_within_the_cgroup(self):
-        with mock_cgroup_commands():
-            processes = SystemdCgroupsApi.get_processes_in_cgroup("/sys/fs/cgroup/cpu/system.slice/walinuxagent.service")
-
-            self.assertTrue(len(processes) >= 2,
-                            "The cgroup should contain at least 2 procceses (daemon and extension handler): [{0}]".format(processes))
-
-            daemon_present = any("waagent -daemon" in command for (pid, command) in processes)
-            self.assertTrue(daemon_present, "Could not find the daemon in the cgroup: [{0}]".format(processes))
-
-            extension_handler_present = any(re.search(r"(WALinuxAgent-.+\.egg|waagent) -run-exthandlers", command) for (pid, command) in processes)
-            self.assertTrue(extension_handler_present, "Could not find the extension handler in the cgroup: [{0}]".format(processes))
-
     @attr('requires_sudo')
     def test_create_extension_cgroups_should_create_extension_slice(self):
         self.assertTrue(i_am_root(), "Test does not run when non-root")
diff --git a/tests/common/test_cgroupconfigurator.py b/tests/common/test_cgroupconfigurator.py
index 1f78a7b6..a0e6b1c4 100644
--- a/tests/common/test_cgroupconfigurator.py
+++ b/tests/common/test_cgroupconfigurator.py
@@ -17,16 +17,21 @@
 
 from __future__ import print_function
 
-import re
+import os
+import random
+import signal
 import subprocess
+import threading
 
+from azurelinuxagent.common.future import ustr
 from azurelinuxagent.common.cgroup import CGroup
-from azurelinuxagent.common.cgroupconfigurator import CGroupConfigurator
+from azurelinuxagent.common.cgroupconfigurator import CGroupConfigurator, UnexpectedProcessesInCGroupException
 from azurelinuxagent.common.cgroupstelemetry import CGroupsTelemetry
 from azurelinuxagent.common.exception import CGroupsException
+from azurelinuxagent.common.utils import shellutil
 from tests.common.mock_cgroup_commands import mock_cgroup_commands
 from tests.tools import AgentTestCase, patch, mock_sleep
-
+from tests.utils.miscellaneous_tools import format_processes, wait_for
 
 class CGroupConfiguratorSystemdTestCase(AgentTestCase):
     @classmethod
@@ -204,21 +209,6 @@ cgroup on /sys/fs/cgroup/blkio type cgroup (rw,nosuid,nodev,noexec,relatime,blki
                     message = args[0]
                     self.assertIn("A TEST EXCEPTION", message)
 
-    def test_get_processes_in_agent_cgroup_should_return_the_processes_within_the_agent_cgroup(self):
-        with mock_cgroup_commands():
-            configurator = CGroupConfiguratorSystemdTestCase._get_new_cgroup_configurator_instance()
-
-            processes = configurator.get_processes_in_agent_cgroup()
-
-            self.assertTrue(len(processes) >= 2,
-                "The cgroup should contain at least 2 procceses (daemon and extension handler): [{0}]".format(processes))
-
-            daemon_present = any("waagent -daemon" in command for (pid, command) in processes)
-            self.assertTrue(daemon_present, "Could not find the daemon in the cgroup: [{0}]".format(processes))
-
-            extension_handler_present = any(re.search(r"(WALinuxAgent-.+\.egg|waagent) -run-exthandlers", command) for (pid, command) in processes)
-            self.assertTrue(extension_handler_present, "Could not find the extension handler in the cgroup: [{0}]".format(processes))
-
     @patch('time.sleep', side_effect=lambda _: mock_sleep())
     def test_start_extension_command_should_not_use_systemd_when_cgroups_are_not_enabled(self, _):
         configurator = CGroupConfiguratorSystemdTestCase._get_new_cgroup_configurator_instance()
@@ -308,3 +298,129 @@ cgroup on /sys/fs/cgroup/blkio type cgroup (rw,nosuid,nodev,noexec,relatime,blki
 
                 self.assertIn("A TEST EXCEPTION", str(context_manager.exception))
 
+    def test_check_processes_in_agent_cgroup_should_raise_when_there_are_unexpected_processes_in_the_agent_cgroup(self):
+        CGroupConfiguratorSystemdTestCase._get_new_cgroup_configurator_instance()
+
+        # the test script recursively creates a number of descendant processes as given by its argument, then it blocks for a long time
+        pids_file = os.path.join(self.tmp_dir, "pids.txt")
+        test_script = os.path.join(self.tmp_dir, "create_descendants.py")
+        AgentTestCase.create_script(test_script, """
+#!/usr/bin/env python3
+import os
+import subprocess
+import sys
+import time
+
+count = int(sys.argv[1])
+output = sys.argv[2]
+
+with open(output, 'a') as output_file:
+    output_file.write('{0}\\n'.format(os.getpid()))
+
+if count > 1:
+    subprocess.Popen([sys.argv[0], str(count - 1), output]).wait()
+else:
+    time.sleep(120)
+""")
+        thread = None
+        child_commands = []
+        extension_process = None
+
+        try:
+            number_of_descendants = 3
+
+            # start the script on a separate thread
+            def run_script():
+                try:
+                    shellutil.run_command([test_script, ustr(number_of_descendants), pids_file])
+                except shellutil.CommandError as command_error:
+                    if command_error.returncode != -9:  # test cleanup terminates the commands, so this is expected
+                        raise
+
+            thread = threading.Thread(target=run_script)
+            thread.start()
+
+            # wait for the child processes
+            def processes_started():
+                if not os.path.exists(pids_file):
+                    return False
+                with open(pids_file, "r") as pids:
+                    processes_started.pids = pids.read().split()
+                    return len(processes_started.pids) == number_of_descendants
+            processes_started.pids = []
+
+            if not wait_for(processes_started):
+                raise Exception("The child processes were not started within the allowed timeout. Expected {0} processes; got: {1}".format(
+                    number_of_descendants, format_processes(processes_started.pids)))
+            child_commands = [int(pid) for pid in processes_started.pids]
+
+            # wait for shellutil to become aware of the process
+            def commands_running():
+                commands_running.commands = shellutil.get_running_commands()
+                return len(commands_running.commands) > 0
+            commands_running.commands = []
+
+            if not wait_for(commands_running):
+                raise Exception("shellutil did not start tracking the child process within the allowed timeout")
+
+            if len(commands_running.commands) != 1 or commands_running.commands[0] != child_commands[0]:
+                raise Exception("shellutil is not tracking the expected process. Expected: {0} Got: {1}".format(
+                    format_processes(child_commands[0]), format_processes(commands_running.commands)))
+
+            #
+            # That was a long test setup. Now let's verify that check_processes_in_agent_cgroup raises when there
+            # are unexpected processes in the agent's cgroup.
+            #
+            # For the agent's processes, we use the current process and its parent (in the actual agent these would be the daemon and the extension
+            # handler), and the commands started by the agent.
+            #
+            # For other processes, we use process 1, an extension (extensions are started using Popen instead of run_command so we just start an
+            # arbitrary process using Popen), and a process that already completed (for which we use a random number).
+            agent_processes = [os.getppid(), os.getpid()] + child_commands
+
+            extension_process = subprocess.Popen("sleep 120", shell=True)
+            random.seed()
+            completed = random.randint(1000, 10000)
+            while os.path.exists("/proc/{0}".format(completed)):  # ensure we do not use an existing process
+                completed = random.randint(1000, 10000)
+            other_processes = [1, extension_process.pid, completed]
+
+            with patch("azurelinuxagent.common.cgroupconfigurator.CGroupsApi.get_processes_in_cgroup", return_value=agent_processes + other_processes):
+                with self.assertRaises(UnexpectedProcessesInCGroupException) as context_manager:
+                    CGroupConfigurator.get_instance().check_processes_in_agent_cgroup()
+
+                reported = context_manager.exception.unexpected
+
+                self.assertEqual(
+                    len(other_processes), len(reported),
+                    "An incorrect number of processes was reported. Expected: {0} Got: {1}".format(format_processes(other_processes), reported))
+                for pid in other_processes:
+                    self.assertTrue(
+                        any(reported_process.startswith("[PID: {0}]".format(pid)) for reported_process in reported),
+                        "Process {0} was not reported. Got: {1}".format(format_processes([pid]), reported))
+
+            #
+            # And now verify that it does not raise when only the expected processes are in the cgroup
+            #
+            error = None
+            try:
+                with patch("azurelinuxagent.common.cgroupconfigurator.CGroupsApi.get_processes_in_cgroup", return_value=agent_processes):
+                    CGroupConfigurator.get_instance().check_processes_in_agent_cgroup()
+            except UnexpectedProcessesInCGroupException as exception:
+                error = exception
+            # we fail outside the except clause, otherwise the failure is reported as "During handling of the above exception, another exception occurred:..."
+            if error is not None:
+                self.fail("The check of the agent's cgroup should not have reported errors. Reported processes: {0}".format(error.unexpected))
+
+        finally:
+            # terminate the child processes, since they are blocked
+            for pid in child_commands:
+                os.kill(pid, signal.SIGKILL)
+            # wait for the thread that was running the test script
+            if thread is not None:
+                thread.join(timeout=5)
+            # terminate the extension process
+            if extension_process is not None:
+                extension_process.terminate()
+
+
diff --git a/tests/ga/test_monitor.py b/tests/ga/test_monitor.py
index b21d2791..fe84e00b 100644
--- a/tests/ga/test_monitor.py
+++ b/tests/ga/test_monitor.py
@@ -24,7 +24,7 @@ import string
 
 from azurelinuxagent.common import event, logger
 from azurelinuxagent.common.cgroup import CGroup, CpuCgroup, MemoryCgroup, MetricValue
-from azurelinuxagent.common.cgroupconfigurator import CGroupConfigurator
+from azurelinuxagent.common.cgroupconfigurator import CGroupConfigurator, UnexpectedProcessesInCGroupException
 from azurelinuxagent.common.cgroupstelemetry import CGroupsTelemetry
 from azurelinuxagent.common.event import EVENTS_DIRECTORY
 from azurelinuxagent.common.logger import Logger
@@ -193,15 +193,13 @@ class TestExtensionMetricsDataTelemetry(AgentTestCase):
         self.assertEqual(3, patch_add_metric.call_count)  # Three metrics being sent.
 
     @patch('azurelinuxagent.common.event.EventLogger.add_metric')
-    @patch('azurelinuxagent.common.event.EventLogger.add_event')
     @patch("azurelinuxagent.common.cgroupstelemetry.CGroupsTelemetry.poll_all_tracked")
     def test_send_extension_metrics_telemetry_for_empty_cgroup(self, patch_poll_all_tracked,  # pylint: disable=unused-argument
-                                                               patch_add_event, patch_add_metric,*args):
+                                                               patch_add_metric,*args):
         patch_poll_all_tracked.return_value = []
 
         PollResourceUsageOperation().run()
         self.assertEqual(1, patch_poll_all_tracked.call_count)
-        self.assertEqual(0, patch_add_event.call_count)
         self.assertEqual(0, patch_add_metric.call_count)
 
     @patch('azurelinuxagent.common.event.EventLogger.add_metric')
@@ -293,26 +291,8 @@ class PollResourceUsageOperationTestCase(AgentTestCase):
         CGroupConfigurator._instance = None  # pylint: disable=protected-access
         AgentTestCase.tearDownClass()
 
-    def test_it_should_report_processes_that_do_not_belong_to_the_agent_cgroup(self):
-        with mock_cgroup_commands() as mock_commands:
-            mock_commands.add_command(r'^systemd-cgls.+/walinuxagent.service$',
-''' 
-Directory /sys/fs/cgroup/cpu/system.slice/walinuxagent.service:
-├─27519 /usr/bin/python3 -u /usr/sbin/waagent -daemon
-├─27547 python3 -u bin/WALinuxAgent-2.2.48.1-py2.7.egg -run-exthandlers
-├─6200 systemd-cgls /sys/fs/cgroup/cpu,cpuacct/system.slice/walinuxagent.service
-├─5821 pidof systemd-networkd
-├─5822 iptables --version
-├─5823 iptables -w -t security -D OUTPUT -d 168.63.129.16 -p tcp -m conntrack --ctstate INVALID,NEW -j ACCEPT
-├─5824 iptables -w -t security -D OUTPUT -d 168.63.129.16 -p tcp -m owner --uid-owner 0 -j ACCEPT
-├─5825 ip route show
-├─5826 ifdown eth0 && ifup eth0
-├─5699 bash /var/lib/waagent/Microsoft.CPlat.Core.RunCommandLinux-1.0.1/bin/run-command-shim enable
-├─5701 tee -ia /var/log/azure/run-command/handler.log
-├─5719 /var/lib/waagent/Microsoft.CPlat.Core.RunCommandLinux-1.0.1/bin/run-command-extension enable
-├─5727 /bin/sh -c /var/lib/waagent/run-command/download/1/script.sh
-└─5728 /bin/sh /var/lib/waagent/run-command/download/1/script.sh
-''')
+    def test_it_should_issue_a_telemetry_event_when_there_are_processes_that_do_not_belong_to_the_agent_cgroup(self):
+        with patch.object(CGroupConfigurator.get_instance(), "check_processes_in_agent_cgroup", side_effect=UnexpectedProcessesInCGroupException(["A-TEST-PROCESS"])):
             with patch("azurelinuxagent.ga.monitor.add_event") as add_event_patcher:
                 PollResourceUsageOperation().run()
 
@@ -320,22 +300,11 @@ Directory /sys/fs/cgroup/cpu/system.slice/walinuxagent.service:
 
                 self.assertEqual(1, len(messages), "Exactly 1 telemetry event should have been reported. Events: {0}".format(messages))
 
-                unexpected_processes = [
-                    'bash /var/lib/waagent/Microsoft.CPlat.Core.RunCommandLinux-1.0.1/bin/run-command-shim enable',
-                    'tee -ia /var/log/azure/run-command/handler.log',
-                    '/var/lib/waagent/Microsoft.CPlat.Core.RunCommandLinux-1.0.1/bin/run-command-extension enable',
-                    '/bin/sh -c /var/lib/waagent/run-command/download/1/script.sh',
-                    '/bin/sh /var/lib/waagent/run-command/download/1/script.sh',
-                ]
-
-                for fp in unexpected_processes:  # pylint: disable=invalid-name
-                    self.assertIn(fp, messages[0], "[{0}] was not reported as an unexpected process. Events: {1}".format(fp, messages))
-
                 # The list of processes in the message is an array of strings: "['foo', ..., 'bar']"
                 search = re.search(r'\[(?P<processes>.+)\]', messages[0])
                 self.assertIsNotNone(search, "The event message is not in the expected format: {0}".format(messages[0]))
                 processes = search.group('processes')
-                self.assertEqual(5, len(processes.split(',')), 'Extra processes were reported as unexpected: {0}'.format(processes))
+                self.assertIn("A-TEST-PROCESS", processes, 'Extra processes were reported as unexpected: {0}'.format(processes))
 
 
 @patch("azurelinuxagent.common.utils.restutil.http_post")
diff --git a/tests/utils/miscellaneous_tools.py b/tests/utils/miscellaneous_tools.py
new file mode 100644
index 00000000..81e86428
--- /dev/null
+++ b/tests/utils/miscellaneous_tools.py
@@ -0,0 +1,62 @@
+# -*- coding: utf-8 -*-
+# Copyright 2018 Microsoft Corporation
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# Requires Python 2.6+ and Openssl 1.0+
+#
+
+#
+# Utility functions for the unit tests.
+#
+# This module is meant for simple, small tools that don't fit elsewhere.
+#
+
+import datetime
+import os
+import time
+
+from azurelinuxagent.common.future import ustr
+
+
+def format_processes(pid_list):
+    """
+    Formats the given PIDs as a sequence of PIDs and their command lines
+    """
+    def get_command_line(pid):
+        try:
+            cmdline = '/proc/{0}/cmdline'.format(pid)
+            if os.path.exists(cmdline):
+                with open(cmdline, "r") as cmdline_file:
+                    return "[PID: {0}] {1}".format(pid, cmdline_file.read())
+        except Exception:
+            pass
+        return "[PID: {0}] UNKNOWN".format(pid)
+
+    return ustr([get_command_line(pid) for pid in pid_list])
+
+
+def wait_for(predicate, timeout=10, frequency=0.01):
+    """
+    Waits until the given predicate is true or the given timeout elapses. Returns the last evaluation of the predicate.
+    Both the timeout and frequency are in seconds; the latter indicates how often the predicate is evaluated.
+    """
+    def to_seconds(time_delta):
+        return (time_delta.microseconds + (time_delta.seconds + time_delta.days * 24 * 3600) * 10 ** 6) / 10 ** 6
+
+    start_time = datetime.datetime.now()
+    while to_seconds(datetime.datetime.now() - start_time) < timeout:
+        if predicate():
+            return True
+        time.sleep(frequency)
+    return False
diff --git a/tests/utils/test_shell_util.py b/tests/utils/test_shell_util.py
index ef423bc4..528322e7 100644
--- a/tests/utils/test_shell_util.py
+++ b/tests/utils/test_shell_util.py
@@ -15,17 +15,16 @@
 #
 # Requires Python 2.6+ and Openssl 1.0+
 #
-import datetime
 import os
 import signal
 import tempfile
 import threading
-import time
 import unittest
 
 from azurelinuxagent.common.future import ustr
 import azurelinuxagent.common.utils.shellutil as shellutil
 from tests.tools import AgentTestCase, patch
+from tests.utils.miscellaneous_tools import wait_for, format_processes
 
 
 class ShellQuoteTestCase(AgentTestCase):
@@ -410,7 +409,7 @@ time.sleep(120)
                     threads.append(thread)
 
                 # now fetch the PIDs in the files created by the commands, but wait until they are created
-                if not self._wait_for(lambda: all(os.path.exists(file) and os.path.getsize(file) > 0 for file in pid_files)):
+                if not wait_for(lambda: all(os.path.exists(file) and os.path.getsize(file) > 0 for file in pid_files)):
                     raise Exception("The child processes did not start within the allowed timeout")
 
                 for sig_file in pid_files:
@@ -424,24 +423,23 @@ time.sleep(120)
                 started_commands = parent_processes[0:1] + child_processes[1:]
 
                 # wait for all the commands to start
-                running_commands = [[]]
-
                 def all_commands_running():
-                    running_commands[0] = shellutil.get_running_commands()
-                    return len(running_commands[0]) >= len(commands_to_execute) + 1  # +1 because run_pipe starts 2 commands
+                    all_commands_running.running_commands = shellutil.get_running_commands()
+                    return len(all_commands_running.running_commands) >= len(commands_to_execute) + 1  # +1 because run_pipe starts 2 commands
+                all_commands_running.running_commands = []
 
-                if not self._wait_for(all_commands_running):
+                if not wait_for(all_commands_running):
                     self.fail("shellutil.get_running_commands() did not report the expected number of commands after the allowed timeout.\nExpected: {0}\nGot: {1}".format(
-                        self._format_pids(started_commands), self._format_pids(running_commands[0])))
+                        format_processes(started_commands), format_processes(all_commands_running.running_commands)))
 
                 started_commands.sort()
-                running_commands[0].sort()
+                all_commands_running.running_commands.sort()
 
                 self.assertEqual(
                     started_commands,
-                    running_commands[0],
+                    all_commands_running.running_commands,
                     "shellutil.get_running_commands() did not return the expected commands.\nExpected: {0}\nGot: {1}".format(
-                        self._format_pids(started_commands), self._format_pids(running_commands[0])))
+                        format_processes(started_commands), format_processes(all_commands_running.running_commands)))
 
             finally:
                 # terminate the child processes, since they are blocked
@@ -449,48 +447,19 @@ time.sleep(120)
                     os.kill(pid, signal.SIGKILL)
 
             # once the processes complete, their PIDs should go away
-            running_commands = [[]]
-
-            def all_commands_completed():
-                running_commands[0] = shellutil.get_running_commands()
-                return len(running_commands[0]) == 0
+            def no_commands_running():
+                no_commands_running.running_commands = shellutil.get_running_commands()
+                return len(no_commands_running.running_commands) == 0
+            no_commands_running.running_commands = []
 
-            if not self._wait_for(all_commands_completed):
+            if not wait_for(no_commands_running):
                 self.fail("shellutil.get_running_commands() should return empty after the commands complete. Got: {0}".format(
-                    self._format_pids(running_commands[0])))
+                    format_processes(no_commands_running.running_commands)))
 
         finally:
             for thread in threads:
                 thread.join(timeout=5)
 
-    @staticmethod
-    def _format_pids(pids):
-        return ustr([RunCommandTestCase._get_command_line(pid) for pid in pids])
-
-    @staticmethod
-    def _get_command_line(pid):
-        try:
-            cmdline = '/proc/{0}/cmdline'.format(pid)
-            if os.path.exists(cmdline):
-                with open(cmdline, "r") as cmdline_file:
-                    return "[PID: {0}] {1}".format(pid, cmdline_file.read())
-        except Exception:
-            pass
-        return "[PID: {0}] UNKNOWN".format(pid)
-
-    @staticmethod
-    def _wait_for(predicate):
-        start_time = datetime.datetime.now()
-        while RunCommandTestCase._to_seconds(datetime.datetime.now() - start_time) < 10:
-            if predicate():
-                return True
-            time.sleep(0.01)
-        return False
-
-    @staticmethod
-    def _to_seconds(time_delta):
-        return (time_delta.microseconds + (time_delta.seconds + time_delta.days * 24 * 3600) * 10**6) / 10**6
-
 
 if __name__ == '__main__':
     unittest.main()