Repository: Azure/WALinuxAgent
Pull request #1199 (2018-06-06 23:27:59)

Cgroup support for monitoring agent and extensions
Passes hand smoke-tests on Ubuntu 14.04 (without systemd) and 16.04 (with systemd). Unit tests are disabled to keep Travis from trying to run them; mounting the cgroup device doesn't seem to work terribly well within containers, at least as Travis has them configured.


Patch:
diff --git a/azurelinuxagent/common/cgroups.py b/azurelinuxagent/common/cgroups.py
new file mode 100644
index 00000000..a68350c7
--- /dev/null
+++ b/azurelinuxagent/common/cgroups.py
@@ -0,0 +1,687 @@
+# 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+
+
+import errno
+import os
+import re
+import time
+
+from azurelinuxagent.common import logger
+from azurelinuxagent.common.osutil import get_osutil
+from azurelinuxagent.common.utils import fileutil
+from azurelinuxagent.common.version import AGENT_NAME, CURRENT_VERSION
+
+
+BASE_CGROUPS = '/sys/fs/cgroup'
+WRAPPER_CGROUP_NAME = "Agent+Extensions"
+METRIC_HIERARCHIES = ['cpu', 'memory']
+MEMORY_DEFAULT = -1
+
+re_user_system_times = re.compile('user (\d+)\nsystem (\d+)\n')
+
+related_services = {
+    "Microsoft.OSTCExtensions.LinuxDiagnostic":    ["omid", "omsagent-LAD", "mdsd-lde"],
+    "Microsoft.Azure.Diagnostics.LinuxDiagnostic": ["omid", "omsagent-LAD", "mdsd-lde"],
+}
+
+
+class CGroupsException(Exception):
+
+    def __init__(self, msg):
+        self.msg = msg
+
+    def __str__(self):
+        return repr(self.msg)
+
+
+# The metric classes (Cpu, Memory, etc) can all assume that CGroups is enabled, as the CGroupTelemetry
+# class is very careful not to call them if CGroups isn't enabled. Any tests should be disabled if the osutil
+# is_cgroups_support() method returns false.
+
+
+class Cpu(object):
+    def __init__(self, cgt):
+        """
+        Initialize data collection for the Cpu hierarchy. User must call update() before attempting to get
+        any useful metrics.
+
+        :param cgt: CGroupsTelemetry
+        :return:
+        """
+        self.cgt = cgt
+        self.osutil = get_osutil()
+        self.current_cpu_total = self.get_current_cpu_total()
+        self.previous_cpu_total = 0
+        self.current_system_cpu = self.osutil.get_total_cpu_ticks_since_boot()
+        self.previous_system_cpu = 0
+
+    def __str__(self):
+        return "Cgroup: Current {0}, previous {1}; System: Current {2}, previous {3}".format(
+            self.current_cpu_total, self.previous_cpu_total, self.current_system_cpu, self.previous_system_cpu
+        )
+
+    def get_current_cpu_total(self):
+        """
+        Compute the number of USER_HZ of CPU time (user and system) consumed by this cgroup since boot.
+
+        :return: int
+        """
+        cpu_total = 0
+        cpu_stat = self.cgt.cgroup.get_file_contents('cpu', 'cpuacct.stat')
+        if cpu_stat is not None:
+            m = re_user_system_times.match(cpu_stat)
+            if m:
+                cpu_total = int(m.groups()[0]) + int(m.groups()[1])
+        return cpu_total
+
+    def update(self):
+        """
+        Update all raw data required to compute metrics of interest. The intent is to call update() once, then
+        call the various get_*() methods which use this data, which we've collected exactly once.
+        """
+        self.previous_cpu_total = self.current_cpu_total
+        self.previous_system_cpu = self.current_system_cpu
+        self.current_cpu_total = self.get_current_cpu_total()
+        self.current_system_cpu = self.osutil.get_total_cpu_ticks_since_boot()
+
+    def get_cpu_percent(self):
+        """
+        Compute the percent CPU time used by this cgroup over the elapsed time since the last time this instance was
+        update()ed.  If the cgroup fully consumed 2 cores on a 4 core system, return 200.
+
+        :return: float
+        """
+        cpu_delta = self.current_cpu_total - self.previous_cpu_total
+        system_delta = max(1, self.current_system_cpu - self.previous_system_cpu)
+
+        return round(float(cpu_delta * self.cgt.cpu_count * 100) / float(system_delta), 3)
+
+    def collect(self):
+        """
+        Collect and return a list of all cpu metrics. If no metrics are collected, return an empty list.
+
+        :return: [(str, str, float)]
+        """
+        self.update()
+        usage = self.get_cpu_percent()
+        return [("Process", "% Processor Time", usage)]
+
+
+class Memory(object):
+    def __init__(self, cgt):
+        """
+        Initialize data collection for the Memory hierarchy
+
+        :param cgt: CGroupsTelemetry
+        :return:
+        """
+        self.cgt = cgt
+
+    def update(self):
+        pass
+
+    def collect(self):
+        """
+        Collect and return a list of all memory metrics
+        """
+        self.update()
+        return []
+
+
+class CGroupsTelemetry(object):
+    """
+    Encapsulate the cgroup-based telemetry for the agent or one of its extensions, or for the aggregation across
+    the agent and all of its extensions. These objects should have lifetimes that span the time window over which
+    measurements are desired; in general, they're not terribly effective at providing instantaneous measurements.
+    """
+    _tracked = {}
+    _metrics = {
+        "cpu": Cpu,
+        "memory": Memory
+    }
+    _hierarchies = list(_metrics.keys())
+    tracked_names = set()
+
+    @staticmethod
+    def metrics_hierarchies():
+        return CGroupsTelemetry._hierarchies
+
+    @staticmethod
+    def track_cgroup(cgroup):
+        """
+        Create a CGroupsTelemetry object to track a particular CGroups instance. Typical usage:
+        1) Create a CGroups object
+        2) Ask CGroupsTelemetry to track it
+        3) Tell the CGroups object to add one or more processes (or let systemd handle that, for its cgroups)
+
+        :param CGroups cgroup: The cgroup to track
+        """
+        name = cgroup.name
+        if CGroups.enabled() and not CGroupsTelemetry.is_tracked(name):
+            tracker = CGroupsTelemetry(name, cgroup=cgroup)
+            CGroupsTelemetry._tracked[name] = tracker
+
+    @staticmethod
+    def track_systemd_service(name):
+        """
+        If not already tracking it, create the CGroups object for a systemd service and track it.
+
+        :param str name: Service name (without .service suffix) to be tracked.
+        """
+        service_name = "{0}.service".format(name)
+        if CGroups.enabled() and not CGroupsTelemetry.is_tracked(service_name):
+            cgroup = CGroups.for_systemd_service(service_name)
+            tracker = CGroupsTelemetry(service_name, cgroup=cgroup)
+            CGroupsTelemetry._tracked[service_name] = tracker
+
+    @staticmethod
+    def track_extension(name, cgroup=None):
+        """
+        Create all required CGroups to track all metrics for an extension and its associated services.
+
+        :param str name: Full name of the extension to be tracked
+        :param CGroups cgroup: CGroup for the extension itself. This method will create it if none is supplied.
+        """
+        if not CGroups.enabled():
+            return
+
+        if not CGroupsTelemetry.is_tracked(name):
+            cgroup = CGroups.for_extension(name) if cgroup is None else cgroup
+            logger.info("Now tracking cgroup {0}".format(name))
+            CGroupsTelemetry.track_cgroup(cgroup)
+        if CGroups.is_systemd_manager():
+            if name in related_services:
+                for service_name in related_services[name]:
+                    CGroupsTelemetry.track_systemd_service(service_name)
+
+    @staticmethod
+    def track_agent(name):
+        """
+        Create and track the correct cgroup for the agent itself. The actual cgroup depends on whether systemd
+        is in use, but the caller doesn't need to know that.
+
+        :param str name: Name for the agent itself
+        """
+        if not CGroups.enabled():
+            return
+        if CGroups.is_systemd_manager():
+            CGroupsTelemetry.track_systemd_service(name)
+        else:
+            CGroupsTelemetry.track_cgroup(CGroups.for_extension(name))
+
+    @staticmethod
+    def is_tracked(name):
+        return name in CGroupsTelemetry._tracked
+
+    @staticmethod
+    def stop_tracking(name):
+        """
+        Stop tracking telemetry for the CGroups associated with an extension. If any system services are being
+        tracked, those will continue to be tracked; multiple extensions might rely upon the same service.
+
+        :param str name: Extension to be dropped from tracking
+        """
+        if CGroupsTelemetry.is_tracked(name):
+            del (CGroupsTelemetry._tracked[name])
+
+    @staticmethod
+    def collect_all_tracked():
+        """
+        Return a dictionary mapping from the name of a tracked cgroup to the list of collected metrics for that cgroup.
+
+        :returns: Dictionary of list collected metrics (metric class, metric name, value), by cgroup
+        :rtype: dict(str: [(str, str, float)])
+        """
+        results = {}
+        for cgroup_name, collector in CGroupsTelemetry._tracked.items():
+            cgroup_name = cgroup_name if cgroup_name else WRAPPER_CGROUP_NAME
+            results[cgroup_name] = collector.collect()
+        return results
+
+    @staticmethod
+    def update_tracked(ext_handlers):
+        """
+        Track CGroups for all enabled extensions.
+        Track CGroups for services created by enabled extensions.
+        Stop tracking CGroups for not-enabled extensions.
+
+        :param List(ExtHandler) ext_handlers:
+        """
+        if not CGroups.enabled():
+            return
+
+        not_enabled_extensions = set()
+        for extension in ext_handlers:
+            if extension.properties.state == u"enabled":
+                CGroupsTelemetry.track_extension(extension.name)
+            else:
+                not_enabled_extensions.add(extension.name)
+
+        names_now_tracked = set(CGroupsTelemetry._tracked.keys())
+        if CGroupsTelemetry.tracked_names != names_now_tracked:
+            now_tracking = " ".join("[{0}]".format(name) for name in sorted(names_now_tracked))
+            if len(now_tracking):
+                logger.info("After updating cgroup telemetry, tracking {0}".format(now_tracking))
+            else:
+                logger.warn("After updating cgroup telemetry, tracking no cgroups.")
+            CGroupsTelemetry.tracked_names = names_now_tracked
+
+    def __init__(self, name, cgroup=None):
+        """
+        Create the necessary state to collect metrics for the agent, one of its extensions, or the aggregation across
+        the agent and all of its extensions. To access aggregated metrics, instantiate this object with an empty string
+        or None.
+
+        :param name: str
+        """
+        if name is None:
+            name = ""
+        self.name = name
+        if cgroup is None:
+            cgroup = CGroups.for_extension(name)
+        self.cgroup = cgroup
+        self.cpu_count = CGroups.get_num_cores()
+        self.current_wall_time = time.time()
+        self.previous_wall_time = 0
+
+        self.data = {}
+        if CGroups.enabled():
+            for hierarchy in CGroupsTelemetry.metrics_hierarchies():
+                self.data[hierarchy] = CGroupsTelemetry._metrics[hierarchy](self)
+
+    def collect(self):
+        """
+        Return a list of collected metrics. Each element is a tuple of
+        (metric group name, metric name, metric value)
+        :return: [(str, str, float)]
+        """
+        results = []
+        for collector in self.data.values():
+            results.extend(collector.collect())
+        return results
+
+
+class CGroups(object):
+    """
+    This class represents the cgroup folders for the agent or an extension. This is a pretty lightweight object
+    without much state worth preserving; it's not unreasonable to create one just when you need it.
+    """
+    # whether cgroup support is enabled
+    _enabled = True
+    _hierarchies = CGroupsTelemetry.metrics_hierarchies()
+    _use_systemd = None     # Tri-state: None (i.e. "unknown"), True, False
+    _osutil = get_osutil()
+
+    @staticmethod
+    def _construct_custom_path_for_hierarchy(hierarchy, cgroup_name):
+        return os.path.join(BASE_CGROUPS, hierarchy, AGENT_NAME, cgroup_name).rstrip(os.path.sep)
+
+    @staticmethod
+    def _construct_systemd_path_for_hierarchy(hierarchy, cgroup_name):
+        return os.path.join(BASE_CGROUPS, hierarchy, 'system.slice', cgroup_name).rstrip(os.path.sep)
+
+    @staticmethod
+    def for_extension(name):
+        return CGroups(name, CGroups._construct_custom_path_for_hierarchy)
+
+    @staticmethod
+    def for_systemd_service(name):
+        return CGroups(name.lower(), CGroups._construct_systemd_path_for_hierarchy)
+
+    @staticmethod
+    def enabled():
+        return CGroups._osutil.is_cgroups_supported() and CGroups._enabled
+
+    @staticmethod
+    def disable():
+        CGroups._enabled = False
+
+    @staticmethod
+    def enable():
+        CGroups._enabled = True
+
+    def __init__(self, name, path_maker):
+        """
+        Construct CGroups object. Create appropriately-named directory for each hierarchy of interest.
+
+        :param str name: Name for the cgroup (usually the full name of the extension)
+        :param path_maker: Function which constructs the root path for a given hierarchy where this cgroup lives
+        """
+        if name == "":
+            self.name = "Agents+Extensions"
+            self.is_wrapper_cgroup = True
+        else:
+            self.name = name
+            self.is_wrapper_cgroup = False
+
+        self.cgroups = {}
+
+        if not self.enabled():
+            return
+
+        system_hierarchies = os.listdir(BASE_CGROUPS)
+        for hierarchy in CGroups._hierarchies:
+            if hierarchy not in system_hierarchies:
+                self.disable()
+                raise CGroupsException("Hierarchy {0} is not mounted".format(hierarchy))
+
+            cgroup_name = "" if self.is_wrapper_cgroup else self.name
+            cgroup_path = path_maker(hierarchy, cgroup_name)
+            if not os.path.isdir(cgroup_path):
+                logger.info("Creating cgroup directory {0}".format(cgroup_path))
+                CGroups._try_mkdir(cgroup_path)
+            self.cgroups[hierarchy] = cgroup_path
+
+    @staticmethod
+    def is_systemd_manager():
+        """
+        Determine if systemd is managing system services. Many extensions are structured as a set of services,
+        including the agent itself; systemd expects those services to remain in the cgroups in which it placed them.
+        If this process (presumed to be the agent) is in a cgroup that looks like one created by systemd, we can
+        assume systemd is in use.
+
+        :return: True if systemd is managing system services
+        :rtype: Bool
+        """
+        if not CGroups.enabled():
+            return False
+        if CGroups._use_systemd is None:
+            hierarchy = METRIC_HIERARCHIES[0]
+            path = CGroups.get_my_cgroup_folder(hierarchy)
+            CGroups._use_systemd = path.startswith(CGroups._construct_systemd_path_for_hierarchy(hierarchy, ""))
+        return CGroups._use_systemd
+
+    @staticmethod
+    def _try_mkdir(path):
+        """
+        Try to create a directory, recursively. If it already exists as such, do nothing. Raise the appropriate
+        exception should an error occur.
+
+        :param path: str
+        """
+        if not os.path.isdir(path):
+            try:
+                os.makedirs(path, 0o755)
+            except OSError as e:
+                if e.errno == errno.EEXIST:
+                    if not os.path.isdir(path):
+                        raise CGroupsException(
+                            "Create directory for cgroup {0}: normal file already exists with that name".format(path)
+                        )
+                    else:
+                        pass    # There was a race to create the directory, but it's there now, and that's fine
+                elif e.errno == errno.EACCES:
+                    # This is unexpected, as the agent runs as root
+                    raise CGroupsException("Create directory for cgroup {0}: permission denied".format(path))
+                else:
+                    raise
+
+    def add(self, pid):
+        """
+        Add a process to the cgroups for this agent/extension.
+        """
+        if not self.enabled():
+            return
+
+        if self.is_wrapper_cgroup:
+            raise CGroupsException("Cannot add a process to the Agents+Extensions wrapper cgroup")
+
+        if not self._osutil.check_pid_alive(pid):
+            raise CGroupsException('PID {0} does not exist'.format(pid))
+        for hierarchy, cgroup in self.cgroups.items():
+            tasks_file = self._get_cgroup_file(hierarchy, 'cgroup.procs')
+            fileutil.append_file(tasks_file, "{0}\n".format(pid))
+
+    def set_limits(self):
+        """
+        Set per-hierarchy limits based on the cgroup name (agent or particular extension)
+        """
+        # TODO: set limits, simply record telemetry for now
+        # cg.set_cpu_limit(50)
+        # cg.set_memory_limit(500)
+        pass
+
+    @staticmethod
+    def _apply_wrapper_limits(path, hierarchy):
+        """
+        Find wrapping limits for the hierarchy and apply them to the cgroup denoted by the path
+
+        :param path: str
+        :param hierarchy: str
+        """
+        pass
+
+    @staticmethod
+    def _setup_wrapper_groups():
+        """
+        For each hierarchy, construct the wrapper cgroup and apply the appropriate limits
+        """
+        for hierarchy in METRIC_HIERARCHIES:
+            root_dir = CGroups._construct_custom_path_for_hierarchy(hierarchy, "")
+            CGroups._try_mkdir(root_dir)
+            CGroups._apply_wrapper_limits(root_dir, hierarchy)
+
+    @staticmethod
+    def setup(suppress_process_add=False):
+        """
+        Only needs to be called once, and should be called from the -daemon instance of the agent.
+            Mount the cgroup fs if necessary
+            Create wrapper cgroups for agent-plus-extensions and set limits on them;
+            Add this process to the "agent" cgroup, if required
+        Actual collection of metrics from cgroups happens in the -run-exthandlers instance
+        """
+        cgroups_enabled = True
+        if CGroups.enabled():
+            try:
+                CGroups._osutil.mount_cgroups()
+                if not suppress_process_add:
+                    CGroups._setup_wrapper_groups()
+                    pid = int(os.getpid())
+                    if not CGroups.is_systemd_manager():
+                        cg = CGroups.for_extension(AGENT_NAME)
+                        cg.add(pid)
+                        logger.info("Add daemon process pid {0} to {1} cgroup".format(pid, cg.name))
+                    else:
+                        logger.info("Daemon process pid {0} cgroup managed by systemd".format(pid))
+                status = "OK"
+            except CGroupsException as cge:
+                status = cge.msg
+                cgroups_enabled = False
+                CGroups.disable()
+        else:
+            status = "Cgroups not supported by platform"
+
+        from azurelinuxagent.common.event import add_event, WALAEventOperation
+        add_event(
+            AGENT_NAME,
+            version=CURRENT_VERSION,
+            op=WALAEventOperation.InitializeCGroups,
+            is_success=cgroups_enabled,
+            message=status,
+            log_event=False)
+
+    @staticmethod
+    def add_to_extension_cgroup(name, pid=int(os.getpid())):
+        """
+        Create cgroup directories for this extension in each of the hierarchies and add this process to the new cgroup.
+        Should only be called when creating sub-processes and invoked inside the fork/exec window. As a result,
+        there's no point in returning the CGroups object itself; the goal is to move the child process into the
+        cgroup before the new code even starts running.
+
+        :param str name: Short name of extension, suitable for naming directories in the filesystem
+        :param int pid: Process id of extension to be added to the cgroup
+        """
+        if not CGroups.enabled():
+            return
+        if name == AGENT_NAME:
+            logger.warn('Extension cgroup name cannot match agent cgroup name ({0})'.format(AGENT_NAME))
+            return
+
+        try:
+            logger.info("Move process {0} into cgroups for extension {1}".format(pid, name))
+            CGroups.for_extension(name).add(pid)
+        except Exception as ex:
+            logger.warn("Unable to move process {0} into cgroups for extension {1}: {2}".format(pid, name, ex))
+
+    @staticmethod
+    def get_my_cgroup_path(hierarchy_id):
+        """
+        Get the cgroup path "suffix" for this process for the given hierarchy ID. The leading "/" is always stripped,
+        so the suffix is suitable for passing to os.path.join(). (If the process is in the root cgroup, an empty
+        string is returned, and os.path.join() will still do the right thing.)
+
+        :param hierarchy_id: str
+        :return: str
+        """
+        cgroup_paths = fileutil.read_file("/proc/self/cgroup")
+        for entry in cgroup_paths.splitlines():
+            fields = entry.split(':')
+            if fields[0] == hierarchy_id:
+                return fields[2].lstrip(os.path.sep)
+        raise CGroupsException("This process belongs to no cgroup for hierarchy ID {0}".format(hierarchy_id))
+
+    @staticmethod
+    def get_hierarchy_id(hierarchy):
+        """
+        Get the cgroups hierarchy ID for a given hierarchy name
+
+        :param hierarchy:
+        :return: str
+        """
+        cgroup_states = fileutil.read_file("/proc/cgroups")
+        for entry in cgroup_states.splitlines():
+            fields = entry.split('\t')
+            if fields[0] == hierarchy:
+                return fields[1]
+        raise CGroupsException("Cgroup hierarchy {0} not found in /proc/cgroups".format(hierarchy))
+
+    @staticmethod
+    def get_my_cgroup_folder(hierarchy):
+        """
+        Find the path of the cgroup in which this process currently lives for the given hierarchy.
+
+        :param hierarchy: str
+        :return: str
+        """
+        hierarchy_id = CGroups.get_hierarchy_id(hierarchy)
+        return os.path.join(BASE_CGROUPS, hierarchy, CGroups.get_my_cgroup_path(hierarchy_id))
+
+    def _get_cgroup_file(self, hierarchy, file_name):
+        return os.path.join(self.cgroups[hierarchy], file_name)
+
+    @staticmethod
+    def _convert_cpu_limit_to_fraction(value):
+        """
+        Convert a CPU limit from percent (e.g. 50 meaning 50%) to a decimal fraction (0.50).
+        :return: Fraction of one CPU to be made available (e.g. 0.5 means half a core)
+        :rtype: float
+        """
+        try:
+            limit = float(value)
+        except ValueError:
+            raise CGroupsException('CPU Limit must be convertible to a float')
+
+        if limit <= float(0) or limit > float(CGroups.get_num_cores() * 100):
+            raise CGroupsException('CPU Limit must be between 0 and 100 * numCores')
+
+        return limit / 100.0
+
+    def get_file_contents(self, hierarchy, file_name):
+        """
+        Retrieve the value of a parameter from a hierarchy.
+
+        :param hierarchy: str
+        :param file_name: str
+        :return: st
+        """
+        if hierarchy in self.cgroups:
+            parameter_file = self._get_cgroup_file(hierarchy, file_name)
+            try:
+                return fileutil.read_file(parameter_file)
+            except Exception:
+                raise CGroupsException("Could not retrieve cgroup file {0}/{1}".format(hierarchy, file_name))
+        else:
+            raise CGroupsException("{0} subsystem not available in cgroup {1}. cgroup paths: {2}".format(
+                hierarchy, self.name, self.cgroups))
+
+    def get_parameter(self, hierarchy, parameter_name):
+        """
+        Retrieve the value of a parameter from a hierarchy.
+
+        :param hierarchy: str
+        :param parameter_name: str
+        :return: str
+        """
+        values = self.get_file_contents(hierarchy, parameter_name).splitlines
+        return values[0]
+
+    def set_cpu_limit(self, limit=None):
+        """
+        Limit this cgroup to a percentage of a single core. limit=10 means 10% of one core; 150 means 150%, which
+        is useful only in multi-core systems.
+        To limit a cgroup to utilize 10% of a single CPU, use the following commands:
+            # echo 10000 > /cgroup/cpu/red/cpu.cfs_quota_us
+            # echo 100000 > /cgroup/cpu/red/cpu.cfs_period_us
+
+        :param limit:
+        """
+        if not CGroups.enabled():
+            return
+
+        if limit is None:
+            return
+
+        if 'cpu' in self.cgroups:
+            total_units = float(self.get_parameter('cpu', 'cpu.cfs_period_us'))
+            limit_units = self._convert_cpu_limit_to_fraction(limit) * total_units
+            cpu_shares_file = self._get_cgroup_file('cpu', 'cpu.cfs_quota_us')
+            fileutil.write_file(cpu_shares_file, "{0}\n".format(limit_units))
+        else:
+            raise CGroupsException("CPU hierarchy not available in this cgroup")
+
+    @staticmethod
+    def get_num_cores():
+        """
+        Return the number of CPU cores exposed to this system.
+
+        :return: int
+        """
+        return CGroups._osutil.get_processor_cores()
+
+    @staticmethod
+    def _format_memory_value(unit, limit=None):
+        units = {'bytes': 1, 'kilobytes': 1024, 'megabytes': 1024*1024, 'gigabytes': 1024*1024*1024}
+        if unit not in units:
+            raise CGroupsException("Unit must be one of {0}".format(units.keys()))
+        if limit is None:
+            value = MEMORY_DEFAULT
+        else:
+            try:
+                limit = float(limit)
+            except ValueError:
+                raise CGroupsException('Limit must be convertible to a float')
+            else:
+                value = int(limit * units[unit])
+        return value
+
+    def set_memory_limit(self, limit=None, unit='megabytes'):
+        if 'memory' in self.cgroups:
+            value = self._format_memory_value(unit, limit)
+            memory_limit_file = self._get_cgroup_file('memory', 'memory.limit_in_bytes')
+            with open(memory_limit_file, 'w+') as f:
+                f.write("{0}\n".format(value))
+        else:
+            raise CGroupsException("Memory hierarchy not available in this cgroup")
diff --git a/azurelinuxagent/common/event.py b/azurelinuxagent/common/event.py
index 2e35d8a2..33e326fc 100644
--- a/azurelinuxagent/common/event.py
+++ b/azurelinuxagent/common/event.py
@@ -60,6 +60,7 @@ class WALAEventOperation:
     HttpErrors = "HttpErrors"
     ImdsHeartbeat = "ImdsHeartbeat"
     Install = "Install"
