Repository: Azure/WALinuxAgent
Pull request #2448 (2021-12-20 18:56:01)

Merge release 2.5.0.0 into master


Patch:
diff --git a/.github/workflows/ci_pr.yml b/.github/workflows/ci_pr.yml
index a6c395a4..a4565198 100644
--- a/.github/workflows/ci_pr.yml
+++ b/.github/workflows/ci_pr.yml
@@ -11,7 +11,7 @@ jobs:
   test-legacy-python-versions:
     
     name: "Python 2.6 Unit Tests"
-    runs-on: ubuntu-16.04
+    runs-on: ubuntu-18.04
 
     strategy:
       fail-fast: false
diff --git a/azurelinuxagent/agent.py b/azurelinuxagent/agent.py
index ad789334..b70429a1 100644
--- a/azurelinuxagent/agent.py
+++ b/azurelinuxagent/agent.py
@@ -28,7 +28,8 @@ import re
 import subprocess
 import sys
 import threading
-import traceback
+from azurelinuxagent.common import cgroupconfigurator, logcollector
+from azurelinuxagent.common.cgroupapi import SystemdCgroupsApi
 
 import azurelinuxagent.common.conf as conf
 import azurelinuxagent.common.event as event
@@ -36,7 +37,7 @@ import azurelinuxagent.common.logger as logger
 from azurelinuxagent.common.future import ustr
 from azurelinuxagent.common.logcollector import LogCollector, OUTPUT_RESULTS_FILE_PATH
 from azurelinuxagent.common.osutil import get_osutil
-from azurelinuxagent.common.utils import fileutil
+from azurelinuxagent.common.utils import fileutil, textutil
 from azurelinuxagent.common.utils.flexible_version import FlexibleVersion
 from azurelinuxagent.common.utils.networkutil import AddFirewallRules
 from azurelinuxagent.common.version import AGENT_NAME, AGENT_LONG_VERSION, AGENT_VERSION, \
@@ -44,6 +45,7 @@ from azurelinuxagent.common.version import AGENT_NAME, AGENT_LONG_VERSION, AGENT
     PY_VERSION_MAJOR, PY_VERSION_MINOR, \
     PY_VERSION_MICRO, GOAL_STATE_AGENT_VERSION, \
     get_daemon_version, set_daemon_version
+from azurelinuxagent.ga.collect_logs import CollectLogsHandler
 from azurelinuxagent.pa.provision.default import ProvisionHandler
 
 
@@ -199,6 +201,22 @@ class Agent(object):
         else:
             print("Running log collector mode normal")
 
+        # Check the cgroups unit
+        if CollectLogsHandler.should_validate_cgroups():
+            cpu_cgroup_path, memory_cgroup_path = SystemdCgroupsApi.get_process_cgroup_relative_paths("self")
+
+            cpu_slice_matches = (cgroupconfigurator.LOGCOLLECTOR_SLICE in cpu_cgroup_path)
+            memory_slice_matches = (cgroupconfigurator.LOGCOLLECTOR_SLICE in memory_cgroup_path)
+
+            if not cpu_slice_matches or not memory_slice_matches:
+                print("The Log Collector process is not in the proper cgroups:")
+                if not cpu_slice_matches:
+                    print("\tunexpected cpu slice")
+                if not memory_slice_matches:
+                    print("\tunexpected memory slice")
+
+                sys.exit(logcollector.INVALID_CGROUPS_ERRCODE)
+
         try:
             log_collector = LogCollector(is_full_mode)
             archive = log_collector.collect_logs_and_get_archive()
@@ -259,10 +277,10 @@ def main(args=None):
                 agent.collect_logs(log_collector_full_mode)
             elif command == AgentCommands.SetupFirewall:
                 agent.setup_firewall(firewall_metadata)
-        except Exception:
+        except Exception as e:
             logger.error(u"Failed to run '{0}': {1}",
                          command,
-                         traceback.format_exc())
+                         textutil.format_exception(e))
 
 
 def parse_args(sys_args):
diff --git a/azurelinuxagent/common/cgroupapi.py b/azurelinuxagent/common/cgroupapi.py
index e4951d45..2bb1d5fd 100644
--- a/azurelinuxagent/common/cgroupapi.py
+++ b/azurelinuxagent/common/cgroupapi.py
@@ -36,7 +36,7 @@ from azurelinuxagent.common.version import get_distro
 
 CGROUPS_FILE_SYSTEM_ROOT = '/sys/fs/cgroup'
 CGROUP_CONTROLLERS = ["cpu", "memory"]
-
+EXTENSION_SLICE_PREFIX = "azure-vmextensions"
 
 class SystemdRunError(CGroupsException):
     """
@@ -66,11 +66,6 @@ class CGroupsApi(object):
             logger.warn("Cannot add cgroup '{0}' to tracking list; resource usage will not be tracked. "
                         "Error: {1}".format(cgroup.path, ustr(exception)))
 
-    @staticmethod
-    def _get_extension_cgroup_name(extension_name):
-        # Since '-' is used as a separator in systemd unit names, we replace it with '_' to prevent side-effects.
-        return extension_name.replace('-', '_')
-
     @staticmethod
     def get_processes_in_cgroup(cgroup_path):
         with open(os.path.join(cgroup_path, "cgroup.procs"), "r") as cgroup_procs:
@@ -234,18 +229,25 @@ class SystemdCgroupsApi(CGroupsApi):
         unit_not_found = "Unit {0} not found.".format(scope_name)
         return unit_not_found in stderr or scope_name not in stderr
 
-    def start_extension_command(self, extension_name, command, timeout, shell, cwd, env, stdout, stderr, error_code=ExtensionErrorCodes.PluginUnknownFailure): 
-        scope = "{0}_{1}".format(self._get_extension_cgroup_name(extension_name), uuid.uuid4())
+    @staticmethod
+    def get_extension_cgroup_name(extension_name):
+        # Since '-' is used as a separator in systemd unit names, we replace it with '_' to prevent side-effects.
+        return EXTENSION_SLICE_PREFIX + "-" + extension_name.replace('-', '_')
 
+    def start_extension_command(self, extension_name, command, cmd_name, timeout, shell, cwd, env, stdout, stderr, error_code=ExtensionErrorCodes.PluginUnknownFailure):
+        scope = "{0}_{1}".format(cmd_name, uuid.uuid4())
+        extension_slice_name = self.get_extension_cgroup_name(extension_name)
         with self._systemd_run_commands_lock:
             process = subprocess.Popen(  # pylint: disable=W1509
-                "systemd-run --unit={0} --scope --slice=azure-vmextensions.slice {1}".format(scope, command),
+                "systemd-run --unit={0} --scope --slice={1}.slice {2}".format(scope, extension_slice_name, command),
                 shell=shell,
                 cwd=cwd,
                 stdout=stdout,
                 stderr=stderr,
                 env=env,
                 preexec_fn=os.setsid)
+
+            # We start systemd-run with shell == True so process.pid is the shell's pid, not the pid for systemd-run
             self._systemd_run_commands.append(process.pid)
 
         scope_name = scope + '.scope'
@@ -253,8 +255,7 @@ class SystemdCgroupsApi(CGroupsApi):
         logger.info("Started extension in unit '{0}'", scope_name)
 
         try:
-            # systemd-run creates the scope under the system slice by default
-            cgroup_relative_path = os.path.join('azure.slice/azure-vmextensions.slice', scope_name)
+            cgroup_relative_path = os.path.join('azure.slice/azure-vmextensions.slice', extension_slice_name + ".slice")
 
             cpu_cgroup_mountpoint, _ = self.get_cgroup_mount_points()
 
diff --git a/azurelinuxagent/common/cgroupconfigurator.py b/azurelinuxagent/common/cgroupconfigurator.py
index b10c9426..cd12449b 100644
--- a/azurelinuxagent/common/cgroupconfigurator.py
+++ b/azurelinuxagent/common/cgroupconfigurator.py
@@ -22,7 +22,7 @@ import subprocess
 from azurelinuxagent.common import conf
 from azurelinuxagent.common import logger
 from azurelinuxagent.common.cgroup import CpuCgroup, AGENT_NAME_TELEMETRY, MetricsCounter
-from azurelinuxagent.common.cgroupapi import CGroupsApi, SystemdCgroupsApi, SystemdRunError
+from azurelinuxagent.common.cgroupapi import CGroupsApi, SystemdCgroupsApi, SystemdRunError, EXTENSION_SLICE_PREFIX
 from azurelinuxagent.common.cgroupstelemetry import CGroupsTelemetry
 from azurelinuxagent.common.exception import ExtensionErrorCodes, CGroupsException
 from azurelinuxagent.common.future import ustr
@@ -32,14 +32,14 @@ from azurelinuxagent.common.utils import shellutil, fileutil
 from azurelinuxagent.common.utils.extensionprocessutil import handle_process_completion
 from azurelinuxagent.common.event import add_event, WALAEventOperation
 
-_AZURE_SLICE = "azure.slice"
+AZURE_SLICE = "azure.slice"
 _AZURE_SLICE_CONTENTS = """
 [Unit]
 Description=Slice for Azure VM Agent and Extensions
 DefaultDependencies=no
 Before=slices.target
 """
-_VMEXTENSIONS_SLICE = "azure-vmextensions.slice"
+_VMEXTENSIONS_SLICE = EXTENSION_SLICE_PREFIX + ".slice"
 _VMEXTENSIONS_SLICE_CONTENTS = """
 [Unit]
 Description=Slice for Azure VM Extensions
@@ -48,6 +48,31 @@ Before=slices.target
 [Slice]
 CPUAccounting=yes
 """
+_EXTENSION_SLICE_CONTENTS = """
+[Unit]
+Description=Slice for Azure VM extension {extension_name}
+DefaultDependencies=no
+Before=slices.target
+[Slice]
+CPUAccounting=yes
+"""
+LOGCOLLECTOR_SLICE = "azure-walinuxagent-logcollector.slice"
+# More info on resource limits properties in systemd here:
+# https://access.redhat.com/documentation/en-us/red_hat_enterprise_linux/7/html/resource_management_guide/sec-modifying_control_groups
+_LOGCOLLECTOR_SLICE_CONTENTS_FMT = """
+[Unit]
+Description=Slice for Azure VM Agent Periodic Log Collector
+DefaultDependencies=no
+Before=slices.target
+[Slice]
+CPUAccounting=yes
+CPUQuota={cpu_quota}
+MemoryAccounting=yes
+MemoryLimit={memory_limit}
+"""
+_LOGCOLLECTOR_CPU_QUOTA = "5%"
+_LOGCOLLECTOR_MEMORY_LIMIT = "30M"  # K for kb, M for mb
+
 _AGENT_DROP_IN_FILE_SLICE = "10-Slice.conf"
 _AGENT_DROP_IN_FILE_SLICE_CONTENTS = """
 # This drop-in unit file was created by the Azure VM Agent.
@@ -129,7 +154,7 @@ class CGroupConfigurator(object):
 
                 agent_unit_name = systemd.get_agent_unit_name()
                 agent_slice = systemd.get_unit_property(agent_unit_name, "Slice")
-                if agent_slice not in (_AZURE_SLICE, "system.slice"):
+                if agent_slice not in (AZURE_SLICE, "system.slice"):
                     _log_cgroup_warning("The agent is within an unexpected slice: {0}", agent_slice)
                     return
 
@@ -275,8 +300,9 @@ class CGroupConfigurator(object):
             CGroupConfigurator._Impl.__cleanup_unit_file("/etc/systemd/system/system-walinuxagent.extensions.slice")
 
             unit_file_install_path = systemd.get_unit_file_install_path()
-            azure_slice = os.path.join(unit_file_install_path, _AZURE_SLICE)
+            azure_slice = os.path.join(unit_file_install_path, AZURE_SLICE)
             vmextensions_slice = os.path.join(unit_file_install_path, _VMEXTENSIONS_SLICE)
+            logcollector_slice = os.path.join(unit_file_install_path, LOGCOLLECTOR_SLICE)
             agent_unit_file = systemd.get_agent_unit_file()
             agent_drop_in_path = systemd.get_agent_drop_in_path()
             agent_drop_in_file_slice = os.path.join(agent_drop_in_path, _AGENT_DROP_IN_FILE_SLICE)
@@ -290,6 +316,12 @@ class CGroupConfigurator(object):
             if not os.path.exists(vmextensions_slice):
                 files_to_create.append((vmextensions_slice, _VMEXTENSIONS_SLICE_CONTENTS))
 
+            if not os.path.exists(logcollector_slice):
+                slice_contents = _LOGCOLLECTOR_SLICE_CONTENTS_FMT.format(cpu_quota=_LOGCOLLECTOR_CPU_QUOTA,
+                    memory_limit=_LOGCOLLECTOR_MEMORY_LIMIT)
+
+                files_to_create.append((logcollector_slice, slice_contents))
+
             if fileutil.findre_in_file(agent_unit_file, r"Slice=") is not None:
                 CGroupConfigurator._Impl.__cleanup_unit_file(agent_drop_in_file_slice)
             else:
@@ -500,39 +532,50 @@ class CGroupConfigurator(object):
                 systemd_run_commands.update(self._cgroups_api.get_systemd_run_commands())
 
                 for process in agent_cgroup:
-                    # Note that the agent uses systemd-run to start extensions; systemd-run belongs to the agent cgroup, though the extensions don't
+                    # Note that the agent uses systemd-run to start extensions; systemd-run belongs to the agent cgroup, though the extensions don't.
                     if process in (daemon, extension_handler) or process in systemd_run_commands:
                         continue
+                    # systemd_run_commands contains the shell that started systemd-run, so we also need to check for the parent
+                    if self._get_parent(process) in systemd_run_commands and self._get_command(process) == 'systemd-run':
+                        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)
+                        unexpected.append(self.__format_process(process))
                         if len(unexpected) >= 5:  # collect just a small sample
                             break
             except Exception as exception:
                 _log_cgroup_warning("Error checking the processes in the agent's cgroup: {0}".format(ustr(exception)))
 
             if len(unexpected) > 0:
-                raise CGroupsException("The agent's cgroup includes unexpected processes: {0}".format(self.__format_processes(unexpected)))
+                raise CGroupsException("The agent's cgroup includes unexpected processes: {0}".format(unexpected))
+
+        @staticmethod
+        def _get_command(pid):
+            try:
+                with open('/proc/{0}/comm'.format(pid), "r") as file_:
+                    comm = file_.read()
+                    if comm and comm[-1] == '\x00':  # if null-terminated, remove the null
+                        comm = comm[:-1]
+                    return comm.rstrip()
+            except Exception:
+                return "UNKNOWN"
 
         @staticmethod
-        def __format_processes(pid_list):
+        def __format_process(pid):
             """
-            Formats the given PIDs as a sequence of strings containing the PIDs and their corresponding command line (truncated to 40 chars)
+            Formats the given PID as a string containing the PID and the corresponding command line truncated to 64 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:64.64}".format(pid, cmdline_file.read())
-                except Exception:
-                    pass
-                return "[PID: {0}] UNKNOWN".format(pid)
-
-            return [get_command_line(pid) for pid in pid_list]
+            try:
+                cmdline = '/proc/{0}/cmdline'.format(pid)
+                if os.path.exists(cmdline):
+                    with open(cmdline, "r") as cmdline_file:
+                        return "[PID: {0}] {1:64.64}".format(pid, cmdline_file.read())
+            except Exception:
+                pass
+            return "[PID: {0}] UNKNOWN".format(pid)
 
         @staticmethod
         def _check_agent_throttled_time(cgroup_metrics):
@@ -555,11 +598,12 @@ class CGroupConfigurator(object):
                 pass
             return 0
 
-        def start_extension_command(self, extension_name, command, timeout, shell, cwd, env, stdout, stderr, error_code=ExtensionErrorCodes.PluginUnknownFailure):
+        def start_extension_command(self, extension_name, command, cmd_name, timeout, shell, cwd, env, stdout, stderr, error_code=ExtensionErrorCodes.PluginUnknownFailure):
             """
             Starts a command (install/enable/etc) for an extension and adds the command's PID to the extension's cgroup
             :param extension_name: The extension executing the command
             :param command: The command to invoke
+            :param cmd_name: The type of the command(enable, install, etc.)
             :param timeout: Number of seconds to wait for command completion
             :param cwd: The working directory for the command
             :param env:  The environment to pass to the command's process
@@ -570,7 +614,7 @@ class CGroupConfigurator(object):
             """
             if self.enabled():
                 try:
-                    return self._cgroups_api.start_extension_command(extension_name, command, timeout, shell=shell, cwd=cwd, env=env, stdout=stdout, stderr=stderr, error_code=error_code)
+                    return self._cgroups_api.start_extension_command(extension_name, command, cmd_name, timeout, shell=shell, cwd=cwd, env=env, stdout=stdout, stderr=stderr, error_code=error_code)
                 except SystemdRunError as exception:
                     reason = 'Failed to start {0} using systemd-run, will try invoking the extension directly. Error: {1}'.format(extension_name, ustr(exception))
                     self.disable(reason)
@@ -580,6 +624,45 @@ class CGroupConfigurator(object):
             process = subprocess.Popen(command, shell=shell, cwd=cwd, env=env, stdout=stdout, stderr=stderr, preexec_fn=os.setsid)  # pylint: disable=W1509
             return handle_process_completion(process=process, command=command, timeout=timeout, stdout=stdout, stderr=stderr, error_code=error_code)
 
+        def setup_extension_slice(self, extension_name):
+            """
+            Each extension runs under its own slice (Ex "Microsoft.CPlat.Extension.slice"). All the slices for
+            extensions are grouped under "azure-vmextensions.slice.
+
+            This method ensures that the extension slice is created. Setup should create
+            under /lib/systemd/system if it is not exist.
+            """
+            if self.enabled():
+                unit_file_install_path = systemd.get_unit_file_install_path()
+                extension_slice_path = os.path.join(unit_file_install_path,
+                                                     SystemdCgroupsApi.get_extension_cgroup_name(extension_name) + ".slice")
+                if not os.path.exists(extension_slice_path):
+                    try:
+                        slice_contents = _EXTENSION_SLICE_CONTENTS.format(extension_name = extension_name)
+                        CGroupConfigurator._Impl.__create_unit_file(extension_slice_path, slice_contents)
+                    except Exception as exception:
+                        _log_cgroup_warning("Failed to create unit files for the extension slice: {0}", ustr(exception))
+                        CGroupConfigurator._Impl.__cleanup_unit_file(extension_slice_path)
+
+        def remove_extension_slice(self, extension_name):
+            """
+            This method ensures that the extension slice gets removed from /lib/systemd/system if it exist
+            Lastly stop the unit. This would ensure the cleanup the /sys/fs/cgroup controller paths
+            """
+            if self.enabled():
+                unit_file_install_path = systemd.get_unit_file_install_path()
+                extension_slice_name = SystemdCgroupsApi.get_extension_cgroup_name(extension_name) + ".slice"
+                extension_slice_path = os.path.join(unit_file_install_path, extension_slice_name)
+                if os.path.exists(extension_slice_path):
+                    CGroupConfigurator._Impl.__cleanup_unit_file(extension_slice_path)
+                # stop the unit gracefully; the extensions slices will be removed from /sys/fs/cgroup path
+                try:
+                    logger.info("Executing systemctl stop {0}".format(extension_slice_name))
+                    shellutil.run_command(["systemctl", "stop", extension_slice_name])
+                except Exception as exception:
+                    _log_cgroup_warning("systemctl stop failed (remove slice): {0}", ustr(exception))
+
+
     # unique instance for the singleton
     _instance = None
 
diff --git a/azurelinuxagent/common/conf.py b/azurelinuxagent/common/conf.py
index 67798a47..b3f4c20d 100644
--- a/azurelinuxagent/common/conf.py
+++ b/azurelinuxagent/common/conf.py
@@ -49,25 +49,43 @@ class ConfigurationProvider(object):
                 value = parts[1].split('#')[0].strip("\" ").strip()
                 self.values[key] = value if value != "None" else None
 
-    def get(self, key, default_val):
+    @staticmethod
+    def _get_default(default):
+        if hasattr(default, '__call__'):
+            return default()
+        return default
+
+    def get(self, key, default_value):
+        """
+        Retrieves a string parameter by key and returns its value. If not found returns the default value,
+        or if the default value is a callable returns the result of invoking the callable.
+        """
         val = self.values.get(key)
-        return val if val is not None else default_val
+        return val if val is not None else self._get_default(default_value)
 
-    def get_switch(self, key, default_val):
+    def get_switch(self, key, default_value):
+        """
+        Retrieves a switch parameter by key and returns its value as a boolean. If not found returns the default value,
+        or if the default value is a callable returns the result of invoking the callable.
+        """
         val = self.values.get(key)
         if val is not None and val.lower() == 'y':
             return True
         elif val is not None and val.lower() == 'n':
             return False
-        return default_val
+        return self._get_default(default_value)
 
-    def get_int(self, key, default_val):
+    def get_int(self, key, default_value):
+        """
+        Retrieves an int parameter by key and returns its value. If not found returns the default value,
+        or if the default value is a callable returns the result of invoking the callable.
+        """
         try:
             return int(self.values.get(key))
         except TypeError:
-            return default_val
+            return self._get_default(default_value)
         except ValueError:
-            return default_val
+            return self._get_default(default_value)
 
 
 __conf__ = ConfigurationProvider()
@@ -118,6 +136,7 @@ __SWITCH_OPTIONS__ = {
     "Debug.CgroupLogMetrics": False,
     "Debug.CgroupDisableOnProcessCheckFailure": True,
     "Debug.CgroupDisableOnQuotaCheckFailure": True,
+    "Debug.EnableFastTrack": False,
 }
 
 
@@ -145,6 +164,7 @@ __STRING_OPTIONS__ = {
 
 __INTEGER_OPTIONS__ = {
     "Extensions.GoalStatePeriod": 6,
+    "Extensions.InitialGoalStatePeriod": 6,
     "Extensions.GoalStateHistoryCleanupPeriod": 1800,
     "OS.EnableFirewallPeriod": 30,
     "OS.RemovePersistentNetRulesPeriod": 30,
@@ -345,6 +365,10 @@ def get_goal_state_period(conf=__conf__):
     return conf.get_int("Extensions.GoalStatePeriod", 6)
 
 
+def get_initial_goal_state_period(conf=__conf__):
+    return conf.get_int("Extensions.InitialGoalStatePeriod", default_value=lambda: get_goal_state_period(conf=conf))
+
+
 def get_goal_state_history_cleanup_period(conf=__conf__):
     return conf.get_int("Extensions.GoalStateHistoryCleanupPeriod", 1800)
 
@@ -507,3 +531,11 @@ def get_cgroup_disable_on_quota_check_failure(conf=__conf__):
     NOTE: This option is experimental and may be removed in later versions of the Agent.
     """
     return conf.get_switch("Debug.CgroupDisableOnQuotaCheckFailure", True)
+
+def get_enable_fast_track(conf=__conf__):
+    """
+    If True, the agent use FastTrack when retrieving goal states
+
+    NOTE: This option is experimental and may be removed in later versions of the Agent.
+    """
+    return conf.get_switch("Debug.EnableFastTrack", False)
diff --git a/azurelinuxagent/common/event.py b/azurelinuxagent/common/event.py
index 1fe90214..0887a924 100644
--- a/azurelinuxagent/common/event.py
+++ b/azurelinuxagent/common/event.py
@@ -87,6 +87,7 @@ class WALAEventOperation:
     ExtensionTelemetryEventProcessing = "ExtensionTelemetryEventProcessing"
     FetchGoalState = "FetchGoalState"
     Firewall = "Firewall"
+    GoalState = "GoalState"
     GoalStateUnsupportedFeatures = "GoalStateUnsupportedFeatures"
     HealthCheck = "HealthCheck"
     HealthObservation = "HealthObservation"
@@ -95,6 +96,7 @@ class WALAEventOperation:
     HostPluginHeartbeat = "HostPluginHeartbeat"
     HostPluginHeartbeatExtended = "HostPluginHeartbeatExtended"
     HttpErrors = "HttpErrors"
+    HttpGet = "HttpGet"
     ImdsHeartbeat = "ImdsHeartbeat"
     Install = "Install"
     InitializeHostPlugin = "InitializeHostPlugin"
@@ -491,6 +493,7 @@ class EventLogger(object):
         self.add_common_event_parameters(event, datetime.utcnow())
 
         data = get_properties(event)
+
         try:
             self.save_event(json.dumps(data))
         except EventError as e:
diff --git a/azurelinuxagent/common/logcollector.py b/azurelinuxagent/common/logcollector.py
index 8048a45d..d71ffcfc 100644
--- a/azurelinuxagent/common/logcollector.py
+++ b/azurelinuxagent/common/logcollector.py
@@ -43,6 +43,10 @@ _TRUNCATED_FILES_DIR = os.path.join(_LOG_COLLECTOR_DIR, "truncated")
 OUTPUT_RESULTS_FILE_PATH = os.path.join(_LOG_COLLECTOR_DIR, "results.txt")
 COMPRESSED_ARCHIVE_PATH = os.path.join(_LOG_COLLECTOR_DIR, "logs.zip")
 