+    InitializeCGroups = "InitializeCGroups"
     InitializeHostPlugin = "InitializeHostPlugin"
     Log = "Log"
     Partition = "Partition"
@@ -303,6 +304,33 @@ class EventLogger(object):
         except EventError:
             pass
 
+    def add_metric(self, category, counter, instance, value, log_event=False):
+        """
+        Create and save an event which contains a telemetry event.
+
+        :param str category: The category of metric (e.g. "cpu", "memory")
+        :param str counter: The specific metric within the category (e.g. "%idle")
+        :param str instance: For instanced metrics, the instance identifier (filesystem name, cpu core#, etc.)
+        :param value: Value of the metric
+        :param bool log_event: If true, log the collected metric in the agent log
+        """
+        if log_event:
+            from azurelinuxagent.common.version import AGENT_NAME
+            message = "Metric {0}/{1} [{2}] = {3}".format(category, counter, instance, value)
+            _log_event(AGENT_NAME, "METRIC", message, 0)
+
+        event = TelemetryEvent(4, "69B669B9-4AF8-4C50-BDC4-6006FA76E975")
+        event.parameters.append(TelemetryEventParam('Category', category))
+        event.parameters.append(TelemetryEventParam('Counter', counter))
+        event.parameters.append(TelemetryEventParam('Instance', instance))
+        event.parameters.append(TelemetryEventParam('Value', value))
+
+        data = get_properties(event)
+        try:
+            self.save_event(json.dumps(data))
+        except EventError as e:
+            logger.error("{0}", e)
+
 
 __event_logger__ = EventLogger()
 
@@ -335,6 +363,25 @@ def report_periodic(delta, op, is_success=True, message=''):
               op=op)
 
 
+def report_metric(category, counter, instance, value, log_event=False, reporter=__event_logger__):
+    """
+    Send a telemetry event reporting a single instance of a performance counter.
+    :param str category: The category of the metric (cpu, memory, etc)
+    :param str counter: The name of the metric ("%idle", etc)
+    :param str instance: For instanced metrics, the identifier of the instance. E.g. a disk drive name, a cpu core#
+    :param     value: The value of the metric
+    :param bool log_event: If True, log the metric in the agent log as well
+    :param EventLogger reporter: The EventLogger instance to which metric events should be sent
+    """
+    if reporter.event_dir is None:
+        from azurelinuxagent.common.version import AGENT_NAME
+        logger.warn("Cannot report metric event -- Event reporter is not initialized.")
+        message = "Metric {0}/{1} [{2}] = {3}".format(category, counter, instance, value)
+        _log_event(AGENT_NAME, "METRIC", message, 0)
+        return
+    reporter.add_metric(category, counter, instance, value, log_event)
+
+
 def add_event(name, op=WALAEventOperation.Unknown, is_success=True, duration=0,
               version=CURRENT_VERSION,
               message="", evt_type="", is_internal=False, log_event=True,
diff --git a/azurelinuxagent/common/osutil/default.py b/azurelinuxagent/common/osutil/default.py
index 8d07ca60..32a90e30 100644
--- a/azurelinuxagent/common/osutil/default.py
+++ b/azurelinuxagent/common/osutil/default.py
@@ -19,6 +19,7 @@
 import array
 import base64
 import datetime
+import errno
 import fcntl
 import glob
 import multiprocessing
@@ -73,6 +74,8 @@ FIREWALL_DELETE_OWNER_ACCEPT = "iptables {0} -t security -D OUTPUT -d {1} -p tcp
 FIREWALL_DELETE_CONNTRACK_DROP = "iptables {0} -t security -D OUTPUT -d {1} -p tcp -m conntrack --ctstate INVALID,NEW -j DROP"
 
 PACKET_PATTERN = "^\s*(\d+)\s+(\d+)\s+DROP\s+.*{0}[^\d]*$"
+ALL_CPUS_REGEX = re.compile('^cpu .*')
+
 
 _enable_firewall = True
 
@@ -87,6 +90,7 @@ IOCTL_SIOCGIFFLAGS = 0x8913
 IOCTL_SIOCGIFHWADDR = 0x8927
 IFNAMSIZ = 16
 
+
 class DefaultOSUtil(object):
     def __init__(self):
         self.agent_conf_file_path = '/etc/waagent.conf'
@@ -281,6 +285,12 @@ class DefaultOSUtil(object):
         return id_that == id_this or \
             id_that == self._correct_instance_id(id_this)
 
+    def is_cgroups_supported(self):
+        return False
+
+    def mount_cgroups(self):
+        pass
+
     def get_agent_conf_file_path(self):
         return self.agent_conf_file_path
 
@@ -618,8 +628,8 @@ class DefaultOSUtil(object):
                 time.sleep(1)
         return False
 
-    def mount(self, dvd, mount_point, option="", chk_err=True):
-        cmd = "mount {0} {1} {2}".format(option, dvd, mount_point)
+    def mount(self, device, mount_point, option="", chk_err=True):
+        cmd = "mount {0} {1} {2}".format(option, device, mount_point)
         retcode, err = shellutil.run_get_output(cmd, chk_err)
         if retcode != 0:
             detail = "[{0}] returned {1}: {2}".format(cmd, retcode, err)
@@ -691,7 +701,6 @@ class DefaultOSUtil(object):
         struct_size = 32 if python_arc == '32bit' else 40
         return struct_size
 
-
     def _get_all_interfaces(self):
         """
         Return a dictionary mapping from interface name to IPv4 address.
@@ -1100,8 +1109,52 @@ class DefaultOSUtil(object):
         return multiprocessing.cpu_count()
 
     def check_pid_alive(self, pid):
-        return pid is not None and os.path.isdir(os.path.join('/proc', pid))
+        try:
+            pid = int(pid)
+            os.kill(pid, 0)
+        except (ValueError, TypeError):
+            return False
+        except OSError as e:
+            if e.errno == errno.EPERM:
+                return True
+            return False
+        return True
 
     @property
     def is_64bit(self):
         return sys.maxsize > 2**32
+
+    @staticmethod
+    def _get_proc_stat():
+        """
+        Get the contents of /proc/stat.
+        # cpu  813599 3940 909253 154538746 874851 0 6589 0 0 0
+        # cpu0 401094 1516 453006 77276738 452939 0 3312 0 0 0
+        # cpu1 412505 2423 456246 77262007 421912 0 3276 0 0 0
+
+        :return: A single string with the contents of /proc/stat
+        :rtype: str
+        """
+        results = None
+        try:
+            results = fileutil.read_file('/proc/stat')
+        except (OSError, IOError) as ex:
+            logger.warn("Couldn't read /proc/stat: {0}".format(ex.strerror))
+
+        return results
+
+    @staticmethod
+    def get_total_cpu_ticks_since_boot():
+        """
+        Compute the number of USER_HZ units of time that have elapsed in all categories, across all cores, since boot.
+
+        :return: int
+        """
+        system_cpu = 0
+        proc_stat = DefaultOSUtil._get_proc_stat()
+        if proc_stat is not None:
+            for line in proc_stat.splitlines():
+                if ALL_CPUS_REGEX.match(line):
+                    system_cpu = sum(int(i) for i in line.split()[1:7])
+                    break
+        return system_cpu
diff --git a/azurelinuxagent/common/osutil/freebsd.py b/azurelinuxagent/common/osutil/freebsd.py
index 38790186..9295f58a 100644
--- a/azurelinuxagent/common/osutil/freebsd.py
+++ b/azurelinuxagent/common/osutil/freebsd.py
@@ -247,3 +247,7 @@ class FreeBSDOSUtil(DefaultOSUtil):
                     if not possible.startswith('pass'):
                         return possible
         return None
+
+    @staticmethod
+    def get_total_cpu_ticks_since_boot():
+        return 0
diff --git a/azurelinuxagent/common/osutil/gaia.py b/azurelinuxagent/common/osutil/gaia.py
index ca50afdf..748816e0 100644
--- a/azurelinuxagent/common/osutil/gaia.py
+++ b/azurelinuxagent/common/osutil/gaia.py
@@ -139,16 +139,16 @@ class GaiaOSUtil(DefaultOSUtil):
     def eject_dvd(self, chk_err=True):
         logger.warn('eject is not supported on GAiA')
 
-    def mount(self, dvd, mount_point, option="", chk_err=True):
-        logger.info('mount {0} {1} {2}', dvd, mount_point, option)
+    def mount(self, device, mount_point, option="", chk_err=True):
+        logger.info('mount {0} {1} {2}', device, mount_point, option)
         if 'udf,iso9660' in option:
             ret, out = super(GaiaOSUtil, self).mount(
-                dvd, mount_point, option=option.replace('udf,iso9660', 'udf'),
+                device, mount_point, option=option.replace('udf,iso9660', 'udf'),
                 chk_err=chk_err)
             if not ret:
                 return ret, out
         return super(GaiaOSUtil, self).mount(
-            dvd, mount_point, option=option, chk_err=chk_err)
+            device, mount_point, option=option, chk_err=chk_err)
 
     def allow_dhcp_broadcast(self):
         logger.info('allow_dhcp_broadcast is ignored on GAiA')
diff --git a/azurelinuxagent/common/osutil/openbsd.py b/azurelinuxagent/common/osutil/openbsd.py
index 62d0dbd8..1b10ab23 100644
--- a/azurelinuxagent/common/osutil/openbsd.py
+++ b/azurelinuxagent/common/osutil/openbsd.py
@@ -345,3 +345,7 @@ class OpenBSDOSUtil(DefaultOSUtil):
         Return device name attached to ide port 'n'.
         """
         return "wd{0}".format(port_id)
+
+    @staticmethod
+    def get_total_cpu_ticks_since_boot():
+        return 0
diff --git a/azurelinuxagent/common/osutil/ubuntu.py b/azurelinuxagent/common/osutil/ubuntu.py
index b06e75c7..ac7bc3a3 100644
--- a/azurelinuxagent/common/osutil/ubuntu.py
+++ b/azurelinuxagent/common/osutil/ubuntu.py
@@ -16,14 +16,21 @@
 # Requires Python 2.6+ and Openssl 1.0+
 #
 
+import os
 import time
 
 import azurelinuxagent.common.logger as logger
+import azurelinuxagent.common.utils.fileutil as fileutil
 import azurelinuxagent.common.utils.shellutil as shellutil
 
+from azurelinuxagent.common.future import ustr
 from azurelinuxagent.common.osutil.default import DefaultOSUtil
 
 
+def _cgroup_path(tail=""):
+    return os.path.join('/sys/fs/cgroup/', tail).rstrip(os.path.sep)
+
+
 class Ubuntu14OSUtil(DefaultOSUtil):
     def __init__(self):
         super(Ubuntu14OSUtil, self).__init__()
@@ -46,6 +53,40 @@ class Ubuntu14OSUtil(DefaultOSUtil):
     def get_dhcp_lease_endpoint(self):
         return self.get_endpoint_from_leases_path('/var/lib/dhcp/dhclient.*.leases')
 
+    def is_cgroups_supported(self):
+        if 'TRAVIS' in os.environ and os.environ['TRAVIS'] == 'true':
+            return False
+        return True
+
+    def mount_cgroups(self):
+        try:
+            if not os.path.exists(_cgroup_path()):
+                fileutil.mkdir(_cgroup_path())
+                self.mount(device='cgroup_root',
+                           mount_point=_cgroup_path(),
+                           option="-t tmpfs",
+                           chk_err=False)
+            elif not os.path.isdir(_cgroup_path()):
+                logger.error("Could not mount cgroups: ordinary file at {0}".format(_cgroup_path()))
+                return
+
+            for metric_hierarchy in ['cpu,cpuacct', 'memory']:
+                target_path = _cgroup_path(metric_hierarchy)
+                if not os.path.exists(target_path):
+                    fileutil.mkdir(target_path)
+                self.mount(device=metric_hierarchy,
+                           mount_point=target_path,
+                           option="-t cgroup -o {0}".format(metric_hierarchy),
+                           chk_err=False)
+
+            for metric_hierarchy in ['cpu', 'cpuacct']:
+                target_path = _cgroup_path(metric_hierarchy)
+                if not os.path.exists(target_path):
+                    os.symlink(_cgroup_path('cpu,cpuacct'), target_path)
+
+        except Exception as e:
+            logger.error("Could not mount cgroups: {0}", ustr(e))
+
 
 class Ubuntu12OSUtil(Ubuntu14OSUtil):
     def __init__(self):
@@ -56,6 +97,8 @@ class Ubuntu12OSUtil(Ubuntu14OSUtil):
         ret = shellutil.run_get_output("pidof dhclient3", chk_err=False)
         return ret[1] if ret[0] == 0 else None
 
+    def mount_cgroups(self):
+        pass
 
 class Ubuntu16OSUtil(Ubuntu14OSUtil):
     """
@@ -70,6 +113,12 @@ class Ubuntu16OSUtil(Ubuntu14OSUtil):
     def unregister_agent_service(self):
         return shellutil.run("systemctl mask walinuxagent", chk_err=False)
 
+    def mount_cgroups(self):
+        """
+        Mounted by default in Ubuntu 16.04
+        """
+        pass
+
 
 class Ubuntu18OSUtil(Ubuntu16OSUtil):
     """
@@ -127,3 +176,9 @@ class UbuntuSnappyOSUtil(Ubuntu14OSUtil):
     def __init__(self):
         super(UbuntuSnappyOSUtil, self).__init__()
         self.conf_file_path = '/apps/walinuxagent/current/waagent.conf'
+
+    def mount_cgroups(self):
+        """
+        Already mounted in Snappy
+        """
+        pass
diff --git a/azurelinuxagent/daemon/main.py b/azurelinuxagent/daemon/main.py
index ce56965d..f09a1bb9 100644
--- a/azurelinuxagent/daemon/main.py
+++ b/azurelinuxagent/daemon/main.py
@@ -26,6 +26,7 @@ import azurelinuxagent.common.conf as conf
 import azurelinuxagent.common.logger as logger
 import azurelinuxagent.common.utils.fileutil as fileutil
 
+from azurelinuxagent.common.cgroups import CGroups
 from azurelinuxagent.common.event import add_event, WALAEventOperation
 from azurelinuxagent.common.future import ustr
 from azurelinuxagent.common.osutil import get_osutil
@@ -65,6 +66,7 @@ class DaemonHandler(object):
                     PY_VERSION_MICRO)
 
         self.check_pid()