+CGROUPS_UNIT = "collect-logs.scope"
+
+INVALID_CGROUPS_ERRCODE = 2
+
 _MUST_COLLECT_FILES = [
     _AGENT_LOG,
     os.path.join(_AGENT_LIB_DIR, "GoalState.*.xml"),
@@ -50,7 +54,7 @@ _MUST_COLLECT_FILES = [
     os.path.join(_AGENT_LIB_DIR, "HostingEnvironmentConfig.*.xml"),
     os.path.join(_AGENT_LIB_DIR, "SharedConfig.*.xml"),
     os.path.join(_AGENT_LIB_DIR, "*manifest.xml"),
-    os.path.join(_AGENT_LIB_DIR, "waagent_status.json"),
+    os.path.join(_AGENT_LIB_DIR, "waagent_status.*.json"),
     os.path.join(_AGENT_LIB_DIR, "history", "*.zip"),
     os.path.join(_EXTENSION_LOG_DIR, "*", "*"),
     os.path.join(_EXTENSION_LOG_DIR, "*", "*", "*"),
diff --git a/azurelinuxagent/common/logcollector_manifests.py b/azurelinuxagent/common/logcollector_manifests.py
index 4d9d1953..27eba6ed 100644
--- a/azurelinuxagent/common/logcollector_manifests.py
+++ b/azurelinuxagent/common/logcollector_manifests.py
@@ -40,7 +40,8 @@ echo,
 
 echo,### Gathering Extension Files ###
 copy,$LIB_DIR/*.xml
-copy,$LIB_DIR/waagent_status.json
+copy,$LIB_DIR/VmSettings.*.json
+copy,$LIB_DIR/waagent_status.*.json
 copy,$LIB_DIR/*/status/*.status
 copy,$LIB_DIR/*/config/*.settings
 copy,$LIB_DIR/*/config/HandlerState
@@ -119,7 +120,7 @@ copy,$LIB_DIR/SharedConfig.xml
 copy,$LIB_DIR/ManagedIdentity-*.json
 copy,$LIB_DIR/*/error.json
 copy,$LIB_DIR/Incarnation
-copy,$LIB_DIR/waagent_status.json
+copy,$LIB_DIR/waagent_status.*.json
 copy,$LIB_DIR/history/*.zip
 echo,
 
diff --git a/azurelinuxagent/common/osutil/factory.py b/azurelinuxagent/common/osutil/factory.py
index b212c382..4a09544c 100644
--- a/azurelinuxagent/common/osutil/factory.py
+++ b/azurelinuxagent/common/osutil/factory.py
@@ -98,7 +98,7 @@ def _get_osutil(distro_name, distro_code_name, distro_version, distro_full_name)
 
         return DebianOSBaseUtil()
 
-    if distro_name in ("redhat", "rhel", "centos", "oracle"):
+    if distro_name in ("redhat", "rhel", "centos", "oracle", "almalinux"):
         if Version(distro_version) < Version("7"):
             return Redhat6xOSUtil()
 
diff --git a/azurelinuxagent/common/persist_firewall_rules.py b/azurelinuxagent/common/persist_firewall_rules.py
index ac336cda..bd143454 100644
--- a/azurelinuxagent/common/persist_firewall_rules.py
+++ b/azurelinuxagent/common/persist_firewall_rules.py
@@ -17,14 +17,13 @@
 #
 import os
 import sys
-import traceback
 
 import azurelinuxagent.common.conf as conf
 from azurelinuxagent.common import logger
 from azurelinuxagent.common.event import add_event, WALAEventOperation
 from azurelinuxagent.common.future import ustr
 from azurelinuxagent.common.osutil import get_osutil, systemd
-from azurelinuxagent.common.utils import shellutil, fileutil
+from azurelinuxagent.common.utils import shellutil, fileutil, textutil
 from azurelinuxagent.common.utils.networkutil import AddFirewallRules
 from azurelinuxagent.common.utils.shellutil import CommandError
 
@@ -275,9 +274,9 @@ if __name__ == '__main__':
             msg = "Unable to fetch service logs, Command: {0} failed with ExitCode: {1}\nStdout: {2}\nStderr: {3}".format(
                 ' '.join(cmd), error.returncode, error.stdout, error.stderr)
             logger.warn(msg)
-        except Exception:
+        except Exception as e:
             msg = "Ran into unexpected error when getting logs for {0} service. Error: {1}".format(
-                self._network_setup_service_name, traceback.format_exc())
+                self._network_setup_service_name, textutil.format_exception(e))
             logger.warn(msg)
 
         # Log service status and logs if we can fetch them from journalctl and send it to Kusto,
diff --git a/azurelinuxagent/common/protocol/extensions_goal_state.py b/azurelinuxagent/common/protocol/extensions_goal_state.py
new file mode 100644
index 00000000..049cf00f
--- /dev/null
+++ b/azurelinuxagent/common/protocol/extensions_goal_state.py
@@ -0,0 +1,29 @@
+# Microsoft Azure Linux Agent
+#
+# Copyright 2020 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 re
+
+
+class ExtensionsGoalState(object):
+    def __init__(self, etag, vm_settings):
+        self.etag = etag
+        self.vm_settings = vm_settings
+
+    def get_redacted_vm_settings(self):
+        return re.sub(r'("protectedSettings"\s*:\s*)"[^"]+"', r'\1"*** REDACTED ***"', self.vm_settings)
+
diff --git a/azurelinuxagent/common/protocol/hostplugin.py b/azurelinuxagent/common/protocol/hostplugin.py
index 4d23a205..438a394c 100644
--- a/azurelinuxagent/common/protocol/hostplugin.py
+++ b/azurelinuxagent/common/protocol/hostplugin.py
@@ -36,6 +36,7 @@ from azurelinuxagent.common.version import AGENT_NAME, AGENT_VERSION, PY_VERSION
 HOST_PLUGIN_PORT = 32526
 
 URI_FORMAT_GET_API_VERSIONS = "http://{0}:{1}/versions"
+URI_FORMAT_VM_SETTINGS = "http://{0}:{1}/vmSettings"
 URI_FORMAT_GET_EXTENSION_ARTIFACT = "http://{0}:{1}/extensionArtifact"
 URI_FORMAT_PUT_VM_STATUS = "http://{0}:{1}/status"
 URI_FORMAT_PUT_LOG = "http://{0}:{1}/vmAgentLog"
@@ -99,7 +100,7 @@ class HostPluginProtocol(object):
             self.api_versions = self.get_api_versions()
             self.is_available = API_VERSION in self.api_versions
             self.is_initialized = self.is_available
-            add_event(WALAEventOperation.InitializeHostPlugin,
+            add_event(op=WALAEventOperation.InitializeHostPlugin,
                       is_success=self.is_available)
         return self.is_available
 
@@ -140,6 +141,21 @@ class HostPluginProtocol(object):
 
         return return_val
 
+    def get_vm_settings_request(self, correlation_id):
+        if not self.ensure_initialized():
+            raise ProtocolError("HostGAPlugin: Host plugin channel is not available")
+
+        url = URI_FORMAT_VM_SETTINGS.format(self.endpoint, HOST_PLUGIN_PORT)
+
+        headers = {
+            _HEADER_VERSION: API_VERSION,
+           _HEADER_CONTAINER_ID: self.container_id,
+           _HEADER_HOST_CONFIG_NAME: self.role_config_name,
+           _HEADER_CORRELATION_ID: correlation_id
+        }
+
+        return url, headers
+
     def get_artifact_request(self, artifact_url, artifact_manifest_url=None):
         if not self.ensure_initialized():
             raise ProtocolError("HostGAPlugin: Host plugin channel is not available")
diff --git a/azurelinuxagent/common/protocol/wire.py b/azurelinuxagent/common/protocol/wire.py
index 7654b675..0191d209 100644
--- a/azurelinuxagent/common/protocol/wire.py
+++ b/azurelinuxagent/common/protocol/wire.py
@@ -20,7 +20,7 @@ import json
 import os
 import random
 import time
-import traceback
+import uuid
 import xml.sax.saxutils as saxutils
 from collections import defaultdict
 from datetime import datetime, timedelta
@@ -35,6 +35,7 @@ from azurelinuxagent.common.event import add_event, WALAEventOperation, report_e
 from azurelinuxagent.common.exception import ProtocolNotFoundError, \
     ResourceGoneError, ExtensionDownloadError, InvalidContainerError, ProtocolError, HttpError
 from azurelinuxagent.common.future import httpclient, bytebuffer, ustr
+from azurelinuxagent.common.protocol.extensions_goal_state import ExtensionsGoalState
 from azurelinuxagent.common.protocol.goal_state import GoalState, TRANSPORT_CERT_FILE_NAME, TRANSPORT_PRV_FILE_NAME
 from azurelinuxagent.common.protocol.hostplugin import HostPluginProtocol
 from azurelinuxagent.common.protocol.restapi import DataContract, ExtHandlerPackage, \
@@ -55,6 +56,7 @@ TELEMETRY_URI = "http://{0}/machine?comp=telemetrydata"
 WIRE_SERVER_ADDR_FILE_NAME = "WireServer"
 INCARNATION_FILE_NAME = "Incarnation"
 GOAL_STATE_FILE_NAME = "GoalState.{0}.xml"
+VM_SETTINGS_FILE_NAME = "VmSettings.{0}.json"
 HOSTING_ENV_FILE_NAME = "HostingEnvironmentConfig.xml"
 SHARED_CONF_FILE_NAME = "SharedConfig.xml"
 REMOTE_ACCESS_FILE_NAME = "RemoteAccess.{0}.xml"
@@ -93,12 +95,12 @@ class WireProtocol(DataContract):
         logger.info('Initializing goal state during protocol detection')
         self.client.update_goal_state(forced=True)
 
+    def update_extensions_goal_state(self):
+        self.client.update_extensions_goal_state()
+
     def update_goal_state(self):
         self.client.update_goal_state()
 
-    def try_update_goal_state(self):
-        return self.client.try_update_goal_state()
-
     def update_host_plugin_from_goal_state(self):
         self.client.update_host_plugin_from_goal_state()
 
@@ -124,6 +126,9 @@ class WireProtocol(DataContract):
     def get_incarnation(self):
         return self.client.get_goal_state().incarnation
 
+    def get_etag(self):
+        return self.client.get_etag()
+
     def get_in_vm_gs_metadata(self):
         return self.client.get_ext_conf().in_vm_gs_metadata
 
@@ -198,6 +203,9 @@ class WireProtocol(DataContract):
     def upload_logs(self, logs):
         self.client.upload_logs(logs)
 
+    def get_status_blob_data(self):
+        return self.client.status_blob.data
+
 
 def _build_role_properties(container_id, role_instance_id, thumbprint):
     xml = (u"<?xml version=\"1.0\" encoding=\"utf-8\"?>"
@@ -572,7 +580,7 @@ class WireClient(object):
         logger.info("Wire server endpoint:{0}", endpoint)
         self._endpoint = endpoint
         self._goal_state = None
-        self._last_try_update_goal_state_failed = False
+        self._extensions_goal_state = None
         self._host_plugin = None
         self.status_blob = StatusBlob(self)
         self.goal_state_flusher = StateFlusher(conf.get_lib_dir())
@@ -643,7 +651,7 @@ class WireClient(object):
     def fetch_manifest_through_host(self, uri):
         host = self.get_host_plugin()
         uri, headers = host.get_artifact_request(uri)
-        response = self.fetch(uri, headers, use_proxy=False, max_retry=1)
+        response, _ = self.fetch(uri, headers, use_proxy=False, max_retry=1)
         return response
 
     def fetch_manifest(self, version_uris, timeout_in_minutes=5, timeout_in_ms=0):
@@ -666,7 +674,7 @@ class WireClient(object):
                 logger.verbose('The specified manifest URL is empty, ignored.')
                 continue
 
-            direct_func = lambda: self.fetch(version.uri, max_retry=1)  # pylint: disable=W0640
+            direct_func = lambda: self.fetch(version.uri, max_retry=1)[0]  # pylint: disable=W0640
             # NOTE: the host_func may be called after refreshing the goal state, be careful about any goal state data
             # in the lambda.
             host_func = lambda: self.fetch_manifest_through_host(version.uri)  # pylint: disable=W0640
@@ -707,19 +715,21 @@ class WireClient(object):
 
         return success
 
-    def fetch(self, uri, headers=None, use_proxy=None, decode=True, max_retry=None):
+    def fetch(self, uri, headers=None, use_proxy=None, decode=True, max_retry=None, ok_codes=None):
         """
         max_retry indicates the maximum number of retries for the HTTP request; None indicates that the default value should be used
+
+        Returns a tuple with the content and headers of the response. The headers are a list of (name, value) tuples.
         """
         logger.verbose("Fetch [{0}] with headers [{1}]", uri, headers)
         content = None
-        response = self._fetch_response(uri, headers, use_proxy, max_retry=max_retry)
-        if response is not None and not restutil.request_failed(response):
+        response = self._fetch_response(uri, headers, use_proxy, max_retry=max_retry, ok_codes=ok_codes)
+        if response is not None and not restutil.request_failed(response, ok_codes=ok_codes):
             response_content = response.read()
             content = self.decode_config(response_content) if decode else response_content
-        return content
+        return content, response.getheaders()
 
-    def _fetch_response(self, uri, headers=None, use_proxy=None, max_retry=None):
+    def _fetch_response(self, uri, headers=None, use_proxy=None, max_retry=None, ok_codes=None):
         """
         max_retry indicates the maximum number of retries for the HTTP request; None indicates that the default value should be used
         """
@@ -734,7 +744,7 @@ class WireClient(object):
 
             host_plugin = self.get_host_plugin()
 
-            if restutil.request_failed(resp):
+            if restutil.request_failed(resp, ok_codes=ok_codes):
                 error_response = restutil.read_response_error(resp)
                 msg = "Fetch failed from [{0}]: {1}".format(uri, error_response)
                 logger.warn(msg)
@@ -752,10 +762,7 @@ class WireClient(object):
         except (HttpError, ProtocolError, IOError) as error:
             msg = "Fetch failed: {0}".format(error)
             logger.warn(msg)
-            report_event(op=WALAEventOperation.Download,
-                            is_success=False,
-                            message=msg,
-                            log_event=False)
+            report_event(op=WALAEventOperation.HttpGet, is_success=False, message=msg, log_event=False)
 
             if isinstance(error, (InvalidContainerError, ResourceGoneError)):
                 # These are retryable errors that should force a goal state refresh in the host plugin
@@ -782,35 +789,56 @@ class WireClient(object):
 
             if new_goal_state is not None:
                 self._goal_state = new_goal_state
-                self._save_goal_state()
                 self._update_host_plugin(new_goal_state.container_id, new_goal_state.role_config_name)
+                if conf.get_enable_fast_track():
+                    self.update_extensions_goal_state()
+                self._save_goal_state()
 
         except Exception as exception:
-            raise ProtocolError("Error processing goal state: {0}".format(ustr(exception)))
+            raise ProtocolError("Error fetching goal state: {0}".format(ustr(exception)))
+
+    def update_extensions_goal_state(self):
+        correlation_id = str(uuid.uuid4())
+        etag = self.get_etag()
 
-    def try_update_goal_state(self):
-        """
-        Attempts to update the goal state and returns True on success or False on failure, sending telemetry events about the failures.
-        """
         try:
-            self.update_goal_state()
+            # TODO: Remove this log statement when invoking this method on each iteration of the goal state loop (currently it is invoked only on a new goal state)
+            logger.info("Fetching vmSettings [correlation ID: {0} eTag: {1}]", correlation_id, etag)
+            def get_vm_settings():
+                url, headers = self.get_host_plugin().get_vm_settings_request(correlation_id)
+                if etag is not None:
+                    headers['if-none-match'] = etag
+                return self.fetch(url, headers, ok_codes=[httpclient.OK, httpclient.NOT_MODIFIED])
 
-            if self._last_try_update_goal_state_failed:
-                self._last_try_update_goal_state_failed = False
-                message = u"Retrieving the goal state recovered from previous errors"
-                add_event(AGENT_NAME, op=WALAEventOperation.FetchGoalState, version=CURRENT_VERSION, is_success=True, message=message, log_event=False)
-                logger.info(message)
-        except Exception as e:
-            if not self._last_try_update_goal_state_failed:
-                self._last_try_update_goal_state_failed = True
-                message = u"An error occurred while retrieving the goal state: {0}".format(ustr(e))
-                add_event(AGENT_NAME, op=WALAEventOperation.FetchGoalState, version=CURRENT_VERSION, is_success=False, message=message, log_event=False)
-                message = u"An error occurred while retrieving the goal state: {0}".format(ustr(traceback.format_exc()))
-                logger.warn(message)
-            message = u"Attempts to retrieve the goal state are failing: {0}".format(ustr(e))
-            logger.periodic_warn(logger.EVERY_SIX_HOURS, "[PERIODIC] {0}".format(message))
-            return False
-        return True
+            try:
+                vm_settings, response_headers = get_vm_settings()
+            except ResourceGoneError:
+                self.update_host_plugin_from_goal_state()
+                vm_settings, response_headers = get_vm_settings()
+
+            # on error, fetch() logs the original exception and returns None as content
+            if vm_settings is None:
+                raise Exception("Failed to retrieve the extensions goal state (vmSettings), the error was reported previously in the log.")
+
+            # The response includes an etag if and only if the VM settings change.
+            response_etag = None
+            for h in response_headers:
+                if h[0].lower() == 'etag':
+                    response_etag = h[1]
+                    break
+
+            if response_etag is not None:
+                logger.info("Fetched new vmSettings [correlation ID: {0} New eTag: {1}]", correlation_id, response_etag)
+                self._extensions_goal_state = ExtensionsGoalState(response_etag, vm_settings)
+            else:
+                # TODO: Remove this log statement (the entire else clause) when invoking this method on each iteration of the goal state loop (currently it is invoked only on a new goal state)
+                logger.error("Expected new vmSettings, but the response did not include an eTag [correlation ID: {0} eTag: {1}]", correlation_id, etag)
+
+        except Exception as exception:
+            # TODO: raise ProtocolError instead of logging a warning
+            message = "Error fetching vmSettings [correlation ID: {0} eTag: {1}]: {2}".format(correlation_id, etag, ustr(exception))
+            logger.warn(message)
+            report_event(op=WALAEventOperation.GoalState, is_success=False, message=message, log_event=False)
 
     def _update_host_plugin(self, container_id, role_config_name):
         if self._host_plugin is not None:
@@ -837,6 +865,9 @@ class WireClient(object):
             if self._goal_state.ext_conf is not None and self._goal_state.ext_conf.xml_text is not None:
                 self._save_cache(self._goal_state.ext_conf.get_redacted_xml_text(), EXT_CONF_FILE_NAME.format(self._goal_state.incarnation))
 
+            if self._extensions_goal_state is not None:
+                self._save_cache(self._extensions_goal_state.get_redacted_vm_settings(), VM_SETTINGS_FILE_NAME.format(self._extensions_goal_state.etag))
+
         except Exception as e:
             logger.warn("Failed to save the goal state to disk: {0}", ustr(e))
 
@@ -845,6 +876,19 @@ class WireClient(object):
             logger.warn("Setting empty Host Plugin object!")
         self._host_plugin = new_host_plugin
 
+    def get_etag(self):
+        """
+        Returns the Etag of the current ExtensionsGoalState, or None, if the goal state has not been retrieved.
+        """
+        if self._extensions_goal_state is None:
+            return None
+        return self._extensions_goal_state.etag
+
+    def get_extensions_goal_state(self):
+        if self._extensions_goal_state is None:
+            raise ProtocolError("Trying to fetch the extensions goal state before initialization!")
+        return self._extensions_goal_state
+
     def get_goal_state(self):
         if self._goal_state is None:
             raise ProtocolError("Trying to fetch goal state before initialization!")
@@ -1043,7 +1087,7 @@ class WireClient(object):
 
         if ext_conf.status_upload_blob is None:
             # the status upload blob is in ExtensionsConfig so force a full goal state refresh
-            self.update_goal_state(forced=True)
+            self.update_goal_state(forced=True)  # FT: This will come from the ExtensionsGoalState
             ext_conf = self.get_ext_conf()
 
         if ext_conf.status_upload_blob is None:
@@ -1204,7 +1248,7 @@ class WireClient(object):
                 events_per_provider[event.providerId] += 1
 
             except Exception as error:
-                logger.warn("Unexpected error when generating Events: {0}, {1}", ustr(error), traceback.format_exc())
+                logger.warn("Unexpected error when generating Events:{0}", textutil.format_exception(error))
 
         # Send out all events left in buffer.
         for provider_id in list(buf.keys()):
@@ -1261,7 +1305,7 @@ class WireClient(object):
     def get_artifacts_profile_through_host(self, blob):
         host = self.get_host_plugin()
         uri, headers = host.get_artifact_request(blob)
-        profile = self.fetch(uri, headers, use_proxy=False)
+        profile, _ = self.fetch(uri, headers, use_proxy=False)
         return profile
 
     def get_artifacts_profile(self):
@@ -1269,7 +1313,7 @@ class WireClient(object):
 
         if self.has_artifacts_profile_blob():
             blob = self.get_ext_conf().artifacts_profile_blob
-            direct_func = lambda: self.fetch(blob)
+            direct_func = lambda: self.fetch(blob)[0]
             # NOTE: the host_func may be called after refreshing the goal state, be careful about any goal state data
             # in the lambda.
             host_func = lambda: self.get_artifacts_profile_through_host(blob)
diff --git a/azurelinuxagent/common/utils/archive.py b/azurelinuxagent/common/utils/archive.py
index e18b1b06..8e484577 100644
--- a/azurelinuxagent/common/utils/archive.py
+++ b/azurelinuxagent/common/utils/archive.py
@@ -41,9 +41,11 @@ ARCHIVE_DIRECTORY_NAME = 'history'
 _MAX_ARCHIVED_STATES = 50
 
 _CACHE_PATTERNS = [
+    re.compile(r"^VmSettings.\d+\.json$"),
     re.compile(r"^(.*)\.(\d+)\.(agentsManifest)$", re.IGNORECASE),
     re.compile(r"^(.*)\.(\d+)\.(manifest\.xml)$", re.IGNORECASE),
-    re.compile(r"^(.*)\.(\d+)\.(xml)$", re.IGNORECASE)
+    re.compile(r"^(.*)\.(\d+)\.(xml)$", re.IGNORECASE),
+    re.compile(r"waagent_status\.(\d+)\.json$")
 ]
 
 _GOAL_STATE_PATTERN = re.compile(r"^(.*)/GoalState\.(\d+)\.xml$", re.IGNORECASE)
diff --git a/azurelinuxagent/common/utils/restutil.py b/azurelinuxagent/common/utils/restutil.py
index 06dbde91..71ffab5b 100644
--- a/azurelinuxagent/common/utils/restutil.py
+++ b/azurelinuxagent/common/utils/restutil.py
@@ -21,7 +21,6 @@ import os
 import re
 import threading
 import time
-import traceback
 import socket
 import struct
 
@@ -626,6 +625,6 @@ def read_response_error(resp):
 
             result = textutil.replace_non_ascii(result)
 
-        except Exception:
-            logger.warn(traceback.format_exc())
+        except Exception as e:
+            logger.warn(textutil.format_exception(e))
     return result
diff --git a/azurelinuxagent/common/utils/textutil.py b/azurelinuxagent/common/utils/textutil.py
index 21d0adf6..153eb806 100644
--- a/azurelinuxagent/common/utils/textutil.py
+++ b/azurelinuxagent/common/utils/textutil.py
@@ -24,6 +24,7 @@ import re
 import string
 import struct
 import sys
+import traceback
 import xml.dom.minidom as minidom
 import zlib
 
@@ -430,3 +431,21 @@ def str_to_encoded_ustr(s, encoding='utf-8'):
             return ustr(s)
     # For Py2, explicitly convert the string to unicode with the specified encoding
     return ustr(s, encoding=encoding)
+
+
+def format_exception(exception):
+    # Function to format exception message
+    e = None
+    if sys.version_info[0] == 2:
+        _, e, tb = sys.exc_info()
+    else:
+        tb = exception.__traceback__
+
+    msg = ustr(exception) + "\n"
+    if tb is None or (sys.version_info[0] == 2 and e != exception):
+        msg += "[Traceback not available]"
+    else:
+        msg += ''.join(traceback.format_exception(etype=type(exception), value=exception, tb=tb))
+
+    return msg
+
diff --git a/azurelinuxagent/common/version.py b/azurelinuxagent/common/version.py
index b686b866..fa10683e 100644
--- a/azurelinuxagent/common/version.py
+++ b/azurelinuxagent/common/version.py
@@ -200,9 +200,9 @@ AGENT_LONG_NAME = "Azure Linux Agent"
 # IMPORTANT: Please be sure that the version is always 9.9.9.9 on the develop branch. Automation requires this, otherwise
 #            DCR may test the wrong agent version.
 #
-#            When doing a release, be sure to use the actual agent version. Current agent version: 2.4.0.0
+#            When doing a release, be sure to use the actual agent version. Current agent version: 2.5.0.2
 #
-AGENT_VERSION = '2.4.0.2'
+AGENT_VERSION = '2.5.0.2'
 AGENT_LONG_VERSION = "{0}-{1}".format(AGENT_NAME, AGENT_VERSION)
 AGENT_DESCRIPTION = """
 The Azure Linux Agent supports the provisioning and running of Linux
diff --git a/azurelinuxagent/daemon/main.py b/azurelinuxagent/daemon/main.py
index 93ba9eb9..7c397dc5 100644
--- a/azurelinuxagent/daemon/main.py
+++ b/azurelinuxagent/daemon/main.py
@@ -20,7 +20,6 @@
 import os
 import sys
 import time
-import traceback
 
 import azurelinuxagent.common.conf as conf
 import azurelinuxagent.common.logger as logger
@@ -30,8 +29,8 @@ from azurelinuxagent.common.event import add_event, WALAEventOperation, initiali
 from azurelinuxagent.common.future import ustr
 from azurelinuxagent.common.osutil import get_osutil
 from azurelinuxagent.common.protocol.util import get_protocol_util
-from azurelinuxagent.common.protocol.wire import WireProtocol
 from azurelinuxagent.common.rdma import setup_rdma_device
+from azurelinuxagent.common.utils import textutil
 from azurelinuxagent.common.version import AGENT_NAME, AGENT_LONG_NAME, \
     AGENT_VERSION, \
     DISTRO_NAME, DISTRO_VERSION, PY_VERSION_MAJOR, PY_VERSION_MINOR, \
@@ -82,7 +81,7 @@ class DaemonHandler(object):
             try:
                 self.daemon(child_args)
             except Exception as e:  # pylint: disable=W0612
-                err_msg = traceback.format_exc()
+                err_msg = textutil.format_exception(e)
                 add_event(name=AGENT_NAME, is_success=False, message=ustr(err_msg),
                           op=WALAEventOperation.UnhandledError)
                 logger.warn("Daemon ended with exception -- Sleep 15 seconds and restart daemon")
@@ -160,8 +159,6 @@ class DaemonHandler(object):
                 #   incarnation number. A forced update ensures the most
                 #   current values.
                 protocol = self.protocol_util.get_protocol()
-                if type(protocol) is not WireProtocol:
-                    raise Exception("Attempt to setup RDMA without Wireserver")
 
                 protocol.client.update_goal_state(forced=True)
 
diff --git a/azurelinuxagent/ga/collect_logs.py b/azurelinuxagent/ga/collect_logs.py
index ad5befdf..5181837b 100644
--- a/azurelinuxagent/ga/collect_logs.py
+++ b/azurelinuxagent/ga/collect_logs.py
@@ -22,6 +22,7 @@ import os
 import sys
 import threading
 import time
+from azurelinuxagent.common import cgroupconfigurator, logcollector
 
 import azurelinuxagent.common.conf as conf
 from azurelinuxagent.common import logger
@@ -29,13 +30,12 @@ from azurelinuxagent.common.event import elapsed_milliseconds, add_event, WALAEv
 from azurelinuxagent.common.future import subprocess_dev_null, ustr
 from azurelinuxagent.common.interfaces import ThreadHandlerInterface
 from azurelinuxagent.common.logcollector import COMPRESSED_ARCHIVE_PATH
-from azurelinuxagent.common.osutil import systemd
+from azurelinuxagent.common.cgroupconfigurator import CGroupConfigurator
 from azurelinuxagent.common.protocol.util import get_protocol_util
 from azurelinuxagent.common.utils import shellutil
 from azurelinuxagent.common.utils.shellutil import CommandError
 from azurelinuxagent.common.version import PY_VERSION_MAJOR, PY_VERSION_MINOR, AGENT_NAME, CURRENT_VERSION
 
-
 def get_collect_logs_handler():
     return CollectLogsHandler()
 
@@ -43,17 +43,17 @@ def get_collect_logs_handler():
 def is_log_collection_allowed():
     # There are three conditions that need to be met in order to allow periodic log collection:
     # 1) It should be enabled in the configuration.
-    # 2) The system must be using systemd to manage services. Needed for resource limiting of the log collection.
+    # 2) The system must be using cgroups to manage services. Needed for resource limiting of the log collection.
     # 3) The python version must be greater than 2.6 in order to support the ZipFile library used when collecting.
     conf_enabled = conf.get_collect_logs()
-    systemd_present = systemd.is_systemd()
+    cgroups_enabled = CGroupConfigurator.get_instance().enabled()
     supported_python = PY_VERSION_MINOR >= 7 if PY_VERSION_MAJOR == 2 else PY_VERSION_MAJOR == 3
-    is_allowed = conf_enabled and systemd_present and supported_python
+    is_allowed = conf_enabled and cgroups_enabled and supported_python
 
     msg = "Checking if log collection is allowed at this time [{0}]. All three conditions must be met: " \
-          "configuration enabled [{1}], systemd present [{2}], python supported: [{3}]".format(is_allowed,
+          "configuration enabled [{1}], cgroups enabled [{2}], python supported: [{3}]".format(is_allowed,
                                                                                                conf_enabled,
-                                                                                               systemd_present,
+                                                                                               cgroups_enabled,
                                                                                                supported_python)
     logger.info(msg)
     add_event(
@@ -73,11 +73,27 @@ class CollectLogsHandler(ThreadHandlerInterface):
     """
 
     _THREAD_NAME = "CollectLogsHandler"
+    __CGROUPS_FLAG_ENV_VARIABLE = "_AZURE_GUEST_AGENT_LOG_COLLECTOR_MONITOR_CGROUPS_"
 
     @staticmethod
     def get_thread_name():
         return CollectLogsHandler._THREAD_NAME
 
+    @staticmethod
+    def enable_cgroups_validation():
+        os.environ[CollectLogsHandler.__CGROUPS_FLAG_ENV_VARIABLE] = "1"
+
+    @staticmethod
+    def disable_cgroups_validation():
+        if CollectLogsHandler.__CGROUPS_FLAG_ENV_VARIABLE in os.environ:
+            del os.environ[CollectLogsHandler.__CGROUPS_FLAG_ENV_VARIABLE]
+
+    @staticmethod
+    def should_validate_cgroups():
+        if CollectLogsHandler.__CGROUPS_FLAG_ENV_VARIABLE in os.environ:
+            return os.environ[CollectLogsHandler.__CGROUPS_FLAG_ENV_VARIABLE] == "1"
+        return False
+
     def __init__(self):
         self.protocol = None
         self.protocol_util = None
@@ -118,6 +134,7 @@ class CollectLogsHandler(ThreadHandlerInterface):
 
     def daemon(self):
         try:
+            CollectLogsHandler.enable_cgroups_validation()
             if self.protocol_util is None or self.protocol is None:
                 self.init_protocols()
 
@@ -131,44 +148,33 @@ class CollectLogsHandler(ThreadHandlerInterface):
                     time.sleep(self.period)
         except Exception as e:
             logger.error("An error occurred in the log collection thread; will exit the thread.\n{0}", ustr(e))
+        finally:
+            CollectLogsHandler.disable_cgroups_validation()
 
     def collect_and_send_logs(self):
         if self._collect_logs():
             self._send_logs()
 
-    @staticmethod
-    def _get_resource_limits():
-        # Define CPU limit (as percentage of CPU time) and memory limit (absolute value in megabytes).
-        cpu_limit = "5%"
-        memory_limit = "30M"  # K for kb, M for mb
-        return cpu_limit, memory_limit
-
-    @staticmethod
-    def _collect_logs():
+    def _collect_logs(self):
         logger.info("Starting log collection...")
 
         # Invoke the command line tool in the agent to collect logs, with resource limits on CPU and memory (RAM).
-        scope_name = "collect-logs-{0}.scope".format(ustr(int(time.time() * 1000000)))
-        systemd_cmd = ["systemd-run", "--unit={0}".format(scope_name), "--scope"]
 
-        # More info on resource limits properties in systemd here:
-        # https://access.redhat.com/documentation/en-us/red_hat_enterprise_linux/7/html/resource_management_guide/sec-modifying_control_groups
-        cpu_limit, memory_limit = CollectLogsHandler._get_resource_limits()
-        resource_limits = ["--property=CPUAccounting=1", "--property=CPUQuota={0}".format(cpu_limit),
-                           "--property=MemoryAccounting=1", "--property=MemoryLimit={0}".format(memory_limit)]
+        systemd_cmd = [
+            "systemd-run", "--unit={0}".format(logcollector.CGROUPS_UNIT),
+            "--slice={0}".format(cgroupconfigurator.LOGCOLLECTOR_SLICE), "--scope"
+        ]
 
         # The log tool is invoked from the current agent's egg with the command line option
         collect_logs_cmd = [sys.executable, "-u", sys.argv[0], "-collect-logs"]
-        final_command = systemd_cmd + resource_limits + collect_logs_cmd
+        final_command = systemd_cmd + collect_logs_cmd
 
         def exec_command(output_file):
             start_time = datetime.datetime.utcnow()
             success = False
             msg = None
             try:
-                # 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=False, track_process=False,
-                    stdout=output_file, stderr=output_file)
+                shellutil.run_command(final_command, log_error=False, stdout=output_file, stderr=output_file)
                 duration = elapsed_milliseconds(start_time)
                 archive_size = os.path.getsize(COMPRESSED_ARCHIVE_PATH)
 
@@ -180,14 +186,22 @@ class CollectLogsHandler(ThreadHandlerInterface):
                 return True
             except Exception as e:
                 duration = elapsed_milliseconds(start_time)
+                err_msg = ustr(e)
 
                 if isinstance(e, CommandError):
-                    exception_message = ustr("[stderr] %s", e.stderr)  # pylint: disable=no-member
-                else:
-                    exception_message = ustr(e)
-
-                msg = "Failed to collect logs. Elapsed time: {0} ms. Error: {1}".format(duration, exception_message)
-                # No need to log to the local log since we ran run_command with logging errors as enabled
+                    # pylint has limited (i.e. no) awareness of control flow w.r.t. typing. we disable=no-member
+                    # here because we know e must be a CommandError but pylint still considers the case where
+                    # e is a different type of exception.
+                    err_msg = ustr("Log Collector exited with code {0}").format(e.returncode) # pylint: disable=no-member
+
+                    if e.returncode == logcollector.INVALID_CGROUPS_ERRCODE: # pylint: disable=no-member
+                        logger.info("Disabling periodic log collection until service restart due to process error.")
+                        self.stop()
+                    else:
+                        logger.info(err_msg)
+
+                msg = "Failed to collect logs. Elapsed time: {0} ms. Error: {1}".format(duration, err_msg)
+                # No need to log to the local log since we logged stdout, stderr from the process.
 
                 return False
             finally:
diff --git a/azurelinuxagent/ga/collect_telemetry_events.py b/azurelinuxagent/ga/collect_telemetry_events.py
index c3a17ca1..0503fd6b 100644
--- a/azurelinuxagent/ga/collect_telemetry_events.py
+++ b/azurelinuxagent/ga/collect_telemetry_events.py
@@ -21,7 +21,6 @@ import json
 import os
 import re
 import threading
-import traceback
 from collections import defaultdict
 
 import azurelinuxagent.common.logger as logger
@@ -35,6 +34,7 @@ from azurelinuxagent.common.future import ustr
 from azurelinuxagent.common.interfaces import ThreadHandlerInterface
 from azurelinuxagent.common.telemetryevent import TelemetryEvent, TelemetryEventParam, \
     GuestAgentGenericLogsSchema, GuestAgentExtensionEventsSchema
+from azurelinuxagent.common.utils import textutil
 from azurelinuxagent.ga.exthandlers import HANDLER_NAME_PATTERN
 from azurelinuxagent.ga.periodic_operation import PeriodicOperation
 
@@ -116,8 +116,8 @@ class _ProcessExtensionEvents(PeriodicOperation):
             # the telemetry service comes back up
             delete_all_event_files = False
         except Exception as error:
-            msg = "Unknown error occurred when trying to collect extension events. Error: {0}, Stack: {1}".format(
-                ustr(error), traceback.format_exc())
+            msg = "Unknown error occurred when trying to collect extension events:{0}".format(
+                textutil.format_exception(error))
             add_event(op=WALAEventOperation.ExtensionTelemetryEventProcessing, message=msg, is_success=False)
         finally:
             # Always ensure that the events directory are being deleted each run except when Telemetry Service is stopped,
@@ -204,8 +204,8 @@ class _ProcessExtensionEvents(PeriodicOperation):
                     # This is a trade-off between data replication vs data loss.
                     raise
                 except Exception as error:
-                    msg = "Failed to process event file {0}: {1}, {2}".format(event_file, ustr(error),
-                                                                              traceback.format_exc())
+                    msg = "Failed to process event file {0}:{1}".format(event_file,
+                                                                              textutil.format_exception(error))
                     logger.warn(msg)
                     add_log_event(level=logger.LogLevel.WARNING, message=msg, forced=True)
                 finally:
diff --git a/azurelinuxagent/ga/exthandlers.py b/azurelinuxagent/ga/exthandlers.py
index 9ccf2669..ed5fe8d8 100644
--- a/azurelinuxagent/ga/exthandlers.py
+++ b/azurelinuxagent/ga/exthandlers.py
@@ -16,7 +16,6 @@
 #
 # Requires Python 2.6+ and Openssl 1.0+
 #
-
 import datetime
 import glob
 import json
@@ -25,10 +24,8 @@ import random
 import re
 import shutil
 import stat
-import sys
 import tempfile
 import time
-import traceback
 import zipfile
 from collections import defaultdict
 from functools import partial
@@ -50,10 +47,11 @@ from azurelinuxagent.common.exception import ExtensionDownloadError, ExtensionEr
 from azurelinuxagent.common.future import ustr, is_file_not_found_error
 from azurelinuxagent.common.protocol.restapi import ExtensionStatus, ExtensionSubStatus, ExtHandler, ExtHandlerStatus, \
     VMStatus, GoalStateAggregateStatus, ExtensionState, ExtHandlerRequestedState, Extension
+from azurelinuxagent.common.utils import textutil
 from azurelinuxagent.common.utils.archive import ARCHIVE_DIRECTORY_NAME
 from azurelinuxagent.common.utils.flexible_version import FlexibleVersion
-from azurelinuxagent.common.version import AGENT_NAME, CURRENT_VERSION, DISTRO_NAME, DISTRO_VERSION, \
-    GOAL_STATE_AGENT_VERSION, PY_VERSION_MAJOR, PY_VERSION_MICRO, PY_VERSION_MINOR
+from azurelinuxagent.common.version import AGENT_NAME, CURRENT_VERSION, \
+    PY_VERSION_MAJOR, PY_VERSION_MICRO, PY_VERSION_MINOR
 
 _HANDLER_NAME_PATTERN = r'^([^-]+)'
 _HANDLER_VERSION_PATTERN = r'(\d+(?:\.\d+)*)'
@@ -67,7 +65,7 @@ HANDLER_NAME_PATTERN = re.compile(_HANDLER_NAME_PATTERN, re.IGNORECASE)
 HANDLER_COMPLETE_NAME_PATTERN = re.compile(_HANDLER_PATTERN + r'$', re.IGNORECASE)
 HANDLER_PKG_EXT = ".zip"
 
-AGENT_STATUS_FILE = "waagent_status.json"
+AGENT_STATUS_FILE = "waagent_status.{0}.json"
 NUMBER_OF_DOWNLOAD_RETRIES = 2
 
 # This is the default value for the env variables, whenever we call a command which is not an update scenario, we
@@ -90,7 +88,18 @@ _STATUS_FILE_RETRY_DELAY = 2  # seconds
 _DEFAULT_SEQ_NO = "0"
 
 
+class HandlerStatus(object):
+    """
+    Statuses for Handlers
+    """
+    ready = "Ready"
+    not_ready = "NotReady"
+
+
 class ValidHandlerStatus(object):
+    """
+    Statuses for Extensions
+    """
     transitioning = "transitioning"
     warning = "warning"
     error = "error"
@@ -115,14 +124,6 @@ class ExtCommandEnvVariable(object):
     ExtensionSupportedFeatures = "{0}_EXTENSION_SUPPORTED_FEATURES".format(Prefix)
 
 
-def get_traceback(e):  # pylint: disable=R1710
-    if sys.version_info[0] == 3:  # pylint: disable=R1705
-        return e.__traceback__
-    elif sys.version_info[0] == 2:
-        ex_type, ex, tb = sys.exc_info()  # pylint: disable=W0612
-        return tb
-
-
 def validate_has_key(obj, key, full_key_path):
     if key not in obj:
         raise ExtensionStatusError(msg="Invalid status format by extension: Missing {0} key".format(full_key_path),
@@ -307,9 +308,9 @@ class ExtHandlersHandler(object):
                 # trying to retrieve is debug only on linux.
                 error_msg = u"Couldn't parse debug metadata value: {0}".format(e)
                 logger.verbose(error_msg)
-            
+
             return "NA"
-        
+
         to_utc = lambda time: time.strftime(logger.Logger.LogTimeFormatInUTC)
         identity = lambda value: value
 
@@ -333,7 +334,7 @@ class ExtHandlersHandler(object):
 
             activity_id, correlation_id, gs_creation_time = self.get_goal_state_debug_metadata()
         except Exception as error:
-            msg = u"ProcessExtensionsInGoalState - Exception processing extension handlers: {0}\n{1}".format(ustr(error), traceback.extract_tb(get_traceback(error)))
+            msg = u"ProcessExtensionsInGoalState - Exception processing extension handlers:{0}".format(textutil.format_exception(error))
             logger.warn(msg)
             add_event(op=WALAEventOperation.ExtensionProcessing, is_success=False, message=msg, log_event=False)
             return
@@ -353,7 +354,7 @@ class ExtHandlersHandler(object):
             self.__process_and_handle_extensions(etag)
             self._cleanup_outdated_handlers()
         except Exception as error:
-            error = u"ProcessExtensionsInGoalState - Exception processing extension handlers: {0}\n{1}".format(ustr(error), traceback.extract_tb(get_traceback(error)))
+            error = u"ProcessExtensionsInGoalState - Exception processing extension handlers:{0}".format(textutil.format_exception(error))
         finally:
             duration = elapsed_milliseconds(utc_start)
             if error is None:
@@ -390,7 +391,7 @@ class ExtHandlersHandler(object):
                                                                       code=GoalStateAggregateStatusCodes.Success,
                                                                       message="GoalState executed successfully")
         except Exception as error:
-            msg = "Unexpected error when processing goal state: {0}; {1}".format(ustr(error), traceback.format_exc())
+            msg = "Unexpected error when processing goal state:{0}".format(textutil.format_exception(error))
             self.__gs_aggregate_status = GoalStateAggregateStatus(status=GoalStateStatus.Failed, seq_no=etag,
                                                                   code=GoalStateAggregateStatusCodes.GoalStateUnknownFailure,
                                                                   message=msg)
@@ -529,9 +530,9 @@ class ExtHandlersHandler(object):
                     if handler_i.get_handler_status() is None:
                         handler_i.set_handler_status(message=depends_on_err_msg, code=-1)
 
-                    handler_i.create_placeholder_status_file(extension, status=ValidHandlerStatus.error, code=-1,
-                                                             operation=WALAEventOperation.ExtensionProcessing,
-                                                             message=depends_on_err_msg)
+                    handler_i.create_status_file_if_not_exist(extension, status=ValidHandlerStatus.error, code=-1,
+                                                              operation=WALAEventOperation.ExtensionProcessing,
+                                                              message=depends_on_err_msg)
 
                 # For SC extensions, overwrite the HandlerStatus with the relevant message
                 else:
@@ -591,9 +592,9 @@ class ExtHandlersHandler(object):
                     break
                 time.sleep(5)
 
-        except Exception:
+        except Exception as e:
             msg = "Failed to wait for Handler completion due to unknown error. Marking the dependent extension as failed: {0}, {1}".format(
-                extension_name, traceback.format_exc())
+                extension_name, textutil.format_exception(e))
             raise Exception(msg)
 
         # In case of timeout or terminal error state, we log it and raise
@@ -654,8 +655,8 @@ class ExtHandlersHandler(object):
             # This error is only thrown for enable operation on MultiConfig extension.
             # Since these are maintained by the extensions, the expectation here is that they would update their status files appropriately with their errors.
             # The extensions should already have a placeholder status file, but incase they dont, setting one here to fail fast.
-            ext_handler_i.create_placeholder_status_file(extension, status=ValidHandlerStatus.error, code=error.code,
-                                                         operation=ext_handler_i.operation, message=err_msg)
+            ext_handler_i.create_status_file_if_not_exist(extension, status=ValidHandlerStatus.error, code=error.code,
+                                                          operation=ext_handler_i.operation, message=err_msg)
             add_event(name=ext_name, version=ext_handler_i.ext_handler.properties.version, op=ext_handler_i.operation,
                       is_success=False, log_event=True, message=err_msg)
         except ExtensionConfigError as error:
@@ -695,8 +696,8 @@ class ExtHandlersHandler(object):
         # file with failure since the extensions wont be called where they can create their status files.
         # This way we guarantee reporting back to CRP
         if ext_handler_i.should_perform_multi_config_op(extension):
-            ext_handler_i.create_placeholder_status_file(extension, status=ValidHandlerStatus.error, code=error.code,
-                                                         operation=report_op, message=message)
+            ext_handler_i.create_status_file_if_not_exist(extension, status=ValidHandlerStatus.error, code=error.code,
+                                                          operation=report_op, message=message)
 
         if report:
             name = ext_handler_i.get_extension_full_name(extension)
@@ -730,17 +731,6 @@ class ExtHandlersHandler(object):
             ext_handler_i.ensure_consistent_data_for_mc()
             ext_handler_i.update_settings(extension)
 
-        # Always create a placeholder file for enable/disable command if not exists
-        # We don't create a placeholder for other commands because we use status file as a way to report Handler level
-        # failures back to CRP. If a placeholder for an extension already exists with Transitioning status, we would
-        # not override it, hence we only create a placeholder for enable/disable commands but the extensions have the
-        # data to create their own if needed.
-
-        # Note: Due to a bug in multiple extensions, we're only creating a default placeholder for Multi-Config extensions.
-        # A fix will follow soon where we will report transitioning status for extensions by default if no status file
-        # found instead of reporting an error.
-        if ext_handler_i.should_perform_multi_config_op(extension):
-            ext_handler_i.create_placeholder_status_file(extension)
         self.__handle_extension(ext_handler_i, extension, uninstall_exit_code)
 
     @staticmethod
@@ -925,7 +915,8 @@ class ExtHandlersHandler(object):
 
     def report_ext_handlers_status(self, incarnation_changed=False):
         """
-        Go through handler_state dir, collect and report status
+        Go through handler_state dir, collect and report status.
+        Returns the status it reported, or None if an error occurred.
         """
         try:
             vm_status = VMStatus(status="Ready", message="Guest Agent is running",
@@ -979,48 +970,54 @@ class ExtHandlersHandler(object):
 
             self.write_ext_handlers_status_to_info_file(vm_status)
 
+            return vm_status
+
         except Exception as error:
-            msg = u"Failed to report status: {0}\n{1}".format(ustr(error), traceback.extract_tb(get_traceback(error)))
+            msg = u"Failed to report status: {0}".format(textutil.format_exception(error))
             logger.warn(msg)
             add_event(AGENT_NAME,
                       version=CURRENT_VERSION,
                       op=WALAEventOperation.ReportStatus,
                       is_success=False,
                       message=msg)
+            return None
 
-    @staticmethod
-    def write_ext_handlers_status_to_info_file(vm_status):
-        status_path = os.path.join(conf.get_lib_dir(), AGENT_STATUS_FILE)
-
-        agent_details = {
-            "agent_name": AGENT_NAME,
-            "current_version": str(CURRENT_VERSION),
-            "goal_state_version": str(GOAL_STATE_AGENT_VERSION),
-            "distro_details": "{0}:{1}".format(DISTRO_NAME, DISTRO_VERSION),
-            "last_successful_status_upload_time": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
-            "python_version": "Python: {0}.{1}.{2}".format(PY_VERSION_MAJOR, PY_VERSION_MINOR, PY_VERSION_MICRO),
-            "crp_supported_features": [name for name, _ in get_agent_supported_features_list_for_crp().items()],
-            "extension_supported_features": [name for name, _ in get_agent_supported_features_list_for_extensions().items()]
+    def write_ext_handlers_status_to_info_file(self, vm_status):
+        status_path = os.path.join(conf.get_lib_dir(), AGENT_STATUS_FILE.format(self.protocol.get_incarnation()))
+        status_blob_data = self.protocol.get_status_blob_data()
+        data = dict()
+        if status_blob_data is not None:
+            data = json.loads(status_blob_data)
+
+        # Populating the fields that does not come from vm_status or status_blob_data
+        _metadataNotSentToCRP = {
+            "agentName": AGENT_NAME,
+            "daemonVersion": str(version.get_daemon_version()),
+            "pythonVersion": "Python: {0}.{1}.{2}".format(PY_VERSION_MAJOR, PY_VERSION_MINOR, PY_VERSION_MICRO),
+            "extensionSupportedFeatures": [name for name, _ in
+                                         get_agent_supported_features_list_for_extensions().items()]
         }
+        data["_metadataNotSentToCRP"] = _metadataNotSentToCRP
 
-        # Convert VMStatus class to Dict.
-        data = get_properties(vm_status)
+        # Consuming supports_multi_config info from vm_status. creating a dict out of it for easy lookup in the next step.
+        support_multi_config = dict()
+        if vm_status is not None:
+            # Convert VMStatus class to Dict.
+            vm_status_data = get_properties(vm_status)
+            vm_handler_statuses = vm_status_data.get('vmAgent', dict()).get('extensionHandlers')
+            for handler_status in vm_handler_statuses:
+                if handler_status.get('name') is not None:
+                    support_multi_config[handler_status.get('name')] = handler_status.get('supports_multi_config')
 
-        # The above class contains vmAgent.extensionHandlers
-        # (more info: azurelinuxagent.common.protocol.restapi.VMAgentStatus)
-        handler_statuses = data['vmAgent']['extensionHandlers']
-        for handler_status in handler_statuses:
-            try:
-                handler_status.pop('code', None)
-                handler_status.pop('message', None)
-                handler_status.pop('extension_status', None)
-            except KeyError:
-                pass
+        handler_aggregate_status = data.get('aggregateStatus', dict()).get('handlerAggregateStatus', dict())
 
-        agent_details['extensions_status'] = handler_statuses
-        agent_details_json = json.dumps(agent_details)
+        for handler_status in handler_aggregate_status:
+            handler_status['supportsMultiConfig'] = support_multi_config.get(handler_status.get('handlerName'))
+            status = handler_status.get('runtimeSettingsStatus', dict()).get('settingsStatus', dict()).get('status', dict())
+            status.pop('formattedMessage', None)
+            status.pop('substatus', None)
 
-        fileutil.write_file(status_path, agent_details_json)
+        fileutil.write_file(status_path, json.dumps(data))
 
     def report_ext_handler_status(self, vm_status, ext_handler, incarnation_changed):
         ext_handler_i = ExtHandlerInstance(ext_handler, self.protocol)
@@ -1403,8 +1400,10 @@ class ExtHandlerInstance(object):
         # Save HandlerEnvironment.json
         self.create_handler_env()
 
-    def create_placeholder_status_file(self, extension=None, status=ValidHandlerStatus.transitioning, code=0,
-                                       operation="Enabling Extension", message="Install/Enable is in progress."):
+        CGroupConfigurator.get_instance().setup_extension_slice(
+            extension_name=self.get_full_name())
+
+    def create_status_file_if_not_exist(self, extension, status, code, operation, message):
         _, status_path = self.get_status_file_path(extension)
         if status_path is not None and not os.path.exists(status_path):
             now = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")
@@ -1440,7 +1439,7 @@ class ExtHandlerInstance(object):
             raise
         # Even if a single extension is enabled for this handler, set the Handler state as Enabled
         self.set_handler_state(ExtHandlerState.Enabled)
-        self.set_handler_status(status="Ready", message="Plugin enabled")
+        self.set_handler_status(status=HandlerStatus.ready, message="Plugin enabled")
 
     def should_perform_multi_config_op(self, extension):
         return self.supports_multi_config and extension is not None
@@ -1456,7 +1455,7 @@ class ExtHandlerInstance(object):
         man = self.load_manifest()
         enable_cmd = man.get_enable_command()
         self.logger.info("Enable extension: [{0}]".format(enable_cmd))
-        self.launch_command(enable_cmd, timeout=300,
+        self.launch_command(enable_cmd, cmd_name="enable", timeout=300,
                             extension_error_code=ExtensionErrorCodes.PluginEnableProcessingFailed, env=env,
                             extension=extension)
 
@@ -1469,7 +1468,7 @@ class ExtHandlerInstance(object):
         man = self.load_manifest()
         disable_cmd = man.get_disable_command()
         self.logger.info("Disable extension: [{0}]".format(disable_cmd))
-        self.launch_command(disable_cmd, timeout=900,
+        self.launch_command(disable_cmd, cmd_name="disable", timeout=900,
                             extension_error_code=ExtensionErrorCodes.PluginDisableProcessingFailed,
                             extension=extension)
 
@@ -1493,7 +1492,7 @@ class ExtHandlerInstance(object):
         # For MultiConfig, Set the handler state to Installed only when all extensions have been disabled
         if not self.supports_multi_config or not any(self.enabled_extensions):
             self.set_handler_state(ExtHandlerState.Installed)
-            self.set_handler_status(status="NotReady", message="Plugin disabled")
+            self.set_handler_status(status=HandlerStatus.not_ready, message="Plugin disabled")
 
     def install(self, uninstall_exit_code=None, extension=None):
         # For Handler level operations, extension just specifies the settings that initiated the install.
@@ -1506,10 +1505,10 @@ class ExtHandlerInstance(object):
         install_cmd = man.get_install_command()
         self.logger.info("Install extension [{0}]".format(install_cmd))
         self.set_operation(WALAEventOperation.Install)
-        self.launch_command(install_cmd, timeout=900, extension=extension,
+        self.launch_command(install_cmd, cmd_name="install", timeout=900, extension=extension,
                             extension_error_code=ExtensionErrorCodes.PluginInstallProcessingFailed, env=env)
         self.set_handler_state(ExtHandlerState.Installed)
-        self.set_handler_status(status="NotReady", message="Plugin installed but not enabled")
+        self.set_handler_status(status=HandlerStatus.not_ready, message="Plugin installed but not enabled")
 
     def uninstall(self, extension=None):
         # For Handler level operations, extension just specifies the settings that initiated the uninstall.
@@ -1519,7 +1518,7 @@ class ExtHandlerInstance(object):
         man = self.load_manifest()
         uninstall_cmd = man.get_uninstall_command()
         self.logger.info("Uninstall extension [{0}]".format(uninstall_cmd))
-        self.launch_command(uninstall_cmd, extension=extension)
+        self.launch_command(uninstall_cmd, cmd_name="uninstall", extension=extension)
 
     def remove_ext_handler(self):
         try:
@@ -1539,6 +1538,10 @@ class ExtHandlerInstance(object):
                         raise exception
 
                 shutil.rmtree(base_dir, onerror=on_rmtree_error)
+
+            self.logger.info("Remove the extension slice: {0}".format(self.get_full_name()))
+            CGroupConfigurator.get_instance().remove_extension_slice(
+                extension_name=self.get_full_name())
         except IOError as e:
             message = "Failed to remove extension handler directory: {0}".format(e)
             self.report_event(message=message, is_success=False)
@@ -1573,7 +1576,7 @@ class ExtHandlerInstance(object):
             man = self.load_manifest()
             update_cmd = man.get_update_command()
             self.logger.info("Update extension [{0}]".format(update_cmd))
-            self.launch_command(update_cmd,
+            self.launch_command(update_cmd, cmd_name="update",
                                 timeout=900,
                                 extension_error_code=ExtensionErrorCodes.PluginUpdateProcessingFailed,
                                 env=env, extension=extension)
@@ -1679,15 +1682,25 @@ class ExtHandlerInstance(object):
             data_str, data = self._read_status_file(ext_status_file)
         except ExtensionStatusError as e:
             msg = ""
+            ext_status.status = ValidHandlerStatus.error
+
             if e.code == ExtensionStatusError.CouldNotReadStatusFile:
                 ext_status.code = ExtensionErrorCodes.PluginUnknownFailure
                 msg = u"We couldn't read any status for {0} extension, for the sequence number {1}. It failed due" \
                       u" to {2}".format(self.get_full_name(ext), seq_no, ustr(e))
-            elif ExtensionStatusError.InvalidJsonFile:
+            elif e.code == ExtensionStatusError.InvalidJsonFile:
                 ext_status.code = ExtensionErrorCodes.PluginSettingsStatusInvalid
                 msg = u"The status reported by the extension {0}(Sequence number {1}), was in an " \
                       u"incorrect format and the agent could not parse it correctly. Failed due to {2}" \
                       .format(self.get_full_name(ext), seq_no, ustr(e))
+            elif e.code == ExtensionStatusError.FileNotExists:
+                msg = "This status is being reported by the Guest Agent since no status file was " \
+                      "reported by extension {0}: {1}".format(self.get_extension_full_name(ext), ustr(e))
+
+                # Reporting a success code and transitioning status to keep in accordance with existing code that
+                # creates default status placeholder file
+                ext_status.code = ExtensionErrorCodes.PluginSuccess
+                ext_status.status = ValidHandlerStatus.transitioning
 
             # This log is periodic due to the verbose nature of the status check. Please make sure that the message
             # constructed above does not change very frequently and includes important info such as sequence number,
@@ -1700,7 +1713,6 @@ class ExtHandlerInstance(object):
                          log_event=False)
 
             ext_status.message = msg
-            ext_status.status = ValidHandlerStatus.error
 
             return ext_status
 
@@ -1798,11 +1810,11 @@ class ExtHandlerInstance(object):
                 ext_handler_status = ExtHandlerStatus()
                 set_properties("ExtHandlerStatus", ext_handler_status, get_properties(handler_status))
             except Exception as error:
-                msg = "Something went wrong when trying to get a copy of the Handler status for {0}: {1}".format(
-                    self.get_extension_full_name(), ustr(error))
+                msg = "Something went wrong when trying to get a copy of the Handler status for {0}".format(
+                    self.get_extension_full_name())
                 self.report_error_on_incarnation_change(incarnation_changed, event_msg=msg,
                                                         log_msg="{0}.\nStack Trace: {1}".format(
-                                                                                     msg, traceback.format_exc()))
+                                                                                     msg, textutil.format_exception(error)))
                 # Since this is a Handler level error and we need to do it per extension, breaking here and logging
                 # error since we wont be able to report error anyways and saving it as a handler status (legacy behavior)
                 self.set_handler_status(message=msg, code=-1)
@@ -1816,11 +1828,11 @@ class ExtHandlerInstance(object):
                 ext_handler_statuses.append(ext_handler_status)
             except ExtensionError as error:
 
-                msg = "Unknown error when trying to fetch status from extension {0}: {1}".format(
-                    self.get_extension_full_name(ext), ustr(error))
+                msg = "Unknown error when trying to fetch status from extension {0}".format(
+                    self.get_extension_full_name(ext))
                 self.report_error_on_incarnation_change(incarnation_changed, event_msg=msg,
                                                         log_msg="{0}.\nStack Trace: {1}".format(
-                                                                                     msg, traceback.format_exc()),
+                                                                                     msg, textutil.format_exception(error)),
                                                         extension=ext)
 
                 # Unexpected error, for single config, keep the behavior as is
@@ -1873,7 +1885,7 @@ class ExtHandlerInstance(object):
         last_update = int(time.time() - os.stat(heartbeat_file).st_mtime)
         return last_update <= 600
 
-    def launch_command(self, cmd, timeout=300, extension_error_code=ExtensionErrorCodes.PluginProcessingError,
+    def launch_command(self, cmd, cmd_name=None, timeout=300, extension_error_code=ExtensionErrorCodes.PluginProcessingError,
                        env=None, extension=None):
         begin_utc = datetime.datetime.utcnow()
         self.logger.verbose("Launch command: [{0}]", cmd)
@@ -1923,6 +1935,7 @@ class ExtHandlerInstance(object):
                     process_output = CGroupConfigurator.get_instance().start_extension_command(
                         extension_name=self.get_full_name(extension),
                         command=command_full_path,
+                        cmd_name=cmd_name,
                         timeout=timeout,
                         shell=True,
                         cwd=base_dir,
@@ -1993,14 +2006,16 @@ class ExtHandlerInstance(object):
 
     def create_handler_env(self):
         handler_env = {
-                HandlerEnvironment.logFolder: self.get_log_dir(), 
-                HandlerEnvironment.configFolder: self.get_conf_dir(), 
-                HandlerEnvironment.statusFolder: self.get_status_dir(), 
-                HandlerEnvironment.heartbeatFile: self.get_heartbeat_file() 
+                HandlerEnvironment.logFolder: self.get_log_dir(),
+                HandlerEnvironment.configFolder: self.get_conf_dir(),
+                HandlerEnvironment.statusFolder: self.get_status_dir(),
+                HandlerEnvironment.heartbeatFile: self.get_heartbeat_file()
             }
 
         if get_supported_feature_by_name(SupportedFeatureNames.ExtensionTelemetryPipeline).is_supported:
             handler_env[HandlerEnvironment.eventsFolder] = self.get_extension_events_dir()
+            # For now, keep the preview key to not break extensions that were using the preview.
+            handler_env[HandlerEnvironment.eventsFolder_preview] = self.get_extension_events_dir()
 
         env = [{
             HandlerEnvironment.name: self.ext_handler.name,
@@ -2072,7 +2087,7 @@ class ExtHandlerInstance(object):
             self.report_event(name=extension_name, message=message, is_success=False, log_event=False)
             self.logger.warn(message)
 
-    def set_handler_status(self, status="NotReady", message="", code=0):
+    def set_handler_status(self, status=HandlerStatus.not_ready, message="", code=0):
         state_dir = self.get_conf_dir()
 
         handler_status = ExtHandlerStatus()
@@ -2095,7 +2110,7 @@ class ExtHandlerInstance(object):
                     self.ext_handler.name, self.ext_handler.properties.version))
         except (IOError, ValueError, ProtocolError) as error:
             fileutil.clean_ioerror(error, paths=[status_file])
-            self.logger.error("Failed to save handler status: {0}, {1}", ustr(error), traceback.format_exc())
+            self.logger.error("Failed to save handler status: {0}", textutil.format_exception(error))
 
     def get_handler_status(self):
         state_dir = self.get_conf_dir()
@@ -2177,6 +2192,9 @@ class ExtHandlerInstance(object):
     @staticmethod
     def _read_and_parse_json_status_file(ext_status_file):
 
+        if not os.path.exists(ext_status_file):
+            raise ExtensionStatusError(msg="Status file {0} does not exist".format(ext_status_file),
+                                       code=ExtensionStatusError.FileNotExists)
         try:
             data_str = fileutil.read_file(ext_status_file)
         except IOError as e:
@@ -2226,7 +2244,8 @@ class HandlerEnvironment(object):
     configFolder = "configFolder"
     statusFolder = "statusFolder"
     heartbeatFile = "heartbeatFile"
-    eventsFolder = "eventsFolder_preview"
+    eventsFolder_preview = "eventsFolder_preview"
+    eventsFolder = "eventsFolder"
     name = "name"
     version = "version"
 
@@ -2282,6 +2301,7 @@ class ExtensionStatusError(ExtensionError):
     InvalidJsonFile = 2
     StatusFileMalformed = 3
     MaxSizeExceeded = 4
+    FileNotExists = 5
 
     def __init__(self, msg=None, inner=None, code=-1):  # pylint: disable=W0235
         super(ExtensionStatusError, self).__init__(msg, inner, code)
diff --git a/azurelinuxagent/ga/remoteaccess.py b/azurelinuxagent/ga/remoteaccess.py
index f12fd0d4..592df1b7 100644
--- a/azurelinuxagent/ga/remoteaccess.py
+++ b/azurelinuxagent/ga/remoteaccess.py
@@ -19,7 +19,6 @@
 
 import os
 import os.path
-import traceback
 from datetime import datetime, timedelta
 
 import azurelinuxagent.common.conf as conf
@@ -27,6 +26,7 @@ import azurelinuxagent.common.logger as logger
 from azurelinuxagent.common.event import add_event, WALAEventOperation
 from azurelinuxagent.common.future import ustr
 from azurelinuxagent.common.osutil import get_osutil
+from azurelinuxagent.common.utils import textutil
 from azurelinuxagent.common.utils.cryptutil import CryptUtil
 from azurelinuxagent.common.version import AGENT_NAME, CURRENT_VERSION
 
@@ -57,7 +57,7 @@ class RemoteAccessHandler(object):
                 self._remote_access = self._protocol.client.get_remote_access()
                 self._handle_remote_access()
         except Exception as e:
-            msg = u"Exception processing goal state for remote access users: {0} {1}".format(ustr(e), traceback.format_exc())
+            msg = u"Exception processing goal state for remote access users: {0}".format(textutil.format_exception(e))
             add_event(AGENT_NAME,
                       version=CURRENT_VERSION,
                       op=WALAEventOperation.RemoteAccessHandling,
diff --git a/azurelinuxagent/ga/send_telemetry_events.py b/azurelinuxagent/ga/send_telemetry_events.py
index 0f52686c..c2e27776 100644
--- a/azurelinuxagent/ga/send_telemetry_events.py
+++ b/azurelinuxagent/ga/send_telemetry_events.py
@@ -19,13 +19,13 @@
 import datetime
 import threading
 import time
-import traceback
 
 from azurelinuxagent.common import logger
 from azurelinuxagent.common.event import add_event, WALAEventOperation
 from azurelinuxagent.common.exception import ServiceStoppedError
 from azurelinuxagent.common.future import ustr, Queue, Empty
 from azurelinuxagent.common.interfaces import ThreadHandlerInterface
+from azurelinuxagent.common.utils import textutil
 
 
 def get_send_telemetry_events_handler(protocol_util):
@@ -135,8 +135,8 @@ class SendTelemetryEventsHandler(ThreadHandlerInterface):
                     self._send_events_in_queue(first_event)
 
         except Exception as error:
-            err_msg = "An unknown error occurred in the {0} thread main loop, stopping thread. Error: {1}, Stack: {2}".format(
-                self.get_thread_name(), ustr(error), traceback.format_exc())
+            err_msg = "An unknown error occurred in the {0} thread main loop, stopping thread.{1}".format(
+                self.get_thread_name(), textutil.format_exception(error))
             add_event(op=WALAEventOperation.UnhandledError, message=err_msg, is_success=False)
 
     def _send_events_in_queue(self, first_event):
@@ -161,5 +161,4 @@ class SendTelemetryEventsHandler(ThreadHandlerInterface):
                 self._queue.task_done()
                 yield event
             except Exception as error:
-                logger.error("Some exception when fetching event from queue: {0}, {1}".format(ustr(error),
-                                                                                              traceback.format_exc()))
\ No newline at end of file
+                logger.error("Some exception when fetching event from queue: {0}".format(textutil.format_exception(error)))
\ No newline at end of file
diff --git a/azurelinuxagent/ga/update.py b/azurelinuxagent/ga/update.py
index c656388d..3dd3fc66 100644
--- a/azurelinuxagent/ga/update.py
+++ b/azurelinuxagent/ga/update.py
@@ -27,7 +27,6 @@ import stat
 import subprocess
 import sys
 import time
-import traceback
 import uuid
 import zipfile
 
@@ -57,7 +56,7 @@ from azurelinuxagent.ga.collect_logs import get_collect_logs_handler, is_log_col
 from azurelinuxagent.ga.env import get_env_handler
 from azurelinuxagent.ga.collect_telemetry_events import get_collect_telemetry_events_handler
 
-from azurelinuxagent.ga.exthandlers import HandlerManifest, get_traceback, ExtHandlersHandler, list_agent_lib_directory
+from azurelinuxagent.ga.exthandlers import HandlerManifest, ExtHandlersHandler, list_agent_lib_directory, ValidHandlerStatus, HandlerStatus
 from azurelinuxagent.ga.monitor import get_monitor_handler
 
 from azurelinuxagent.ga.send_telemetry_events import get_send_telemetry_events_handler
@@ -73,13 +72,18 @@ CHILD_POLL_INTERVAL = 60
 
 MAX_FAILURE = 3  # Max failure allowed for agent before blacklisted
 
-GOAL_STATE_INTERVAL_DISABLED = 5 * 60
+GOAL_STATE_PERIOD_EXTENSIONS_DISABLED = 5 * 60
 
 ORPHAN_POLL_INTERVAL = 3
 ORPHAN_WAIT_INTERVAL = 15 * 60
 
 AGENT_SENTINEL_FILE = "current_version"
 
+# This file marks that the first goal state (after provisioning) has been completed, either because it converged or because we received another goal
+# state before it converged. The contents will be an instance of ExtensionsSummary. If the file does not exist then we have not finished processing
+# the goal state.
+INITIAL_GOAL_STATE_FILE = "initial_goal_state"
+
 READONLY_FILE_GLOBS = [
     "*.crt",
     "*.p7m",
@@ -89,6 +93,33 @@ READONLY_FILE_GLOBS = [
 ]
 
 
+class ExtensionsSummary(object):
+    """
+    The extensions summary is a list of (extension name, extension status) tuples for the current goal state; it is
+    used to report changes in the status of extensions and to keep track of when the goal state converges (i.e. when
+    all extensions in the goal state reach a terminal state: success or error.)
+    The summary is computed from the VmStatus reported to blob storage.
+    """
+    def __init__(self, vm_status=None):
+        if vm_status is None:
+            self.summary = []
+            self.converged = True
+        else:
+            # take the name and status of the extension if is it not None, else use the handler's
+            self.summary = [(o.name, o.status) for o in map(lambda h: h.extension_status if h.extension_status is not None else h, vm_status.vmAgent.extensionHandlers)]
+            self.summary.sort(key=lambda s: s[0])  # sort by extension name to make comparisons easier
+            self.converged = all(status in (ValidHandlerStatus.success, ValidHandlerStatus.error, HandlerStatus.ready, HandlerStatus.not_ready) for _, status in self.summary)
+
+    def __eq__(self, other):
+        return self.summary == other.summary
+
+    def __ne__(self, other):
+        return not (self.summary == other.summary)
+
+    def __str__(self):
+        return ustr(self.summary)
+
+
 def get_update_handler():
     return UpdateHandler()
 
@@ -115,10 +146,26 @@ class UpdateHandler(object):
         self._last_telemetry_heartbeat = None
         self._heartbeat_id = str(uuid.uuid4()).upper()
         self._heartbeat_counter = 0
+
+        # these members are used to avoid reporting errors too frequently
         self._heartbeat_update_goal_state_error_count = 0
+        self._last_try_update_goal_state_failed = False
+        self._report_status_last_failed_incarnation = -1
 
         self.last_incarnation = None
 
+        self._extensions_summary = ExtensionsSummary()
+
+        self._is_initial_goal_state = not os.path.exists(self._initial_goal_state_file_path())
+
+        if not conf.get_extensions_enabled():
+            self._goal_state_period = GOAL_STATE_PERIOD_EXTENSIONS_DISABLED
+        else:
+            if self._is_initial_goal_state:
+                self._goal_state_period = conf.get_initial_goal_state_period()
+            else:
+                self._goal_state_period = conf.get_goal_state_period()
+
     def run_latest(self, child_args=None):
         """
         This method is called from the daemon to find and launch the most
@@ -228,12 +275,11 @@ class UpdateHandler(object):
         except Exception as e:
             # Ignore child errors during termination
             if self.is_running:
-                msg = u"Agent {0} launched with command '{1}' failed with exception: {2}".format(
+                msg = u"Agent {0} launched with command '{1}' failed with exception: \n".format(
                     agent_name,
-                    agent_cmd,
-                    ustr(e))
+                    agent_cmd)
                 logger.warn(msg)
-                detailed_message = '{0} {1}'.format(msg, traceback.format_exc())
+                detailed_message = '{0} {1}'.format(msg, textutil.format_exception(e))
                 add_event(
                     AGENT_NAME,
                     version=agent_version,
@@ -317,14 +363,14 @@ class UpdateHandler(object):
             for thread_handler in all_thread_handlers:
                 thread_handler.run()
 
-            goal_state_interval = conf.get_goal_state_period() if conf.get_extensions_enabled() else GOAL_STATE_INTERVAL_DISABLED
+            logger.info("Goal State Period: {0} sec. This indicates how often the agent checks for new goal states and reports status.", self._goal_state_period)
 
             while self.is_running:
                 self._check_daemon_running(debug)
                 self._check_threads_running(all_thread_handlers)
-                self._process_goal_state(protocol, exthandlers_handler, remote_access_handler)
+                self._process_goal_state(exthandlers_handler, remote_access_handler)
                 self._send_heartbeat_telemetry(protocol)
-                time.sleep(goal_state_interval)
+                time.sleep(self._goal_state_period)
 
         except ExitException as exitException:
             logger.info(exitException.reason)
@@ -332,7 +378,7 @@ class UpdateHandler(object):
             msg = u"Agent {0} failed with exception: {1}".format(CURRENT_AGENT, ustr(error))
             self._set_sentinel(msg=msg)
             logger.warn(msg)
-            logger.warn(traceback.format_exc())
+            logger.warn(textutil.format_exception(error))
             sys.exit(1)
             # additional return here because sys.exit is mocked in unit tests
             return
@@ -352,8 +398,33 @@ class UpdateHandler(object):
                 logger.warn("{0} thread died, restarting".format(thread_handler.get_thread_name()))
                 thread_handler.start()
 
-    def _process_goal_state(self, protocol, exthandlers_handler, remote_access_handler):
-        if not protocol.try_update_goal_state():
+    def _try_update_goal_state(self, protocol):
+        """
+        Attempts to update the goal state and returns True on success or False on failure, sending telemetry events about the failures.
+        """
+        try:
+            protocol.update_goal_state()
+
+            if self._last_try_update_goal_state_failed:
+                self._last_try_update_goal_state_failed = False
+                message = u"Retrieving the goal state recovered from previous errors"
+                add_event(AGENT_NAME, op=WALAEventOperation.FetchGoalState, version=CURRENT_VERSION, is_success=True, message=message, log_event=False)
+                logger.info(message)
+
+        except Exception as e:
+            if not self._last_try_update_goal_state_failed:
+                self._last_try_update_goal_state_failed = True
+                message = u"An error occurred while retrieving the goal state: {0}".format(textutil.format_exception(e))
+                logger.warn(message)
+                add_event(AGENT_NAME, op=WALAEventOperation.FetchGoalState, version=CURRENT_VERSION, is_success=False, message=message, log_event=False)
+            message = u"Attempts to retrieve the goal state are failing: {0}".format(ustr(e))
+            logger.periodic_warn(logger.EVERY_SIX_HOURS, "[PERIODIC] {0}".format(message))
+            return False
+        return True
+
+    def _process_goal_state(self, exthandlers_handler, remote_access_handler):
+        protocol = exthandlers_handler.protocol
+        if not self._try_update_goal_state(protocol):
             self._heartbeat_update_goal_state_error_count += 1
             return
 
@@ -368,18 +439,59 @@ class UpdateHandler(object):
         incarnation = protocol.get_incarnation()
 
         try:
-            if incarnation != self.last_incarnation:
+            if incarnation != self.last_incarnation:  # TODO: This check should be based in the etag for the extensions goal state
+                if not self._extensions_summary.converged:
+                    message = "A new goal state was received, but not all the extensions in the previous goal state have completed: {0}".format(self._extensions_summary)
+                    logger.warn(message)
+                    add_event(op=WALAEventOperation.GoalState, message=message, is_success=False, log_event=False)
+                    if self._is_initial_goal_state:
+                        self._on_initial_goal_state_completed(self._extensions_summary)
+                self._extensions_summary = ExtensionsSummary()
                 exthandlers_handler.run()
 
             # report status always, even if the goal state did not change
             # do it before processing the remote access, since that operation can take a long time
-            exthandlers_handler.report_ext_handlers_status(incarnation_changed=incarnation != self.last_incarnation)
+            self._report_status(exthandlers_handler, incarnation_changed=incarnation != self.last_incarnation)
 
             if incarnation != self.last_incarnation:
                 remote_access_handler.run()
         finally:
             self.last_incarnation = incarnation
 
+    def _report_status(self, exthandlers_handler, incarnation_changed):
+        # report_ext_handlers_status does its own error handling and returns None if an error occurred
+        vm_status = exthandlers_handler.report_ext_handlers_status(incarnation_changed=incarnation_changed)
+        if vm_status is None:
+            return
+
+        try:
+            extensions_summary = ExtensionsSummary(vm_status)
+            if self._extensions_summary != extensions_summary:
+                self._extensions_summary = extensions_summary
+                message = "Extension status: {0}".format(self._extensions_summary)
+                logger.info(message)
+                add_event(op=WALAEventOperation.GoalState, message=message)
+                if self._extensions_summary.converged:
+                    message = "All extensions in the goal state have reached a terminal state: {0}".format(extensions_summary)
+                    logger.info(message)
+                    add_event(op=WALAEventOperation.GoalState, message=message)
+                    if self._is_initial_goal_state:
+                        self._on_initial_goal_state_completed(self._extensions_summary)
+        except Exception as error:
+            # report errors only once per incarnation
+            if self._report_status_last_failed_incarnation != exthandlers_handler.protocol.get_incarnation():
+                self._report_status_last_failed_incarnation = exthandlers_handler.protocol.get_incarnation()
+                msg = u"Error logging the goal state summary: {0}".format(textutil.format_exception(error))
+                logger.warn(msg)
+                add_event(op=WALAEventOperation.GoalState, is_success=False, message=msg)
+
+    def _on_initial_goal_state_completed(self, extensions_summary):
+        fileutil.write_file(self._initial_goal_state_file_path(), ustr(extensions_summary))
+        if conf.get_extensions_enabled() and self._goal_state_period != conf.get_goal_state_period():
+            self._goal_state_period = conf.get_goal_state_period()
+            logger.info("Initial goal state completed, switched the goal state period to {0}", self._goal_state_period)
+        self._is_initial_goal_state = False
+
     def forward_signal(self, signum, frame):
         if signum == signal.SIGTERM:
             self._shutdown()
@@ -441,19 +553,22 @@ class UpdateHandler(object):
     def _emit_changes_in_default_configuration():
         try:
             def log_event(msg):
-                logger.info(msg)
+                logger.info("******** {0} ********", msg)
                 add_event(AGENT_NAME, op=WALAEventOperation.ConfigurationChange, message=msg)
 
-            def log_if_int_changed_from_default(name, current):
+            def log_if_int_changed_from_default(name, current, message=""):
                 default = conf.get_int_default_value(name)
                 if default != current:
-                    log_event("{0} changed from its default: {1}. New value: {2}".format(name, default, current))
+                    log_event("{0} changed from its default: {1}. New value: {2}. {3}".format(name, default, current, message))
 
             def log_if_op_disabled(name, value):
                 if not value:
                     log_event("{0} is set to False, not processing the operation".format(name))
 
-            log_if_int_changed_from_default("Extensions.GoalStatePeriod", conf.get_goal_state_period())
+            log_if_int_changed_from_default("Extensions.GoalStatePeriod", conf.get_goal_state_period(),
+                "Changing this value affects how often extensions are processed and status for the VM is reported. Too small a value may report the VM as unresponsive")
+            log_if_int_changed_from_default("Extensions.InitialGoalStatePeriod", conf.get_initial_goal_state_period(),
+                "Changing this value affects how often extensions are processed and status for the VM is reported. Too small a value may report the VM as unresponsive")
             log_if_op_disabled("OS.EnableFirewall", conf.enable_firewall())
             log_if_op_disabled("Extensions.Enabled", conf.get_extensions_enabled())
 
@@ -676,6 +791,10 @@ class UpdateHandler(object):
     def _sentinel_file_path(self):
         return os.path.join(conf.get_lib_dir(), AGENT_SENTINEL_FILE)
 
+    @staticmethod
+    def _initial_goal_state_file_path():
+        return os.path.join(conf.get_lib_dir(), INITIAL_GOAL_STATE_FILE)
+
     def _shutdown(self):
         # Todo: Ensure all threads stopped when shutting down the main extension handler to ensure that the state of
         # all threads is clean.
@@ -743,7 +862,7 @@ class UpdateHandler(object):
                    or (len(self.agents) > 0 and self.agents[0].version > base_version)
 
         except Exception as e:  # pylint: disable=W0612
-            msg = u"Exception retrieving agent manifests: {0}".format(ustr(traceback.format_exc()))
+            msg = u"Exception retrieving agent manifests: {0}".format(textutil.format_exception(e))
             add_event(AGENT_NAME, op=WALAEventOperation.Download, version=CURRENT_VERSION, is_success=False,
                       message=msg)
             return False
@@ -899,9 +1018,9 @@ class GuestAgent(object):
             #   is corrupt (e.g., missing the HandlerManifest.json file)
             self.mark_failure(is_fatal=os.path.isfile(self.get_agent_pkg_path()))
 
-            msg = u"Agent {0} install failed with exception: {1}".format(
-                self.name, ustr(e))
-            detailed_msg = '{0} {1}'.format(msg, traceback.extract_tb(get_traceback(e)))
+            msg = u"Agent {0} install failed with exception:".format(
+                self.name)
+            detailed_msg = '{0} {1}'.format(msg, textutil.format_exception(e))
             add_event(
                 AGENT_NAME,
                 version=self.version,
diff --git a/azurelinuxagent/pa/provision/cloudinit.py b/azurelinuxagent/pa/provision/cloudinit.py
index b3980065..79316212 100644
--- a/azurelinuxagent/pa/provision/cloudinit.py
+++ b/azurelinuxagent/pa/provision/cloudinit.py
@@ -50,7 +50,7 @@ class CloudInitProvisionHandler(ProvisionHandler):
             utc_start = datetime.utcnow()
             logger.info("Running CloudInit provisioning handler")
             self.wait_for_ovfenv()
-            self.protocol_util.get_protocol()
+            self.protocol_util.get_protocol()  # Trigger protocol detection
             self.report_not_ready("Provisioning", "Starting")
 
             thumbprint = self.wait_for_ssh_host_key()  # pylint: disable=W0612
diff --git a/azurelinuxagent/pa/provision/default.py b/azurelinuxagent/pa/provision/default.py
index 1baa4097..91fe04ed 100644
--- a/azurelinuxagent/pa/provision/default.py
+++ b/azurelinuxagent/pa/provision/default.py
@@ -78,7 +78,7 @@ class ProvisionHandler(object):
             logger.info("Copying ovf-env.xml")
             ovf_env = self.protocol_util.copy_ovf_env()
 
-            self.protocol_util.get_protocol()
+            self.protocol_util.get_protocol()  # Trigger protocol detection
             self.report_not_ready("Provisioning", "Starting")
             logger.info("Starting provisioning")
 
diff --git a/azurelinuxagent/pa/rdma/factory.py b/azurelinuxagent/pa/rdma/factory.py
index d86d8b8f..c6d80811 100644
--- a/azurelinuxagent/pa/rdma/factory.py
+++ b/azurelinuxagent/pa/rdma/factory.py
@@ -38,7 +38,7 @@ def get_rdma_handler(
     ):
         return SUSERDMAHandler()
 
-    if distro_full_name == 'CentOS Linux' or distro_full_name == 'CentOS' or distro_full_name == 'Red Hat Enterprise Linux Server':
+    if distro_full_name in ('CentOS Linux', 'CentOS', 'Red Hat Enterprise Linux Server', 'AlmaLinux'):
         return CentOSRDMAHandler(distro_version)
 
     if distro_full_name == 'Ubuntu':
diff --git a/config/debian/waagent.conf b/config/debian/waagent.conf
index 74676419..6bc36604 100644
--- a/config/debian/waagent.conf
+++ b/config/debian/waagent.conf
@@ -64,7 +64,7 @@ Logs.Verbose=n
 # Logs.Console=y
 
 # Enable periodic log collection, default is n
-Logs.Collect=n
+Logs.Collect=y
 
 # How frequently to collect logs, default is each hour
 Logs.CollectPeriod=3600
diff --git a/config/suse/waagent.conf b/config/suse/waagent.conf
index fdcb6027..ac9f11a2 100644
--- a/config/suse/waagent.conf
+++ b/config/suse/waagent.conf
@@ -67,7 +67,7 @@ Logs.Verbose=n
 # Logs.Console=y
 
 # Enable periodic log collection, default is n
-Logs.Collect=n
+Logs.Collect=y
 
 # How frequently to collect logs, default is each hour
 Logs.CollectPeriod=3600
diff --git a/config/ubuntu/waagent.conf b/config/ubuntu/waagent.conf
index 81f79842..63635a81 100644
--- a/config/ubuntu/waagent.conf
+++ b/config/ubuntu/waagent.conf
@@ -67,7 +67,7 @@ Logs.Verbose=n
 # Logs.Console=y
 
 # Enable periodic log collection, default is n
-Logs.Collect=n
+Logs.Collect=y
 
 # How frequently to collect logs, default is each hour
 Logs.CollectPeriod=3600
diff --git a/config/waagent.conf b/config/waagent.conf
index 75a79b06..c7d10e2f 100644
--- a/config/waagent.conf
+++ b/config/waagent.conf
@@ -73,7 +73,7 @@ Logs.Verbose=n
 # Logs.Console=y
 
 # Enable periodic log collection, default is n
-Logs.Collect=n
+Logs.Collect=y
 
 # How frequently to collect logs, default is each hour
 Logs.CollectPeriod=3600
diff --git a/init/redhat/waagent.service b/init/redhat/waagent.service
new file mode 100644
index 00000000..39d01d1c
--- /dev/null
+++ b/init/redhat/waagent.service
@@ -0,0 +1,16 @@
+[Unit]
+Description=Azure Linux Agent
+Wants=network-online.target sshd.service sshd-keygen.service
+After=network-online.target
+
+ConditionFileIsExecutable=/usr/sbin/waagent
+ConditionPathExists=/etc/waagent.conf
+
+[Service]
+Type=simple
+ExecStart=/usr/bin/python3 -u /usr/sbin/waagent -daemon
+Restart=always
+RestartSec=5
+
+[Install]
+WantedBy=multi-user.target
diff --git a/setup.py b/setup.py
index 23680ed8..a359b39a 100755
--- a/setup.py
+++ b/setup.py
@@ -95,12 +95,21 @@ def get_data_files(name, version, fullname):  # pylint: disable=R0912
     systemd_dir_path = osutil.get_systemd_unit_file_install_path()
     agent_bin_path = osutil.get_agent_bin_path()
 
-    if name == 'redhat' or name == 'centos':  # pylint: disable=R1714
-        set_bin_files(data_files, dest=agent_bin_path)
+    if name == 'redhat' or name == 'centos' or name == 'almalinux':  # pylint: disable=R1714
+        if version.startswith("8.2"):
+            # redhat8+ default to py3
+            set_bin_files(data_files, dest=agent_bin_path,
+                          src=["bin/py3/waagent", "bin/waagent2.0"])
+        else:
+            set_bin_files(data_files, dest=agent_bin_path)
         set_conf_files(data_files)
         set_logrotate_files(data_files)
         set_udev_files(data_files)
-        if version.startswith("6"):
+        if version.startswith("8.2"):
+            # redhat 8+ uses systemd and python3
+            set_systemd_files(data_files, dest=systemd_dir_path,
+                              src=["init/redhat/waagent.service"])
+        elif version.startswith("6"):
             set_sysv_files(data_files)
         else:
             # redhat7.0+ use systemd
@@ -135,20 +144,23 @@ def get_data_files(name, version, fullname):  # pylint: disable=R0912
         set_systemd_files(data_files, dest=systemd_dir_path,
                           src=["init/mariner/waagent.service"])
     elif name == 'ubuntu':
-        set_bin_files(data_files, dest=agent_bin_path)
         set_conf_files(data_files, src=["config/ubuntu/waagent.conf"])
         set_logrotate_files(data_files)
         set_udev_files(data_files)
         if version.startswith("12") or version.startswith("14"):
             # Ubuntu12.04/14.04 - uses upstart
+            if version.startswith("12"):
+                set_bin_files(data_files, dest=agent_bin_path)
+            else:
+                set_bin_files(data_files, dest=agent_bin_path,
+                              src=["bin/py3/waagent", "bin/waagent2.0"])
             set_files(data_files, dest="/etc/init",
                       src=["init/ubuntu/walinuxagent.conf"])
             set_files(data_files, dest='/etc/default',
                       src=['init/ubuntu/walinuxagent'])
-        elif fullname == 'Snappy Ubuntu Core':
-            set_files(data_files, dest="<TODO>",
-                      src=["init/ubuntu/snappy/walinuxagent.yml"])
         else:
+            set_bin_files(data_files, dest=agent_bin_path,
+                          src=["bin/py3/waagent", "bin/waagent2.0"])
             # Ubuntu15.04+ uses systemd
             set_systemd_files(data_files, dest=systemd_dir_path,
                               src=[
@@ -188,7 +200,8 @@ def get_data_files(name, version, fullname):  # pylint: disable=R0912
         set_conf_files(data_files, src=["config/openbsd/waagent.conf"])
         set_openbsd_rc_files(data_files)
     elif name == 'debian':
-        set_bin_files(data_files, dest=agent_bin_path)
+        set_bin_files(data_files, dest=agent_bin_path,
+                      src=["bin/py3/waagent", "bin/waagent2.0"])
         set_conf_files(data_files, src=["config/debian/waagent.conf"])
         set_logrotate_files(data_files)
         set_udev_files(data_files, dest="/lib/udev/rules.d")


Test patch:
diff --git a/tests/common/mock_cgroup_environment.py b/tests/common/mock_cgroup_environment.py
index 720cba93..c10805f9 100644
--- a/tests/common/mock_cgroup_environment.py
+++ b/tests/common/mock_cgroup_environment.py
@@ -63,6 +63,8 @@ cgroup on /sys/fs/cgroup/blkio type cgroup (rw,nosuid,nodev,noexec,relatime,blki
 
     MockCommand(r"^systemctl daemon-reload", ""),
 
+    MockCommand(r"^systemctl stop ([^\s]+)"),
+
     MockCommand(r"^systemd-run --unit=([^\s]+) --scope ([^\s]+)",
 ''' 
 Running scope as unit: TEST_UNIT.scope
@@ -87,6 +89,7 @@ class UnitFilePaths:
     walinuxagent = "/lib/systemd/system/walinuxagent.service"
     azure = "/lib/systemd/system/azure.slice"
     vmextensions = "/lib/systemd/system/azure-vmextensions.slice"
+    extensionslice = "/lib/systemd/system/azure-vmextensions-Microsoft.CPlat.Extension.slice"
     slice = "/lib/systemd/system/walinuxagent.service.d/10-Slice.conf"
     cpu_accounting = "/lib/systemd/system/walinuxagent.service.d/11-CPUAccounting.conf"
     cpu_quota = "/lib/systemd/system/walinuxagent.service.d/12-CPUQuota.conf"
diff --git a/tests/common/osutil/test_default.py b/tests/common/osutil/test_default.py
index d6eae68f..a736974e 100644
--- a/tests/common/osutil/test_default.py
+++ b/tests/common/osutil/test_default.py
@@ -20,7 +20,6 @@ import os
 import socket
 import subprocess
 import tempfile
-import traceback
 import unittest
 
 import mock
@@ -291,7 +290,7 @@ class TestOSUtil(AgentTestCase):
             try:
                 osutil.DefaultOSUtil().get_first_if()[0]
             except Exception as e:  # pylint: disable=unused-variable
-                print(traceback.format_exc())
+                print(textutil.format_exception(e))
                 exception = True
             self.assertFalse(exception)
 
diff --git a/tests/common/osutil/test_factory.py b/tests/common/osutil/test_factory.py
index f7d46b21..4ab23413 100644
--- a/tests/common/osutil/test_factory.py
+++ b/tests/common/osutil/test_factory.py
@@ -237,6 +237,13 @@ class TestOsUtilFactory(AgentTestCase):
         self.assertTrue(isinstance(ret, RedhatOSUtil))
         self.assertEqual(ret.get_service_name(), "waagent")
 
+        ret = _get_osutil(distro_name="almalinux",
+                          distro_code_name="",
+                          distro_full_name="",
+                          distro_version="8")
+        self.assertTrue(isinstance(ret, RedhatOSUtil))
+        self.assertEqual(ret.get_service_name(), "waagent")
+
     def test_get_osutil_it_should_return_euleros(self):
         ret = _get_osutil(distro_name="euleros",
                           distro_code_name="",
diff --git a/tests/common/osutil/test_freebsd.py b/tests/common/osutil/test_freebsd.py
index 0be36a48..38518360 100644
--- a/tests/common/osutil/test_freebsd.py
+++ b/tests/common/osutil/test_freebsd.py
@@ -15,11 +15,11 @@
 # Requires Python 2.6+ and Openssl 1.0+
 #
 
-import traceback
 import unittest
 
 import azurelinuxagent.common.utils.shellutil as shellutil
 from azurelinuxagent.common.osutil.freebsd import FreeBSDOSUtil
+from azurelinuxagent.common.utils import textutil
 from tests.tools import AgentTestCase, patch
 from .test_default import osutil_get_dhcp_pid_should_return_a_list_of_pids
 
@@ -120,7 +120,7 @@ Destination        Gateway            Flags     Netif Expire
             try:
                 freebsdosutil.get_first_if()[0]
             except Exception as e:  # pylint: disable=unused-variable
-                print(traceback.format_exc())
+                print(textutil.format_exception(e))
                 exception = True  # pylint: disable=unused-variable
 
 if __name__ == '__main__':
diff --git a/tests/common/test_cgroupapi.py b/tests/common/test_cgroupapi.py
index 854f15d9..54df7c93 100644
--- a/tests/common/test_cgroupapi.py
+++ b/tests/common/test_cgroupapi.py
@@ -129,7 +129,7 @@ class SystemdCgroupsApiTestCase(AgentTestCase):
         original_popen = subprocess.Popen
 
         def mock_popen(command, *args, **kwargs):
-            if command.startswith('systemd-run --unit=Microsoft.Compute.TestExtension_1.2.3'):
+            if command.startswith('systemd-run --unit'):
                 command = "echo TEST_OUTPUT"
             return original_popen(command, *args, **kwargs)
 
@@ -139,6 +139,7 @@ class SystemdCgroupsApiTestCase(AgentTestCase):
                     command_output = SystemdCgroupsApi().start_extension_command(
                         extension_name="Microsoft.Compute.TestExtension-1.2.3",
                         command="A_TEST_COMMAND",
+                        cmd_name="test",
                         shell=True,
                         timeout=300,
                         cwd=self.tmp_dir,
@@ -154,6 +155,7 @@ class SystemdCgroupsApiTestCase(AgentTestCase):
             SystemdCgroupsApi().start_extension_command(
                 extension_name="Microsoft.Compute.TestExtension-1.2.3",
                 command="test command",
+                cmd_name="test",
                 shell=False,
                 timeout=300,
                 cwd=self.tmp_dir,
@@ -174,6 +176,7 @@ class SystemdCgroupsApiTestCase(AgentTestCase):
                 SystemdCgroupsApi().start_extension_command(
                     extension_name="Microsoft.Compute.TestExtension-1.2.3",
                     command="the-test-extension-command",
+                    cmd_name="test",
                     timeout=300,
                     shell=True,
                     cwd=self.tmp_dir,
@@ -184,7 +187,7 @@ class SystemdCgroupsApiTestCase(AgentTestCase):
                 extension_calls = [args[0] for (args, _) in popen_patch.call_args_list if "the-test-extension-command" in args[0]]
 
                 self.assertEqual(1, len(extension_calls), "The extension should have been invoked exactly once")
-                self.assertIn("systemd-run --unit=Microsoft.Compute.TestExtension_1.2.3", extension_calls[0], "The extension should have been invoked using systemd")
+                self.assertIn("systemd-run", extension_calls[0], "The extension should have been invoked using systemd")
 
 
 class SystemdCgroupsApiMockedFileSystemTestCase(_MockedFileSystemTestCase):
diff --git a/tests/common/test_cgroupconfigurator.py b/tests/common/test_cgroupconfigurator.py
index 793c40ff..ac71b1ff 100644
--- a/tests/common/test_cgroupconfigurator.py
+++ b/tests/common/test_cgroupconfigurator.py
@@ -180,6 +180,24 @@ cgroup on /sys/fs/cgroup/blkio type cgroup (rw,nosuid,nodev,noexec,relatime,blki
             self.assertTrue(os.path.exists(agent_drop_in_file_slice), "{0} was not created".format(agent_drop_in_file_slice))
             self.assertTrue(os.path.exists(agent_drop_in_file_cpu_accounting), "{0} was not created".format(agent_drop_in_file_cpu_accounting))
 
+    def test_setup_extension_slice_should_create_unit_files(self):
+        with self._get_cgroup_configurator() as configurator:
+            # get the paths to the mocked files
+            extension_slice_unit_file = configurator.mocks.get_mapped_path(UnitFilePaths.extensionslice)
+
+            configurator.setup_extension_slice(extension_name="Microsoft.CPlat.Extension")
+
+            self.assertTrue(os.path.exists(extension_slice_unit_file), "{0} was not created".format(extension_slice_unit_file))
+
+    def test_remove_extension_slice_should_remove_unit_files(self):
+        with self._get_cgroup_configurator() as configurator:
+            # get the paths to the mocked files
+            extension_slice_unit_file = configurator.mocks.get_mapped_path(UnitFilePaths.extensionslice)
+
+            configurator.remove_extension_slice(extension_name="Microsoft.CPlat.Extension")
+
+            self.assertFalse(os.path.exists(extension_slice_unit_file), "{0} should not be present".format(extension_slice_unit_file))
+
     def test_enable_should_raise_cgroups_exception_when_cgroups_are_not_supported(self):
         with self._get_cgroup_configurator(enable=False) as configurator:
             with patch.object(configurator, "supported", return_value=False):
@@ -239,6 +257,7 @@ cgroup on /sys/fs/cgroup/blkio type cgroup (rw,nosuid,nodev,noexec,relatime,blki
                 configurator.start_extension_command(
                     extension_name="Microsoft.Compute.TestExtension-1.2.3",
                     command="date",
+                    cmd_name="test",
                     timeout=300,
                     shell=False,
                     cwd=self.tmp_dir,
@@ -258,6 +277,7 @@ cgroup on /sys/fs/cgroup/blkio type cgroup (rw,nosuid,nodev,noexec,relatime,blki
                 configurator.start_extension_command(
                     extension_name="Microsoft.Compute.TestExtension-1.2.3",
                     command="the-test-extension-command",
+                    cmd_name="test",
                     timeout=300,
                     shell=False,
                     cwd=self.tmp_dir,
@@ -268,7 +288,7 @@ cgroup on /sys/fs/cgroup/blkio type cgroup (rw,nosuid,nodev,noexec,relatime,blki
                 command_calls = [args[0] for (args, _) in popen_patch.call_args_list if "the-test-extension-command" in args[0]]
 
                 self.assertEqual(len(command_calls), 1, "The test command should have been called exactly once [{0}]".format(command_calls))
-                self.assertIn("systemd-run --unit=Microsoft.Compute.TestExtension_1.2.3", command_calls[0], "The extension should have been invoked using systemd")
+                self.assertIn("systemd-run", command_calls[0], "The extension should have been invoked using systemd")
 
     @patch('time.sleep', side_effect=lambda _: mock_sleep())
     def test_start_extension_command_should_start_tracking_the_extension_cgroups(self, _):
@@ -278,6 +298,7 @@ cgroup on /sys/fs/cgroup/blkio type cgroup (rw,nosuid,nodev,noexec,relatime,blki
             configurator.start_extension_command(
                 extension_name="Microsoft.Compute.TestExtension-1.2.3",
                 command="test command",
+                cmd_name="test",
                 timeout=300,
                 shell=False,
                 cwd=self.tmp_dir,
@@ -305,6 +326,7 @@ cgroup on /sys/fs/cgroup/blkio type cgroup (rw,nosuid,nodev,noexec,relatime,blki
                     configurator.start_extension_command(
                         extension_name="Microsoft.Compute.TestExtension-1.2.3",
                         command="test command",
+                        cmd_name="test",
                         timeout=300,
                         shell=False,
                         cwd=self.tmp_dir,
@@ -335,6 +357,7 @@ cgroup on /sys/fs/cgroup/blkio type cgroup (rw,nosuid,nodev,noexec,relatime,blki
                         command_output = configurator.start_extension_command(
                             extension_name="Microsoft.Compute.TestExtension-1.2.3",
                             command=command,
+                            cmd_name="test",
                             timeout=300,
                             shell=True,
                             cwd=self.tmp_dir,
@@ -355,7 +378,7 @@ cgroup on /sys/fs/cgroup/blkio type cgroup (rw,nosuid,nodev,noexec,relatime,blki
                         extension_calls = [args[0] for (args, _) in popen_patch.call_args_list if command in args[0]]
 
                         self.assertEqual(2, len(extension_calls), "The extension should have been invoked exactly twice")
-                        self.assertIn("systemd-run --unit=Microsoft.Compute.TestExtension_1.2.3", extension_calls[0],
+                        self.assertIn("systemd-run", extension_calls[0],
                                       "The first call to the extension should have used systemd")
                         self.assertEqual(command, extension_calls[1],
                                           "The second call to the extension should not have used systemd")
@@ -380,6 +403,7 @@ cgroup on /sys/fs/cgroup/blkio type cgroup (rw,nosuid,nodev,noexec,relatime,blki
                         configurator.start_extension_command(
                             extension_name="Microsoft.Compute.TestExtension-1.2.3",
                             command="echo 'success'",
+                            cmd_name="test",
                             timeout=300,
                             shell=True,
                             cwd=self.tmp_dir,
@@ -391,7 +415,7 @@ cgroup on /sys/fs/cgroup/blkio type cgroup (rw,nosuid,nodev,noexec,relatime,blki
 
                         extension_calls = [args[0] for (args, _) in popen_patch.call_args_list if "echo 'success'" in args[0]]
                         self.assertEqual(2, len(extension_calls), "The extension should have been called twice. Got: {0}".format(extension_calls))
-                        self.assertIn("systemd-run --unit=Microsoft.Compute.TestExtension_1.2.3", extension_calls[0], "The first call to the extension should have used systemd")
+                        self.assertIn("systemd-run", extension_calls[0], "The first call to the extension should have used systemd")
                         self.assertNotIn("systemd-run", extension_calls[1], "The second call to the extension should not have used systemd")
 
                         self.assertEqual(len(CGroupsTelemetry._tracked), 0, "No cgroups should have been created")
@@ -413,6 +437,7 @@ cgroup on /sys/fs/cgroup/blkio type cgroup (rw,nosuid,nodev,noexec,relatime,blki
                         configurator.start_extension_command(
                             extension_name="Microsoft.Compute.TestExtension-1.2.3",
                             command=command,
+                            cmd_name="test",
                             timeout=300,
                             shell=True,
                             cwd=self.tmp_dir,
@@ -423,14 +448,14 @@ cgroup on /sys/fs/cgroup/blkio type cgroup (rw,nosuid,nodev,noexec,relatime,blki
                     extension_calls = [args[0] for (args, _) in popen_patch.call_args_list if command in args[0]]
 
                     self.assertEqual(1, len(extension_calls), "The extension should have been invoked exactly once")
-                    self.assertIn("systemd-run --unit=Microsoft.Compute.TestExtension_1.2.3", extension_calls[0],
+                    self.assertIn("systemd-run", extension_calls[0],
                                   "The first call to the extension should have used systemd")
 
                     self.assertEqual(context_manager.exception.code, ExtensionErrorCodes.PluginUnknownFailure)
                     self.assertIn("Non-zero exit code", ustr(context_manager.exception))
                     # The scope name should appear in the process output since systemd-run was invoked and stderr
                     # wasn't truncated.
-                    self.assertIn("Microsoft.Compute.TestExtension_1.2.3", ustr(context_manager.exception))
+                    self.assertIn("Running scope as unit", ustr(context_manager.exception))
 
     @attr('requires_sudo')
     @patch('time.sleep', side_effect=lambda _: mock_sleep())
@@ -451,6 +476,7 @@ cgroup on /sys/fs/cgroup/blkio type cgroup (rw,nosuid,nodev,noexec,relatime,blki
                         configurator.start_extension_command(
                             extension_name="Microsoft.Compute.TestExtension-1.2.3",
                             command=long_stdout_stderr_command,
+                            cmd_name="test",
                             timeout=300,
                             shell=True,
                             cwd=self.tmp_dir,
@@ -461,14 +487,14 @@ cgroup on /sys/fs/cgroup/blkio type cgroup (rw,nosuid,nodev,noexec,relatime,blki
                     extension_calls = [args[0] for (args, _) in popen_patch.call_args_list if long_stdout_stderr_command in args[0]]
 
                     self.assertEqual(1, len(extension_calls), "The extension should have been invoked exactly once")
-                    self.assertIn("systemd-run --unit=Microsoft.Compute.TestExtension_1.2.3", extension_calls[0],
+                    self.assertIn("systemd-run", extension_calls[0],
                                   "The first call to the extension should have used systemd")
 
                     self.assertEqual(context_manager.exception.code, ExtensionErrorCodes.PluginUnknownFailure)
                     self.assertIn("Non-zero exit code", ustr(context_manager.exception))
                     # stdout and stderr should have been truncated, so the scope name doesn't appear in stderr
                     # even though systemd-run ran
-                    self.assertNotIn("Microsoft.Compute.TestExtension_1.2.3", ustr(context_manager.exception))
+                    self.assertNotIn("Running scope as unit", ustr(context_manager.exception))
 
     @attr('requires_sudo')
     def test_start_extension_command_should_not_use_fallback_option_if_extension_times_out(self, *args):  # pylint: disable=unused-argument
@@ -487,6 +513,7 @@ cgroup on /sys/fs/cgroup/blkio type cgroup (rw,nosuid,nodev,noexec,relatime,blki
                             configurator.start_extension_command(
                                 extension_name="Microsoft.Compute.TestExtension-1.2.3",
                                 command="date",
+                                cmd_name="test",
                                 timeout=300,
                                 shell=True,
                                 cwd=self.tmp_dir,
@@ -518,6 +545,7 @@ cgroup on /sys/fs/cgroup/blkio type cgroup (rw,nosuid,nodev,noexec,relatime,blki
                     process_output = configurator.start_extension_command(
                         extension_name="Microsoft.Compute.TestExtension-1.2.3",
                         command="echo 'very specific test message'",
+                        cmd_name="test",
                         timeout=300,
                         shell=True,
                         cwd=self.tmp_dir,
@@ -609,6 +637,7 @@ exit 0
                                 configurator.start_extension_command(
                                     extension_name="TestExtension",
                                     command="{0} {1} {2}".format(test_script, extension_output, number_of_descendants),
+                                    cmd_name="test",
                                     timeout=30,
                                     shell=True,
                                     cwd=self.tmp_dir,
diff --git a/tests/common/test_conf.py b/tests/common/test_conf.py
index 8f98f3d7..ebc57ffe 100644
--- a/tests/common/test_conf.py
+++ b/tests/common/test_conf.py
@@ -75,6 +75,18 @@ class TestConf(AgentTestCase):
                 os.path.join(data_dir, "test_waagent.conf"),
                 self.conf)
 
+    def test_get_should_return_default_when_key_is_not_found(self):
+        self.assertEqual("The Default Value", self.conf.get("this-key-does-not-exist", "The Default Value"))
+        self.assertEqual("The Default Value", self.conf.get("this-key-does-not-exist", lambda: "The Default Value"))
+
+    def test_get_switch_should_return_default_when_key_is_not_found(self):
+        self.assertEqual(True, self.conf.get_switch("this-key-does-not-exist", True))
+        self.assertEqual(True, self.conf.get_switch("this-key-does-not-exist", lambda: True))
+
+    def test_get_int_should_return_default_when_key_is_not_found(self):
+        self.assertEqual(123456789, self.conf.get_int("this-key-does-not-exist", 123456789))
+        self.assertEqual(123456789, self.conf.get_int("this-key-does-not-exist", lambda: 123456789))
+
     def test_key_value_handling(self):
         self.assertEqual("Value1", self.conf.get("FauxKey1", "Bad"))
         self.assertEqual("Value2 Value2", self.conf.get("FauxKey2", "Bad"))
diff --git a/tests/data/hostgaplugin/ext_conf.xml b/tests/data/hostgaplugin/ext_conf.xml
new file mode 100644
index 00000000..3ec94101
--- /dev/null
+++ b/tests/data/hostgaplugin/ext_conf.xml
@@ -0,0 +1,130 @@
+<?xml version="1.0" encoding="utf-8"?>
+<Extensions version="1.0.0.0" goalStateIncarnation="1"><GuestAgentExtension xmlns:i="http://www.w3.org/2001/XMLSchema-instance">
+  <GAFamilies>
+    <GAFamily>
+      <Name>Prod</Name>
+      <Uris>
+        <Uri>https://zrdfepirv2cbn09pr02a.blob.core.windows.net/7d89d439b79f4452950452399add2c90/Microsoft.OSTCLinuxAgent_Prod_useast2euap_manifest.xml</Uri>
+        <Uri>https://zrdfepirv2cbz06prdstr01a.blob.core.windows.net/7d89d439b79f4452950452399add2c90/Microsoft.OSTCLinuxAgent_Prod_useast2euap_manifest.xml</Uri>
+        <Uri>https://zrdfepirv2cbn04prdstr01a.blob.core.windows.net/7d89d439b79f4452950452399add2c90/Microsoft.OSTCLinuxAgent_Prod_useast2euap_manifest.xml</Uri>
+        <Uri>https://zrdfepirv2cbn06prdstr01a.blob.core.windows.net/7d89d439b79f4452950452399add2c90/Microsoft.OSTCLinuxAgent_Prod_useast2euap_manifest.xml</Uri>
+      </Uris>
+      <Version>2.4.0.1</Version>
+    </GAFamily>
+    <GAFamily>
+      <Name>Test</Name>
+      <Uris>
+        <Uri>https://zrdfepirv2cbn06prdstr01a.blob.core.windows.net/7d89d439b79f4452950452399add2c90/Microsoft.OSTCLinuxAgent_Test_useast2euap_manifest.xml</Uri>
+        <Uri>https://zrdfepirv2cbn09pr02a.blob.core.windows.net/7d89d439b79f4452950452399add2c90/Microsoft.OSTCLinuxAgent_Test_useast2euap_manifest.xml</Uri>
+        <Uri>https://zrdfepirv2cbn04prdstr01a.blob.core.windows.net/7d89d439b79f4452950452399add2c90/Microsoft.OSTCLinuxAgent_Test_useast2euap_manifest.xml</Uri>
+        <Uri>https://zrdfepirv2cbz06prdstr01a.blob.core.windows.net/7d89d439b79f4452950452399add2c90/Microsoft.OSTCLinuxAgent_Test_useast2euap_manifest.xml</Uri>
+      </Uris>
+      <Version>2.4.0.1</Version>
+    </GAFamily>
+  </GAFamilies>
+  <Location>EastUS2EUAP</Location>
+  <ServiceName>CRP</ServiceName>
+</GuestAgentExtension>
+<RequiredFeatures>
+  <RequiredFeature>
+    <Name>MultipleExtensionsPerHandler</Name>
+  </RequiredFeature>
+</RequiredFeatures>
+<StatusUploadBlob statusBlobType="PageBlob">https://md-jcv4sdpnxtrz.z47.blob.storage.azure.net/$system/nam-canary.fc9e847a-29fd-4ca8-a5ae-59e581425ca7.status?sv=2018-03-28&amp;sr=b&amp;sk=system-1&amp;sig=6n%2fhcx6RYxkk2tX1q1J4R9Ls1xJ8%2bQ3WjJn96x%2b%2fnEg%3d&amp;se=9999-01-01T00%3a00%3a00Z&amp;sp=rw</StatusUploadBlob>
+<Plugins>
+  <Plugin name="Microsoft.Azure.Extensions.customScript" version="2.1.3" location="https://zrdfepirv2cbn04prdstr01a.blob.core.windows.net/708cdbbfb9ad4802ac80a012992930b1/Microsoft.Azure.Extensions_CustomScript_useast2euap_manifest.xml" state="enabled" autoUpgrade="true" failoverlocation="https://zrdfepirv2cbn06prdstr01a.blob.core.windows.net/708cdbbfb9ad4802ac80a012992930b1/Microsoft.Azure.Extensions_CustomScript_useast2euap_manifest.xml" runAsStartupTask="false" isJson="true" useExactVersion="true">
+    <additionalLocations>
+      <additionalLocation>https://zrdfepirv2cbn09pr02a.blob.core.windows.net/708cdbbfb9ad4802ac80a012992930b1/Microsoft.Azure.Extensions_CustomScript_useast2euap_manifest.xml</additionalLocation>
+    </additionalLocations>
+  </Plugin>
+  <Plugin name="Microsoft.Azure.Monitor.AzureMonitorLinuxAgent" version="1.9.1" location="https://zrdfepirv2cbn04prdstr01a.blob.core.windows.net/a47f0806d764480a8d989d009c75007d/Microsoft.Azure.Monitor_AzureMonitorLinuxAgent_useast2euap_manifest.xml" state="enabled" autoUpgrade="true" failoverlocation="https://zrdfepirv2cbn06prdstr01a.blob.core.windows.net/a47f0806d764480a8d989d009c75007d/Microsoft.Azure.Monitor_AzureMonitorLinuxAgent_useast2euap_manifest.xml" runAsStartupTask="false" isJson="true" useExactVersion="true">
+    <additionalLocations>
+      <additionalLocation>https://zrdfepirv2cbn09pr02a.blob.core.windows.net/a47f0806d764480a8d989d009c75007d/Microsoft.Azure.Monitor_AzureMonitorLinuxAgent_useast2euap_manifest.xml</additionalLocation>
+    </additionalLocations>
+  </Plugin>
+  <Plugin name="Microsoft.Azure.Security.Monitoring.AzureSecurityLinuxAgent" version="2.15.112" location="https://zrdfepirv2cbn04prdstr01a.blob.core.windows.net/4ef06ad957494df49c807a5334f2b5d2/Microsoft.Azure.Security.Monitoring_AzureSecurityLinuxAgent_useast2euap_manifest.xml" state="enabled" autoUpgrade="true" failoverlocation="https://zrdfepirv2cbz06prdstr01a.blob.core.windows.net/4ef06ad957494df49c807a5334f2b5d2/Microsoft.Azure.Security.Monitoring_AzureSecurityLinuxAgent_useast2euap_manifest.xml" runAsStartupTask="false" isJson="true" useExactVersion="true">
+    <additionalLocations>
+      <additionalLocation>https://zrdfepirv2cbn06prdstr01a.blob.core.windows.net/4ef06ad957494df49c807a5334f2b5d2/Microsoft.Azure.Security.Monitoring_AzureSecurityLinuxAgent_useast2euap_manifest.xml</additionalLocation>
+    </additionalLocations>
+  </Plugin>
+  <Plugin name="Microsoft.CPlat.Core.RunCommandLinux" version="1.0.1" location="https://umsahsv3lrjswx2ggpxc.blob.core.windows.net/4beefb67-5a1e-aa37-7ede-3180ba34ba0d/4beefb67-5a1e-aa37-7ede-3180ba34ba0d_manifest.xml" state="enabled" autoUpgrade="true" failoverlocation="https://umsanjfkjbj4tsg50hbz.blob.core.windows.net/4beefb67-5a1e-aa37-7ede-3180ba34ba0d/4beefb67-5a1e-aa37-7ede-3180ba34ba0d_manifest.xml" runAsStartupTask="false" isJson="true" useExactVersion="true">
+    <additionalLocations>
+      <additionalLocation>https://umsaw3s1v4ftjvfg5qzl.blob.core.windows.net/4beefb67-5a1e-aa37-7ede-3180ba34ba0d/4beefb67-5a1e-aa37-7ede-3180ba34ba0d_manifest.xml</additionalLocation>
+    </additionalLocations>
+  </Plugin>
+</Plugins>
+<PluginSettings>
+  <Plugin name="Microsoft.Azure.Extensions.customScript" version="2.1.3">
+    <RuntimeSettings seqNo="1">{
+  "runtimeSettings": [
+    {
+      "handlerSettings": {
+        "protectedSettingsCertThumbprint": "4C4F304667711036E64AF4894B76EB208A863BD4",
+        "protectedSettings": "MIIBsAYJKoZIhvcNAQcDoIIBoTCCAZ0CAQAxggFpMIIBZQIBADBNMDkxNzA1BgoJkiaJk/IsZAEZFidXaW5kb3dzIEF6dXJlIENSUCBDZXJ0aWZpY2F0ZSBHZW5lcmF0b3ICEFpB/HKM/7evRk+DBz754wUwDQYJKoZIhvcNAQEBBQAEggEADPJwniDeIUXzxNrZCloitFdscQ59Bz1dj9DLBREAiM8jmxM0LLicTJDUv272Qm/4ZQgdqpFYBFjGab/9MX+Ih2x47FkVY1woBkckMaC/QOFv84gbboeQCmJYZC/rZJdh8rCMS+CEPq3uH1PVrvtSdZ9uxnaJ+E4exTPPviIiLIPtqWafNlzdbBt8HZjYaVw+SSe+CGzD2pAQeNttq3Rt/6NjCzrjG8ufKwvRoqnrInMs4x6nnN5/xvobKIBSv4/726usfk8Ug+9Q6Benvfpmre2+1M5PnGTfq78cO3o6mI3cPoBUjp5M0iJjAMGeMt81tyHkimZrEZm6pLa4NQMOEjArBgkqhkiG9w0BBwEwFAYIKoZIhvcNAwcECC5nVaiJaWt+gAhgeYvxUOYHXw=="
+      }
+    }
+  ]
+}</RuntimeSettings>
+  </Plugin>
+  <Plugin name="Microsoft.Azure.Monitor.AzureMonitorLinuxAgent" version="1.9.1">
+    <RuntimeSettings seqNo="0">{
+  "runtimeSettings": [
+    {
+      "handlerSettings": {
+        "protectedSettingsCertThumbprint": "4C4F304667711036E64AF4894B76EB208A863BD4",
+        "protectedSettings": "MIIBsAYJKoZIhvcNAQcDoIIBoTCCAZ0CAQAxggFpMIIBZQIBADBNMDkxNzA1BgoJkiaJk/IsZAEZFidXaW5kb3dzIEF6dXJlIENSUCBDZXJ0aWZpY2F0ZSBHZW5lcmF0b3ICEFpB/HKM/7evRk+DBz754wUwDQYJKoZIhvcNAQEBBQAEggEADPJwniDeIUXzxNrZCloitFdscQ59Bz1dj9DLBREAiM8jmxM0LLicTJDUv272Qm/4ZQgdqpFYBFjGab/9MX+Ih2x47FkVY1woBkckMaC/QOFv84gbboeQCmJYZC/rZJdh8rCMS+CEPq3uH1PVrvtSdZ9uxnaJ+E4exTPPviIiLIPtqWafNlzdbBt8HZjYaVw+SSe+CGzD2pAQeNttq3Rt/6NjCzrjG8ufKwvRoqnrInMs4x6nnN5/xvobKIBSv4/726usfk8Ug+9Q6Benvfpmre2+1M5PnGTfq78cO3o6mI3cPoBUjp5M0iJjAMGeMt81tyHkimZrEZm6pLa4NQMOEjArBgkqhkiG9w0BBwEwFAYIKoZIhvcNAwcECC5nVaiJaWt+gAhgeYvxUOYHXw==",
+        "publicSettings": {"GCS_AUTO_CONFIG":true}
+      }
+    }
+  ]
+}</RuntimeSettings>
+  </Plugin>
+  <Plugin name="Microsoft.Azure.Security.Monitoring.AzureSecurityLinuxAgent" version="2.15.112">
+    <RuntimeSettings seqNo="0">{
+  "runtimeSettings": [
+    {
+      "handlerSettings": {
+        "protectedSettingsCertThumbprint": "4C4F304667711036E64AF4894B76EB208A863BD4",
+        "protectedSettings": "MIIBsAYJKoZIhvcNAQcDoIIBoTCCAZ0CAQAxggFpMIIBZQIBADBNMDkxNzA1BgoJkiaJk/IsZAEZFidXaW5kb3dzIEF6dXJlIENSUCBDZXJ0aWZpY2F0ZSBHZW5lcmF0b3ICEFpB/HKM/7evRk+DBz754wUwDQYJKoZIhvcNAQEBBQAEggEADPJwniDeIUXzxNrZCloitFdscQ59Bz1dj9DLBREAiM8jmxM0LLicTJDUv272Qm/4ZQgdqpFYBFjGab/9MX+Ih2x47FkVY1woBkckMaC/QOFv84gbboeQCmJYZC/rZJdh8rCMS+CEPq3uH1PVrvtSdZ9uxnaJ+E4exTPPviIiLIPtqWafNlzdbBt8HZjYaVw+SSe+CGzD2pAQeNttq3Rt/6NjCzrjG8ufKwvRoqnrInMs4x6nnN5/xvobKIBSv4/726usfk8Ug+9Q6Benvfpmre2+1M5PnGTfq78cO3o6mI3cPoBUjp5M0iJjAMGeMt81tyHkimZrEZm6pLa4NQMOEjArBgkqhkiG9w0BBwEwFAYIKoZIhvcNAwcECC5nVaiJaWt+gAhgeYvxUOYHXw==",
+        "publicSettings": {"enableGenevaUpload":true}
+      }
+    }
+  ]
+}</RuntimeSettings>
+  </Plugin>
+  <Plugin name="Microsoft.CPlat.Core.RunCommandLinux" version="1.0.1">
+    <RuntimeSettings seqNo="2">{
+  "runtimeSettings": [
+    {
+      "handlerSettings": {
+        "protectedSettingsCertThumbprint": "4C4F304667711036E64AF4894B76EB208A863BD4",
+        "protectedSettings": "MIIBsAYJKoZIhvcNAQcDoIIBoTCCAZ0CAQAxggFpMIIBZQIBADBNMDkxNzA1BgoJkiaJk/IsZAEZFidXaW5kb3dzIEF6dXJlIENSUCBDZXJ0aWZpY2F0ZSBHZW5lcmF0b3ICEFpB/HKM/7evRk+DBz754wUwDQYJKoZIhvcNAQEBBQAEggEADPJwniDeIUXzxNrZCloitFdscQ59Bz1dj9DLBREAiM8jmxM0LLicTJDUv272Qm/4ZQgdqpFYBFjGab/9MX+Ih2x47FkVY1woBkckMaC/QOFv84gbboeQCmJYZC/rZJdh8rCMS+CEPq3uH1PVrvtSdZ9uxnaJ+E4exTPPviIiLIPtqWafNlzdbBt8HZjYaVw+SSe+CGzD2pAQeNttq3Rt/6NjCzrjG8ufKwvRoqnrInMs4x6nnN5/xvobKIBSv4/726usfk8Ug+9Q6Benvfpmre2+1M5PnGTfq78cO3o6mI3cPoBUjp5M0iJjAMGeMt81tyHkimZrEZm6pLa4NQMOEjArBgkqhkiG9w0BBwEwFAYIKoZIhvcNAwcECC5nVaiJaWt+gAhgeYvxUOYHXw==",
+        "publicSettings": {}
+      }
+    }
+  ]
+}</RuntimeSettings>
+  </Plugin>
+</PluginSettings>
+<SecureIdentity>
+  <IdentityClientId p2:nil="true" xmlns:p2="http://www.w3.org/2001/XMLSchema-instance" />
+  <IdentityClientTenantId p2:nil="true" xmlns:p2="http://www.w3.org/2001/XMLSchema-instance" />
+  <IdentityObjectId p2:nil="true" xmlns:p2="http://www.w3.org/2001/XMLSchema-instance" />
+  <IdentitySecretUrl>https://control-eastus2euap.identity.azure.net/subscriptions/a53f7094-a16c-47af-abe4-b05c05d0d79a/resourcegroups/narrieta-rg/providers/Microsoft.Compute/virtualMachines/nam-canary/credentials/v2/systemassigned?arpid=2ebf1440-8fc3-413b-abea-249d5af1488b</IdentitySecretUrl>
+  <IdentityInternalId>2ebf1440-8fc3-413b-abea-249d5af1488b</IdentityInternalId>
+  <IdentityTag>f1b49e76-b675-459b-a493-29de3a6fe978</IdentityTag>
+  <UserAssignedIdentities>
+    <UserAssignedIdentity>
+      <IdentityClientId>f40a031a-27f4-4d98-9850-49d58028f3ae</IdentityClientId>
+      <IdentityClientTenantId>72f988bf-86f1-41af-91ab-2d7cd011db47</IdentityClientTenantId>
+      <IdentityObjectId>20152205-d986-4a5e-a1f7-e38876f27dea</IdentityObjectId>
+      <IdentitySecretUrl>https://control-eastus2euap.identity.azure.net/subscriptions/a53f7094-a16c-47af-abe4-b05c05d0d79a/resourcegroups/narrieta-rg/providers/Microsoft.Compute/virtualMachines/nam-canary/credentials/v2/userassigned?arpid=2ebf1440-8fc3-413b-abea-249d5af1488b&amp;uaid=f40a031a-27f4-4d98-9850-49d58028f3ae&amp;tid=72f988bf-86f1-41af-91ab-2d7cd011db47</IdentitySecretUrl>
+      <IdentityResourceId>/subscriptions/a53f7094-a16c-47af-abe4-b05c05d0d79a/resourceGroups/AzSecPackAutoConfigRG/providers/Microsoft.ManagedIdentity/userAssignedIdentities/AzSecPackAutoConfigUA-eastus2euap</IdentityResourceId>
+    </UserAssignedIdentity>
+  </UserAssignedIdentities>
+</SecureIdentity>
+<VaultSecrets>
+  <VaultCertificates />
+</VaultSecrets>
+<InVMArtifactsProfileBlob>https://md-jcv4sdpnxtrz.z47.blob.storage.azure.net/$system/nam-canary.fc9e847a-29fd-4ca8-a5ae-59e581425ca7.vmSettings?sv=2018-03-28&amp;sr=b&amp;sk=system-1&amp;sig=TMVEkKTY3%2bDpMJIjnO90uGaayI1t%2fd3yPYoxAFAjCcQ%3d&amp;se=9999-01-01T00%3a00%3a00Z&amp;sp=r</InVMArtifactsProfileBlob>
+<InVMGoalStateMetaData inSvdSeqNo="20" createdOnTicks="637629365237217593" activityId="b3f493e2-cab8-487d-820a-5718504f2f3f" correlationId="4f4939fb-35e6-411f-98f1-aeaf8796ed30" /></Extensions>
\ No newline at end of file
diff --git a/tests/data/hostgaplugin/vm_settings.json b/tests/data/hostgaplugin/vm_settings.json
new file mode 100644
index 00000000..0742c8c3
--- /dev/null
+++ b/tests/data/hostgaplugin/vm_settings.json
@@ -0,0 +1,108 @@
+{
+    "activityId": "b3f493e2-cab8-487d-820a-5718504f2f3f",
+    "correlationId": "4f4939fb-35e6-411f-98f1-aeaf8796ed30",
+    "extensionsLastModifiedTickCount": 637629365237217593,
+    "inVMMetadata": {
+        "subscriptionId": "a53f7094-a16c-47af-abe4-b05c05d0d79a",
+        "resourceGroupName": "narrieta-rg",
+        "vmName": "nam-canary",
+        "location": "EastUS2EUAP",
+        "vmId": "fc9e847a-29fd-4ca8-a5ae-59e581425ca7",
+        "vmSize": "Standard_DS1_v2",
+        "osType": "Linux",
+        "vmImage": {
+            "publisher": "Canonical",
+            "offer": "UbuntuServer",
+            "sku": "18.04-LTS",
+            "version": "18.04.202104150"
+        }
+    },
+    "requiredFeatures": [
+        {
+            "name": "MultipleExtensionsPerHandler"
+        }
+    ],
+    "gaFamilies": [
+        {
+            "name": "Prod",
+            "version": "2.4.0.1"
+        },
+        {
+            "name": "Test",
+            "version": "2.4.0.1"
+        }
+    ],
+    "extensionGoalStates": [
+        {
+            "name": "Microsoft.Azure.Extensions.customScript",
+            "version": "2.1.3",
+            "location": "https://zrdfepirv2cbn04prdstr01a.blob.core.windows.net/708cdbbfb9ad4802ac80a012992930b1/Microsoft.Azure.Extensions_CustomScript_useast2euap_manifest.xml",
+            "state": "enabled",
+            "autoUpgrade": true,
+            "runAsStartupTask": false,
+            "isJson": true,
+            "useExactVersion": true,
+            "settingsSeqNo": 1,
+            "settings": [
+                {
+                    "protectedSettingsCertThumbprint": "4C4F304667711036E64AF4894B76EB208A863BD4",
+                    "protectedSettings": "MIIBsAYJKoZIhvcNAQcDoIIBoTCCAZ0CAQAxggFpMIIBZQIBADBNMDkxNzA1BgoJkiaJk/IsZAEZFidXaW5kb3dzIEF6dXJlIENSUCBDZXJ0aWZpY2F0ZSBHZW5lcmF0b3ICEFpB/HKM/7evRk+DBz754wUwDQYJKoZIhvcNAQEBBQAEggEADPJwniDeIUXzxNrZCloitFdscQ59Bz1dj9DLBREAiM8jmxM0LLicTJDUv272Qm/4ZQgdqpFYBFjGab/9MX+Ih2x47FkVY1woBkckMaC/QOFv84gbboeQCmJYZC/rZJdh8rCMS+CEPq3uH1PVrvtSdZ9uxnaJ+E4exTPPviIiLIPtqWafNlzdbBt8HZjYaVw+SSe+CGzD2pAQeNttq3Rt/6NjCzrjG8ufKwvRoqnrInMs4x6nnN5/xvobKIBSv4/726usfk8Ug+9Q6Benvfpmre2+1M5PnGTfq78cO3o6mI3cPoBUjp5M0iJjAMGeMt81tyHkimZrEZm6pLa4NQMOEjArBgkqhkiG9w0BBwEwFAYIKoZIhvcNAwcECC5nVaiJaWt+gAhgeYvxUOYHXw=="
+                }
+            ]
+        },
+        {
+            "name": "Microsoft.Azure.Monitor.AzureMonitorLinuxAgent",
+            "version": "1.9.1",
+            "location": "https://zrdfepirv2cbn04prdstr01a.blob.core.windows.net/a47f0806d764480a8d989d009c75007d/Microsoft.Azure.Monitor_AzureMonitorLinuxAgent_useast2euap_manifest.xml",
+            "state": "enabled",
+            "autoUpgrade": true,
+            "runAsStartupTask": false,
+            "isJson": true,
+            "useExactVersion": true,
+            "settingsSeqNo": 0,
+            "settings": [
+                {
+                    "protectedSettingsCertThumbprint": "4C4F304667711036E64AF4894B76EB208A863BD4",
+                    "protectedSettings": "MIIBsAYJKoZIhvcNAQcDoIIBoTCCAZ0CAQAxggFpMIIBZQIBADBNMDkxNzA1BgoJkiaJk/IsZAEZFidXaW5kb3dzIEF6dXJlIENSUCBDZXJ0aWZpY2F0ZSBHZW5lcmF0b3ICEFpB/HKM/7evRk+DBz754wUwDQYJKoZIhvcNAQEBBQAEggEADPJwniDeIUXzxNrZCloitFdscQ59Bz1dj9DLBREAiM8jmxM0LLicTJDUv272Qm/4ZQgdqpFYBFjGab/9MX+Ih2x47FkVY1woBkckMaC/QOFv84gbboeQCmJYZC/rZJdh8rCMS+CEPq3uH1PVrvtSdZ9uxnaJ+E4exTPPviIiLIPtqWafNlzdbBt8HZjYaVw+SSe+CGzD2pAQeNttq3Rt/6NjCzrjG8ufKwvRoqnrInMs4x6nnN5/xvobKIBSv4/726usfk8Ug+9Q6Benvfpmre2+1M5PnGTfq78cO3o6mI3cPoBUjp5M0iJjAMGeMt81tyHkimZrEZm6pLa4NQMOEjArBgkqhkiG9w0BBwEwFAYIKoZIhvcNAwcECC5nVaiJaWt+gAhgeYvxUOYHXw==",
+                    "publicSettings": "{\"GCS_AUTO_CONFIG\":true}"
+                }
+            ]
+        },
+        {
+            "name": "Microsoft.Azure.Security.Monitoring.AzureSecurityLinuxAgent",
+            "version": "2.15.112",
+            "location": "https://zrdfepirv2cbn04prdstr01a.blob.core.windows.net/4ef06ad957494df49c807a5334f2b5d2/Microsoft.Azure.Security.Monitoring_AzureSecurityLinuxAgent_useast2euap_manifest.xml",
+            "state": "enabled",
+            "autoUpgrade": true,
+            "runAsStartupTask": false,
+            "isJson": true,
+            "useExactVersion": true,
+            "settingsSeqNo": 0,
+            "settings": [
+                {
+                    "protectedSettingsCertThumbprint": "4C4F304667711036E64AF4894B76EB208A863BD4",
+                    "protectedSettings": "MIIBsAYJKoZIhvcNAQcDoIIBoTCCAZ0CAQAxggFpMIIBZQIBADBNMDkxNzA1BgoJkiaJk/IsZAEZFidXaW5kb3dzIEF6dXJlIENSUCBDZXJ0aWZpY2F0ZSBHZW5lcmF0b3ICEFpB/HKM/7evRk+DBz754wUwDQYJKoZIhvcNAQEBBQAEggEADPJwniDeIUXzxNrZCloitFdscQ59Bz1dj9DLBREAiM8jmxM0LLicTJDUv272Qm/4ZQgdqpFYBFjGab/9MX+Ih2x47FkVY1woBkckMaC/QOFv84gbboeQCmJYZC/rZJdh8rCMS+CEPq3uH1PVrvtSdZ9uxnaJ+E4exTPPviIiLIPtqWafNlzdbBt8HZjYaVw+SSe+CGzD2pAQeNttq3Rt/6NjCzrjG8ufKwvRoqnrInMs4x6nnN5/xvobKIBSv4/726usfk8Ug+9Q6Benvfpmre2+1M5PnGTfq78cO3o6mI3cPoBUjp5M0iJjAMGeMt81tyHkimZrEZm6pLa4NQMOEjArBgkqhkiG9w0BBwEwFAYIKoZIhvcNAwcECC5nVaiJaWt+gAhgeYvxUOYHXw==",
+                    "publicSettings": "{\"enableGenevaUpload\":true}"
+                }
+            ]
+        },
+        {
+            "name": "Microsoft.CPlat.Core.RunCommandLinux",
+            "version": "1.0.1",
+            "location": "https://umsahsv3lrjswx2ggpxc.blob.core.windows.net/4beefb67-5a1e-aa37-7ede-3180ba34ba0d/4beefb67-5a1e-aa37-7ede-3180ba34ba0d_manifest.xml",
+            "state": "enabled",
+            "autoUpgrade": true,
+            "runAsStartupTask": false,
+            "isJson": true,
+            "useExactVersion": true,
+            "settingsSeqNo": 2,
+            "settings": [
+                {
+                    "protectedSettingsCertThumbprint": "4C4F304667711036E64AF4894B76EB208A863BD4",
+                    "protectedSettings": "MIIBsAYJKoZIhvcNAQcDoIIBoTCCAZ0CAQAxggFpMIIBZQIBADBNMDkxNzA1BgoJkiaJk/IsZAEZFidXaW5kb3dzIEF6dXJlIENSUCBDZXJ0aWZpY2F0ZSBHZW5lcmF0b3ICEFpB/HKM/7evRk+DBz754wUwDQYJKoZIhvcNAQEBBQAEggEADPJwniDeIUXzxNrZCloitFdscQ59Bz1dj9DLBREAiM8jmxM0LLicTJDUv272Qm/4ZQgdqpFYBFjGab/9MX+Ih2x47FkVY1woBkckMaC/QOFv84gbboeQCmJYZC/rZJdh8rCMS+CEPq3uH1PVrvtSdZ9uxnaJ+E4exTPPviIiLIPtqWafNlzdbBt8HZjYaVw+SSe+CGzD2pAQeNttq3Rt/6NjCzrjG8ufKwvRoqnrInMs4x6nnN5/xvobKIBSv4/726usfk8Ug+9Q6Benvfpmre2+1M5PnGTfq78cO3o6mI3cPoBUjp5M0iJjAMGeMt81tyHkimZrEZm6pLa4NQMOEjArBgkqhkiG9w0BBwEwFAYIKoZIhvcNAwcECC5nVaiJaWt+gAhgeYvxUOYHXw==",
+                    "publicSettings": "{}"
+                }
+            ]
+        }
+    ]
+}
\ No newline at end of file
diff --git a/tests/data/test_waagent.conf b/tests/data/test_waagent.conf
index dbdc017d..a386228a 100644
--- a/tests/data/test_waagent.conf
+++ b/tests/data/test_waagent.conf
@@ -68,7 +68,7 @@ ResourceDisk.MountOptions=None
 Logs.Verbose=n
 
 # Enable periodic log collection, default is n
-Logs.Collect=n
+Logs.Collect=y
 
 # How frequently to collect logs, default is each hour
 Logs.CollectPeriod=3600
diff --git a/tests/ga/test_collect_logs.py b/tests/ga/test_collect_logs.py
index 6e4e4e43..2cefd8cc 100644
--- a/tests/ga/test_collect_logs.py
+++ b/tests/ga/test_collect_logs.py
@@ -18,6 +18,7 @@ import contextlib
 import os
 
 from azurelinuxagent.common import logger, conf
+from azurelinuxagent.common.cgroupconfigurator import CGroupConfigurator
 from azurelinuxagent.common.logger import Logger
 from azurelinuxagent.common.protocol.util import ProtocolUtil
 from azurelinuxagent.ga.collect_logs import get_collect_logs_handler, is_log_collection_allowed
@@ -28,7 +29,7 @@ from tests.tools import Mock, MagicMock, patch, AgentTestCase, clear_singleton_i
 
 
 @contextlib.contextmanager
-def _create_collect_logs_handler(iterations=1, systemd_present=True):
+def _create_collect_logs_handler(iterations=1, cgroups_enabled=True, collect_logs_conf=True):
     """
     Creates an instance of CollectLogsHandler that
         * Uses a mock_wire_protocol for network requests,
@@ -46,8 +47,11 @@ def _create_collect_logs_handler(iterations=1, systemd_present=True):
         with patch("azurelinuxagent.ga.collect_logs.get_protocol_util", return_value=protocol_util):
             with patch("azurelinuxagent.ga.collect_logs.CollectLogsHandler.stopped", side_effect=[False] * iterations + [True]):
                 with patch("time.sleep"):
-                    with patch("azurelinuxagent.common.osutil.systemd.is_systemd", return_value=systemd_present):
-                        with patch("azurelinuxagent.ga.collect_logs.conf.get_collect_logs", return_value=True):
+
+                    # Grab the singleton to patch it
+                    cgroups_configurator_singleton = CGroupConfigurator.get_instance()
+                    with patch.object(cgroups_configurator_singleton, "enabled", return_value=cgroups_enabled):
+                        with patch("azurelinuxagent.ga.collect_logs.conf.get_collect_logs", return_value=collect_logs_conf):
                             def run_and_wait():
                                 collect_logs_handler.run()
                                 collect_logs_handler.join()
@@ -93,40 +97,25 @@ class TestCollectLogs(AgentTestCase, HttpRequestPredicates):
     def test_it_should_only_collect_logs_if_conditions_are_met(self):
         # In order to collect logs, three conditions have to be met:
         # 1) the flag must be set to true in the conf file
-        # 2) systemd must be managing services
+        # 2) cgroups must be managing services
         # 3) python version 2.7+ which is automatically true for these tests since they are disabled on py2.6
 
-        # systemd not present, config flag false
-        with _create_collect_logs_handler(systemd_present=False):
-            with patch("azurelinuxagent.ga.collect_logs.conf.get_collect_logs", return_value=False):
-                self.assertEqual(False, is_log_collection_allowed(), "Log collection should not have been enabled")
+        # cgroups not enabled, config flag false
+        with _create_collect_logs_handler(cgroups_enabled=False, collect_logs_conf=False):
+            self.assertEqual(False, is_log_collection_allowed(), "Log collection should not have been enabled")
 
-        # systemd present, config flag false
-        with _create_collect_logs_handler():
-            with patch("azurelinuxagent.ga.collect_logs.conf.get_collect_logs", return_value=False):
-                self.assertEqual(False, is_log_collection_allowed(), "Log collection should not have been enabled")
+        # cgroups enabled, config flag false
+        with _create_collect_logs_handler(cgroups_enabled=True, collect_logs_conf=False):
+            self.assertEqual(False, is_log_collection_allowed(), "Log collection should not have been enabled")
 
-        # systemd not present, config flag true
-        with _create_collect_logs_handler(systemd_present=False):
-            with patch("azurelinuxagent.ga.collect_logs.conf.get_collect_logs", return_value=True):
-                self.assertEqual(False, is_log_collection_allowed(), "Log collection should not have been enabled")
+        # cgroups not enabled, config flag true
+        with _create_collect_logs_handler(cgroups_enabled=False, collect_logs_conf=True):
+            self.assertEqual(False, is_log_collection_allowed(), "Log collection should not have been enabled")
 
-        # systemd present, config flag true
-        with _create_collect_logs_handler():
+        # cgroups enabled, config flag true
+        with _create_collect_logs_handler(cgroups_enabled=True, collect_logs_conf=True):
             self.assertEqual(True, is_log_collection_allowed(), "Log collection should have been enabled")
 
-    def test_log_collection_is_invoked_with_resource_limits(self):
-        with _create_collect_logs_handler() as collect_logs_handler:
-            with patch("azurelinuxagent.ga.collect_logs.shellutil.run_command") as patch_run_command:
-                collect_logs_handler.run_and_wait()
-
-        args, _ = patch_run_command.call_args
-        self.assertIn("systemd-run", args[0], "The log collector should have been invoked with systemd-run")
-        self.assertIn("--property=CPUAccounting=1", args[0], "The log collector should have been invoked with CPUAccounting turned on")
-        self.assertIn("--property=MemoryAccounting=1", args[0], "The log collector should have been invoked with MemoryAccounting turned on")
-        self.assertIn("--property=CPUQuota=5%", args[0], "The log collector should have been invoked with a CPU limit")
-        self.assertIn("--property=MemoryLimit=30M", args[0], "The log collector should have been invoked with a memory limit")
-
     def test_it_uploads_logs_when_collection_is_successful(self):
         archive_size = 42
 
diff --git a/tests/ga/test_extension.py b/tests/ga/test_extension.py
index 29546a1c..e5f110e3 100644
--- a/tests/ga/test_extension.py
+++ b/tests/ga/test_extension.py
@@ -27,16 +27,17 @@ import time
 import unittest
 
 from azurelinuxagent.common import conf
-from azurelinuxagent.common.agent_supported_feature import get_agent_supported_features_list_for_crp, \
-    get_agent_supported_features_list_for_extensions
+from azurelinuxagent.common.agent_supported_feature import get_agent_supported_features_list_for_extensions, \
+    get_agent_supported_features_list_for_crp
 from azurelinuxagent.common.cgroupconfigurator import CGroupConfigurator
 from azurelinuxagent.common.datacontract import get_properties
 from azurelinuxagent.common.event import WALAEventOperation
 from azurelinuxagent.common.utils import fileutil
+from azurelinuxagent.common.utils.archive import StateArchiver
 from azurelinuxagent.common.utils.fileutil import read_file
 from azurelinuxagent.common.utils.flexible_version import FlexibleVersion
 from azurelinuxagent.common.version import PY_VERSION_MAJOR, PY_VERSION_MINOR, PY_VERSION_MICRO, AGENT_NAME, \
-    GOAL_STATE_AGENT_VERSION, CURRENT_VERSION, DISTRO_NAME, DISTRO_VERSION
+    AGENT_VERSION
 from azurelinuxagent.common.exception import ResourceGoneError, ExtensionDownloadError, ProtocolError, \
     ExtensionErrorCodes, ExtensionError, GoalStateAggregateStatusCodes
 from azurelinuxagent.common.protocol.restapi import Extension, ExtHandler, ExtHandlerStatus, \
@@ -54,6 +55,7 @@ from tests.protocol.mockwiredata import DATA_FILE, DATA_FILE_EXT_ADDITIONAL_LOCA
 from tests.tools import AgentTestCase, data_dir, MagicMock, Mock, patch, mock_sleep
 from tests.ga.extension_emulator import Actions, ExtensionCommandNames, extension_emulator, \
     enable_invocations, generate_put_handler
+from tests.utils.test_archive import TestArchive
 
 # Mocking the original sleep to reduce test execution time
 SLEEP = time.sleep
@@ -433,11 +435,11 @@ class TestHandlerStateMigration(AgentTestCase):
             self.assertTrue(False, "Unexpected exception: {0}".format(str(e)))  # pylint: disable=redundant-unittest-assert
         return
 
-
+# Deprecated. New tests should be added to the TestExtension class
 @patch('time.sleep', side_effect=lambda _: mock_sleep(0.001))
 @patch("azurelinuxagent.common.protocol.wire.CryptUtil")
 @patch("azurelinuxagent.common.utils.restutil.http_get")
-class TestExtension(AgentTestCase):
+class TestExtension_Deprecated(AgentTestCase):
     def setUp(self):
         AgentTestCase.setUp(self)
 
@@ -1125,63 +1127,6 @@ class TestExtension(AgentTestCase):
         self.assertEqual(exthandlers_handler.ext_handlers.extHandlers[0].properties.extensions[0].dependencyLevel, 0)
         self.assertEqual(exthandlers_handler.ext_handlers.extHandlers[0].properties.extensions[0].dependencyLevel, 0)
 
-    @patch('time.gmtime', MagicMock(return_value=time.gmtime(0)))
-    def test_ext_handler_reporting_status_file(self, *args):
-
-        expected_status = {
-            "agent_name": AGENT_NAME,
-            "current_version": str(CURRENT_VERSION),
-            "goal_state_version": str(GOAL_STATE_AGENT_VERSION),
-            "distro_details": "{0}:{1}".format(DISTRO_NAME, DISTRO_VERSION),
-            "last_successful_status_upload_time": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
-            "python_version": "Python: {0}.{1}.{2}".format(PY_VERSION_MAJOR, PY_VERSION_MINOR, PY_VERSION_MICRO),
-            "crp_supported_features": [name for name, _ in get_agent_supported_features_list_for_crp().items()],
-            "extension_supported_features": [name for name, _ in get_agent_supported_features_list_for_extensions().items()],
-            "extensions_status": [
-                {
-                    "name": "OSTCExtensions.ExampleHandlerLinux",
-                    "version": "1.0.0",
-                    "status": "Ready",
-                    "supports_multi_config": False
-                },
-                {
-                    "name": "Microsoft.Powershell.ExampleExtension",
-                    "version": "1.0.0",
-                    "status": "Ready",
-                    "supports_multi_config": False
-                },
-                {
-                    "name": "Microsoft.EnterpriseCloud.Monitoring.ExampleHandlerLinux",
-                    "version": "1.0.0",
-                    "status": "Ready",
-                    "supports_multi_config": False
-                },
-                {
-                    "name": "Microsoft.CPlat.Core.ExampleExtensionLinux",
-                    "version": "1.0.0",
-                    "status": "Ready",
-                    "supports_multi_config": False
-                },
-                {
-                    "name": "Microsoft.OSTCExtensions.Edp.ExampleExtensionLinuxInTest",
-                    "version": "1.0.0",
-                    "status": "Ready",
-                    "supports_multi_config": False
-                }
-            ]
-        }
-
-        test_data = mockwiredata.WireProtocolData(mockwiredata.DATA_FILE_MULTIPLE_EXT)
-        exthandlers_handler, protocol = self._create_mock(test_data, *args)  # pylint: disable=unused-variable,no-value-for-parameter
-        exthandlers_handler.run()
-        exthandlers_handler.report_ext_handlers_status()
-
-
-        status_path = os.path.join(conf.get_lib_dir(), AGENT_STATUS_FILE)
-        actual_status_json = json.loads(fileutil.read_file(status_path))
-
-        self.assertEqual(expected_status, actual_status_json)
-
     def test_ext_handler_rollingupgrade(self, *args):
         # Test enable scenario.
         test_data = mockwiredata.WireProtocolData(mockwiredata.DATA_FILE_EXT_ROLLINGUPGRADE)
@@ -1622,7 +1567,9 @@ class TestExtension(AgentTestCase):
         exthandlers_handler.report_ext_handlers_status()
 
         self._assert_handler_status(protocol.report_vm_status, "Ready", 1, "1.0.0")
-        self._assert_ext_status(protocol.report_vm_status, ValidHandlerStatus.error, 0)
+        self._assert_ext_status(protocol.report_vm_status, ValidHandlerStatus.transitioning, 0,
+                                expected_msg="This status is being reported by the Guest Agent since no status "
+                                             "file was reported by extension OSTCExtensions.ExampleHandlerLinux")
 
     def test_wait_for_handler_completion_no_status(self, mock_http_get, mock_crypt_util, *args):
         """
@@ -1649,14 +1596,14 @@ class TestExtension(AgentTestCase):
                 exthandlers_handler.run()
                 exthandlers_handler.report_ext_handlers_status()
 
-
                 # The Handler Status for the base extension should be ready as it was executed successfully by the agent
                 self._assert_handler_status(protocol.report_vm_status, "Ready", 1, "1.0.0",
                                             expected_handler_name="OSTCExtensions.OtherExampleHandlerLinux")
-                # The extension status reported by the Handler should be an error since no status file was found
-                self._assert_ext_status(protocol.report_vm_status, ValidHandlerStatus.error, 0,
+                # The extension status reported by the Handler should be transitioning since no status file was found
+                self._assert_ext_status(protocol.report_vm_status, ValidHandlerStatus.transitioning, 0,
                                         expected_handler_name="OSTCExtensions.OtherExampleHandlerLinux",
-                                        expected_msg="No such file or directory: '{0}'".format(mock_popen.deleted_status_file))
+                                        expected_msg="This status is being reported by the Guest Agent since no status "
+                                                     "file was reported by extension OSTCExtensions.OtherExampleHandlerLinux")
 
                 # The Handler Status for the dependent extension should be NotReady as it was not executed at all
                 # And since it was not executed, it should not report any extension status either
@@ -3190,23 +3137,30 @@ class TestExtensionUpdateOnFailure(AgentTestCase):
             "UninstallAction's return code in installAction's env should be 0.")
 
 
-@patch('time.sleep', side_effect=lambda _: mock_sleep(0.001))
 class TestCollectExtensionStatus(AgentTestCase):
     def setUp(self):
         AgentTestCase.setUp(self)
         self.lib_dir = tempfile.mkdtemp()
+        self.mock_sleep = patch("time.sleep", lambda *_: mock_sleep(0.001))
+        self.mock_sleep.start()
 
-    def _setup_extension_for_validating_collect_ext_status(self, mock_lib_dir, status_file, *args):  # pylint: disable=unused-argument
+    def tearDown(self):
+        self.mock_sleep.stop()
+        AgentTestCase.tearDown(self)
+
+    def _setup_extension_for_validating_collect_ext_status(self, mock_lib_dir, status_file=None):
         handler_name = "TestHandler"
         handler_version = "1.0.0"
         mock_lib_dir.return_value = self.lib_dir
         fileutil.mkdir(os.path.join(self.lib_dir, handler_name + "-" + handler_version, "config"))
         fileutil.mkdir(os.path.join(self.lib_dir, handler_name + "-" + handler_version, "status"))
-        shutil.copy(os.path.join(data_dir, "ext", status_file),
-                    os.path.join(self.lib_dir, handler_name + "-" + handler_version, "status", "0.status"))
         shutil.copy(tempfile.mkstemp(prefix="test-file")[1],
                     os.path.join(self.lib_dir, handler_name + "-" + handler_version, "config", "0.settings"))
 
+        if status_file is not None:
+            shutil.copy(os.path.join(data_dir, "ext", status_file),
+                        os.path.join(self.lib_dir, handler_name + "-" + handler_version, "status", "0.status"))
+
         with mock_wire_protocol(DATA_FILE) as protocol:
             exthandler = ExtHandler(name=handler_name)
             exthandler.properties.version = handler_version
@@ -3216,13 +3170,13 @@ class TestCollectExtensionStatus(AgentTestCase):
             return ExtHandlerInstance(exthandler, protocol), extension
 
     @patch("azurelinuxagent.common.conf.get_lib_dir")
-    def test_collect_ext_status(self, mock_lib_dir, *args):
+    def test_collect_ext_status(self, mock_lib_dir):
         """
         This test validates that collect_ext_status correctly picks up the status file (sample-status.json) and then
         parses it correctly.
         """
         ext_handler_i, extension = self._setup_extension_for_validating_collect_ext_status(mock_lib_dir,
-                                                                                           "sample-status.json", *args)
+                                                                                           "sample-status.json")
         ext_status = ext_handler_i.collect_ext_status(extension)
 
         self.assertEqual(ext_status.code, SUCCESS_CODE_FROM_STATUS_FILE)
@@ -3240,14 +3194,14 @@ class TestCollectExtensionStatus(AgentTestCase):
         self.assertEqual(sub_status.status, ValidHandlerStatus.success)
 
     @patch("azurelinuxagent.common.conf.get_lib_dir")
-    def test_collect_ext_status_for_invalid_json(self, mock_lib_dir, *args):
+    def test_collect_ext_status_for_invalid_json(self, mock_lib_dir):
         """
         This test validates that collect_ext_status correctly picks up the status file (sample-status-invalid-json-format.json)
         and then since the Json cannot be parsed correctly it extension status message should include 2000 bytes of status file
         and the line number in which it failed to parse. The uniqueMachineId tag comes from status file.
         """
         ext_handler_i, extension = self._setup_extension_for_validating_collect_ext_status(mock_lib_dir,
-                                                                                           "sample-status-invalid-json-format.json", *args)
+                                                                                           "sample-status-invalid-json-format.json")
         ext_status = ext_handler_i.collect_ext_status(extension)
 
         self.assertEqual(ext_status.code, ExtensionErrorCodes.PluginSettingsStatusInvalid)
@@ -3257,15 +3211,15 @@ class TestCollectExtensionStatus(AgentTestCase):
         self.assertRegex(ext_status.message, r".*The status reported by the extension TestHandler-1.0.0\(Sequence number 0\), "
                                              r"was in an incorrect format and the agent could not parse it correctly."
                                              r" Failed due to.*")
-        self.assertIn("\"uniqueMachineId\": \"e5e5602b-48a6-4c35-9f96-752043777af1\"",ext_status.message)
+        self.assertIn("\"uniqueMachineId\": \"e5e5602b-48a6-4c35-9f96-752043777af1\"", ext_status.message)
         self.assertEqual(ext_status.status, ValidHandlerStatus.error)
         self.assertEqual(len(ext_status.substatusList), 0)
 
     @patch("azurelinuxagent.common.conf.get_lib_dir")
-    def test_it_should_collect_ext_status_even_when_config_dir_deleted(self, mock_lib_dir, *args):
+    def test_it_should_collect_ext_status_even_when_config_dir_deleted(self, mock_lib_dir):
 
         ext_handler_i, extension = self._setup_extension_for_validating_collect_ext_status(mock_lib_dir,
-                                                                                           "sample-status.json", *args)
+                                                                                           "sample-status.json")
 
         shutil.rmtree(ext_handler_i.get_conf_dir(), ignore_errors=True)
         ext_status = ext_handler_i.collect_ext_status(extension)
@@ -3279,14 +3233,13 @@ class TestCollectExtensionStatus(AgentTestCase):
         self.assertEqual(ext_status.status, ValidHandlerStatus.success)
 
     @patch("azurelinuxagent.common.conf.get_lib_dir")
-    def test_collect_ext_status_very_large_status_message(self, mock_lib_dir, *args):
+    def test_collect_ext_status_very_large_status_message(self, mock_lib_dir):
         """
         Testing collect_ext_status() with a very large status file (>128K) to see if it correctly parses the status
         without generating a really large message.
         """
         ext_handler_i, extension = self._setup_extension_for_validating_collect_ext_status(mock_lib_dir,
-                                                                                           "sample-status-very-large.json",
-                                                                                           *args)
+                                                                                           "sample-status-very-large.json")
         ext_status = ext_handler_i.collect_ext_status(extension)
 
         self.assertEqual(ext_status.code, SUCCESS_CODE_FROM_STATUS_FILE)
@@ -3308,14 +3261,14 @@ class TestCollectExtensionStatus(AgentTestCase):
             self.assertEqual(sub_status.status, ValidHandlerStatus.success)
 
     @patch("azurelinuxagent.common.conf.get_lib_dir")
-    def test_collect_ext_status_very_large_status_file_with_multiple_substatus_nodes(self, mock_lib_dir, *args):
+    def test_collect_ext_status_very_large_status_file_with_multiple_substatus_nodes(self, mock_lib_dir):
         """
         Testing collect_ext_status() with a very large status file (>128K) to see if it correctly parses the status
         without generating a really large message. This checks if the multiple substatus messages are correctly parsed
         and truncated.
         """
         ext_handler_i, extension = self._setup_extension_for_validating_collect_ext_status(
-            mock_lib_dir, "sample-status-very-large-multiple-substatuses.json", *args)  # ~470K bytes.
+            mock_lib_dir, "sample-status-very-large-multiple-substatuses.json")  # ~470K bytes.
         ext_status = ext_handler_i.collect_ext_status(extension)
 
         self.assertEqual(ext_status.code, SUCCESS_CODE_FROM_STATUS_FILE)
@@ -3335,25 +3288,24 @@ class TestCollectExtensionStatus(AgentTestCase):
             self.assertEqual(ValidHandlerStatus.success, sub_status.status)
 
     @patch("azurelinuxagent.common.conf.get_lib_dir")
-    def test_collect_ext_status_read_file_read_exceptions(self, mock_lib_dir, *args):
+    def test_collect_ext_status_read_file_read_exceptions(self, mock_lib_dir):
         """
         Testing collect_ext_status to validate the readfile exceptions.
         """
-        ext_handler_i, extension = self._setup_extension_for_validating_collect_ext_status(mock_lib_dir,
-                                                                                           "sample-status.json", *args)
+        ext_handler_i, extension = self._setup_extension_for_validating_collect_ext_status(mock_lib_dir, "sample-status.json")
         original_read_file = read_file
 
-        def mock_read_file(file, *args, **kwargs):  # pylint: disable=redefined-builtin
+        def mock_read_file(file_, *args, **kwargs):
             expected_status_file_path = os.path.join(self.lib_dir,
                                                      ext_handler_i.ext_handler.name + "-" +
                                                      ext_handler_i.ext_handler. properties.version,
                                                      "status", "0.status")
-            if file == expected_status_file_path:
+            if file_ == expected_status_file_path:
                 raise IOError("No such file or directory: {0}".format(expected_status_file_path))
             else:
-                original_read_file(file, *args, **kwargs)
+                original_read_file(file_, *args, **kwargs)
 
-        with patch('azurelinuxagent.common.utils.fileutil.read_file', mock_read_file) as patch_read_file:  # pylint: disable=unused-variable
+        with patch('azurelinuxagent.common.utils.fileutil.read_file', mock_read_file):
             ext_status = ext_handler_i.collect_ext_status(extension)
 
             self.assertEqual(ext_status.code, ExtensionErrorCodes.PluginUnknownFailure)
@@ -3367,12 +3319,12 @@ class TestCollectExtensionStatus(AgentTestCase):
             self.assertEqual(len(ext_status.substatusList), 0)
 
     @patch("azurelinuxagent.common.conf.get_lib_dir")
-    def test_collect_ext_status_json_exceptions(self, mock_lib_dir, *args):
+    def test_collect_ext_status_json_exceptions(self, mock_lib_dir):
         """
         Testing collect_ext_status() with a malformed json status file.
         """
         ext_handler_i, extension = self._setup_extension_for_validating_collect_ext_status(mock_lib_dir,
-                                        "sample-status-invalid-format-emptykey-line7.json", *args)
+                                        "sample-status-invalid-format-emptykey-line7.json")
         ext_status = ext_handler_i.collect_ext_status(extension)
 
         self.assertEqual(ext_status.code, ExtensionErrorCodes.PluginSettingsStatusInvalid)
@@ -3387,12 +3339,12 @@ class TestCollectExtensionStatus(AgentTestCase):
         self.assertEqual(len(ext_status.substatusList), 0)
 
     @patch("azurelinuxagent.common.conf.get_lib_dir")
-    def test_collect_ext_status_parse_ext_status_exceptions(self, mock_lib_dir, *args):
+    def test_collect_ext_status_parse_ext_status_exceptions(self, mock_lib_dir):
         """
         Testing collect_ext_status() with a malformed json status file.
         """
         ext_handler_i, extension = self._setup_extension_for_validating_collect_ext_status(mock_lib_dir,
-                                        "sample-status-invalid-status-no-status-status-key.json", *args)
+                                        "sample-status-invalid-status-no-status-status-key.json")
         ext_status = ext_handler_i.collect_ext_status(extension)
 
         self.assertEqual(ext_status.code, ExtensionErrorCodes.PluginSettingsStatusInvalid)
@@ -3404,6 +3356,24 @@ class TestCollectExtensionStatus(AgentTestCase):
         self.assertEqual(ext_status.status, ValidHandlerStatus.error)
         self.assertEqual(len(ext_status.substatusList), 0)
 
+    @patch("azurelinuxagent.common.conf.get_lib_dir")
+    def test_it_should_report_transitioning_if_status_file_not_found(self, mock_lib_dir):
+        """
+        Testing collect_ext_status() with a missing status file.
+        """
+        ext_handler_i, extension = self._setup_extension_for_validating_collect_ext_status(mock_lib_dir)
+        ext_status = ext_handler_i.collect_ext_status(extension)
+
+        self.assertEqual(ext_status.code, ExtensionErrorCodes.PluginSuccess)
+        self.assertEqual(ext_status.configurationAppliedTime, None)
+        self.assertEqual(ext_status.operation, None)
+        self.assertEqual(ext_status.sequenceNumber, 0)
+        self.assertIn("This status is being reported by the Guest Agent since no status file was reported by extension {0}".
+                      format("TestHandler"), ext_status.message)
+        self.assertEqual(ext_status.status, ValidHandlerStatus.transitioning)
+        self.assertEqual(len(ext_status.substatusList), 0)
+
+
 class TestAdditionalLocationsExtensions(AgentTestCase):
 
     def setUp(self):
@@ -3491,5 +3461,159 @@ class TestAdditionalLocationsExtensions(AgentTestCase):
                     timeout_in_minutes=0, timeout_in_ms=200)
 
 
+# New test cases should be added here.This class uses mock_wire_protocol
+class TestExtension(AgentTestCase):
+
+    def setUp(self):
+        AgentTestCase.setUp(self)
+
+    def tearDown(self):
+        AgentTestCase.tearDown(self)
+
+    @patch('time.gmtime', MagicMock(return_value=time.gmtime(0)))
+    def test_ext_handler_reporting_status_file(self):
+        with mock_wire_protocol(mockwiredata.DATA_FILE) as protocol:
+
+            def mock_http_put(url, *args, **_):
+                if HttpRequestPredicates.is_host_plugin_status_request(url):
+                    # Skip reading the HostGA request data as its encoded
+                    return MockHttpResponse(status=500)
+                protocol.aggregate_status = json.loads(args[0])
+                return MockHttpResponse(status=201)
+
+            protocol.aggregate_status = None
+            protocol.set_http_handlers(http_put_handler=mock_http_put)
+            exthandlers_handler = get_exthandlers_handler(protocol)
+
+            # creating supported features list that is sent to crp
+            supported_features = []
+            for _, feature in get_agent_supported_features_list_for_crp().items():
+                supported_features.append(
+                    {
+                        "Key": feature.name,
+                        "Value": feature.version
+                    }
+                )
+
+            expected_status = {
+                "version": "1.1",
+                "timestampUTC": "1970-01-01T00:00:00Z",
+                "aggregateStatus": {
+                    "guestAgentStatus": {
+                        "version": AGENT_VERSION,
+                        "status": "Ready",
+                        "formattedMessage": {
+                            "lang": "en-US",
+                            "message": "Guest Agent is running"
+                        }
+                    },
+                    "handlerAggregateStatus": [
+                        {
+                            "handlerVersion": "1.0.0",
+                            "handlerName": "OSTCExtensions.ExampleHandlerLinux",
+                            "status": "Ready",
+                            "code": 0,
+                            "useExactVersion": True,
+                            "formattedMessage": {
+                                "lang": "en-US",
+                                "message": "Plugin enabled"
+                            },
+                            "runtimeSettingsStatus": {
+                                "settingsStatus": {
+                                    "status": {
+                                        "name": "OSTCExtensions.ExampleHandlerLinux",
+                                        "configurationAppliedTime": None,
+                                        "operation": None,
+                                        "status": "success",
+                                        "code": 0
+                                    },
+                                    "version": 1,
+                                    "timestampUTC": "1970-01-01T00:00:00Z"
+                                },
+                                "sequenceNumber": 0
+                            },
+                            "supportsMultiConfig": False
+                        }
+                    ],
+                    "vmArtifactsAggregateStatus": {
+                        "goalStateAggregateStatus": {
+                            "formattedMessage": {
+                                "lang": "en-US",
+                                "message": "GoalState executed successfully"
+                            },
+                            "timestampUTC": "1970-01-01T00:00:00Z",
+                            "inSvdSeqNo": "1",
+                            "status": "Success",
+                            "code": 0
+                        }
+                    }
+                },
+                "supportedFeatures": supported_features,
+                "_metadataNotSentToCRP": {
+                    "agentName": AGENT_NAME,
+                    "daemonVersion": "0.0.0.0",
+                    "pythonVersion": "Python: {0}.{1}.{2}".format(PY_VERSION_MAJOR, PY_VERSION_MINOR, PY_VERSION_MICRO),
+                    "extensionSupportedFeatures": [name for name, _ in
+                                                   get_agent_supported_features_list_for_extensions().items()]
+
+                }
+
+            }
+
+            exthandlers_handler.run()
+            exthandlers_handler.report_ext_handlers_status()
+
+            status_path = os.path.join(conf.get_lib_dir(), AGENT_STATUS_FILE.format(1))
+            actual_status_json = json.loads(fileutil.read_file(status_path))
+
+            # Popping run time attributes
+            actual_status_json.pop('guestOSInfo', None)
+
+            self.assertEqual(expected_status, actual_status_json)
+
+    def test_it_should_zip_waagent_status_when_incarnation_changes(self):
+        with mock_wire_protocol(mockwiredata.DATA_FILE) as protocol:
+
+            # This test checks when the incarnation changes the waagent_status file for the previous incarnation
+            # is added into the history folder for the previous incarnation and gets zipped
+
+            exthandlers_handler = get_exthandlers_handler(protocol)
+
+            temp_files = [
+                'ExtensionsConfig.1.xml',
+                'GoalState.1.xml',
+                'OSTCExtensions.ExampleHandlerLinux.1.manifest.xml',
+                'waagent_status.1.json'
+            ]
+
+            exthandlers_handler.run()
+            exthandlers_handler.report_ext_handlers_status()
+
+            # Updating incarnation to 2 , hence the history folder should have waaagent_status.1.json added under
+            # incarnation 1
+            protocol.mock_wire_data.set_incarnation(2)
+            protocol.update_goal_state()
+
+            test_subject = StateArchiver(self.tmp_dir)
+            test_subject.archive()
+
+            timestamp_zips = os.listdir(os.path.join(self.tmp_dir, "history"))
+            self.assertEqual(1, len(timestamp_zips), "Expected number of zips in history is 1 for"
+                                                     " incarnation 1(previous incarnation)")
+
+            zip_fn = timestamp_zips[0]
+            zip_fullname = os.path.join(self.tmp_dir, "history", zip_fn)
+            self.assertEqual(TestArchive.assert_zip_contains(zip_fullname, temp_files), None)
+            exthandlers_handler.run()
+            exthandlers_handler.report_ext_handlers_status()
+
+            # Updating incarnation to 3 , hence the history folder should have 2 zips files corresponding to incarnation
+            # 1 and 2
+            protocol.mock_wire_data.set_incarnation(3)
+            protocol.update_goal_state()
+            test_subject.archive()
+            self.assertEqual(2, len(os.listdir(os.path.join(self.tmp_dir, "history"))))
+
+
 if __name__ == '__main__':
     unittest.main()
diff --git a/tests/ga/test_multi_config_extension.py b/tests/ga/test_multi_config_extension.py
index 274fe4b1..2c787f31 100644
--- a/tests/ga/test_multi_config_extension.py
+++ b/tests/ga/test_multi_config_extension.py
@@ -8,10 +8,10 @@ from azurelinuxagent.common import conf
 from azurelinuxagent.common.event import WALAEventOperation
 from azurelinuxagent.common.exception import GoalStateAggregateStatusCodes
 from azurelinuxagent.common.future import ustr
-from azurelinuxagent.common.protocol.restapi import ExtHandlerRequestedState, ExtensionState, ExtensionStatus
+from azurelinuxagent.common.protocol.restapi import ExtHandlerRequestedState, ExtensionState
 from azurelinuxagent.common.utils import fileutil
 from azurelinuxagent.ga.exthandlers import get_exthandlers_handler, ValidHandlerStatus, ExtCommandEnvVariable, \
-    parse_ext_status, GoalStateStatus, ExtHandlerInstance
+    GoalStateStatus, ExtHandlerInstance
 from tests.ga.extension_emulator import enable_invocations, extension_emulator, ExtensionCommandNames, Actions, \
     extract_extension_info_from_command
 from tests.protocol.mocks import mock_wire_protocol, HttpRequestPredicates, MockHttpResponse
@@ -894,29 +894,12 @@ class TestMultiConfigExtensions(_MultiConfigBaseTestClass):
                         fail_code in kwargs['message'] for args, kwargs in patch_report_event.call_args_list if
                         kwargs['name'] == third_ext.name), "Error not reported")
 
-    def test_it_should_always_create_placeholder_for_multi_config_extensions(self):
+    def test_it_should_report_transitioning_if_status_file_not_found(self):
         original_popen = subprocess.Popen
-        handler_statuses = {}
-
-        def __assert_status_file_in_handlers():
-            for handler in mc_handlers:
-                file_name = "{0}.{1}.status".format(handler['runtimeSettingsStatus']['extensionName'],
-                                                    handler['runtimeSettingsStatus']['sequenceNumber'])
-                __assert_status_file(handler['handlerName'], status_file=file_name)
-
-        def __assert_status_file(handler_name, status_file):
-            status = handler_statuses["{0}.{1}.enable".format(handler_name, status_file)]
-            self.assertIsNotNone(status, "No status found")
-            # Assert the format of the placeholder is correct
-            ext_status = ExtensionStatus()
-            # If the format is wrong or unexpected, this would throw and fail the test
-            parse_ext_status(ext_status, status)
-            self.assertIn(ext_status.status, [ValidHandlerStatus.success, ValidHandlerStatus.transitioning],
-                          "Incorrect status")
 
         def mock_popen(cmd, *_, **kwargs):
             if 'env' in kwargs:
-                handler_name, handler_version, command_name = extract_extension_info_from_command(cmd)
+                handler_name, handler_version, __ = extract_extension_info_from_command(cmd)
                 ext_name = None
                 if ExtCommandEnvVariable.ExtensionName in kwargs['env']:
                     ext_name = kwargs['env'][ExtCommandEnvVariable.ExtensionName]
@@ -924,34 +907,43 @@ class TestMultiConfigExtensions(_MultiConfigBaseTestClass):
                 status_file_name = "{0}.status".format(seq_no)
                 status_file_name = "{0}.{1}".format(ext_name, status_file_name) if ext_name is not None else status_file_name
                 status_file = os.path.join(self.tmp_dir, "{0}-{1}".format(handler_name, handler_version), "status", status_file_name)
-                contents = None
                 if os.path.exists(status_file):
-                    contents = json.loads(fileutil.read_file(status_file))
-                handler_statuses["{0}.{1}.{2}".format(handler_name, status_file_name, command_name)] = contents
+                    os.remove(status_file)
 
-            return original_popen(cmd, *_, **kwargs)
+            return original_popen("echo " + cmd, *_, **kwargs)
 
         self.test_data['ext_conf'] = os.path.join(self._MULTI_CONFIG_TEST_DATA,
                                                   "ext_conf_multi_config_no_dependencies.xml")
         with self._setup_test_env(mock_manifest=True) as (exthandlers_handler, protocol, no_of_extensions):
             with patch('azurelinuxagent.common.cgroupapi.subprocess.Popen', side_effect=mock_popen):
-                mc_handlers, _ = self.__run_and_assert_generic_case(exthandlers_handler, protocol,
-                                                                             no_of_extensions)
-
-                # Ensure we dont create a placeholder for Install command
-                self.assertTrue(
-                    all(handler_statuses[status] is None for status in handler_statuses if "install" in status),
-                    "Incorrect status file found for install")
-
-                # Ensure we create a valid status file for Enable
-                # Note: As part of our test, the sample-ext creates a status file after install due to which a placeholder
-                # is not created. We will verify a valid status file exists for all extensions instead since that's the
-                # main scenario.
-                __assert_status_file_in_handlers()
+                exthandlers_handler.run()
+                exthandlers_handler.report_ext_handlers_status()
+                self.assertEqual(no_of_extensions,
+                                 len(protocol.aggregate_status['aggregateStatus']['handlerAggregateStatus']),
+                                 "incorrect extensions reported")
+                mc_handlers = self._assert_and_get_handler_status(aggregate_status=protocol.aggregate_status,
+                                                                  handler_name="OSTCExtensions.ExampleHandlerLinux",
+                                                                  expected_count=3)
+                agent_status_message = "This status is being reported by the Guest Agent since no status file was " \
+                                       "reported by extension {0}: " \
+                                       "[ExtensionStatusError] Status file"
+                expected_extensions = {
+                    "firstExtension": {"status": ValidHandlerStatus.transitioning, "seq_no": 1,
+                                       "message": agent_status_message.format("OSTCExtensions.ExampleHandlerLinux.firstExtension")},
+                    "secondExtension": {"status": ValidHandlerStatus.transitioning, "seq_no": 2,
+                                        "message": agent_status_message.format("OSTCExtensions.ExampleHandlerLinux.secondExtension")},
+                    "thirdExtension": {"status": ValidHandlerStatus.transitioning, "seq_no": 3,
+                                       "message": agent_status_message.format("OSTCExtensions.ExampleHandlerLinux.thirdExtension")},
+                }
+                self._assert_extension_status(mc_handlers[:], expected_extensions, multi_config=True)
 
-                # Update GS, remove 2 extensions and add 3
-                mc_handlers, _ = self.__setup_and_assert_disable_scenario(exthandlers_handler, protocol)
-                __assert_status_file_in_handlers()
+                sc_handler = self._assert_and_get_handler_status(aggregate_status=protocol.aggregate_status,
+                                                                 handler_name="Microsoft.Powershell.ExampleExtension")
+                expected_extensions = {
+                    "Microsoft.Powershell.ExampleExtension": {"status": ValidHandlerStatus.transitioning, "seq_no": 9,
+                                                              "message": agent_status_message.format("Microsoft.Powershell.ExampleExtension")}
+                }
+                self._assert_extension_status(sc_handler[:], expected_extensions)
 
     def test_it_should_report_status_correctly_for_unsupported_goal_state(self):
         with self.__setup_generic_test_env() as (exthandlers_handler, protocol, _):
diff --git a/tests/ga/test_update.py b/tests/ga/test_update.py
index e700f43d..3a99a859 100644
--- a/tests/ga/test_update.py
+++ b/tests/ga/test_update.py
@@ -7,6 +7,7 @@ import contextlib
 import glob
 import json
 import os
+import re
 import shutil
 import stat
 import subprocess
@@ -18,18 +19,20 @@ import zipfile
 from datetime import datetime, timedelta
 from threading import currentThread
 
+_ORIGINAL_POPEN = subprocess.Popen
+
 from mock import PropertyMock
 
 from azurelinuxagent.common import conf
 from azurelinuxagent.common.event import EVENTS_DIRECTORY
-from azurelinuxagent.common.exception import ProtocolError, UpdateError, ResourceGoneError
+from azurelinuxagent.common.exception import ProtocolError, UpdateError, ResourceGoneError, HttpError
 from azurelinuxagent.common.future import ustr
 from azurelinuxagent.common.persist_firewall_rules import PersistFirewallRulesHandler
 from azurelinuxagent.common.protocol.goal_state import ExtensionsConfig
 from azurelinuxagent.common.protocol.hostplugin import URI_FORMAT_GET_API_VERSIONS, HOST_PLUGIN_PORT, \
     URI_FORMAT_GET_EXTENSION_ARTIFACT, HostPluginProtocol
 from azurelinuxagent.common.protocol.restapi import ExtHandlerPackageUri, VMAgentManifest, VMAgentManifestUri, \
-    VMAgentManifestList, ExtHandlerPackage, ExtHandlerPackageList, ExtHandler
+    VMAgentManifestList, ExtHandlerPackage, ExtHandlerPackageList, ExtHandler, VMStatus, ExtHandlerStatus, ExtensionStatus
 from azurelinuxagent.common.protocol.util import ProtocolUtil
 from azurelinuxagent.common.protocol.wire import WireProtocol
 from azurelinuxagent.common.utils import fileutil, restutil, textutil
@@ -37,14 +40,16 @@ from azurelinuxagent.common.utils.flexible_version import FlexibleVersion
 from azurelinuxagent.common.utils.networkutil import FirewallCmdDirectCommands
 from azurelinuxagent.common.version import AGENT_PKG_GLOB, AGENT_DIR_GLOB, AGENT_NAME, AGENT_DIR_PATTERN, \
     AGENT_VERSION, CURRENT_AGENT, CURRENT_VERSION
-from azurelinuxagent.ga.exthandlers import ExtHandlerInstance, HandlerEnvironment
+from azurelinuxagent.ga.exthandlers import ExtHandlersHandler, ExtHandlerInstance, HandlerEnvironment, ValidHandlerStatus
 from azurelinuxagent.ga.update import GuestAgent, GuestAgentError, MAX_FAILURE, AGENT_MANIFEST_FILE, \
     get_update_handler, ORPHAN_POLL_INTERVAL, AGENT_PARTITION_FILE, AGENT_ERROR_FILE, ORPHAN_WAIT_INTERVAL, \
-    CHILD_LAUNCH_RESTART_MAX, CHILD_HEALTH_INTERVAL, UpdateHandler
+    CHILD_LAUNCH_RESTART_MAX, CHILD_HEALTH_INTERVAL, GOAL_STATE_PERIOD_EXTENSIONS_DISABLED, UpdateHandler, ExtensionsSummary
 from tests.protocol.mocks import mock_wire_protocol
 from tests.protocol.mockwiredata import DATA_FILE, DATA_FILE_MULTIPLE_EXT
 from tests.tools import AgentTestCase, data_dir, DEFAULT, patch, load_bin_data, load_data, Mock, MagicMock, \
-    clear_singleton_instances, mock_sleep
+    clear_singleton_instances, mock_sleep, skip_if_predicate_true
+from tests.protocol import mockwiredata
+from tests.protocol.mocks import HttpRequestPredicates
 
 NO_ERROR = {
     "last_failure": 0.0,
@@ -1102,11 +1107,11 @@ class TestUpdate(UpdateTestCase):
         if mock_time is None:
             mock_time = TimeMock()
 
-        with patch('subprocess.Popen', return_value=mock_child) as mock_popen:
+        with patch('azurelinuxagent.ga.update.subprocess.Popen', return_value=mock_child) as mock_popen:
             with patch('time.time', side_effect=mock_time.time):
                 with patch('time.sleep', side_effect=mock_time.sleep):
                     self.update_handler.run_latest(child_args=child_args)
-                    self.assertEqual(1, mock_popen.call_count)
+                    self.assertEqual(1, mock_popen.call_count, "Expected a single call to the latest agent; got: {0}".format([call.args for call in mock_popen.call_args_list]))
 
                     return mock_popen.call_args
 
@@ -1498,14 +1503,6 @@ class TestUpdate(UpdateTestCase):
         self.update_handler._upgrade_available = Mock(return_value=True)
         self._test_run(invocations=0, calls=0, enable_updates=True, sleep_interval=(300,))
 
-    @patch('azurelinuxagent.common.conf.get_extensions_enabled', return_value=False)
-    def test_interval_changes_when_extensions_disabled(self, _):
-        """
-        When extension processing is disabled, the goal state interval should be larger.
-        """
-        self.update_handler._upgrade_available = Mock(return_value=False)
-        self._test_run(invocations=1, calls=1, sleep_interval=(300,))
-
     @patch("azurelinuxagent.common.logger.info")
     @patch("azurelinuxagent.ga.update.add_event")
     def test_telemetry_heartbeat_creates_event(self, patch_add_event, patch_info, *_):
@@ -1601,14 +1598,13 @@ class TestUpdate(UpdateTestCase):
                           "{0} not found in HandlerEnv file".format(HandlerEnvironment.eventsFolder))
 
     def test_it_should_not_setup_persistent_firewall_rules_if_EnableFirewall_is_disabled(self):
-        original_popen = subprocess.Popen
         executed_firewall_commands = []
 
         def _mock_popen(cmd, *args, **kwargs):
             if 'firewall-cmd' in cmd:
                 executed_firewall_commands.append(cmd)
                 cmd = ["echo", "running"]
-            return original_popen(cmd, *args, **kwargs)
+            return _ORIGINAL_POPEN(cmd, *args, **kwargs)
 
         with patch("azurelinuxagent.common.logger.info") as patch_info:
             with self._get_update_handler(iterations=1) as (update_handler, _):
@@ -1623,14 +1619,13 @@ class TestUpdate(UpdateTestCase):
 
     def test_it_should_setup_persistent_firewall_rules_on_startup(self):
         iterations = 1
-        original_popen = subprocess.Popen
         executed_commands = []
 
         def _mock_popen(cmd, *args, **kwargs):
             if 'firewall-cmd' in cmd:
                 executed_commands.append(cmd)
                 cmd = ["echo", "running"]
-            return original_popen(cmd, *args, **kwargs)
+            return _ORIGINAL_POPEN(cmd, *args, **kwargs)
 
         with self._get_update_handler(iterations) as (update_handler, _):
             with patch("azurelinuxagent.common.utils.shellutil.subprocess.Popen", side_effect=_mock_popen):
@@ -1892,39 +1887,338 @@ class TimeMock(Mock):
         return current_time
 
 
-class TestProcessGoalState(AgentTestCase):
+class TryUpdateGoalStateTestCase(HttpRequestPredicates, AgentTestCase):
+    """
+    Tests for UpdateHandler._try_update_goal_state()
+    """
+    def test_it_should_return_true_on_success(self):
+        update_handler = get_update_handler()
+        with mock_wire_protocol(mockwiredata.DATA_FILE) as protocol:
+            self.assertTrue(update_handler._try_update_goal_state(protocol), "try_update_goal_state should have succeeded")
+
+    def test_it_should_return_false_on_failure(self):
+        with mock_wire_protocol(mockwiredata.DATA_FILE) as protocol:
+            def http_get_handler(url, *_, **__):
+                if self.is_goal_state_request(url):
+                    return HttpError('Exception to fake an error retrieving the goal state')
+                return None
+            protocol.set_http_handlers(http_get_handler=http_get_handler)
+
+            update_handler = get_update_handler()
+            self.assertFalse(update_handler._try_update_goal_state(protocol), "try_update_goal_state should have failed")
+
+    @skip_if_predicate_true(lambda: True, "TODO: Enable this test once we start retrieving the ExtensionsGoalState (vmSettings)")
+    def test_it_should_update_the_goal_state(self):
+        update_handler = get_update_handler()
+        with mock_wire_protocol(mockwiredata.DATA_FILE) as protocol:
+            protocol.mock_wire_data.set_incarnation(12345)
+            protocol.mock_wire_data.set_etag('54321')
+
+            # the first goal state should produce an update
+            update_handler._try_update_goal_state(protocol)
+            self.assertEqual(protocol.get_incarnation(), '12345', "The goal state was not updated (received unexpected incarnation)")
+            self.assertEqual(protocol.get_etag(), '54321', "The extensions goal state was not updated (received unexpected ETag)")
+
+            # no changes in the goal state should not produce an update
+            update_handler._try_update_goal_state(protocol)
+            self.assertEqual(protocol.get_incarnation(), '12345', "The goal state should not be updated (received unexpected incarnation)")
+            self.assertEqual(protocol.get_etag(), '54321', "The extensions goal state should not be updated (received unexpected ETag)")
+
+            # a new  goal state should produce an update
+            protocol.mock_wire_data.set_incarnation(6789)
+            protocol.mock_wire_data.set_etag('9876')
+            update_handler._try_update_goal_state(protocol)
+            self.assertEqual(protocol.get_incarnation(), '6789', "The goal state was not updated (received unexpected incarnation)")
+            self.assertEqual(protocol.get_etag(), '9876', "The extensions goal state was not updated (received unexpected ETag)")
+
+    def test_it_should_log_errors_only_when_the_error_state_changes(self):
+        with mock_wire_protocol(mockwiredata.DATA_FILE) as protocol:
+            def http_get_handler(url, *_, **__):
+                if self.is_goal_state_request(url):
+                    if fail_goal_state_request:
+                        return HttpError('Exception to fake an error retrieving the goal state')
+                return None
+
+            protocol.set_http_handlers(http_get_handler=http_get_handler)
+
+            @contextlib.contextmanager
+            def create_log_and_telemetry_mocks():
+                with patch("azurelinuxagent.ga.update.logger", autospec=True) as logger_patcher:
+                    with patch("azurelinuxagent.ga.update.add_event") as add_event_patcher:
+                        yield logger_patcher, add_event_patcher
+
+            calls_to_strings = lambda calls: (str(c) for c in calls)
+            filter_calls = lambda calls, regex=None: (c for c in calls_to_strings(calls) if regex is None or re.match(regex, c))
+            logger_calls = lambda regex=None: [m for m in filter_calls(logger.method_calls, regex)]  # pylint: disable=used-before-assignment,unnecessary-comprehension
+            warnings = lambda: logger_calls(r'call.warn\(.*An error occurred while retrieving the goal state.*')
+            periodic_warnings = lambda: logger_calls(r'call.periodic_warn\(.*Attempts to retrieve the goal state are failing.*')
+            success_messages = lambda: logger_calls(r'call.info\(.*Retrieving the goal state recovered from previous errors.*')
+            telemetry_calls = lambda regex=None: [m for m in filter_calls(add_event.mock_calls, regex)]  # pylint: disable=used-before-assignment,unnecessary-comprehension
+            goal_state_events = lambda: telemetry_calls(r".*op='FetchGoalState'.*")
+
+            #
+            # Initially calls to retrieve the goal state are successful...
+            #
+            update_handler = get_update_handler()
+            fail_goal_state_request = False
+            with create_log_and_telemetry_mocks() as (logger, add_event):
+                update_handler._try_update_goal_state(protocol)
+
+                lc = logger_calls()
+                self.assertTrue(len(lc) == 0, "A successful call should not produce any log messages: [{0}]".format(lc))
+
+                tc = telemetry_calls()
+                self.assertTrue(len(tc) == 0, "A successful call should not produce any telemetry events: [{0}]".format(tc))
+
+            #
+            # ... then an error happens...
+            #
+            fail_goal_state_request = True
+            with create_log_and_telemetry_mocks() as (logger, add_event):
+                update_handler._try_update_goal_state(protocol)
+
+                w = warnings()
+                pw = periodic_warnings()
+                self.assertEqual(1, len(w), "A failure should have produced a warning: [{0}]".format(w))
+                self.assertEqual(1, len(pw), "A failure should have produced a periodic warning: [{0}]".format(pw))
+
+                gs = goal_state_events()
+                self.assertTrue(len(gs) == 1 and 'is_success=False' in gs[0], "A failure should produce a telemetry event (success=false): [{0}]".format(gs))
+
+            #
+            # ... and errors continue happening...
+            #
+            with create_log_and_telemetry_mocks() as (logger, add_event):
+                update_handler._try_update_goal_state(protocol)
+                update_handler._try_update_goal_state(protocol)
+                update_handler._try_update_goal_state(protocol)
+
+                w = warnings()
+                pw = periodic_warnings()
+                self.assertTrue(len(w) == 0, "Subsequent failures should not produce warnings: [{0}]".format(w))
+                self.assertEqual(len(pw), 3, "Subsequent failures should produce periodic warnings: [{0}]".format(pw))
+
+                tc = telemetry_calls()
+                self.assertTrue(len(tc) == 0, "Subsequent failures should not produce any telemetry events: [{0}]".format(tc))
+
+            #
+            # ... until we finally succeed
+            #
+            fail_goal_state_request = False
+            with create_log_and_telemetry_mocks() as (logger, add_event):
+                update_handler._try_update_goal_state(protocol)
+
+                s = success_messages()
+                w = warnings()
+                pw = periodic_warnings()
+                self.assertEqual(len(s), 1, "Recovering after failures should have produced an info message: [{0}]".format(s))
+                self.assertTrue(len(w) == 0 and len(pw) == 0, "Recovering after failures should have not produced any warnings: [{0}] [{1}]".format(w, pw))
+
+                gs = goal_state_events()
+                self.assertTrue(len(gs) == 1 and 'is_success=True' in gs[0], "Recovering after failures should produce a telemetry event (success=true): [{0}]".format(gs))
+
+def _create_update_handler():
+    """
+    Creates an UpdateHandler in which agent updates are mocked as a no-op.
+    """
+    update_handler = get_update_handler()
+    update_handler._upgrade_available = Mock(return_value=False)
+    return update_handler
+
+@contextlib.contextmanager
+def _mock_exthandlers_handler(extension_statuses=None):
     """
-    Tests for UpdateHandler._process_goal_state
+    Creates an ExtHandlersHandler that doesn't actually handle any extensions, but that returns status for 1 extension.
+    The returned ExtHandlersHandler uses a mock WireProtocol, and both the run() and report_ext_handlers_status() are
+    mocked. The mock run() is a no-op. If a list of extension_statuses is given, successive calls to the mock
+    report_ext_handlers_status() returns a single extension with each of the statuses in the list. If extension_statuses
+    is ommitted all calls to report_ext_handlers_status() return a single extension with a success status.
+    """
+    def create_vm_status(extension_status):
+        vm_status = VMStatus(status="Ready", message="Ready")
+        vm_status.vmAgent.extensionHandlers = [ExtHandlerStatus()]
+        vm_status.vmAgent.extensionHandlers[0].extension_status = ExtensionStatus(name="TestExtension")
+        vm_status.vmAgent.extensionHandlers[0].extension_status.status = extension_status
+        return vm_status
+
+    with mock_wire_protocol(DATA_FILE) as protocol:
+        exthandlers_handler = ExtHandlersHandler(protocol)
+        exthandlers_handler.run = Mock()
+        if extension_statuses is None:
+            exthandlers_handler.report_ext_handlers_status = Mock(return_value=create_vm_status(ValidHandlerStatus.success))
+        else:
+            exthandlers_handler.report_ext_handlers_status = Mock(side_effect=[create_vm_status(s) for s in extension_statuses])
+        yield exthandlers_handler
+
+
+class ProcessGoalStateTestCase(AgentTestCase):
+    """
+    Tests for UpdateHandler._process_goal_state()
     """
     def test_it_should_process_goal_state_only_on_new_goal_state(self):
-        with mock_wire_protocol(DATA_FILE) as protocol:
+        with _mock_exthandlers_handler() as exthandlers_handler:
+            update_handler = _create_update_handler()
+            remote_access_handler = Mock()
+            remote_access_handler.run = Mock()
+
+            # process a goal state
+            update_handler._process_goal_state(exthandlers_handler, remote_access_handler)
+            self.assertEqual(1, exthandlers_handler.run.call_count, "exthandlers_handler.run() should have been called on the first goal state")
+            self.assertEqual(1, exthandlers_handler.report_ext_handlers_status.call_count, "exthandlers_handler.report_ext_handlers_status() should have been called on the first goal state")
+            self.assertEqual(1, remote_access_handler.run.call_count, "remote_access_handler.run() should have been called on the first goal state")
+
+            # process the same goal state
+            update_handler._process_goal_state(exthandlers_handler, remote_access_handler)
+            self.assertEqual(1, exthandlers_handler.run.call_count, "exthandlers_handler.run() should have not been called on the same goal state")
+            self.assertEqual(2, exthandlers_handler.report_ext_handlers_status.call_count, "exthandlers_handler.report_ext_handlers_status() should have been called on the same goal state")
+            self.assertEqual(1, remote_access_handler.run.call_count, "remote_access_handler.run() should not have been called on the same goal state")
+
+            # process a new goal state
+            exthandlers_handler.protocol.mock_wire_data.set_incarnation(999)
+            exthandlers_handler.protocol.client.update_goal_state()
+            update_handler._process_goal_state(exthandlers_handler, remote_access_handler)
+            self.assertEqual(2, exthandlers_handler.run.call_count, "exthandlers_handler.run() should have been called on a new goal state")
+            self.assertEqual(3, exthandlers_handler.report_ext_handlers_status.call_count, "exthandlers_handler.report_ext_handlers_status() should have been called on a new goal state")
+            self.assertEqual(2, remote_access_handler.run.call_count, "remote_access_handler.run() should have been called on a new goal state")
+
+
+class ReportStatusTestCase(AgentTestCase):
+    """
+    Tests for UpdateHandler._report_status()
+    """
+    def test_report_status_should_log_errors_only_once_per_goal_state(self):
+        update_handler = _create_update_handler()
+        with _mock_exthandlers_handler() as exthandlers_handler:
+            with patch("azurelinuxagent.ga.update.logger.warn") as logger_warn:
+                update_handler._report_status(exthandlers_handler, False)
+                self.assertEqual(0, logger_warn.call_count, "UpdateHandler._report_status() should not report WARNINGS when there are no errors")
+
+                with patch("azurelinuxagent.ga.update.ExtensionsSummary.__init__", return_value=Exception("TEST EXCEPTION")):  # simulate an error during _report_status()
+                    update_handler._report_status(exthandlers_handler, False)
+                    update_handler._report_status(exthandlers_handler, False)
+                    update_handler._report_status(exthandlers_handler, False)
+                    self.assertEqual(1, logger_warn.call_count, "UpdateHandler._report_status() should report only 1 WARNING when there are multiple errors within the same goal state")
+
+                    exthandlers_handler.protocol.mock_wire_data.set_incarnation(999)
+                    update_handler._try_update_goal_state(exthandlers_handler.protocol)
+                    update_handler._report_status(exthandlers_handler, True)
+                    self.assertEqual(2, logger_warn.call_count, "UpdateHandler._report_status() should continue reporting errors after a new goal state")
+
+
+class GoalStateIntervalTestCase(AgentTestCase):
+    def test_initial_goal_state_period_should_default_to_goal_state_period(self):
+        configuration_provider = conf.ConfigurationProvider()
+        test_file = os.path.join(self.tmp_dir, "waagent.conf")
+        with open(test_file, "w") as file_:
+            file_.write("Extensions.GoalStatePeriod=987654321\n")
+        conf.load_conf_from_file(test_file, configuration_provider)
+
+        self.assertEqual(987654321, conf.get_initial_goal_state_period(conf=configuration_provider))
+
+    def test_update_handler_should_use_the_default_goal_state_period(self):
+        update_handler = get_update_handler()
+        default = conf.get_int_default_value("Extensions.GoalStatePeriod")
+        self.assertEqual(default, update_handler._goal_state_period, "The UpdateHanlder is not using the default goal state period")
+
+    def test_update_handler_should_not_use_the_default_goal_state_period_when_extensions_are_disabled(self):
+        with patch('azurelinuxagent.common.conf.get_extensions_enabled', return_value=False):
+            update_handler = get_update_handler()
+            self.assertEqual(GOAL_STATE_PERIOD_EXTENSIONS_DISABLED, update_handler._goal_state_period, "Incorrect goal state period when extensions are disabled")
+
+    def test_the_default_goal_state_period_and_initial_goal_state_period_should_be_the_same(self):
+        update_handler = get_update_handler()
+        default = conf.get_int_default_value("Extensions.GoalStatePeriod")
+        self.assertEqual(default, update_handler._goal_state_period, "The UpdateHanlder is not using the default goal state period")
+
+    def test_update_handler_should_use_the_initial_goal_state_period_when_it_is_different_to_the_goal_state_period(self):
+        with patch('azurelinuxagent.common.conf.get_initial_goal_state_period', return_value=99999):
             update_handler = get_update_handler()
-            with patch.object(update_handler, "_upgrade_available", return_value=False):  # skip the upgrade logic
-                exthandlers_handler = Mock()
-                remote_access_handler = Mock()
-
-                def get_method_calls(mock, method):
-                    return [call[0] for call in mock.method_calls if call[0] == method]
-
-                # process a goal state
-                update_handler._process_goal_state(protocol, exthandlers_handler, remote_access_handler)
-                self.assertEqual(1, len(get_method_calls(exthandlers_handler, 'run')), "exthandlers_handler.run() should have been called on the first goal state")
-                self.assertEqual(1, len(get_method_calls(exthandlers_handler, 'report_ext_handlers_status')), "exthandlers_handler.report_ext_handlers_status() should have been called on the first goal state")
-                self.assertEqual(1, len(get_method_calls(remote_access_handler, 'run')), "remote_access_handler.run() should have been called on the first goal state")
-
-                # process the same goal state
-                update_handler._process_goal_state(protocol, exthandlers_handler, remote_access_handler)
-                self.assertEqual(1, len(get_method_calls(exthandlers_handler, 'run')), "exthandlers_handler.run() should have not been called on the same goal state")
-                self.assertEqual(2, len(get_method_calls(exthandlers_handler, 'report_ext_handlers_status')), "exthandlers_handler.report_ext_handlers_status() should have been called on the same goal state")
-                self.assertEqual(1, len(get_method_calls(remote_access_handler, 'run')), "remote_access_handler.run() should not have been called on the same goal state")
-
-                # process a new goal state
-                protocol.mock_wire_data.set_incarnation(999)
-                protocol.client.update_goal_state()
-                update_handler._process_goal_state(protocol, exthandlers_handler, remote_access_handler)
-                self.assertEqual(2, len(get_method_calls(exthandlers_handler, 'run')), "exthandlers_handler.run() should have been called on a new goal state")
-                self.assertEqual(3, len(get_method_calls(exthandlers_handler, 'report_ext_handlers_status')), "exthandlers_handler.report_ext_handlers_status() should have been called on a new goal state")
-                self.assertEqual(2, len(get_method_calls(remote_access_handler, 'run')), "remote_access_handler.run() should have been called on a new goal state")
+            self.assertEqual(99999, update_handler._goal_state_period, "Expected the initial goal state period")
+
+    def test_update_handler_should_use_the_initial_goal_state_period_until_the_goal_state_converges(self):
+        initial_goal_state_period, goal_state_period = 11111, 22222
+        with patch('azurelinuxagent.common.conf.get_initial_goal_state_period', return_value=initial_goal_state_period):
+            with patch('azurelinuxagent.common.conf.get_goal_state_period', return_value=goal_state_period):
+                with _mock_exthandlers_handler([ValidHandlerStatus.transitioning, ValidHandlerStatus.success]) as exthandlers_handler:
+                    remote_access_handler = Mock()
+
+                    update_handler = _create_update_handler()
+                    self.assertEqual(initial_goal_state_period, update_handler._goal_state_period, "Expected the initial goal state period")
+
+                    # the extension is transisioning, so we should still be using the initial goal state period
+                    update_handler._process_goal_state(exthandlers_handler, remote_access_handler)
+                    self.assertEqual(initial_goal_state_period, update_handler._goal_state_period, "Expected the initial goal state period when the extension is transitioning")
+
+                    # the goal state converged (the extension succeeded), so we should switch to the regular goal state period
+                    update_handler._process_goal_state(exthandlers_handler, remote_access_handler)
+                    self.assertEqual(goal_state_period, update_handler._goal_state_period, "Expected the regular goal state period after the goal state converged")
+
+    def test_update_handler_should_switch_to_the_regular_goal_state_period_when_the_goal_state_does_not_converges(self):
+        initial_goal_state_period, goal_state_period = 11111, 22222
+        with patch('azurelinuxagent.common.conf.get_initial_goal_state_period', return_value=initial_goal_state_period):
+            with patch('azurelinuxagent.common.conf.get_goal_state_period', return_value=goal_state_period):
+                with _mock_exthandlers_handler([ValidHandlerStatus.transitioning, ValidHandlerStatus.transitioning]) as exthandlers_handler:
+                    remote_access_handler = Mock()
+
+                    update_handler = _create_update_handler()
+                    self.assertEqual(initial_goal_state_period, update_handler._goal_state_period, "Expected the initial goal state period")
+
+                    # the extension is transisioning, so we should still be using the initial goal state period
+                    update_handler._process_goal_state(exthandlers_handler, remote_access_handler)
+                    self.assertEqual(initial_goal_state_period, update_handler._goal_state_period, "Expected the initial goal state period when the extension is transitioning")
+
+                    # a new goal state arrives before the current goal state converged (the extension is transitioning), so we should switch to the regular goal state period
+                    exthandlers_handler.protocol.mock_wire_data.set_incarnation(100)
+                    update_handler._process_goal_state(exthandlers_handler, remote_access_handler)
+                    self.assertEqual(goal_state_period, update_handler._goal_state_period, "Expected the regular goal state period when the goal state does not converge")
+
+
+class ExtensionsSummaryTestCase(AgentTestCase):
+    @staticmethod
+    def _create_extensions_summary(extension_statuses):
+        """
+        Creates an ExtensionsSummary from an array of (extension name, extension status) tuples
+        """
+        vm_status = VMStatus(status="Ready", message="Ready")
+        vm_status.vmAgent.extensionHandlers = [ExtHandlerStatus()] * len(extension_statuses)
+        for i in range(len(extension_statuses)):
+            vm_status.vmAgent.extensionHandlers[i].extension_status = ExtensionStatus(name=extension_statuses[i][0])
+            vm_status.vmAgent.extensionHandlers[0].extension_status.status = extension_statuses[i][1]
+        return ExtensionsSummary(vm_status)
+
+    def test_equality_operator_should_return_true_on_items_with_the_same_value(self):
+        summary1 = ExtensionsSummaryTestCase._create_extensions_summary([("Extension 1", ValidHandlerStatus.success), ("Extension 2", ValidHandlerStatus.transitioning)])
+        summary2 = ExtensionsSummaryTestCase._create_extensions_summary([("Extension 1", ValidHandlerStatus.success), ("Extension 2", ValidHandlerStatus.transitioning)])
+
+        self.assertTrue(summary1 == summary2, "{0} == {1} should be True".format(summary1, summary2))
+
+    def test_equality_operator_should_return_false_on_items_with_different_values(self):
+        summary1 = ExtensionsSummaryTestCase._create_extensions_summary([("Extension 1", ValidHandlerStatus.success), ("Extension 2", ValidHandlerStatus.transitioning)])
+        summary2 = ExtensionsSummaryTestCase._create_extensions_summary([("Extension 1", ValidHandlerStatus.success), ("Extension 2", ValidHandlerStatus.success)])
+
+        self.assertFalse(summary1 == summary2, "{0} == {1} should be False")
+
+        summary1 = ExtensionsSummaryTestCase._create_extensions_summary([("Extension 1", ValidHandlerStatus.success)])
+        summary2 = ExtensionsSummaryTestCase._create_extensions_summary([("Extension 1", ValidHandlerStatus.success), ("Extension 2", ValidHandlerStatus.success)])
+
+        self.assertFalse(summary1 == summary2, "{0} == {1} should be False")
+
+    def test_inequality_operator_should_return_true_on_items_with_different_values(self):
+        summary1 = ExtensionsSummaryTestCase._create_extensions_summary([("Extension 1", ValidHandlerStatus.success), ("Extension 2", ValidHandlerStatus.transitioning)])
+        summary2 = ExtensionsSummaryTestCase._create_extensions_summary([("Extension 1", ValidHandlerStatus.success), ("Extension 2", ValidHandlerStatus.success)])
+
+        self.assertTrue(summary1 != summary2, "{0} != {1} should be True".format(summary1, summary2))
+
+        summary1 = ExtensionsSummaryTestCase._create_extensions_summary([("Extension 1", ValidHandlerStatus.success)])
+        summary2 = ExtensionsSummaryTestCase._create_extensions_summary([("Extension 1", ValidHandlerStatus.success), ("Extension 2", ValidHandlerStatus.success)])
+
+        self.assertTrue(summary1 != summary2, "{0} != {1} should be True")
+
+    def test_inequality_operator_should_return_false_on_items_with_same_value(self):
+        summary1 = ExtensionsSummaryTestCase._create_extensions_summary([("Extension 1", ValidHandlerStatus.success), ("Extension 2", ValidHandlerStatus.transitioning)])
+        summary2 = ExtensionsSummaryTestCase._create_extensions_summary([("Extension 1", ValidHandlerStatus.success), ("Extension 2", ValidHandlerStatus.transitioning)])
+
+        self.assertFalse(summary1 != summary2, "{0} != {1} should be False".format(summary1, summary2))
 
 
 if __name__ == '__main__':
diff --git a/tests/protocol/mocks.py b/tests/protocol/mocks.py
index 88cf65d3..f3ebc314 100644
--- a/tests/protocol/mocks.py
+++ b/tests/protocol/mocks.py
@@ -177,6 +177,10 @@ class HttpRequestPredicates(object):
             raise ValueError('Host plugin request is missing the x-ms-artifact-location header ({0})'.format(url))
         return headers['x-ms-artifact-location']
 
+    @staticmethod
+    def is_host_plugin_vm_settings_request(url):
+        return url.lower() == 'http://{0}:{1}/vmsettings'.format(restutil.KNOWN_WIRESERVER_IP, restutil.HOST_PLUGIN_PORT)
+
     @staticmethod
     def is_host_plugin_health_request(url):
         return url.lower() == 'http://{0}:{1}/health'.format(restutil.KNOWN_WIRESERVER_IP, restutil.HOST_PLUGIN_PORT)
@@ -210,9 +214,12 @@ class HttpRequestPredicates(object):
 
 
 class MockHttpResponse:
-    def __init__(self, status, body=''):
+    def __init__(self, status, body=b''):
         self.body = body
         self.status = status
 
     def read(self, *_):
         return self.body
+
+    def getheaders(self):
+        return {}
diff --git a/tests/protocol/mockwiredata.py b/tests/protocol/mockwiredata.py
index 337b678d..5c411109 100644
--- a/tests/protocol/mockwiredata.py
+++ b/tests/protocol/mockwiredata.py
@@ -36,7 +36,9 @@ DATA_FILE = {
         "trans_cert": "wire/trans_cert",
         "test_ext": "ext/sample_ext-1.3.0.zip",
         "remote_access": None,
-        "in_vm_artifacts_profile": None
+        "in_vm_artifacts_profile": None,
+        "vm_settings": "hostgaplugin/vm_settings.json",
+        "ETag": "1"
 }
 
 DATA_FILE_IN_VM_ARTIFACTS_PROFILE = DATA_FILE.copy()
@@ -109,6 +111,8 @@ DATA_FILE_PLUGIN_SETTINGS_MISMATCH["ext_conf"] = "wire/invalid_config/ext_conf_p
 DATA_FILE_REQUIRED_FEATURES = DATA_FILE.copy()
 DATA_FILE_REQUIRED_FEATURES["ext_conf"] = "wire/ext_conf_required_features.xml"
 
+DATA_FILE_VM_SETTINGS = DATA_FILE.copy()
+DATA_FILE_VM_SETTINGS["ext_conf"] = "hostgaplugin/ext_conf.xml"
 
 class WireProtocolData(object):
     def __init__(self, data_files=None):
@@ -131,7 +135,8 @@ class WireProtocolData(object):
             "manifest.xml": 0,
             "manifest_of_ga.xml": 0,
             "ExampleHandlerLinux": 0,
-            "in_vm_artifacts_profile": 0
+            "in_vm_artifacts_profile": 0,
+            "vm_settings": 0
         }
         self.data_files = data_files
         self.version_info = None
@@ -147,6 +152,8 @@ class WireProtocolData(object):
         self.ext = None
         self.remote_access = None
         self.in_vm_artifacts_profile = None
+        self.vm_settings = None
+        self.etag = None
 
         self.reload()
 
@@ -164,6 +171,8 @@ class WireProtocolData(object):
         self.trans_prv = load_data(self.data_files.get("trans_prv"))
         self.trans_cert = load_data(self.data_files.get("trans_cert"))
         self.ext = load_bin_data(self.data_files.get("test_ext"))
+        self.vm_settings = load_data(self.data_files.get("vm_settings"))
+        self.etag = self.data_files.get("ETag")
 
         remote_access_data_file = self.data_files.get("remote_access")
         if remote_access_data_file is not None:
@@ -175,6 +184,7 @@ class WireProtocolData(object):
 
     def mock_http_get(self, url, *args, **kwargs):  # pylint: disable=unused-argument
         content = None
+        response_headers = []
 
         resp = MagicMock()
         resp.status = httpclient.OK
@@ -209,6 +219,10 @@ class WireProtocolData(object):
         elif ".vmSettings" in url or ".settings" in url:
             content = self.in_vm_artifacts_profile
             self.call_counts["in_vm_artifacts_profile"] += 1
+        elif "/vmSettings" in url:
+            content = self.vm_settings
+            response_headers = [('ETag', self.etag)]
+            self.call_counts["vm_settings"] += 1
 
         else:
             # A stale GoalState results in a 400 from the HostPlugin
@@ -249,6 +263,7 @@ class WireProtocolData(object):
                 raise Exception("Bad url {0}".format(url))
 
         resp.read = Mock(return_value=content.encode("utf-8"))
+        resp.getheaders = Mock(return_value=response_headers)
         return resp
 
     def mock_http_post(self, url, *args, **kwargs):  # pylint: disable=unused-argument
@@ -334,6 +349,12 @@ class WireProtocolData(object):
             raise Exception("Could not match attribute '{0}' of element '{1}'".format(attribute_name, element_name))
         return new_xml_document
 
+    def set_etag(self, etag):
+        '''
+        Sets the ETag for the mock response
+        '''
+        self.etag = etag
+
     def set_incarnation(self, incarnation):
         '''
         Sets the incarnation in the goal state, but not on its subcomponents (e.g. hosting env, shared config)
diff --git a/tests/protocol/test_goal_state.py b/tests/protocol/test_goal_state.py
new file mode 100644
index 00000000..7cbfe527
--- /dev/null
+++ b/tests/protocol/test_goal_state.py
@@ -0,0 +1,101 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the Apache License.
+
+import json
+import os
+
+from azurelinuxagent.common.future import httpclient
+from azurelinuxagent.common import conf
+from azurelinuxagent.common.exception import IncompleteGoalStateError
+from azurelinuxagent.common.protocol.goal_state import GoalState, ExtensionsConfig
+from azurelinuxagent.common.protocol import hostplugin
+from azurelinuxagent.common.utils import restutil
+from tests.protocol.mocks import mock_wire_protocol
+from tests.protocol import mockwiredata
+from tests.protocol.mocks import HttpRequestPredicates, MockHttpResponse
+from tests.tools import AgentTestCase, patch
+
+_original_http_request = restutil.http_request
+
+@patch("azurelinuxagent.common.protocol.wire.conf.get_enable_fast_track", return_value=True)
+class GoalStateTestCase(HttpRequestPredicates, AgentTestCase):
+    def test_fetch_goal_state_should_raise_on_incomplete_goal_state(self, _):
+        with mock_wire_protocol(mockwiredata.DATA_FILE) as protocol:
+            GoalState.fetch_full_goal_state(protocol.client)
+
+            protocol.mock_wire_data.data_files = mockwiredata.DATA_FILE_NOOP_GS
+            protocol.mock_wire_data.reload()
+            protocol.mock_wire_data.set_incarnation(2)
+
+            with self.assertRaises(IncompleteGoalStateError):
+                GoalState.fetch_full_goal_state_if_incarnation_different_than(protocol.client, 1)
+
+    def test_update_goal_state_should_save_goal_state(self, _):
+        with mock_wire_protocol(mockwiredata.DATA_FILE_VM_SETTINGS) as protocol:
+            protocol.mock_wire_data.set_incarnation(999)
+            protocol.mock_wire_data.set_etag(888)
+            protocol.update_goal_state()
+
+        extensions_config_file = os.path.join(conf.get_lib_dir(), "ExtensionsConfig.999.xml")
+        vm_settings_file = os.path.join(conf.get_lib_dir(), "VmSettings.888.json")
+        expected_files = [
+            os.path.join(conf.get_lib_dir(), "GoalState.999.xml"),
+            os.path.join(conf.get_lib_dir(), "SharedConfig.xml"),
+            os.path.join(conf.get_lib_dir(), "Certificates.xml"),
+            os.path.join(conf.get_lib_dir(), "HostingEnvironmentConfig.xml"),
+            extensions_config_file,
+            vm_settings_file
+        ]
+
+        for f in expected_files:
+            self.assertTrue(os.path.exists(f), "{0} was not saved".format(f))
+
+        with open(extensions_config_file, "r") as file_:
+            extensions_config = ExtensionsConfig(file_.read())
+        self.assertEqual(4, len(extensions_config.ext_handlers.extHandlers), "Expected 4 extensions in the test ExtensionsConfig")
+        for e in extensions_config.ext_handlers.extHandlers:
+            self.assertEqual(e.properties.extensions[0].protectedSettings, "*** REDACTED ***", "The protected settings for {0} were not redacted".format(e.name))
+
+        # TODO: Use azurelinuxagent.common.protocol.ExtensionsGoalState once it implements parsing
+        with open(vm_settings_file, "r") as file_:
+            vm_settings = json.load(file_)
+        extensions = vm_settings["extensionGoalStates"]
+        self.assertEqual(4, len(extensions), "Expected 4 extensions in the test vmSettings")
+        for e in extensions:
+            self.assertEqual(e["settings"][0]["protectedSettings"], "*** REDACTED ***", "The protected settings for {0} were not redacted".format(e["name"]))
+
+
+    def test_update_extensions_goal_state_should_should_retry_on_resource_gone_error(self, _):
+        """
+        Requests to the hostgaplugin incude the Container ID and the RoleConfigName as headers; when the hostgaplugin returns GONE (HTTP status 410) the agent
+        needs to get a new goal state and retry the request with updated values for the Container ID and RoleConfigName headers.
+        """
+        with mock_wire_protocol(mockwiredata.DATA_FILE_VM_SETTINGS) as protocol:
+            # The mock_wire_protocol mocks azurelinuxagent.common.utils.restutil.http_request, but the GONE status is handled at a lower level, in the internal _http_request.
+            # To mock this, we bypass the mock http_request by calling the original http_request (function http_get_handler) and then mocking the internal _http_request (function
+            # http_get_vm_settings).
+            def http_get_handler(url, *_, **kwargs):
+                if self.is_host_plugin_vm_settings_request(url):
+                    return _original_http_request("GET", url, None, **kwargs)
+                return None
+            protocol.set_http_handlers(http_get_handler=http_get_handler)
+
+            request_headers = []  # we expect a retry with new headers and use this array to persist the headers of each request
+            def http_get_vm_settings(_method, _host, _relative_url, **kwargs):
+                request_headers.append(kwargs["headers"])
+                if len(request_headers) == 1:
+                    # Fail the first request with status GONE and update the mock data to return the new Container ID and RoleConfigName that should be
+                    # used in the headers of the retry request.
+                    protocol.mock_wire_data.set_container_id("GET_VM_SETTINGS_TEST_CONTAINER_ID")
+                    protocol.mock_wire_data.set_role_config_name("GET_VM_SETTINGS_TEST_ROLE_CONFIG_NAME")
+                    return MockHttpResponse(status=httpclient.GONE)
+                # For this test we are interested only on the retry logic, so the second request (the retry) is not important; we use NOT_MODIFIED (304) for simplicity.
+                return MockHttpResponse(status=httpclient.NOT_MODIFIED)
+
+            with patch("azurelinuxagent.common.utils.restutil._http_request", side_effect=http_get_vm_settings):
+                protocol.update_extensions_goal_state()
+
+            self.assertEqual(2, len(request_headers), "We expected 2 requests for vmSettings: the original request and the retry request")
+            self.assertEqual("GET_VM_SETTINGS_TEST_CONTAINER_ID", request_headers[1][hostplugin._HEADER_CONTAINER_ID], "The retry request did not include the expected header for the ContainerId")
+            self.assertEqual("GET_VM_SETTINGS_TEST_ROLE_CONFIG_NAME", request_headers[1][hostplugin._HEADER_HOST_CONFIG_NAME], "The retry request did not include the expected header for the RoleConfigName")
+
diff --git a/tests/protocol/test_hostplugin.py b/tests/protocol/test_hostplugin.py
index c93e5ab8..641cb7d7 100644
--- a/tests/protocol/test_hostplugin.py
+++ b/tests/protocol/test_hostplugin.py
@@ -186,7 +186,7 @@ class TestHostPlugin(HttpRequestPredicates, AgentTestCase):
         self.assertEqual(should_initialize, host.is_initialized)
 
         self.assertEqual(1, patch_event.call_count)
-        self.assertEqual('InitializeHostPlugin', patch_event.call_args[0][0])
+        self.assertEqual('InitializeHostPlugin', patch_event.call_args[1]['op'])
 
         self.assertEqual(should_initialize, patch_event.call_args[1]['is_success'])
         self.assertEqual(1, patch_report_health.call_count)
diff --git a/tests/protocol/test_wire.py b/tests/protocol/test_wire.py
index d85ed095..0ca57934 100644
--- a/tests/protocol/test_wire.py
+++ b/tests/protocol/test_wire.py
@@ -31,7 +31,7 @@ from azurelinuxagent.common.agent_supported_feature import SupportedFeatureNames
     get_agent_supported_features_list_for_crp
 from azurelinuxagent.common.exception import ResourceGoneError, ProtocolError, \
     ExtensionDownloadError, HttpError
-from azurelinuxagent.common.protocol.goal_state import ExtensionsConfig, GoalState
+from azurelinuxagent.common.protocol.goal_state import ExtensionsConfig
 from azurelinuxagent.common.protocol.hostplugin import HostPluginProtocol
 from azurelinuxagent.common.protocol.restapi import VMAgentManifestUri
 from azurelinuxagent.common.protocol.wire import WireProtocol, WireClient, \
@@ -39,7 +39,6 @@ from azurelinuxagent.common.protocol.wire import WireProtocol, WireClient, \
 from azurelinuxagent.common.telemetryevent import GuestAgentExtensionEventsSchema, \
     TelemetryEventParam, TelemetryEvent
 from azurelinuxagent.common.utils import restutil
-from azurelinuxagent.common.exception import IncompleteGoalStateError
 from azurelinuxagent.common.version import CURRENT_VERSION, DISTRO_NAME, DISTRO_VERSION
 from azurelinuxagent.ga.exthandlers import get_exthandlers_handler
 from tests.ga.test_monitor import random_generator
@@ -1205,123 +1204,6 @@ class UpdateGoalStateTestCase(AgentTestCase):
                     "Could not find the expected number of redacted settings. Expected {0}.\n{1}".format(len(protected_settings), extensions_config))
 
 
-class TryUpdateGoalStateTestCase(HttpRequestPredicates, AgentTestCase):
-    """
-    Tests for WireClient.try_update_goal_state()
-    """
-    def test_it_should_return_true_on_success(self):
-        with mock_wire_protocol(mockwiredata.DATA_FILE) as protocol:
-            self.assertTrue(protocol.client.try_update_goal_state(), "try_update_goal_state should have succeeded")
-
-    def test_incomplete_gs_should_fail(self):
-
-        with mock_wire_protocol(mockwiredata.DATA_FILE) as protocol:
-            GoalState.fetch_full_goal_state(protocol.client)
-
-            protocol.mock_wire_data.data_files = mockwiredata.DATA_FILE_NOOP_GS
-            protocol.mock_wire_data.reload()
-            protocol.mock_wire_data.set_incarnation(2)
-
-            with self.assertRaises(IncompleteGoalStateError):
-                GoalState.fetch_full_goal_state_if_incarnation_different_than(protocol.client, 1)
-
-    def test_it_should_return_false_on_failure(self):
-        with mock_wire_protocol(mockwiredata.DATA_FILE) as protocol:
-            def http_get_handler(url, *_, **__):
-                if self.is_goal_state_request(url):
-                    return HttpError('Exception to fake an error retrieving the goal state')
-                return None
-
-            protocol.set_http_handlers(http_get_handler=http_get_handler)
-
-            self.assertFalse(protocol.client.try_update_goal_state(), "try_update_goal_state should have failed")
-
-    def test_it_should_log_errors_only_when_the_error_state_changes(self):
-        with mock_wire_protocol(mockwiredata.DATA_FILE) as protocol:
-            def http_get_handler(url, *_, **__):
-                if self.is_goal_state_request(url):
-                    if fail_goal_state_request:
-                        return HttpError('Exception to fake an error retrieving the goal state')
-                return None
-
-            protocol.set_http_handlers(http_get_handler=http_get_handler)
-
-            @contextlib.contextmanager
-            def create_log_and_telemetry_mocks():
-                with patch("azurelinuxagent.common.protocol.wire.logger", autospec=True) as logger_patcher:
-                    with patch("azurelinuxagent.common.protocol.wire.add_event") as add_event_patcher:
-                        yield logger_patcher, add_event_patcher
-
-            calls_to_strings = lambda calls: (str(c) for c in calls)
-            filter_calls = lambda calls, regex=None: (c for c in calls_to_strings(calls) if regex is None or re.match(regex, c))
-            logger_calls = lambda regex=None: [m for m in filter_calls(logger.method_calls, regex)]  # pylint: disable=used-before-assignment,unnecessary-comprehension
-            warnings = lambda: logger_calls(r'call.warn\(.*An error occurred while retrieving the goal state.*')
-            periodic_warnings = lambda: logger_calls(r'call.periodic_warn\(.*Attempts to retrieve the goal state are failing.*')
-            success_messages = lambda: logger_calls(r'call.info\(.*Retrieving the goal state recovered from previous errors.*')
-            telemetry_calls = lambda regex=None: [m for m in filter_calls(add_event.mock_calls, regex)]  # pylint: disable=used-before-assignment,unnecessary-comprehension
-            goal_state_events = lambda: telemetry_calls(r".*op='FetchGoalState'.*")
-
-            #
-            # Initially calls to retrieve the goal state are successful...
-            #
-            fail_goal_state_request = False
-            with create_log_and_telemetry_mocks() as (logger, add_event):
-                protocol.client.try_update_goal_state()
-
-                lc = logger_calls()
-                self.assertTrue(len(lc) == 0, "A successful call should not produce any log messages: [{0}]".format(lc))
-
-                tc = telemetry_calls()
-                self.assertTrue(len(tc) == 0, "A successful call should not produce any telemetry events: [{0}]".format(tc))
-
-            #
-            # ... then an error happens...
-            #
-            fail_goal_state_request = True
-            with create_log_and_telemetry_mocks() as (logger, add_event):
-                protocol.client.try_update_goal_state()
-
-                w = warnings()
-                pw = periodic_warnings()
-                self.assertEqual(len(w), 1, "A failure should have produced a warning: [{0}]".format(w))
-                self.assertEqual(len(pw), 1, "A failure should have produced a periodic warning: [{0}]".format(pw))
-
-                gs = goal_state_events()
-                self.assertTrue(len(gs) == 1 and 'is_success=False' in gs[0], "A failure should produce a telemetry event (success=false): [{0}]".format(gs))
-
-            #
-            # ... and errors continue happening...
-            #
-            with create_log_and_telemetry_mocks() as (logger, add_event):
-                protocol.client.try_update_goal_state()
-                protocol.client.try_update_goal_state()
-                protocol.client.try_update_goal_state()
-
-                w = warnings()
-                pw = periodic_warnings()
-                self.assertTrue(len(w) == 0, "Subsequent failures should not produce warnings: [{0}]".format(w))
-                self.assertEqual(len(pw), 3, "Subsequent failures should produce periodic warnings: [{0}]".format(pw))
-
-                tc = telemetry_calls()
-                self.assertTrue(len(tc) == 0, "Subsequent failures should not produce any telemetry events: [{0}]".format(tc))
-
-            #
-            # ... until we finally succeed
-            #
-            fail_goal_state_request = False
-            with create_log_and_telemetry_mocks() as (logger, add_event):
-                protocol.client.try_update_goal_state()
-
-                s = success_messages()
-                w = warnings()
-                pw = periodic_warnings()
-                self.assertEqual(len(s), 1, "Recovering after failures should have produced an info message: [{0}]".format(s))
-                self.assertTrue(len(w) == 0 and len(pw) == 0, "Recovering after failures should have not produced any warnings: [{0}] [{1}]".format(w, pw))
-
-                gs = goal_state_events()
-                self.assertTrue(len(gs) == 1 and 'is_success=True' in gs[0], "Recovering after failures should produce a telemetry event (success=true): [{0}]".format(gs))
-
-
 class UpdateHostPluginFromGoalStateTestCase(AgentTestCase):
     """
     Tests for WireClient.update_host_plugin_from_goal_state()
@@ -1368,6 +1250,8 @@ class MockResponse:
     def read(self, *_):
         return self.body
 
+    def getheaders(self):
+        return []
 
 if __name__ == '__main__':
     unittest.main()
diff --git a/tests/test_agent.py b/tests/test_agent.py
index 558532f8..ee36061b 100644
--- a/tests/test_agent.py
+++ b/tests/test_agent.py
@@ -18,8 +18,10 @@
 import os.path
 
 from azurelinuxagent.agent import parse_args, Agent, usage, AgentCommands
-from azurelinuxagent.common import conf
+from azurelinuxagent.common import cgroupconfigurator, conf, logcollector
+from azurelinuxagent.common.cgroupapi import SystemdCgroupsApi
 from azurelinuxagent.common.utils import fileutil
+from azurelinuxagent.ga.collect_logs import CollectLogsHandler
 from tests.tools import AgentTestCase, data_dir, Mock, patch
 
 EXPECTED_CONFIGURATION = \
@@ -31,16 +33,18 @@ Debug.CgroupCheckPeriod = 300
 Debug.CgroupDisableOnProcessCheckFailure = True
 Debug.CgroupDisableOnQuotaCheckFailure = True
 Debug.CgroupLogMetrics = False
+Debug.EnableFastTrack = False
 DetectScvmmEnv = False
 EnableOverProvisioning = True
 Extension.LogDir = /var/log/azure
 Extensions.Enabled = True
 Extensions.GoalStateHistoryCleanupPeriod = 1800
 Extensions.GoalStatePeriod = 6
+Extensions.InitialGoalStatePeriod = 6
 HttpProxy.Host = None
 HttpProxy.Port = None
 Lib.Dir = /var/lib/waagent
-Logs.Collect = False
+Logs.Collect = True
 Logs.CollectPeriod = 3600
 Logs.Console = True
 Logs.Verbose = False
@@ -213,6 +217,49 @@ class TestAgent(AgentTestCase):
         full_mode = mock_log_collector.call_args_list[1][0][0]
         self.assertFalse(full_mode)
 
+    @patch("azurelinuxagent.agent.LogCollector")
+    def test_calls_collect_logs_on_valid_cgroups(self, mock_log_collector):
+        try:
+            CollectLogsHandler.enable_cgroups_validation()
+
+            @staticmethod
+            def mock_cgroup_paths(*args, **kwargs):
+                if args and args[0] == "self":
+                    relative_path = "{0}/{1}".format(cgroupconfigurator.LOGCOLLECTOR_SLICE, logcollector.CGROUPS_UNIT)
+                    return (cgroupconfigurator.LOGCOLLECTOR_SLICE, relative_path)
+                return SystemdCgroupsApi.get_process_cgroup_relative_paths(*args, **kwargs)
+
+            with patch.object(SystemdCgroupsApi, "get_process_cgroup_relative_paths", mock_cgroup_paths):
+                agent = Agent(False, conf_file_path=os.path.join(data_dir, "test_waagent.conf"))
+                agent.collect_logs(is_full_mode=True)
+                
+                mock_log_collector.assert_called_once()
+        finally:
+            CollectLogsHandler.disable_cgroups_validation()
+
+    def test_doesnt_call_collect_logs_on_invalid_cgroups(self):
+        try:
+            CollectLogsHandler.enable_cgroups_validation()
+
+            @staticmethod
+            def mock_cgroup_paths(*args, **kwargs):
+                if args and args[0] == "self":
+                    return ("NOT_THE_CORRECT_PATH", "NOT_THE_CORRECT_PATH")
+                return SystemdCgroupsApi.get_process_cgroup_relative_paths(*args, **kwargs)
+
+            with patch.object(SystemdCgroupsApi, "get_process_cgroup_relative_paths", mock_cgroup_paths):
+                agent = Agent(False, conf_file_path=os.path.join(data_dir, "test_waagent.conf"))
+
+                exit_error = RuntimeError("Exiting")
+                with patch("sys.exit", return_value=exit_error) as mock_exit:
+                    try:
+                        agent.collect_logs(is_full_mode=True)
+                    except RuntimeError as re:
+                        mock_exit.assert_called_once_with(logcollector.INVALID_CGROUPS_ERRCODE)
+                        self.assertEqual(exit_error, re)
+        finally:
+            CollectLogsHandler.disable_cgroups_validation()
+        
     def test_it_should_parse_setup_firewall_properly(self):
 
         test_firewall_meta = {
diff --git a/tests/utils/test_archive.py b/tests/utils/test_archive.py
index 85d08746..2e468487 100644
--- a/tests/utils/test_archive.py
+++ b/tests/utils/test_archive.py
@@ -124,7 +124,7 @@ class TestArchive(AgentTestCase):
         self.assertEqual("0", incarnation)
 
         zip_full = os.path.join(self.history_dir, zip_fn)
-        self.assert_zip_contains(zip_full, temp_files)
+        self.assertEqual(self.assert_zip_contains(zip_full, temp_files), None)
 
     def test_archive02(self):
         """
@@ -236,14 +236,18 @@ class TestArchive(AgentTestCase):
         if secs < 0:
             self.fail("the timestamps are outside of the tolerance of by {0} seconds".format(secs))
 
-    def assert_zip_contains(self, zip_filename, files):
+    @staticmethod
+    def assert_zip_contains(zip_filename, files):
+
         ziph = None
         try:
             # contextmanager for zipfile.ZipFile doesn't exist for py2.6, manually closing it
             ziph = zipfile.ZipFile(zip_filename, 'r')
             zip_files = [x.filename for x in ziph.filelist]
             for current_file in files:
-                self.assertTrue(current_file in zip_files, "'{0}' was not found in {1}".format(current_file, zip_filename))
+                if current_file not in zip_files:
+                    return "'{0}' was not found in {1}".format(current_file, zip_filename)
+            return None
         finally:
             if ziph is not None:
-                ziph.close()
+                ziph.close()
\ No newline at end of file