+        CGroups.setup()
 
         # If FIPS is enabled, set the OpenSSL environment variable
         # Note:
diff --git a/azurelinuxagent/ga/exthandlers.py b/azurelinuxagent/ga/exthandlers.py
index 1d49187d..986a7ca4 100644
--- a/azurelinuxagent/ga/exthandlers.py
+++ b/azurelinuxagent/ga/exthandlers.py
@@ -22,7 +22,6 @@ import glob
 import json
 import operator
 import os
-import os.path
 import random
 import re
 import shutil
@@ -36,8 +35,9 @@ import azurelinuxagent.common.conf as conf
 import azurelinuxagent.common.logger as logger
 import azurelinuxagent.common.utils.fileutil as fileutil
 import azurelinuxagent.common.version as version
-from azurelinuxagent.common.errorstate import ErrorState, ERROR_STATE_DELTA_DEFAULT, ERROR_STATE_DELTA_INSTALL
 
+from azurelinuxagent.common.cgroups import CGroups, CGroupsTelemetry
+from azurelinuxagent.common.errorstate import ErrorState, ERROR_STATE_DELTA_DEFAULT, ERROR_STATE_DELTA_INSTALL
 from azurelinuxagent.common.event import add_event, WALAEventOperation, elapsed_milliseconds, report_event
 from azurelinuxagent.common.exception import ExtensionError, ProtocolError
 from azurelinuxagent.common.future import ustr
@@ -986,17 +986,29 @@ class ExtHandlerInstance(object):
             # of Python versions we must support we must use .communicate().
             # Some extensions erroneously begin cmd with a slash; don't interpret those
             # as root-relative. (Issue #1170)
-            full_path = os.path.join(base_dir, cmd.lstrip(os.sep))
+            full_path = os.path.join(base_dir, cmd.lstrip(os.path.sep))
+
+            def pre_exec_function():
+                """
+                Change process state before the actual target process is started. Effectively, this runs between
+                the fork() and the exec() of sub-process creation.
+                :return:
+                """
+                os.setsid()
+                CGroups.add_to_extension_cgroup(self.ext_handler.name)
+
             process = subprocess.Popen(full_path,
                                   shell=True,
                                   cwd=base_dir,
                                   stdout=subprocess.PIPE,
                                   stderr=subprocess.PIPE,
                                   env=os.environ,
-                                  preexec_fn=os.setsid)
+                                  preexec_fn=pre_exec_function)
         except OSError as e:
             raise ExtensionError("Failed to launch '{0}': {1}".format(full_path, e.strerror))
 
+        cg = CGroups.for_extension(self.ext_handler.name)
+        CGroupsTelemetry.track_extension(self.ext_handler.name, cg)
         msg = capture_from_process(process, cmd, timeout)
 
         ret = process.poll()
diff --git a/azurelinuxagent/ga/monitor.py b/azurelinuxagent/ga/monitor.py
index f43cb01a..d6089799 100644
--- a/azurelinuxagent/ga/monitor.py
+++ b/azurelinuxagent/ga/monitor.py
@@ -21,14 +21,15 @@ import os
 import platform
 import time
 import threading
+import traceback
 import uuid
 
 import azurelinuxagent.common.conf as conf
-import azurelinuxagent.common.utils.fileutil as fileutil
 import azurelinuxagent.common.logger as logger
 from azurelinuxagent.common.errorstate import ErrorState
 
-from azurelinuxagent.common.event import add_event, WALAEventOperation
+from azurelinuxagent.common.cgroups import CGroups, CGroupsTelemetry
+from azurelinuxagent.common.event import add_event, report_metric, WALAEventOperation
 from azurelinuxagent.common.exception import EventError, ProtocolError, OSUtilError, HttpError
 from azurelinuxagent.common.future import ustr
 from azurelinuxagent.common.osutil import get_osutil
@@ -96,6 +97,7 @@ class MonitorHandler(object):
 
     EVENT_COLLECTION_PERIOD = datetime.timedelta(minutes=1)
     TELEMETRY_HEARTBEAT_PERIOD = datetime.timedelta(minutes=30)
+    CGROUP_TELEMETRY_PERIOD = datetime.timedelta(minutes=5)
     # host plugin
     HOST_PLUGIN_HEARTBEAT_PERIOD = datetime.timedelta(minutes=1)
     HOST_PLUGIN_HEALTH_PERIOD = datetime.timedelta(minutes=5)
@@ -111,6 +113,7 @@ class MonitorHandler(object):
         self.event_thread = None
         self.last_event_collection = None
         self.last_telemetry_heartbeat = None
+        self.last_cgroup_telemetry = None
         self.last_host_plugin_heartbeat = None
         self.last_imds_heartbeat = None
         self.protocol = None
@@ -126,6 +129,7 @@ class MonitorHandler(object):
     def run(self):
         self.init_protocols()
         self.init_sysinfo()
+        self.init_cgroups()
         self.start()
 
     def stop(self):
@@ -247,11 +251,13 @@ class MonitorHandler(object):
 
     def daemon(self):
         min_delta = min(MonitorHandler.TELEMETRY_HEARTBEAT_PERIOD,
+                        MonitorHandler.CGROUP_TELEMETRY_PERIOD,
                         MonitorHandler.EVENT_COLLECTION_PERIOD,
                         MonitorHandler.HOST_PLUGIN_HEARTBEAT_PERIOD,
                         MonitorHandler.IMDS_HEARTBEAT_PERIOD).seconds
         while self.should_run:
             self.send_telemetry_heartbeat()
+            self.send_cgroup_telemetry()
             self.collect_and_send_events()
             self.send_host_plugin_heartbeat()
             self.send_imds_heartbeat()
@@ -379,3 +385,34 @@ class MonitorHandler(object):
                 logger.warn("Failed to send heartbeat: {0}", e)
 
             self.last_telemetry_heartbeat = datetime.datetime.utcnow()
+
+    @staticmethod
+    def init_cgroups():
+        # Track metrics for the roll-up cgroup and for the agent cgroup
+        try:
+            CGroupsTelemetry.track_cgroup(CGroups.for_extension(""))
+            CGroupsTelemetry.track_agent(CGroups.for_extension(AGENT_NAME))
+        except Exception as e:
+            logger.error("monitor: Exception tracking wrapper and agent: {0} [{1}]", e, traceback.format_exc())
+
+    def send_cgroup_telemetry(self):
+        if self.last_cgroup_telemetry is None:
+            self.last_cgroup_telemetry = datetime.datetime.utcnow()
+
+        if datetime.datetime.utcnow() >= (self.last_telemetry_heartbeat + MonitorHandler.CGROUP_TELEMETRY_PERIOD):
+            try:
+                for cgroup_name, metrics in CGroupsTelemetry.collect_all_tracked().items():
+                    for metric_group, metric_name, value in metrics:
+                        if value > 0:
+                            report_metric(metric_group, metric_name, cgroup_name, value)
+            except Exception as e:
+                logger.warn("Failed to collect performance metrics: {0} [{1}]", e, traceback.format_exc())
+
+            # Look for extension cgroups we're not already tracking and track them
+            try:
+                ext_handlers_list, incarnation = self.protocol.get_ext_handlers()
+                CGroupsTelemetry.update_tracked(ext_handlers_list.extHandlers)
+            except Exception as e:
+                logger.warn("Monitor: updating tracked extensions raised {0}: {1}", e, traceback.format_exc())
+
+            self.last_cgroup_telemetry = datetime.datetime.utcnow()


Test patch:
diff --git a/tests/common/test_cgroups.py b/tests/common/test_cgroups.py
new file mode 100644
index 00000000..d73c8d09
--- /dev/null
+++ b/tests/common/test_cgroups.py
@@ -0,0 +1,177 @@
+# 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.4+ and Openssl 1.0+
+#
+
+from __future__ import print_function
+
+from azurelinuxagent.common.cgroups import CGroupsTelemetry, CGroups, CGroupsException, BASE_CGROUPS, Cpu
+from tests.tools import *
+
+import os
+import random
+import time
+
+
+def consume_cpu_time():
+    waste = 0
+    for x in range(1, 200000):
+        waste += random.random()
+    return waste
+
+
+def make_self_cgroups():
+    """
+    Build a CGroups object for the cgroup to which this process already belongs
+
+    :return: CGroups containing this process
+    :rtype: CGroups
+    """
+    def path_maker(hierarchy, __):
+        suffix = CGroups.get_my_cgroup_path(CGroups.get_hierarchy_id('cpu'))
+        return os.path.join(BASE_CGROUPS, hierarchy, suffix)
+
+    return CGroups("inplace", path_maker)
+
+
+def make_root_cgroups():
+    """
+    Build a CGroups object for the topmost cgroup
+
+    :return: CGroups for most-encompassing cgroup
+    :rtype: CGroups
+    """
+    def path_maker(hierarchy, _):
+        return os.path.join(BASE_CGROUPS, hierarchy)
+
+    return CGroups("root", path_maker)
+
+
+@skip_if_predicate_false(CGroups.enabled, "CGroups not supported in this environment")
+class TestCGroups(AgentTestCase):
+    @classmethod
+    def setUpClass(cls):
+        CGroups.setup(True)
+        super(AgentTestCase, cls).setUpClass()
+
+    def test_cgroup_utilities(self):
+        """
+        Test utilities for querying cgroup metadata
+        """
+        cpu_id = CGroups.get_hierarchy_id('cpu')
+        self.assertGreater(int(cpu_id), 0)
+        memory_id = CGroups.get_hierarchy_id('memory')
+        self.assertGreater(int(memory_id), 0)
+        self.assertNotEqual(cpu_id, memory_id)
+
+    def test_telemetry_inplace(self):
+        """
+        Test raw measures and basic statistics for the cgroup in which this process is currently running.
+        """
+        cg = make_self_cgroups()
+        self.assertIn('cpu', cg.cgroups)
+        self.assertIn('memory', cg.cgroups)
+        ct = CGroupsTelemetry("test", cg)
+        cpu = Cpu(ct)
+        self.assertGreater(cpu.current_system_cpu, 0)
+
+        consume_cpu_time()  # Eat some CPU
+        cpu.update()
+
+        self.assertGreater(cpu.current_cpu_total, cpu.previous_cpu_total)
+        self.assertGreater(cpu.current_system_cpu, cpu.previous_system_cpu)
+
+        percent_used = cpu.get_cpu_percent()
+        self.assertGreater(percent_used, 0)
+
+    def test_telemetry_in_place_non_root(self):
+        """
+        Ensure this (non-root) cgroup has distinct metrics from the root cgroup.
+        """
+        # Does nothing on systems where the default cgroup for a randomly-created process (like this test invocation)
+        # is the root cgroup.
+        cg = make_self_cgroups()
+        root = make_root_cgroups()
+        if cg.cgroups['cpu'] != root.cgroups['cpu']:
+            ct = CGroupsTelemetry("test", cg)
+            cpu = Cpu(ct)
+            self.assertLess(cpu.current_cpu_total, cpu.current_system_cpu)
+
+            consume_cpu_time()  # Eat some CPU
+            time.sleep(1)       # Generate some idle time
+            cpu.update()
+            self.assertLess(cpu.current_cpu_total, cpu.current_system_cpu)
+
+    def test_telemetry_instantiation(self):
+        """
+        Tracking a cgroup for an extension; collect all metrics.
+        """
+        # Record initial state
+        initial_cgroup = make_self_cgroups()
+
+        # Put the process into a different cgroup, consume some resources, ensure we see them end-to-end
+        test_cgroup = CGroups.for_extension("agent_unittest")
+        self.assertIn('cpu', test_cgroup.cgroups)
+        self.assertIn('memory', test_cgroup.cgroups)
+        test_cgroup.add(os.getpid())
+        self.assertNotEqual(initial_cgroup.cgroups['cpu'], test_cgroup.cgroups['cpu'])
+        CGroupsTelemetry.track_cgroup(test_cgroup)
+        self.assertTrue(CGroupsTelemetry.is_tracked("agent_unittest"))
+        consume_cpu_time()
+        time.sleep(1)
+        metrics = CGroupsTelemetry.collect_all_tracked()
+        my_metrics = metrics["agent_unittest"]
+        self.assertEqual(len(my_metrics), 1)
+        metric_family, metric_name, metric_value = my_metrics[0]
+        self.assertEqual(metric_family, "Process")
+        self.assertEqual(metric_name, "% Processor Time")
+        self.assertGreater(metric_value, 0.0)
+
+        # Restore initial state
+        CGroupsTelemetry.stop_tracking("agent_unittest")
+        initial_cgroup.add(os.getpid())
+
+    def test_cpu_telemetry(self):
+        """
+        Test Cpu telemetry class
+        """
+        cg = make_self_cgroups()
+        self.assertIn('cpu', cg.cgroups)
+        self.assertIn('memory', cg.cgroups)
+        ct = CGroupsTelemetry('test', cg)
+        self.assertIs(cg, ct.cgroup)
+        cpu = Cpu(ct)
+        self.assertIs(cg, cpu.cgt.cgroup)
+        ticks_before = cpu.current_cpu_total
+        consume_cpu_time()
+        time.sleep(1)
+        cpu.update()
+        ticks_after = cpu.current_cpu_total
+        self.assertGreater(ticks_after, ticks_before)
+        p2 = cpu.get_cpu_percent()
+        self.assertGreater(p2, 0)
+        self.assertLess(p2, 100)
+
+    def test_format_memory_value(self):
+        """
+        Test formatting of memory amounts into human-readable units
+        """
+        self.assertEqual(-1, CGroups._format_memory_value('bytes', None))
+        self.assertEqual(2048, CGroups._format_memory_value('kilobytes', 2))
+        self.assertEqual(0, CGroups._format_memory_value('kilobytes', 0))
+        self.assertEqual(2048000, CGroups._format_memory_value('kilobytes', 2000))
+        self.assertEqual(2048*1024, CGroups._format_memory_value('megabytes', 2))
+        self.assertEqual((1024 + 512) * 1024 * 1024, CGroups._format_memory_value('gigabytes', 1.5))
+        self.assertRaises(CGroupsException, CGroups._format_memory_value, 'KiloBytes', 1)
diff --git a/tests/common/test_event.py b/tests/common/test_event.py
index 1acd4bcc..201e1b6e 100644
--- a/tests/common/test_event.py
+++ b/tests/common/test_event.py
@@ -219,3 +219,20 @@ class TestEvent(AgentTestCase):
     def test_elapsed_milliseconds(self):
         utc_start = datetime.utcnow() + timedelta(days=1)
         self.assertEqual(0, elapsed_milliseconds(utc_start))
+
+    @patch('azurelinuxagent.common.event.EventLogger.save_event')
+    def test_report_metric(self, mock_event):
+        event.report_metric("cpu", "%idle", "_total", 10.0)
+        self.assertEqual(1, mock_event.call_count)
+        event_json = mock_event.call_args[0][0]
+        self.assertIn("69B669B9-4AF8-4C50-BDC4-6006FA76E975", event_json)
+        self.assertIn("%idle", event_json)
+        import json
+        event_dictionary = json.loads(event_json)
+        self.assertEqual(event_dictionary['providerId'], "69B669B9-4AF8-4C50-BDC4-6006FA76E975")
+        for parameter in event_dictionary["parameters"]:
+            if parameter['name'] == 'Counter':
+                self.assertEqual(parameter['value'], '%idle')
+                break
+        else:
+            self.fail("Counter '%idle' not found in event parameters: {0}".format(repr(event_dictionary)))
diff --git a/tests/ga/test_monitor.py b/tests/ga/test_monitor.py
index 995365ef..5d53b1e7 100644
--- a/tests/ga/test_monitor.py
+++ b/tests/ga/test_monitor.py
@@ -31,9 +31,9 @@ class TestMonitor(AgentTestCase):
     def test_parse_xml_event(self, *args):
         data_str = load_data('ext/event.xml')
         event = parse_xml_event(data_str)
-        self.assertNotEquals(None, event)
-        self.assertNotEquals(0, event.parameters)
-        self.assertNotEquals(None, event.parameters[0])
+        self.assertNotEqual(None, event)
+        self.assertNotEqual(0, event.parameters)
+        self.assertNotEqual(None, event.parameters[0])
 
     def test_add_sysinfo(self, *args):
         data_str = load_data('ext/event.xml')
@@ -60,35 +60,37 @@ class TestMonitor(AgentTestCase):
         monitor_handler.sysinfo = sysinfo
         monitor_handler.add_sysinfo(event)
 
-        self.assertNotEquals(None, event)
-        self.assertNotEquals(0, event.parameters)
-        self.assertNotEquals(None, event.parameters[0])
+        self.assertNotEqual(None, event)
+        self.assertNotEqual(0, event.parameters)
+        self.assertNotEqual(None, event.parameters[0])
         counter = 0
         for p in event.parameters:
             if p.name == vm_name_param:
-                self.assertEquals(vm_name, p.value)
+                self.assertEqual(vm_name, p.value)
                 counter += 1
             elif p.name == tenant_name_param:
-                self.assertEquals(tenant_name, p.value)
+                self.assertEqual(tenant_name, p.value)
                 counter += 1
             elif p.name == role_name_param:
-                self.assertEquals(role_name, p.value)
+                self.assertEqual(role_name, p.value)
                 counter += 1
             elif p.name == role_instance_name_param:
-                self.assertEquals(role_instance_name, p.value)
+                self.assertEqual(role_instance_name, p.value)
                 counter += 1
             elif p.name == container_id_param:
-                self.assertEquals(container_id, p.value)
+                self.assertEqual(container_id, p.value)
                 counter += 1
 
-        self.assertEquals(5, counter)
+        self.assertEqual(5, counter)
 
     @patch("azurelinuxagent.ga.monitor.MonitorHandler.send_telemetry_heartbeat")
     @patch("azurelinuxagent.ga.monitor.MonitorHandler.collect_and_send_events")
     @patch("azurelinuxagent.ga.monitor.MonitorHandler.send_host_plugin_heartbeat")
+    @patch("azurelinuxagent.ga.monitor.MonitorHandler.send_cgroup_telemetry")
     @patch("azurelinuxagent.ga.monitor.MonitorHandler.send_imds_heartbeat")
     def test_heartbeats(self,
                         patch_imds_heartbeat,
+                        patch_cgroup_telemetry,
                         patch_hostplugin_heartbeat,
                         patch_send_events,
                         patch_telemetry_heartbeat,
@@ -104,6 +106,7 @@ class TestMonitor(AgentTestCase):
         self.assertEqual(0, patch_send_events.call_count)
         self.assertEqual(0, patch_telemetry_heartbeat.call_count)
         self.assertEqual(0, patch_imds_heartbeat.call_count)
+        self.assertEqual(0, patch_cgroup_telemetry.call_count)
 
         monitor_handler.start()
         time.sleep(1)
@@ -113,9 +116,11 @@ class TestMonitor(AgentTestCase):
         self.assertNotEqual(0, patch_send_events.call_count)
         self.assertNotEqual(0, patch_telemetry_heartbeat.call_count)
         self.assertNotEqual(0, patch_imds_heartbeat.call_count)
+        self.assertNotEqual(0, patch_cgroup_telemetry.call_count)
 
         monitor_handler.stop()
 
+    @patch("azurelinuxagent.ga.monitor.MonitorHandler.send_cgroup_telemetry")
     def test_heartbeat_timings_updates_after_window(self, *args):
         monitor_handler = get_monitor_handler()
 
@@ -152,6 +157,7 @@ class TestMonitor(AgentTestCase):
 
         monitor_handler.stop()
 
+    @patch("azurelinuxagent.ga.monitor.MonitorHandler.send_cgroup_telemetry")
     def test_heartbeat_timings_no_updates_within_window(self, *args):
         monitor_handler = get_monitor_handler()
 
diff --git a/tests/tools.py b/tests/tools.py
index 4226c1f9..e36b707a 100644
--- a/tests/tools.py
+++ b/tests/tools.py
@@ -28,7 +28,7 @@ from functools import wraps
 
 import time
 
-import azurelinuxagent.common.event
+import azurelinuxagent.common.event as event
 import azurelinuxagent.common.conf as conf
 import azurelinuxagent.common.logger as logger
 from azurelinuxagent.common.utils import fileutil
@@ -58,13 +58,47 @@ def _do_nothing():
     pass
 
 
+_MAX_LENGTH = 120
+
+
+def skip_if_predicate_false(predicate, message):
+    if not predicate():
+        if hasattr(unittest, "skip"):
+            return unittest.skip(message)
+        return lambda func: None
+    return lambda func: func
+
+
+def _safe_repr(obj, short=False):
+    """
+    Copied from Python 3.x
+    """
+    try:
+        result = repr(obj)
+    except Exception:
+        result = object.__repr__(obj)
+    if not short or len(result) < _MAX_LENGTH:
+        return result
+    return result[:_MAX_LENGTH] + ' [truncated]...'
+
+
 class AgentTestCase(unittest.TestCase):
     @classmethod
     def setUpClass(cls):
+        # Setup newer unittest assertions missing in prior versions of Python
+
         if not hasattr(cls, "assertRegex"):
             cls.assertRegex = cls.assertRegexpMatches if hasattr(cls, "assertRegexpMatches") else _do_nothing
         if not hasattr(cls, "assertNotRegex"):
             cls.assertNotRegex = cls.assertNotRegexpMatches if hasattr(cls, "assertNotRegexpMatches") else _do_nothing
+        if not hasattr(cls, "assertIn"):
+            cls.assertIn = cls.emulate_assertIn
+        if not hasattr(cls, "assertNotIn"):
+            cls.assertNotIn = cls.emulate_assertNotIn
+        if not hasattr(cls, "assertGreater"):
+            cls.assertGreater = cls.emulate_assertGreater
+        if not hasattr(cls, "assertLess"):
+            cls.assertLess = cls.emulate_assertLess
 
     def setUp(self):
         prefix = "{0}_".format(self.__class__.__name__)
@@ -80,13 +114,33 @@ class AgentTestCase(unittest.TestCase):
 
         conf.get_agent_pid_file_path = Mock(return_value=os.path.join(self.tmp_dir, "waagent.pid"))
 
-        azurelinuxagent.common.event.init_event_status(self.tmp_dir)
-        azurelinuxagent.common.event.init_event_logger(self.tmp_dir)
+        event.init_event_status(self.tmp_dir)
+        event.init_event_logger(self.tmp_dir)
 
     def tearDown(self):
         if not debug and self.tmp_dir is not None:
             shutil.rmtree(self.tmp_dir)
 
+    def emulate_assertIn(self, a, b, msg=None):
+        if a not in b:
+            msg = msg if msg is not None else "{0} not found in {1}".format(_safe_repr(a), _safe_repr(b))
+            self.fail(msg)
+
+    def emulate_assertNotIn(self, a, b, msg=None):
+        if a in b:
+            msg = msg if msg is not None else "{0} unexpectedly found in {1}".format(_safe_repr(a), _safe_repr(b))
+            self.fail(msg)
+
+    def emulate_assertGreater(self, a, b, msg=None):
+        if not a > b:
+            msg = msg if msg is not None else '{0} not greater than {1}'.format(_safe_repr(a), _safe_repr(b))
+            self.fail(msg)
+
+    def emulate_assertLess(self, a, b, msg=None):
+        if not a < b:
+            msg = msg if msg is not None else '{0} not less than {1}'.format(_safe_repr(a), _safe_repr(b))
+            self.fail(msg)
+
     @staticmethod
     def _create_files(tmp_dir, prefix, suffix, count, with_sleep=0):
         for i in range(count):