Repository: Azure/WALinuxAgent
Pull request #2495 (2022-02-08 00:10:30)

Merge fast-track into develop


Patch:
diff --git a/azurelinuxagent/common/conf.py b/azurelinuxagent/common/conf.py
index 32cda350..f733663d 100644
--- a/azurelinuxagent/common/conf.py
+++ b/azurelinuxagent/common/conf.py
@@ -136,7 +136,7 @@ __SWITCH_OPTIONS__ = {
     "Debug.CgroupLogMetrics": False,
     "Debug.CgroupDisableOnProcessCheckFailure": True,
     "Debug.CgroupDisableOnQuotaCheckFailure": True,
-    "Debug.EnableFastTrack": True,
+    "Debug.EnableFastTrack": False,
     "Debug.EnableGAVersioning": False
 }
 
@@ -572,7 +572,7 @@ def get_enable_fast_track(conf=__conf__):
 
     NOTE: This option is experimental and may be removed in later versions of the Agent.
     """
-    return conf.get_switch("Debug.EnableFastTrack", True)
+    return conf.get_switch("Debug.EnableFastTrack", False)
 
 
 def get_etp_collection_period(conf=__conf__):
diff --git a/azurelinuxagent/common/exception.py b/azurelinuxagent/common/exception.py
index d39b1b95..bfeb0396 100644
--- a/azurelinuxagent/common/exception.py
+++ b/azurelinuxagent/common/exception.py
@@ -184,12 +184,6 @@ class ProtocolNotFoundError(ProtocolError):
     """
 
 
-class IncompleteGoalStateError(ProtocolError):
-    """
-    Goal state is returned incomplete.
-    """
-
-
 class HttpError(AgentError):
     """
     Http request failure
diff --git a/azurelinuxagent/common/logcollector.py b/azurelinuxagent/common/logcollector.py
index 84055777..a462c5e2 100644
--- a/azurelinuxagent/common/logcollector.py
+++ b/azurelinuxagent/common/logcollector.py
@@ -49,12 +49,7 @@ INVALID_CGROUPS_ERRCODE = 2
 
 _MUST_COLLECT_FILES = [
     _AGENT_LOG,
-    os.path.join(_AGENT_LIB_DIR, "GoalState.*.xml"),
-    os.path.join(_AGENT_LIB_DIR, "ExtensionsConfig.*.xml"),
-    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 cdeed984..e77da3d4 100644
--- a/azurelinuxagent/common/logcollector_manifests.py
+++ b/azurelinuxagent/common/logcollector_manifests.py
@@ -39,16 +39,13 @@ copy,$LOG_DIR/custom-script/handler.log
 echo,
 
 echo,### Gathering Extension Files ###
-copy,$LIB_DIR/*.xml
-copy,$LIB_DIR/VmSettings.*.json
-copy,$LIB_DIR/waagent_status.*.json
+copy,$LIB_DIR/ovf-env.xml
+copy,$LIB_DIR/waagent_status.json
 copy,$LIB_DIR/*/status/*.status
 copy,$LIB_DIR/*/config/*.settings
 copy,$LIB_DIR/*/config/HandlerState
 copy,$LIB_DIR/*/config/HandlerStatus
-copy,$LIB_DIR/*.agentsManifest
 copy,$LIB_DIR/error.json
-copy,$LIB_DIR/Incarnation
 copy,$LIB_DIR/history/*.zip
 echo,
 """
@@ -108,19 +105,15 @@ copy,$LOG_DIR/run-command/handler.log
 echo,
 
 echo,### Gathering Extension Files ###
-copy,$LIB_DIR/ExtensionsConfig.*.xml
+copy,$LIB_DIR/ovf-env.xml
 copy,$LIB_DIR/*/status/*.status
 copy,$LIB_DIR/*/config/*.settings
 copy,$LIB_DIR/*/config/HandlerState
 copy,$LIB_DIR/*/config/HandlerStatus
-copy,$LIB_DIR/GoalState.*.xml
-copy,$LIB_DIR/HostingEnvironmentConfig.xml
-copy,$LIB_DIR/*.manifest.xml
 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/protocol/extensions_goal_state.py b/azurelinuxagent/common/protocol/extensions_goal_state.py
index ee707927..d2d84212 100644
--- a/azurelinuxagent/common/protocol/extensions_goal_state.py
+++ b/azurelinuxagent/common/protocol/extensions_goal_state.py
@@ -19,16 +19,9 @@ import datetime
 
 import azurelinuxagent.common.logger as logger
 from azurelinuxagent.common.AgentGlobals import AgentGlobals
-from azurelinuxagent.common.exception import AgentError
 from azurelinuxagent.common.utils import textutil
 
 
-class GoalStateMismatchError(AgentError):
-    def __init__(self, message, attribute):
-        super(GoalStateMismatchError, self).__init__(message)
-        self.attribute = attribute
-
-
 class ExtensionsGoalState(object):
     """
     ExtensionsGoalState represents the extensions information in the goal state; that information can originate from
@@ -90,85 +83,6 @@ class ExtensionsGoalState(object):
         """
         raise NotImplementedError()
 
-    @staticmethod
-    def compare(from_extensions_config, from_vm_settings):
-        """
-        Compares the two instances given as argument and logs a GoalStateMismatch message if they are different.
-
-        NOTE: The order of the two instances is important for the debug info to be logged correctly (ExtensionsConfig first, vmSettings second)
-        """
-        context = []  # used to keep track of the attribute that is being compared
-
-        def compare_goal_states(first, second):
-            # A mismatch on the timestamp or the activity ID (and maybe also on the correlation ID) most likely indicate that we are comparing two
-            # different goal states so we check them first (we raise an exception as soon as a mismatch is detected). A mismatch on the other
-            # attributes likely indicates an actual issue on vmSettings or extensionsConfig).
-            compare_attributes(first, second, "created_on_timestamp")
-            compare_attributes(first, second, "activity_id")
-            compare_attributes(first, second, "correlation_id")
-            compare_attributes(first, second, "status_upload_blob")
-            compare_attributes(first, second, "status_upload_blob_type")
-            compare_attributes(first, second, "required_features")
-            compare_attributes(first, second, "on_hold")
-            compare_array(first.agent_manifests, second.agent_manifests, compare_agent_manifests, "agent_manifests")
-            compare_array(first.extensions, second.extensions, compare_extensions, "extensions")
-
-        def compare_agent_manifests(first, second):
-            compare_attributes(first, second, "family")
-            compare_attributes(first, second, "requested_version_string")
-            compare_attributes(first, second, "uris", ignore_order=True)
-
-        def compare_extensions(first, second):
-            compare_attributes(first, second, "name")
-            compare_attributes(first, second, "version")
-            compare_attributes(first, second, "state")
-            compare_attributes(first, second, "supports_multi_config")
-            compare_attributes(first, second, "manifest_uris", ignore_order=True)
-            compare_array(first.settings, second.settings, compare_settings, "settings")
-
-        def compare_settings(first, second):
-            # Note that we do not compare protectedSettings since the same settings can be re-encrypted, resulting
-            # on different encrypted text for the same plain text.
-            compare_attributes(first, second, "name")
-            compare_attributes(first, second, "sequenceNumber")
-            compare_attributes(first, second, "publicSettings")
-            compare_attributes(first, second, "certificateThumbprint")
-            compare_attributes(first, second, "dependencyLevel")
-            compare_attributes(first, second, "state")
-
-        def compare_array(first, second, comparer, name):
-            if len(first) != len(second):
-                raise Exception("Number of items in {0} mismatch: {1} != {2}".format(name, len(first), len(second)))
-            for i in range(len(first)):
-                context.append("{0}[{1}]".format(name, i))
-                try:
-                    comparer(first[i], second[i])
-                finally:
-                    context.pop()
-
-        def compare_attributes(first, second, attribute, ignore_order=False):
-            context.append(attribute)
-            try:
-                first_value = getattr(first, attribute)
-                second_value = getattr(second, attribute)
-                if ignore_order:
-                    first_value = first_value[:]
-                    first_value.sort()
-                    second_value = second_value[:]
-                    second_value.sort()
-
-                if first_value != second_value:
-                    if attribute.lower() == 'publicsettings':
-                        mistmatch = "[REDACTED] != [REDACTED] (Attribute: {0})".format(".".join(context))
-                    else:
-                        mistmatch = "[{0}] != [{1}] (Attribute: {2})".format(first_value, second_value, ".".join(context))
-                    message = "Mismatch in Goal States [Incarnation {0}] != [Etag: {1}]: {2}".format(from_extensions_config.id, from_vm_settings.id, mistmatch)
-                    raise GoalStateMismatchError(message, attribute)
-            finally:
-                context.pop()
-
-        compare_goal_states(from_extensions_config, from_vm_settings)
-
     def _do_common_validations(self):
         """
         Does validations common to vmSettings and ExtensionsConfig
diff --git a/azurelinuxagent/common/protocol/extensions_goal_state_from_vm_settings.py b/azurelinuxagent/common/protocol/extensions_goal_state_from_vm_settings.py
index 12a23173..aabf1889 100644
--- a/azurelinuxagent/common/protocol/extensions_goal_state_from_vm_settings.py
+++ b/azurelinuxagent/common/protocol/extensions_goal_state_from_vm_settings.py
@@ -81,6 +81,9 @@ class ExtensionsGoalStateFromVmSettings(ExtensionsGoalState):
 
     @property
     def created_on_timestamp(self):
+        """
+        Timestamp assigned by the CRP (time at which the Fast Track goal state was created)
+        """
         return self._created_on_timestamp
 
     @property
diff --git a/azurelinuxagent/common/protocol/goal_state.py b/azurelinuxagent/common/protocol/goal_state.py
index b92e0f96..f4c96043 100644
--- a/azurelinuxagent/common/protocol/goal_state.py
+++ b/azurelinuxagent/common/protocol/goal_state.py
@@ -15,7 +15,7 @@
 # limitations under the License.
 #
 # Requires Python 2.6+ and Openssl 1.0+
-
+import datetime
 import os
 import re
 import time
@@ -24,12 +24,13 @@ import azurelinuxagent.common.conf as conf
 import azurelinuxagent.common.logger as logger
 from azurelinuxagent.common.AgentGlobals import AgentGlobals
 from azurelinuxagent.common.datacontract import set_properties
-from azurelinuxagent.common.exception import IncompleteGoalStateError
-from azurelinuxagent.common.exception import ProtocolError
+from azurelinuxagent.common.exception import ProtocolError, ResourceGoneError
 from azurelinuxagent.common.future import ustr
 from azurelinuxagent.common.protocol.extensions_goal_state_factory import ExtensionsGoalStateFactory
+from azurelinuxagent.common.protocol.hostplugin import VmSettingsNotSupported
 from azurelinuxagent.common.protocol.restapi import Cert, CertList, RemoteAccessUser, RemoteAccessUsersList
 from azurelinuxagent.common.utils import fileutil
+from azurelinuxagent.common.utils.archive import GoalStateHistory
 from azurelinuxagent.common.utils.cryptutil import CryptUtil
 from azurelinuxagent.common.utils.textutil import parse_doc, findall, find, findtext, getattrib
 
@@ -40,7 +41,7 @@ PEM_FILE_NAME = "Certificates.pem"
 TRANSPORT_CERT_FILE_NAME = "TransportCert.pem"
 TRANSPORT_PRV_FILE_NAME = "TransportPrivate.pem"
 
-_NUM_GS_FETCH_RETRIES = 6
+_GET_GOAL_STATE_MAX_ATTEMPTS = 6
 
 
 class GoalState(object):
@@ -48,73 +49,231 @@ class GoalState(object):
         """
         Fetches the goal state using the given wire client.
 
-        __init__ fetches only the goal state itself, not including inner properties such as ExtensionsConfig; to fetch the entire goal state
-        use the fetch_full_goal_state().
+        Fetching the goal state involves several HTTP requests to the WireServer and the HostGAPlugin. There is an initial request to WireServer's goalstate API,
+        which response includes the incarnation, role instance, container ID, role config, and URIs to the rest of the goal state (ExtensionsConfig, Certificates,
+        Remote Access users, etc.). Additional requests are done using those URIs (all of them point to APIs in the WireServer). Additionally, there is a
+        request to the HostGAPlugin for the vmSettings, which determines the goal state for extensions when using the Fast Track pipeline.
+
+        To reduce the number of requests, when possible, create a single instance of GoalState and use the update() method to keep it up to date.
+        """
+        try:
+            self._wire_client = wire_client
+
+            # These "basic" properties come from the initial request to WireServer's goalstate API
+            self._timestamp = None
+            self._incarnation = None
+            self._role_instance_id = None
+            self._role_config_name = None
+            self._container_id = None
+
+            xml_text, xml_doc = GoalState._fetch_goal_state(self._wire_client)
+
+            self._initialize_basic_properties(xml_doc)
+
+            # The goal state for extensions can come from vmSettings when using FastTrack or from extensionsConfig otherwise, self._fetch_extended_goal_state
+            # populates the '_extensions_goal_state' property.
+            self._extensions_goal_state = None
+            vm_settings = self._fetch_vm_settings()
+
+            # These "extended" properties come from additional HTTP requests to the URIs included in the basic goal state
+            self._hosting_env = None
+            self._shared_conf = None
+            self._certs = None
+            self._remote_access = None
+
+            self._fetch_extended_goal_state(xml_text, xml_doc, vm_settings)
+
+        except Exception as exception:
+            # We don't log the error here since fetching the goal state is done every few seconds
+            raise ProtocolError(msg="Error fetching goal state", inner=exception)
+
+    @property
+    def timestamp(self):
+        return self._timestamp
+
+    @property
+    def incarnation(self):
+        return self._incarnation
+
+    @property
+    def container_id(self):
+        return self._container_id
+
+    @property
+    def role_instance_id(self):
+        return self._role_instance_id
+
+    @property
+    def role_config_name(self):
+        return self._role_config_name
+
+    @property
+    def extensions_goal_state(self):
+        return self._extensions_goal_state
+
+    @property
+    def certs(self):
+        return self._certs
+
+    @property
+    def hosting_env(self):
+        return self._hosting_env
+
+    @property
+    def shared_conf(self):
+        return self._shared_conf
+
+    @property
+    def remote_access(self):
+        return self._remote_access
+
+    @staticmethod
+    def update_host_plugin_headers(wire_client):
+        """
+        Updates the container ID and role config name that are send in the headers of HTTP requests to the HostGAPlugin
+        """
+        # Fetching the goal state updates the HostGAPlugin so simply trigger the request
+        GoalState._fetch_goal_state(wire_client)
+
+    def update(self, force_update=False):
+        """
+        Updates the current GoalState instance fetching values from the WireServer/HostGAPlugin as needed
+        """
+        xml_text, xml_doc = GoalState._fetch_goal_state(self._wire_client)
+
+        vm_settings = self._fetch_vm_settings(force_update=force_update)
+
+        if force_update or self._incarnation != findtext(xml_doc, "Incarnation"):
+            # update the extended goal state, using vm_settings for the extensions (unless they are None, then use extensionsConfig)
+            self._initialize_basic_properties(xml_doc)
+            self._fetch_extended_goal_state(xml_text, xml_doc,  vm_settings)
+        else:
+            # else just ensure the extensions are using the latest vm_settings
+            if vm_settings is not None:
+                self._extensions_goal_state = vm_settings
+
+    def save_to_history(self, data, file_name):
+        self._history.save(data, file_name)
+
+    def _initialize_basic_properties(self, xml_doc):
+        self._timestamp = datetime.datetime.utcnow().isoformat()
+        self._incarnation = findtext(xml_doc, "Incarnation")
+        self._history = GoalStateHistory(self._timestamp, self._incarnation)  # history for the WireServer goal state; vmSettings are separate
+        role_instance = find(xml_doc, "RoleInstance")
+        self._role_instance_id = findtext(role_instance, "InstanceId")
+        role_config = find(role_instance, "Configuration")
+        self._role_config_name = findtext(role_config, "ConfigName")
+        container = find(xml_doc, "Container")
+        self._container_id = findtext(container, "ContainerId")
+
+    @staticmethod
+    def _fetch_goal_state(wire_client):
+        """
+        Issues an HTTP request for the goal state (WireServer) and returns a tuple containing the response as text and as an XML Document
         """
         uri = GOAL_STATE_URI.format(wire_client.get_endpoint())
 
-        for _ in range(0, _NUM_GS_FETCH_RETRIES):
-            self.xml_text = wire_client.fetch_config(uri, wire_client.get_header())
-            xml_doc = parse_doc(self.xml_text)
-            self.incarnation = findtext(xml_doc, "Incarnation")
+        # In some environments a few goal state requests return a missing RoleInstance; these retries are used to work around that issue
+        # TODO: Consider retrying on 410 (ResourceGone) as well
+        for _ in range(0, _GET_GOAL_STATE_MAX_ATTEMPTS):
+            xml_text = wire_client.fetch_config(uri, wire_client.get_header())
+            xml_doc = parse_doc(xml_text)
 
             role_instance = find(xml_doc, "RoleInstance")
             if role_instance:
                 break
             time.sleep(0.5)
         else:
-            raise IncompleteGoalStateError("Fetched goal state without a RoleInstance [incarnation {inc}]".format(inc=self.incarnation))
+            incarnation = findtext(xml_doc, "Incarnation")
+            raise ProtocolError("Fetched goal state without a RoleInstance [incarnation {inc}]".format(inc=incarnation))
 
-        try:
-            self.role_instance_id = findtext(role_instance, "InstanceId")
-            role_config = find(role_instance, "Configuration")
-            self.role_config_name = findtext(role_config, "ConfigName")
-            container = find(xml_doc, "Container")
-            self.container_id = findtext(container, "ContainerId")
-
-            AgentGlobals.update_container_id(self.container_id)
-
-            # these properties are populated by fetch_full_goal_state()
-            self._hosting_env_uri = findtext(xml_doc, "HostingEnvironmentConfig")
-            self.hosting_env = None
-            self._shared_conf_uri = findtext(xml_doc, "SharedConfig")
-            self.shared_conf = None
-            self._certs_uri = findtext(xml_doc, "Certificates")
-            self.certs = None
-            self._remote_access_uri = findtext(container, "RemoteAccessInfo")
-            self.remote_access = None
-            # TODO: extensions_config is an instance member only temporarily. Once we stop comparing extensionsConfig with
-            # vmSettings, it will be replaced with the extensions goal state
-            self.extensions_config = None
-            self._extensions_config_uri = findtext(xml_doc, "ExtensionsConfig")
+        # Telemetry and the HostGAPlugin depend on the container id/role config; keep them up-to-date each time we fetch the goal state
+        # (note that these elements can change even if the incarnation of the goal state does not change)
+        container = find(xml_doc, "Container")
+        container_id = findtext(container, "ContainerId")
+        role_config = find(role_instance, "Configuration")
+        role_config_name = findtext(role_config, "ConfigName")
 
-        except Exception as exception:
-            # We don't log the error here since fetching the goal state is done every few seconds
-            raise ProtocolError(msg="Error fetching goal state", inner=exception)
+        AgentGlobals.update_container_id(container_id)  # Telemetry uses this global to pick up the container id
 
-    def fetch_full_goal_state(self, wire_client):
-        try:
-            logger.info('Fetching goal state [incarnation {0}]', self.incarnation)
+        wire_client.update_host_plugin(container_id, role_config_name)
+
+        return xml_text, xml_doc
+
+    def _fetch_vm_settings(self, force_update=False):
+        """
+        Issues an HTTP request (HostGAPlugin) for the vm settings and returns the response as an ExtensionsGoalStateFromVmSettings.
+        """
+        vm_settings, vm_settings_updated = (None, False)
+
+        if conf.get_enable_fast_track():
+            try:
+                vm_settings, vm_settings_updated = self._wire_client.get_host_plugin().fetch_vm_settings(force_update=force_update)
+
+            except VmSettingsNotSupported:
+                pass
+            except ResourceGoneError:
+                # retry after refreshing the HostGAPlugin
+                GoalState.update_host_plugin_headers(self._wire_client)
+                vm_settings, vm_settings_updated = self._wire_client.get_host_plugin().fetch_vm_settings(force_update=force_update)
+
+            if vm_settings_updated:
+                # The vmSettings are updated independently of the WireServer goal state and they are saved to a separate directory
+                history = GoalStateHistory(datetime.datetime.utcnow().isoformat(), vm_settings.etag)
+                history.save_vm_settings(vm_settings.get_redacted_text())
 
-            xml_text = wire_client.fetch_config(self._hosting_env_uri, wire_client.get_header())
-            self.hosting_env = HostingEnv(xml_text)
+        return vm_settings
 
-            xml_text = wire_client.fetch_config(self._shared_conf_uri, wire_client.get_header())
-            self.shared_conf = SharedConfig(xml_text)
+    def _fetch_extended_goal_state(self, xml_text, xml_doc, vm_settings):
+        """
+        Issues HTTP requests (WireServer) for each of the URIs in the goal state (ExtensionsConfig, Certificate, Remote Access users, etc)
+        and populates the corresponding properties. If the given 'vm_settings' are not None they are used for the extensions goal state,
+        otherwise extensionsConfig is used instead.
+        """
+        try:
+            logger.info('Fetching goal state [incarnation {0}]', self._incarnation)
 
-            if self._certs_uri is not None:
-                xml_text = wire_client.fetch_config(self._certs_uri, wire_client.get_header_for_cert())
-                self.certs = Certificates(xml_text)
+            self._history.save_goal_state(xml_text)
 
-            if self._remote_access_uri is not None:
-                xml_text = wire_client.fetch_config(self._remote_access_uri, wire_client.get_header_for_cert())
-                self.remote_access = RemoteAccess(xml_text)
+            # TODO: at this point we always fetch the extensionsConfig, even if it is not needed, and save it for debugging purposes. Once
+            #       FastTrack is stable this code can be updated to fetch it only when actually needed.
+            extensions_config_uri = findtext(xml_doc, "ExtensionsConfig")
 
-            if self._extensions_config_uri is None:
-                self.extensions_config = ExtensionsGoalStateFactory.create_empty()
+            if extensions_config_uri is None:
+                extensions_config = ExtensionsGoalStateFactory.create_empty()
             else:
-                xml_text = wire_client.fetch_config(self._extensions_config_uri, wire_client.get_header())
-                self.extensions_config = ExtensionsGoalStateFactory.create_from_extensions_config(self.incarnation, xml_text, wire_client)
+                xml_text = self._wire_client.fetch_config(extensions_config_uri, self._wire_client.get_header())
+                extensions_config = ExtensionsGoalStateFactory.create_from_extensions_config(self._incarnation, xml_text, self._wire_client)
+                self._history.save_extensions_config(extensions_config.get_redacted_text())
+
+            if vm_settings is not None:
+                self._extensions_goal_state = vm_settings
+            else:
+                self._extensions_goal_state = extensions_config
+
+            hosting_env_uri = findtext(xml_doc, "HostingEnvironmentConfig")
+            xml_text = self._wire_client.fetch_config(hosting_env_uri, self._wire_client.get_header())
+            self._hosting_env = HostingEnv(xml_text)
+            self._history.save_hosting_env(xml_text)
+
+            shared_conf_uri = findtext(xml_doc, "SharedConfig")
+            xml_text = self._wire_client.fetch_config(shared_conf_uri, self._wire_client.get_header())
+            self._shared_conf = SharedConfig(xml_text)
+            self._history.save_shared_conf(xml_text)
+
+            certs_uri = findtext(xml_doc, "Certificates")
+            if certs_uri is not None:
+                # Note that we do not save the certificates to the goal state history
+                xml_text = self._wire_client.fetch_config(certs_uri, self._wire_client.get_header_for_cert())
+                self._certs = Certificates(xml_text)
+
+            container = find(xml_doc, "Container")
+            remote_access_uri = findtext(container, "RemoteAccessInfo")
+            if remote_access_uri is not None:
+                xml_text = self._wire_client.fetch_config(remote_access_uri, self._wire_client.get_header_for_cert())
+                self._remote_access = RemoteAccess(xml_text)
+                self._history.save_remote_access(xml_text)
+
         except Exception as exception:
             logger.warn("Fetching the goal state failed: {0}", ustr(exception))
             raise ProtocolError(msg="Error fetching goal state", inner=exception)
diff --git a/azurelinuxagent/common/protocol/hostplugin.py b/azurelinuxagent/common/protocol/hostplugin.py
index 31981612..81d9b2f0 100644
--- a/azurelinuxagent/common/protocol/hostplugin.py
+++ b/azurelinuxagent/common/protocol/hostplugin.py
@@ -25,9 +25,11 @@ import uuid
 from azurelinuxagent.common import logger
 from azurelinuxagent.common.errorstate import ErrorState, ERROR_STATE_HOST_PLUGIN_FAILURE
 from azurelinuxagent.common.event import WALAEventOperation, add_event
-from azurelinuxagent.common.exception import HttpError, ProtocolError
-from azurelinuxagent.common.future import ustr
+from azurelinuxagent.common.exception import HttpError, ProtocolError, ResourceGoneError
+from azurelinuxagent.common.utils.flexible_version import FlexibleVersion
+from azurelinuxagent.common.future import ustr, httpclient
 from azurelinuxagent.common.protocol.healthservice import HealthService
+from azurelinuxagent.common.protocol.extensions_goal_state_factory import ExtensionsGoalStateFactory
 from azurelinuxagent.common.utils import restutil
 from azurelinuxagent.common.utils import textutil
 from azurelinuxagent.common.utils.textutil import remove_bom
@@ -63,22 +65,31 @@ class HostPluginProtocol(object):
     FETCH_REPORTING_PERIOD = datetime.timedelta(minutes=1)
     STATUS_REPORTING_PERIOD = datetime.timedelta(minutes=1)
 
-    def __init__(self, endpoint, container_id, role_config_name):
+    def __init__(self, endpoint):
+        """
+        NOTE: Before using the HostGAPlugin be sure to invoke GoalState.update_host_plugin_headers() to initialize
+              the container id and role config name
+        """
         if endpoint is None:
             raise ProtocolError("HostGAPlugin: Endpoint not provided")
         self.is_initialized = False
         self.is_available = False
         self.api_versions = None
         self.endpoint = endpoint
-        self.container_id = container_id
-        self.deployment_id = self._extract_deployment_id(role_config_name)
-        self.role_config_name = role_config_name
+        self.container_id = None
+        self.deployment_id = None
+        self.role_config_name = None
         self.manifest_uri = None
         self.health_service = HealthService(endpoint)
         self.fetch_error_state = ErrorState(min_timedelta=ERROR_STATE_HOST_PLUGIN_FAILURE)
         self.status_error_state = ErrorState(min_timedelta=ERROR_STATE_HOST_PLUGIN_FAILURE)
         self.fetch_last_timestamp = None
         self.status_last_timestamp = None
+        self._host_plugin_version = FlexibleVersion("0.0.0.0")  # Version 0 means "unknown"
+        self._host_plugin_supports_vm_settings = False
+        self._host_plugin_supports_vm_settings_next_check = datetime.datetime.now()
+        self._vm_settings_error_reporter = _VmSettingsErrorReporter()
+        self._cached_vm_settings = None  # Cached value of the most recent ExtensionsGoalStateFromVmSettings
 
     @staticmethod
     def _extract_deployment_id(role_config_name):
@@ -383,3 +394,174 @@ class HostPluginProtocol(object):
         if PY_VERSION_MAJOR > 2:
             return s.decode('utf-8')
         return s
+
+    def fetch_vm_settings(self, force_update):
+        """
+        Queries the vmSettings from the HostGAPlugin and returns an (ExtensionsGoalStateFromVmSettings, bool) tuple with the vmSettings and
+        a boolean indicating if they are an updated (True) or a cached value (False).
+
+        Raises VmSettingsNotSupported if the HostGAPlugin does not support the vmSettings API, ResourceGoneError if the container ID and roleconfig name
+        need to be refreshed, or ProtocolError if the request fails for any other reason (e.g. not supported, time out, server error).
+        """
+        def raise_not_supported(reset_state=False):
+            if reset_state:
+                self._host_plugin_supports_vm_settings = False
+                self._host_plugin_supports_vm_settings_next_check = datetime.datetime.now() + datetime.timedelta(hours=6)  # check again in 6 hours
+                # "Not supported" is not considered an error, so don't use self._vm_settings_error_reporter to report it
+                logger.info("vmSettings is not supported")
+                add_event(op=WALAEventOperation.HostPlugin, message="vmSettings is not supported", is_success=True)
+            raise VmSettingsNotSupported()
+
+        def format_message(msg):
+            return "GET vmSettings [correlation ID: {0} eTag: {1}]: {2}".format(correlation_id, etag, msg)
+
+        try:
+            # Raise if VmSettings are not supported but check for periodically since the HostGAPlugin could have been updated since the last check
+            if not self._host_plugin_supports_vm_settings and self._host_plugin_supports_vm_settings_next_check > datetime.datetime.now():
+                raise_not_supported()
+
+            etag = None if force_update or self._cached_vm_settings is None else self._cached_vm_settings.etag
+            correlation_id = str(uuid.uuid4())
+
+            self._vm_settings_error_reporter.report_request()
+
+            url, headers = self.get_vm_settings_request(correlation_id)
+            if etag is not None:
+                headers['if-none-match'] = etag
+            response = restutil.http_get(url, headers=headers, use_proxy=False, max_retry=1, return_raw_response=True)
+
+            if response.status == httpclient.GONE:
+                raise ResourceGoneError()
+
+            if response.status == httpclient.NOT_FOUND:  # the HostGAPlugin does not support FastTrack
+                raise_not_supported(reset_state=True)
+
+            if response.status == httpclient.NOT_MODIFIED:  # The goal state hasn't changed, return the current instance
+                return self._cached_vm_settings, False
+
+            if response.status != httpclient.OK:
+                error_description = restutil.read_response_error(response)
+                # For historical reasons the HostGAPlugin returns 502 (BAD_GATEWAY) for internal errors instead of using
+                # 500 (INTERNAL_SERVER_ERROR). We add a short prefix to the error message in the hope that it will help
+                # clear any confusion produced by the poor choice of status code.
+                if response.status == httpclient.BAD_GATEWAY:
+                    error_description = "[Internal error in HostGAPlugin] {0}".format(error_description)
+                error_description = format_message(error_description)
+
+                if 400 <= response.status <= 499:
+                    self._vm_settings_error_reporter.report_error(error_description, _VmSettingsError.ClientError)
+                elif 500 <= response.status <= 599:
+                    self._vm_settings_error_reporter.report_error(error_description, _VmSettingsError.ServerError)
+                else:
+                    self._vm_settings_error_reporter.report_error(error_description)
+
+                raise ProtocolError(error_description)
+
+            for h in response.getheaders():
+                if h[0].lower() == 'etag':
+                    response_etag = h[1]
+                    break
+            else:  # since the vmSettings were updated, the response must include an etag
+                message = format_message("The vmSettings response does not include an Etag header")
+                self._vm_settings_error_reporter.report_error(message)
+                raise ProtocolError(message)
+
+            response_content = ustr(response.read(), encoding='utf-8')
+
+            vm_settings = ExtensionsGoalStateFactory.create_from_vm_settings(response_etag, response_content)
+
+            # log the HostGAPlugin version
+            if vm_settings.host_ga_plugin_version != self._host_plugin_version:
+                self._host_plugin_version = vm_settings.host_ga_plugin_version
+                message = "HostGAPlugin version: {0}".format(vm_settings.host_ga_plugin_version)
+                logger.info(message)
+                add_event(op=WALAEventOperation.HostPlugin, message=message, is_success=True)
+
+            # Don't support HostGAPlugin versions older than 115
+            if vm_settings.host_ga_plugin_version < FlexibleVersion("1.0.8.115"):
+                raise_not_supported(reset_state=True)
+
+            logger.info("Fetched new vmSettings [correlation ID: {0} New eTag: {1}]", correlation_id, vm_settings.etag)
+            self._host_plugin_supports_vm_settings = True
+            self._cached_vm_settings = vm_settings
+            return vm_settings, True
+
+        except (ProtocolError, ResourceGoneError, VmSettingsNotSupported):
+            raise
+        except Exception as exception:
+            if isinstance(exception, IOError) and "timed out" in ustr(exception):
+                message = format_message("Timeout")
+                self._vm_settings_error_reporter.report_error(message, _VmSettingsError.Timeout)
+            else:
+                message = format_message("Request failed: {0}".format(textutil.format_exception(exception)))
+                self._vm_settings_error_reporter.report_error(message, _VmSettingsError.RequestFailed)
+            raise ProtocolError(message)
+        finally:
+            self._vm_settings_error_reporter.report_summary()
+
+
+class VmSettingsNotSupported(TypeError):
+    pass
+
+
+class _VmSettingsError(object):
+    ServerError   = 'ServerError'
+    ClientError   = 'ClientError'
+    Timeout       = 'Timeout'
+    RequestFailed = 'RequestFailed'
+
+
+class _VmSettingsErrorReporter(object):
+    _MaxLogErrors = 1  # Max number of errors by period reported to the local log
+    _MaxTelemetryErrors = 3  # Max number of errors  by period reported to telemetry
+    _Period = datetime.timedelta(hours=1)  # How often to report the summary
+
+    def __init__(self):
+        self._reset()
+
+    def _reset(self):
+        self._request_count = 0  # Total number of vmSettings HTTP requests
+        self._error_count = 0   # Total number of errors issuing vmSettings requests (includes all kinds of errors)
+        self._server_error_count = 0  # Count of server side errors (HTTP status in the 500s)
+        self._client_error_count = 0  # Count of client side errors (HTTP status in the 400s)
+        self._timeout_count = 0  # Count of timeouts on vmSettings requests
+        self._request_failure_count = 0  # Total count of requests that could not be issued (does not include timeouts or requests that were actually issued and failed, for example, with 500 or 400 statuses)
+        self._next_period = datetime.datetime.now() + _VmSettingsErrorReporter._Period
+
+    def report_request(self):
+        self._request_count += 1
+
+    def report_error(self, error, category=None):
+        self._error_count += 1
+
+        if self._error_count <= _VmSettingsErrorReporter._MaxLogErrors:
+            logger.info("[VmSettings] [Informational only, the Agent will continue normal operation] {0}", error)
+
+        if self._error_count <= _VmSettingsErrorReporter._MaxTelemetryErrors:
+            add_event(op=WALAEventOperation.VmSettings, message=error, is_success=False, log_event=False)
+
+        if category == _VmSettingsError.ServerError:
+            self._server_error_count += 1
+        elif category == _VmSettingsError.ClientError:
+            self._client_error_count += 1
+        elif category == _VmSettingsError.Timeout:
+            self._timeout_count += 1
+        elif category == _VmSettingsError.RequestFailed:
+            self._request_failure_count += 1
+
+    def report_summary(self):
+        if datetime.datetime.now() >= self._next_period:
+            summary = {
+                "requests":       self._request_count,
+                "errors":         self._error_count,
+                "serverErrors":   self._server_error_count,
+                "clientErrors":   self._client_error_count,
+                "timeouts":       self._timeout_count,
+                "failedRequests": self._request_failure_count
+            }
+            message = json.dumps(summary)
+            add_event(op=WALAEventOperation.VmSettingsSummary, message=message, is_success=False, log_event=False)
+            if self._error_count > 0:
+                logger.info("[VmSettingsSummary] {0}", message)
+
+            self._reset()
diff --git a/azurelinuxagent/common/protocol/wire.py b/azurelinuxagent/common/protocol/wire.py
index cda6381f..7ee8e372 100644
--- a/azurelinuxagent/common/protocol/wire.py
+++ b/azurelinuxagent/common/protocol/wire.py
@@ -20,7 +20,6 @@ import json
 import os
 import random
 import time
-import uuid
 import xml.sax.saxutils as saxutils
 from collections import defaultdict
 from datetime import datetime, timedelta
@@ -35,19 +34,16 @@ 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, GoalStateMismatchError
-from azurelinuxagent.common.protocol.extensions_goal_state_factory import ExtensionsGoalStateFactory
 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, \
     ExtHandlerPackageList, ProvisionStatus, VMInfo, VMStatus
 from azurelinuxagent.common.telemetryevent import GuestAgentExtensionEventsSchema
 from azurelinuxagent.common.utils import fileutil, restutil
-from azurelinuxagent.common.utils.archive import StateFlusher
+from azurelinuxagent.common.utils.archive import _MANIFEST_FILE_NAME
 from azurelinuxagent.common.utils.cryptutil import CryptUtil
-from azurelinuxagent.common.utils.flexible_version import FlexibleVersion
 from azurelinuxagent.common.utils.textutil import parse_doc, findall, find, \
-    findtext, gettext, remove_bom, get_bytes_from_pem, parse_json, format_exception
+    findtext, gettext, remove_bom, get_bytes_from_pem, parse_json
 from azurelinuxagent.common.version import AGENT_NAME, CURRENT_VERSION
 
 VERSION_INFO_URI = "http://{0}/?comp=versions"
@@ -55,16 +51,6 @@ HEALTH_REPORT_URI = "http://{0}/machine?comp=health"
 ROLE_PROP_URI = "http://{0}/machine?comp=roleProperties"
 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"
-EXT_CONF_FILE_NAME = "ExtensionsConfig.{0}.xml"
-MANIFEST_FILE_NAME = "{0}.{1}.manifest.xml"
-
 PROTOCOL_VERSION = "2012-11-30"
 ENDPOINT_FINE_NAME = "WireServer"
 
@@ -127,7 +113,7 @@ class WireProtocol(DataContract):
 
     def get_vmagent_manifests(self):
         goal_state = self.client.get_goal_state()
-        ext_conf = self.client.get_extensions_goal_state()
+        ext_conf = goal_state.extensions_goal_state
         return ext_conf.agent_manifests, goal_state.incarnation
 
     def get_vmagent_pkgs(self, vmagent_manifest):
@@ -141,8 +127,8 @@ class WireProtocol(DataContract):
         man = self.client.get_ext_manifest(ext_handler)
         return man.pkg_list
 
-    def get_extensions_goal_state(self):
-        return self.client.get_extensions_goal_state()
+    def get_goal_state(self):
+        return self.client.get_goal_state()
 
     def _download_ext_handler_pkg_through_host(self, uri, destination):
         host = self.client.get_host_plugin()
@@ -576,15 +562,8 @@ class WireClient(object):
         logger.info("Wire server endpoint:{0}", endpoint)
         self._endpoint = endpoint
         self._goal_state = None
-        self._extensions_goal_state = None  # The goal state to use for extensions; can be an ExtensionsGoalStateFromVmSettings or ExtensionsGoalStateFromExtensionsConfig
-        self._cached_vm_settings = None  # Cached value of the most recent ExtensionsGoalStateFromVmSettings
         self._host_plugin = None
-        self._host_plugin_version = FlexibleVersion("0.0.0.0")  # Version 0 means "unknown"
-        self._host_plugin_supports_vm_settings = False
-        self._host_plugin_supports_vm_settings_next_check = datetime.now()
         self.status_blob = StatusBlob(self)
-        self.goal_state_flusher = StateFlusher(conf.get_lib_dir())
-        self._vm_settings_error_reporter = _VmSettingsErrorReporter()
 
     def get_endpoint(self):
         return self._endpoint
@@ -632,15 +611,6 @@ class WireClient(object):
         except IOError as e:
             raise ProtocolError("Failed to read cache: {0}".format(e))
 
-    @staticmethod
-    def _save_cache(data, file_name):
-        try:
-            file_path = os.path.join(conf.get_lib_dir(), file_name)
-            fileutil.write_file(file_path, data)
-        except IOError as e:
-            fileutil.clean_ioerror(e, paths=[file_name])
-            raise ProtocolError("Failed to write cache: {0}".format(e))
-
     @staticmethod
     def call_storage_service(http_req, *args, **kwargs):
         # Default to use the configured HTTP proxy
@@ -777,244 +747,32 @@ class WireClient(object):
         """
         Fetches a new goal state and updates the Container ID and Role Config Name of the host plugin client
         """
-        goal_state = GoalState(self)
-        self._update_host_plugin(goal_state.container_id, goal_state.role_config_name)
+        if self._host_plugin is not None:
+            GoalState.update_host_plugin_headers(self)
+
+    def update_host_plugin(self, container_id, role_config_name):
+        if self._host_plugin is not None:
+            self._host_plugin.update_container_id(container_id)
+            self._host_plugin.update_role_config_name(role_config_name)
 
-    def update_goal_state(self, force_update=False, is_retry=False):
+    def update_goal_state(self, force_update=False):
         """
         Updates the goal state if the incarnation or etag changed or if 'force_update' is True
         """
         try:
-            #
-            # The entire goal state needs to be retrieved from the WireServer (via the GoalState class), and the HostGAPlugin
-            # (via the self._fetch_vm_settings_goal_state method).
-            #
-            # We fetch it in 3 parts:
-            #
-            # 1) The "main" goal state from the WireServer, which includes the incarnation, container ID, role config, and URLs
-            #    to the rest of the goal state (certificates, remote users, extensions config, etc). We do this first because
-            #    we need to initialize the HostGAPlugin with the container ID and role config.
-            #
-            goal_state = GoalState(self)
-
-            self._update_host_plugin(goal_state.container_id, goal_state.role_config_name)
-
-            #
-            # 2) Then we fetch the vmSettings from the HostGAPlugin. We do this before fetching the rest of the goal state from the
-            #    WireServer to minimize the time between the initial call to the WireServer and the call to the HostGAPlugin (and hence
-            #    reduce the window in which a new goal state may arrive in-between the 2 calls)
-            #
-            vm_settings_goal_state, vm_settings_goal_state_updated = (None, False)
-
-            if conf.get_enable_fast_track():
-                try:
-                    vm_settings_goal_state, vm_settings_goal_state_updated = self._fetch_vm_settings_goal_state(force_update=force_update)
-
-                except Exception as error:
-                    # _fetch_vm_settings_goal_state() does its own detailed error reporting and raises ProtocolError; do not report those
-                    if not isinstance(error, ProtocolError):
-                        self._vm_settings_error_reporter.report_error(format_exception(error))
-                self._vm_settings_error_reporter.report_summary()
-
-            #
-            # 3) Lastly we, fetch the rest of the goal state from the WireServer (but ony if needed: initialization, a "forced" update, or
-            #    a change in the incarnation). Note that if we fetch the full goal state we also update self._goal_state.
-            #
             if force_update:
                 logger.info("Forcing an update of the goal state..")
 
-            fetch_full_goal_state = force_update or self._goal_state is None or self._goal_state.incarnation != goal_state.incarnation
-
-            if not fetch_full_goal_state:
-                goal_state_updated = False
+            if self._goal_state is None or force_update:
+                self._goal_state = GoalState(self)
             else:
-                goal_state.fetch_full_goal_state(self)
-                self._goal_state = goal_state
-                goal_state_updated = True
-
-            #
-            # If we fetched the vmSettings then compare them against extensionsConfig and use them for the extensions goal state if
-            # everything matches, otherwise use extensionsConfig.
-            #
-            use_vm_settings = False
-            if vm_settings_goal_state is not None:
-                if not goal_state_updated and not vm_settings_goal_state_updated:  # no need to compare them, just use vmSettings
-                    use_vm_settings = True
-                else:
-                    try:
-                        ExtensionsGoalState.compare(self._goal_state.extensions_config, vm_settings_goal_state)
-                        use_vm_settings = True
-                    except GoalStateMismatchError as mismatch:
-                        if not is_retry and mismatch.attribute in ("created_on_timestamp", "activity_id"):
-                            # this may be OK; a new goal state may have arrived in-between the calls to the HostGAPlugin and the WireServer;
-                            # retry one time after a delay and then report the error if it happens again.
-                            time.sleep(conf.get_goal_state_period())
-                            self.update_goal_state(is_retry=True)
-                            return
-                        self._vm_settings_error_reporter.report_error(ustr(mismatch))
-                        self._vm_settings_error_reporter.report_summary()
-
-            if use_vm_settings:
-                self._extensions_goal_state = vm_settings_goal_state
-            else:
-                self._extensions_goal_state = self._goal_state.extensions_config
-
-            #
-            # If either goal state changed (goal_state or vm_settings_goal_state) save them
-            #
-            if goal_state_updated or vm_settings_goal_state_updated:
-                self._save_goal_state()
+                self._goal_state.update()
 
         except ProtocolError:
             raise
         except Exception as exception:
             raise ProtocolError("Error fetching goal state: {0}".format(ustr(exception)))
 
-    def _fetch_vm_settings_goal_state(self, force_update):
-        """
-        Queries the vmSettings from the HostGAPlugin and returns an (ExtensionsGoalStateFromVmSettings, bool) tuple with the vmSettings and
-        a boolean indicating if they are an updated (True) or a cached value (False).
-
-        Raises ProtocolError if the request fails for any reason (e.g. not supported, time out, server error)
-        """
-        def raise_not_supported(reset_state=False):
-            if reset_state:
-                self._host_plugin_supports_vm_settings = False
-                self._host_plugin_supports_vm_settings_next_check = datetime.now() + timedelta(hours=6)  # check again in 6 hours
-                # "Not supported" is not considered an error, so don't use self._vm_settings_error_reporter to report it
-                logger.info("vmSettings is not supported")
-                add_event(op=WALAEventOperation.HostPlugin, message="vmSettings is not supported", is_success=True)
-            raise ProtocolError("VmSettings not supported")
-
-        # Raise if VmSettings are not supported but check for periodically since the HostGAPlugin could have been updated since the last check
-        if not self._host_plugin_supports_vm_settings and self._host_plugin_supports_vm_settings_next_check > datetime.now():
-            raise_not_supported()
-
-        etag = None if force_update or self._cached_vm_settings is None else self._cached_vm_settings.etag
-        correlation_id = str(uuid.uuid4())
-
-        def format_message(msg):
-            return "GET vmSettings [correlation ID: {0} eTag: {1} HGAP: {2}]: {3}".format(correlation_id, etag, self._host_plugin_version, msg)
-
-        try:
-            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 restutil.http_get(url, headers=headers, use_proxy=False, max_retry=1, return_raw_response=True)
-
-            self._vm_settings_error_reporter.report_request()
-
-            response = get_vm_settings()
-
-            if response.status == httpclient.GONE:  # retry after refreshing the HostGAPlugin
-                self.update_host_plugin_from_goal_state()
-                response = get_vm_settings()
-
-            if response.status == httpclient.NOT_FOUND:  # the HostGAPlugin does not support FastTrack
-                raise_not_supported(reset_state=True)
-
-            if response.status == httpclient.NOT_MODIFIED:  # The goal state hasn't changed, return the current instance
-                return self._cached_vm_settings, False
-
-            if response.status != httpclient.OK:
-                error_description = restutil.read_response_error(response)
-                # For historical reasons the HostGAPlugin returns 502 (BAD_GATEWAY) for internal errors instead of using
-                # 500 (INTERNAL_SERVER_ERROR). We add a short prefix to the error message in the hope that it will help
-                # clear any confusion produced by the poor choice of status code.
-                if response.status == httpclient.BAD_GATEWAY:
-                    error_description = "[Internal error in HostGAPlugin] {0}".format(error_description)
-                error_description = format_message(error_description)
-
-                if 400 <= response.status <= 499:
-                    self._vm_settings_error_reporter.report_error(error_description, _VmSettingsError.ClientError)
-                elif 500 <= response.status <= 599:
-                    self._vm_settings_error_reporter.report_error(error_description, _VmSettingsError.ServerError)
-                else:
-                    self._vm_settings_error_reporter.report_error(error_description)
-
-                raise ProtocolError(error_description)
-
-            for h in response.getheaders():
-                if h[0].lower() == 'etag':
-                    response_etag = h[1]
-                    break
-            else:  # since the vmSettings were updated, the response must include an etag
-                message = format_message("The vmSettings response does not include an Etag header")
-                self._vm_settings_error_reporter.report_error(message)
-                raise ProtocolError(message)
-
-            response_content = self.decode_config(response.read())
-            vm_settings = ExtensionsGoalStateFactory.create_from_vm_settings(response_etag, response_content)
-
-            # log the HostGAPlugin version
-            if vm_settings.host_ga_plugin_version != self._host_plugin_version:
-                self._host_plugin_version = vm_settings.host_ga_plugin_version
-                message = "HostGAPlugin version: {0}".format(vm_settings.host_ga_plugin_version)
-                logger.info(message)
-                add_event(op=WALAEventOperation.HostPlugin, message=message, is_success=True)
-
-            # Don't support HostGAPlugin versions older than 115
-            if vm_settings.host_ga_plugin_version < FlexibleVersion("1.0.8.115"):
-                raise_not_supported(reset_state=True)
-
-            logger.info("Fetched new vmSettings [correlation ID: {0} New eTag: {1}]", correlation_id, vm_settings.etag)
-            self._host_plugin_supports_vm_settings = True
-            self._cached_vm_settings = vm_settings
-            return vm_settings, True
-
-        except ProtocolError:
-            raise
-        except Exception as exception:
-            if isinstance(exception, IOError) and "timed out" in ustr(exception):
-                message = format_message("Timeout")
-                self._vm_settings_error_reporter.report_error(message, _VmSettingsError.Timeout)
-            else:
-                message = format_message("Request failed: {0}".format(textutil.format_exception(exception)))
-                self._vm_settings_error_reporter.report_error(message, _VmSettingsError.RequestFailed)
-            raise ProtocolError(message)
-
-    def _update_host_plugin(self, container_id, role_config_name):
-        if self._host_plugin is not None:
-            self._host_plugin.update_container_id(container_id)
-            self._host_plugin.update_role_config_name(role_config_name)
-
-    def _save_goal_state(self):
-        try:
-            self.goal_state_flusher.flush()
-        except Exception as e:
-            logger.warn("Failed to save the previous goal state to the history folder: {0}", ustr(e))
-
-        try:
-            def save_if_not_none(goal_state_property, file_name):
-                if goal_state_property is not None and goal_state_property.xml_text is not None:
-                    self._save_cache(goal_state_property.xml_text, file_name)
-
-            # NOTE: Certificates are saved in Certificate.__init__
-            self._save_cache(self._goal_state.incarnation, INCARNATION_FILE_NAME)
-            save_if_not_none(self._goal_state, GOAL_STATE_FILE_NAME.format(self._goal_state.incarnation))
-            save_if_not_none(self._goal_state.hosting_env, HOSTING_ENV_FILE_NAME)
-            save_if_not_none(self._goal_state.shared_conf, SHARED_CONF_FILE_NAME)
-            save_if_not_none(self._goal_state.remote_access, REMOTE_ACCESS_FILE_NAME.format(self._goal_state.incarnation))
-            if self._goal_state.extensions_config is not None:
-                text = self._goal_state.extensions_config.get_redacted_text()
-                if text != '':
-                    self._save_cache(text, EXT_CONF_FILE_NAME.format(self._goal_state.extensions_config.incarnation))
-            # TODO: When Fast Track is fully enabled self._cached_vm_settings will go away and this can be deleted
-            if self._cached_vm_settings is not None:
-                text = self._cached_vm_settings.get_redacted_text()
-                if text != '':
-                    self._save_cache(text, VM_SETTINGS_FILE_NAME.format(self._cached_vm_settings.id))
-            # END TODO
-
-        except Exception as e:
-            logger.warn("Failed to save the goal state to disk: {0}", ustr(e))
-
-    def _set_host_plugin(self, new_host_plugin):
-        if new_host_plugin is None:
-            logger.warn("Setting empty Host Plugin object!")
-        self._host_plugin = new_host_plugin
-
     def get_goal_state(self):
         if self._goal_state is None:
             raise ProtocolError("Trying to fetch goal state before initialization!")
@@ -1035,19 +793,13 @@ class WireClient(object):
             raise ProtocolError("Trying to fetch Certificates before initialization!")
         return self._goal_state.certs
 
-    def get_extensions_goal_state(self):
-        if self._extensions_goal_state is None:
-            raise ProtocolError("Trying to fetch ExtensionsGoalState before initialization!")
-
-        return self._extensions_goal_state
-
     def get_ext_manifest(self, ext_handler):
         if self._goal_state is None:
             raise ProtocolError("Trying to fetch Extension Manifest before initialization!")
 
         try:
             xml_text = self.fetch_manifest(ext_handler.manifest_uris)
-            self._save_cache(xml_text, MANIFEST_FILE_NAME.format(ext_handler.name, self.get_goal_state().incarnation))
+            self._goal_state.save_to_history(xml_text, _MANIFEST_FILE_NAME.format(ext_handler.name))
             return ExtensionManifest(xml_text)
         except Exception as e:
             raise ExtensionDownloadError("Failed to retrieve extension manifest. Error: {0}".format(ustr(e)))
@@ -1058,12 +810,9 @@ class WireClient(object):
         return self._goal_state.remote_access
 
     def fetch_gafamily_manifest(self, vmagent_manifest, goal_state):
-        local_file = MANIFEST_FILE_NAME.format(vmagent_manifest.family, goal_state.incarnation)
-        local_file = os.path.join(conf.get_lib_dir(), local_file)
-
         try:
             xml_text = self.fetch_manifest(vmagent_manifest.uris)
-            fileutil.write_file(local_file, xml_text)
+            goal_state.save_to_history(xml_text, _MANIFEST_FILE_NAME.format(vmagent_manifest.family))
             return ExtensionManifest(xml_text)
         except Exception as e:
             raise ProtocolError("Failed to retrieve GAFamily manifest. Error: {0}".format(ustr(e)))
@@ -1210,12 +959,12 @@ class WireClient(object):
         return ret
 
     def upload_status_blob(self):
-        extensions_goal_state = self.get_extensions_goal_state()
+        extensions_goal_state = self.get_goal_state().extensions_goal_state
 
         if extensions_goal_state.status_upload_blob is None:
             # the status upload blob is in ExtensionsConfig so force a full goal state refresh
             self.update_goal_state(force_update=True)
-            extensions_goal_state = self.get_extensions_goal_state()
+            extensions_goal_state = self.get_goal_state().extensions_goal_state
 
         if extensions_goal_state.status_upload_blob is None:
             raise ProtocolNotFoundError("Status upload uri is missing")
@@ -1415,12 +1164,12 @@ class WireClient(object):
 
     def get_host_plugin(self):
         if self._host_plugin is None:
-            goal_state = GoalState(self)
-            self._set_host_plugin(HostPluginProtocol(self.get_endpoint(), goal_state.container_id, goal_state.role_config_name))
+            self._host_plugin = HostPluginProtocol(self.get_endpoint())
+            GoalState.update_host_plugin_headers(self)
         return self._host_plugin
 
     def get_on_hold(self):
-        return self.get_extensions_goal_state().on_hold
+        return self.get_goal_state().extensions_goal_state.on_hold
 
     def upload_logs(self, content):
         host_func = lambda: self._upload_logs_through_host(content)
@@ -1527,67 +1276,3 @@ class InVMArtifactsProfile(object):
         if 'onHold' in self.__dict__:
             return str(self.onHold).lower() == 'true'  # pylint: disable=E1101
         return False
-
-
-class _VmSettingsError(object):
-    ServerError   = 'ServerError'
-    ClientError   = 'ClientError'
-    Timeout       = 'Timeout'
-    RequestFailed = 'RequestFailed'
-
-
-class _VmSettingsErrorReporter(object):
-    _MaxLogErrors = 1  # Max number of errors by period reported to the local log
-    _MaxTelemetryErrors = 3  # Max number of errors  by period reported to telemetry
-    _Period = timedelta(hours=1)  # How often to report the summary
-
-    def __init__(self):
-        self._reset()
-
-    def _reset(self):
-        self._request_count = 0  # Total number of vmSettings HTTP requests
-        self._error_count = 0   # Total number of errors issuing vmSettings requests (includes all kinds of errors)
-        self._server_error_count = 0  # Count of server side errors (HTTP status in the 500s)
-        self._client_error_count = 0  # Count of client side errors (HTTP status in the 400s)
-        self._timeout_count = 0  # Count of timeouts on vmSettings requests
-        self._request_failure_count = 0  # Total count of requests that could not be issued (does not include timeouts or requests that were actually issued and failed, for example, with 500 or 400 statuses)
-        self._next_period = datetime.now() + _VmSettingsErrorReporter._Period
-
-    def report_request(self):
-        self._request_count += 1
-
-    def report_error(self, error, category=None):
-        self._error_count += 1
-
-        if self._error_count <= _VmSettingsErrorReporter._MaxLogErrors:
-            logger.info("[VmSettings] [Informational only, the Agent will continue normal operation] {0}", error)
-
-        if self._error_count <= _VmSettingsErrorReporter._MaxTelemetryErrors:
-            add_event(op=WALAEventOperation.VmSettings, message=error, is_success=False, log_event=False)
-
-        if category == _VmSettingsError.ServerError:
-            self._server_error_count += 1
-        elif category == _VmSettingsError.ClientError:
-            self._client_error_count += 1
-        elif category == _VmSettingsError.Timeout:
-            self._timeout_count += 1
-        elif category == _VmSettingsError.RequestFailed:
-            self._request_failure_count += 1
-
-    def report_summary(self):
-        if datetime.now() >= self._next_period:
-            summary = {
-                "requests":       self._request_count,
-                "errors":         self._error_count,
-                "serverErrors":   self._server_error_count,
-                "clientErrors":   self._client_error_count,
-                "timeouts":       self._timeout_count,
-                "failedRequests": self._request_failure_count
-            }
-            # always send telemetry, but log errors only
-            message = json.dumps(summary)
-            add_event(op=WALAEventOperation.VmSettingsSummary, message=message, is_success=False, log_event=False)
-            if self._error_count > 0:
-                logger.info("[VmSettingsSummary] {0}", message)
-
-            self._reset()
diff --git a/azurelinuxagent/common/utils/archive.py b/azurelinuxagent/common/utils/archive.py
index 8e484577..9b60e32a 100644
--- a/azurelinuxagent/common/utils/archive.py
+++ b/azurelinuxagent/common/utils/archive.py
@@ -5,9 +5,9 @@ import os
 import re
 import shutil
 import zipfile
-from datetime import datetime
 
 import azurelinuxagent.common.logger as logger
+import azurelinuxagent.common.conf as conf
 from azurelinuxagent.common.utils import fileutil
 
 # pylint: disable=W0105
@@ -38,110 +38,39 @@ The timestamp is an ISO8601 formatted value.
 
 ARCHIVE_DIRECTORY_NAME = 'history'
 
-_MAX_ARCHIVED_STATES = 50
+_MAX_ARCHIVED_STATES = 100
 
 _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"waagent_status\.(\d+)\.json$")
+    re.compile(r"^(.*)\.(\d+)\.(xml)$", re.IGNORECASE)
 ]
 
-_GOAL_STATE_PATTERN = re.compile(r"^(.*)/GoalState\.(\d+)\.xml$", re.IGNORECASE)
-
-# Old names didn't have incarnation, new ones do. Ensure the regex captures both cases.
-# 2018-04-06T08:21:37.142697_incarnation_N
-# 2018-04-06T08:21:37.142697_incarnation_N.zip
-_ARCHIVE_PATTERNS_DIRECTORY = re.compile(r"^\d{4}\-\d{2}\-\d{2}T\d{2}:\d{2}:\d{2}\.\d+(_incarnation_(\d+))?$$")
-_ARCHIVE_PATTERNS_ZIP = re.compile(r"^\d{4}\-\d{2}\-\d{2}T\d{2}:\d{2}:\d{2}\.\d+(_incarnation_(\d+))?\.zip$")
-
-
-class StateFlusher(object):
-    def __init__(self, lib_dir):
-        self._source = lib_dir
-
-        directory = os.path.join(self._source, ARCHIVE_DIRECTORY_NAME)
-        if not os.path.exists(directory):
-            try:
-                fileutil.mkdir(directory)
-            except OSError as exception:
-                if exception.errno != errno.EEXIST:
-                    logger.error("{0} : {1}", self._source, exception.strerror)
-
-    def flush(self):
-        files = self._get_files_to_archive()
-        if not files:
-            return
-
-        archive_name = self._get_archive_name(files)
-        if archive_name is None:
-            return
-
-        if self._mkdir(archive_name):
-            self._archive(files, archive_name)
-        else:
-            self._purge(files)
-
-    def history_dir(self, name):
-        return os.path.join(self._source, ARCHIVE_DIRECTORY_NAME, name)
-
-    @staticmethod
-    def _get_archive_name(files):
-        """
-        Gets the most recently modified GoalState.*.xml and uses that timestamp and incarnation for the archive name.
-        In a normal workflow, we expect there to be only one GoalState.*.xml at a time, but if the previous one
-        wasn't purged for whatever reason, we take the most recently modified goal state file.
-        If there are no GoalState.*.xml files, we return None.
-        """
-        latest_timestamp_ms = None
-        incarnation = None
-
-        for current_file in files:
-            match = _GOAL_STATE_PATTERN.match(current_file)
-            if not match:
-                continue
-
-            modification_time_ms = os.path.getmtime(current_file)
-            if latest_timestamp_ms is None or latest_timestamp_ms < modification_time_ms:
-                latest_timestamp_ms = modification_time_ms
-                incarnation = match.groups()[1]
-
-        if latest_timestamp_ms is not None and incarnation is not None:
-            return datetime.utcfromtimestamp(latest_timestamp_ms).isoformat() + "_incarnation_{0}".format(incarnation)
-        return None
-
-    def _get_files_to_archive(self):
-        files = []
-        for current_file in os.listdir(self._source):
-            full_path = os.path.join(self._source, current_file)
-            for pattern in _CACHE_PATTERNS:
-                match = pattern.match(current_file)
-                if match is not None:
-                    files.append(full_path)
-                    break
-
-        return files
-
-    def _archive(self, files, timestamp):
-        for current_file in files:
-            dst = os.path.join(self.history_dir(timestamp), os.path.basename(current_file))
-            shutil.move(current_file, dst)
-
-    def _purge(self, files):
-        for current_file in files:
-            os.remove(current_file)
-
-    def _mkdir(self, name):
-        directory = self.history_dir(name)
-
-        try:
-            fileutil.mkdir(directory, mode=0o700)
-            return True
-        except IOError as exception:
-            logger.error("{0} : {1}".format(directory, exception.strerror))
-            return False
-
+#
+# Legacy names
+#   2018-04-06T08:21:37.142697_incarnation_N
+#   2018-04-06T08:21:37.142697_incarnation_N.zip
+#
+# Current names
+#
+#   2018-04-06T08:21:37.142697
+#   2018-04-06T08:21:37.142697.zip
+#   2018-04-06T08:21:37.142697_N
+#   2018-04-06T08:21:37.142697_N.zip
+#
+_ARCHIVE_PATTERNS_DIRECTORY = re.compile(r"^\d{4}\-\d{2}\-\d{2}T\d{2}:\d{2}:\d{2}\.\d+((_incarnation)?_(\d+))?$")
+_ARCHIVE_PATTERNS_ZIP = re.compile(r"^\d{4}\-\d{2}\-\d{2}T\d{2}:\d{2}:\d{2}\.\d+((_incarnation)?_(\d+))?\.zip$")
+
+_GOAL_STATE_FILE_NAME = "GoalState.xml"
+_VM_SETTINGS_FILE_NAME = "VmSettings.json"
+_HOSTING_ENV_FILE_NAME = "HostingEnvironmentConfig.xml"
+_SHARED_CONF_FILE_NAME = "SharedConfig.xml"
+_REMOTE_ACCESS_FILE_NAME = "RemoteAccess.xml"
+_EXT_CONF_FILE_NAME = "ExtensionsConfig.xml"
+_MANIFEST_FILE_NAME = "{0}.manifest.xml"
+
+AGENT_STATUS_FILE = "waagent_status.json"
 
 # TODO: use @total_ordering once RHEL/CentOS and SLES 11 are EOL.
 # @total_ordering first appeared in Python 2.7 and 3.2
@@ -218,13 +147,13 @@ class StateArchiver(object):
                 fileutil.mkdir(self._source, mode=0o700)
             except IOError as exception:
                 if exception.errno != errno.EEXIST:
-                    logger.error("{0} : {1}", self._source, exception.strerror)
+                    logger.warn("{0} : {1}", self._source, exception.strerror)
 
     def purge(self):
         """
         Delete "old" archive directories and .zip archives.  Old
         is defined as any directories or files older than the X
-        newest ones.
+        newest ones. Also, clean up any legacy history files.
         """
         states = self._get_archive_states()
         states.sort(reverse=True)
@@ -232,6 +161,18 @@ class StateArchiver(object):
         for state in states[_MAX_ARCHIVED_STATES:]:
             state.delete()
 
+        # legacy history files
+        for current_file in os.listdir(self._source):
+            full_path = os.path.join(self._source, current_file)
+            for pattern in _CACHE_PATTERNS:
+                match = pattern.match(current_file)
+                if match is not None:
+                    try:
+                        os.remove(full_path)
+                    except Exception as e:
+                        logger.warn("Cannot delete legacy history file '{0}': {1}".format(full_path, e))
+                    break
+
     def archive(self):
         states = self._get_archive_states()
         for state in states:
@@ -250,3 +191,41 @@ class StateArchiver(object):
                 states.append(StateZip(full_path, match.group(0)))
 
         return states
+
+
+class GoalStateHistory(object):
+    def __init__(self, timestamp, tag=None):
+        self._errors = False
+        self._root = os.path.join(conf.get_lib_dir(), ARCHIVE_DIRECTORY_NAME, "{0}_{1}".format(timestamp, tag) if tag is not None else timestamp)
+
+    def save(self, data, file_name):
+        try:
+            if not os.path.exists(self._root):
+                fileutil.mkdir(self._root, mode=0o700)
+            full_file_name = os.path.join(self._root, file_name)
+            fileutil.write_file(full_file_name, data)
+        except IOError as e:
+            if not self._errors:  # report only 1 error per directory
+                self._errors = True
+                logger.warn("Failed to save goal state file {0}: {1} [no additional errors saving the goal state will be reported]".format(file_name, e))
+
+    def save_goal_state(self, text):
+        self.save(text, _GOAL_STATE_FILE_NAME)
+
+    def save_extensions_config(self, text):
+        self.save(text, _EXT_CONF_FILE_NAME)
+
+    def save_vm_settings(self, text):
+        self.save(text, _VM_SETTINGS_FILE_NAME)
+
+    def save_remote_access(self, text):
+        self.save(text, _REMOTE_ACCESS_FILE_NAME)
+
+    def save_hosting_env(self, text):
+        self.save(text, _HOSTING_ENV_FILE_NAME)
+
+    def save_shared_conf(self, text):
+        self.save(text, _SHARED_CONF_FILE_NAME)
+
+    def save_status(self, text):
+        self.save(text, AGENT_STATUS_FILE)
diff --git a/azurelinuxagent/ga/exthandlers.py b/azurelinuxagent/ga/exthandlers.py
index 29133c59..cd87225b 100644
--- a/azurelinuxagent/ga/exthandlers.py
+++ b/azurelinuxagent/ga/exthandlers.py
@@ -50,7 +50,7 @@ from azurelinuxagent.common.future import ustr, is_file_not_found_error
 from azurelinuxagent.common.protocol.restapi import ExtensionStatus, ExtensionSubStatus, Extension, ExtHandlerStatus, \
     VMStatus, GoalStateAggregateStatus, ExtensionState, ExtensionRequestedState, ExtensionSettings
 from azurelinuxagent.common.utils import textutil
-from azurelinuxagent.common.utils.archive import ARCHIVE_DIRECTORY_NAME
+from azurelinuxagent.common.utils.archive import ARCHIVE_DIRECTORY_NAME, AGENT_STATUS_FILE, GoalStateHistory
 from azurelinuxagent.common.utils.flexible_version import FlexibleVersion
 from azurelinuxagent.common.version import AGENT_NAME, CURRENT_VERSION, \
     PY_VERSION_MAJOR, PY_VERSION_MICRO, PY_VERSION_MINOR
@@ -67,7 +67,6 @@ 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.{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
@@ -298,7 +297,7 @@ class ExtHandlersHandler(object):
         etag, activity_id, correlation_id, gs_creation_time = None, None, None, None
 
         try:
-            extensions_goal_state = self.protocol.get_extensions_goal_state()
+            extensions_goal_state = self.protocol.get_goal_state().extensions_goal_state
 
             # self.ext_handlers and etag need to be initialized first, since status reporting depends on them; also
             # we make a deep copy of the extensions, since changes are made to self.ext_handlers while processing the extensions
@@ -344,7 +343,7 @@ class ExtHandlersHandler(object):
             add_event(op=WALAEventOperation.ExtensionProcessing, is_success=(error is None), message=message, log_event=False, duration=duration)
 
     def __get_unsupported_features(self):
-        required_features = self.protocol.client.get_extensions_goal_state().required_features
+        required_features = self.protocol.get_goal_state().extensions_goal_state.required_features
         supported_features = get_agent_supported_features_list_for_crp()
         return [feature for feature in required_features if feature not in supported_features]
 
@@ -453,7 +452,7 @@ class ExtHandlersHandler(object):
             return False
 
         if conf.get_enable_overprovisioning():
-            if self.protocol.get_extensions_goal_state().on_hold:
+            if self.protocol.get_goal_state().extensions_goal_state.on_hold:
                 logger.info("Extension handling is on hold")
                 return False
 
@@ -945,7 +944,7 @@ class ExtHandlersHandler(object):
 
                 self.report_status_error_state.reset()
 
-            self.write_ext_handlers_status_to_info_file(vm_status)
+            self.write_ext_handlers_status_to_info_file(vm_status, incarnation_changed)
 
             return vm_status
 
@@ -959,42 +958,52 @@ class ExtHandlersHandler(object):
                       message=msg)
             return None
 
-    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)
+    def write_ext_handlers_status_to_info_file(self, vm_status, incarnation_changed):
+        status_file = os.path.join(conf.get_lib_dir(), AGENT_STATUS_FILE)
 
-        # 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
+        if os.path.exists(status_file) and incarnation_changed:
+            # On new goal state, move the last status report for the previous goal state to the history folder
+            last_modified = os.path.getmtime(status_file)
+            timestamp = datetime.datetime.utcfromtimestamp(last_modified).isoformat()
+            GoalStateHistory(timestamp).save_status(status_file)
+
+        # Now create/overwrite the status file; this file is kept for debugging purposes only
+        status_blob_text = self.protocol.get_status_blob_data()
+        if status_blob_text is None:
+            status_blob_text = ""
+
+        debug_info = ExtHandlersHandler._get_status_debug_info(vm_status)
+
+        status_file_text = \
+'''{{
+    "__comment__": "The __status__ property is the actual status reported to CRP",
+    "__status__": {0},
+    "__debug__": {1}
+}}
+'''.format(status_blob_text, debug_info)
+
+        fileutil.write_file(status_file, status_file_text)
 
-        # Consuming supports_multi_config info from vm_status. creating a dict out of it for easy lookup in the next step.
+    @staticmethod
+    def _get_status_debug_info(vm_status):
         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')
 
-        handler_aggregate_status = data.get('aggregateStatus', dict()).get('handlerAggregateStatus', dict())
-
-        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)
+        debug_info = {
+            "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()],
+            "supportsMultiConfig": support_multi_config
+        }
 
-        fileutil.write_file(status_path, json.dumps(data))
+        return json.dumps(debug_info)
 
     def report_ext_handler_status(self, vm_status, ext_handler, incarnation_changed):
         ext_handler_i = ExtHandlerInstance(ext_handler, self.protocol)
diff --git a/dcr/scenario_utils/check_waagent_log.py b/dcr/scenario_utils/check_waagent_log.py
index 0dcb8972..be539ae0 100644
--- a/dcr/scenario_utils/check_waagent_log.py
+++ b/dcr/scenario_utils/check_waagent_log.py
@@ -41,9 +41,8 @@ def check_waagent_log_for_errors(waagent_log=AGENT_LOG_FILE, ignore=None):
             'if': lambda _: re.match(r"((sles15\.2)|suse12)\D*", distro, flags=re.IGNORECASE) is not None
         },
         # This warning is expected on when WireServer gives us the incomplete goalstate without roleinstance data
-        # raise IncompleteGoalStateError("Fetched goal state without a RoleInstance [incarnation {inc}]".format(inc=self.incarnation))
         {
-            'message': r"\[IncompleteGoalStateError\] Fetched goal state without a RoleInstance",
+            'message': r"\[ProtocolError\] Fetched goal state without a RoleInstance",
         },
         # The following message is expected to log an error if systemd is not enabled on it
         {
diff --git a/dcr/scenarios/agent-bvt/get_blob_content.py b/dcr/scenarios/agent-bvt/get_blob_content.py
deleted file mode 100644
index 91fcce1d..00000000
--- a/dcr/scenarios/agent-bvt/get_blob_content.py
+++ /dev/null
@@ -1,50 +0,0 @@
-import glob
-import re
-from html.parser import HTMLParser
-from time import sleep
-from urllib.parse import unquote_plus
-from urllib.request import urlopen
-
-
-def show_blob_content(description, key):
-    config_files = glob.glob('/var/lib/waagent/ExtensionsConfig*.xml')
-    if len(config_files) == 0:
-        raise Exception('no extension config files found')
-
-    config_files.sort()
-    with open(config_files[-1], 'r') as fh:
-        config = fh.readlines()
-
-    status_line = list(filter(lambda s: key in s, config))[0]
-    status_pattern = '<{0}.*>(.*\?)(.*)<.*'.format(key)
-    match = re.match(status_pattern, status_line)
-
-    if not match:
-        raise Exception(description + ' not found')
-
-    decoded_url = match.groups()[0]
-    encoded_params = match.groups()[1].split('&amp;')
-    for param in encoded_params:
-        kvp = param.split('=')
-        name = kvp[0]
-        skip = name == 'sig'
-        val = HTMLParser().unescape(unquote_plus(kvp[1])) if not skip else kvp[1]
-        decoded_param = '&{0}={1}'.format(name, val)
-        decoded_url += decoded_param
-
-    print("\n{0} uri: {1}\n".format(description, decoded_url))
-    status = None
-    retries = 3
-    while status is None:
-        try:
-            status = urlopen(decoded_url).read()
-        except Exception as e:
-            if retries > 0:
-                retries -= 1
-                sleep(60)
-            else:
-                # we are only collecting information, so do not fail the test
-                status = 'Error reading {0}: {1}'.format(description, e)
-
-    return "\n{0} content: {1}\n".format(description, status)
-
diff --git a/dcr/scenarios/agent-bvt/run2.py b/dcr/scenarios/agent-bvt/run2.py
index c2cf87e9..a563b62c 100644
--- a/dcr/scenarios/agent-bvt/run2.py
+++ b/dcr/scenarios/agent-bvt/run2.py
@@ -3,7 +3,6 @@ from check_firewall import check_firewall
 from dcr.scenario_utils.check_waagent_log import check_waagent_log_for_errors
 from dcr.scenario_utils.models import get_vm_data_from_env
 from dcr.scenario_utils.test_orchestrator import TestFuncObj, TestOrchestrator
-from get_blob_content import show_blob_content
 from test_agent_basics import check_agent_processes, check_sudoers
 
 if __name__ == '__main__':
@@ -11,8 +10,6 @@ if __name__ == '__main__':
     tests = [
         TestFuncObj("check agent processes", check_agent_processes),
         TestFuncObj("check agent log", check_waagent_log_for_errors),
-        TestFuncObj("Verify status blob", lambda: show_blob_content('Status', 'StatusUploadBlob')),
-        TestFuncObj("Verify status blob", lambda: show_blob_content('InVMArtifacts', 'InVMArtifactsProfileBlob')),
         TestFuncObj("verify extension timing", verify_extension_timing),
         TestFuncObj("Check Firewall", lambda: check_firewall(admin_username)),
         TestFuncObj("Check Sudoers", lambda: check_sudoers(admin_username))


Test patch:
diff --git a/tests/data/hostgaplugin/ext_conf.xml b/tests/data/hostgaplugin/ext_conf.xml
index 166e4a71..eac5d636 100644
--- a/tests/data/hostgaplugin/ext_conf.xml
+++ b/tests/data/hostgaplugin/ext_conf.xml
@@ -59,7 +59,7 @@
     {
       "handlerSettings": {
         "protectedSettingsCertThumbprint": "4C4F304667711036E64AF4894B76EB208A863BD4",
-        "protectedSettings": "MIIBsAYJKoZIhvcNAQcDoIIBoTCCAZ0CAQAxggFpMIIBZQIBADBNMDkxNzA1BgoJkiaJk/IsZAEZFidXaW5kb3dzIEF6dXJlIENSUCBDZXJ0aWZpY2F0ZSBHZW5lcmF0b3ICEFpB/HKM/7evRk+DBz754wUwDQYJKoZIhvcNAQEBBQAEggEADPJwniDeIUXzxNrZCloitFdscQ59Bz1dj9DLBREAiM8jmxM0LLicTJDUv272Qm/4ZQgdqpFYBFjGab/9MX+Ih2x47FkVY1woBkckMaC/QOFv84gbboeQCmJYZC/rZJdh8rCMS+CEPq3uH1PVrvtSdZ9uxnaJ+E4exTPPviIiLIPtqWafNlzdbBt8HZjYaVw+SSe+CGzD2pAQeNttq3Rt/6NjCzrjG8ufKwvRoqnrInMs4x6nnN5/xvobKIBSv4/726usfk8Ug+9Q6Benvfpmre2+1M5PnGTfq78cO3o6mI3cPoBUjp5M0iJjAMGeMt81tyHkimZrEZm6pLa4NQMOEjArBgkqhkiG9w0BBwEwFAYIKoZIhvcNAwcECC5nVaiJaWt+gAhgeYvxUOYHXw==",
+        "protectedSettings": "MIIBsAYJKoZIhvcNAQcDoIIBoTCCAZ0CAQAxggFpMIIBZQIBADBNMDkxNzA1BgoJkiaJk/Microsoft.Azure.Monitor.AzureMonitorLinuxAgent==",
         "publicSettings": {"GCS_AUTO_CONFIG":true}
       }
     }
@@ -72,7 +72,7 @@
     {
       "handlerSettings": {
         "protectedSettingsCertThumbprint": "4C4F304667711036E64AF4894B76EB208A863BD4",
-        "protectedSettings": "MIIBsAYJKoZIhvcNAQcDoIIBoTCCAZ0CAQAxggFpMIIBZQIBADBNMDkxNzA1BgoJkiaJk/IsZAEZFidXaW5kb3dzIEF6dXJlIENSUCBDZXJ0aWZpY2F0ZSBHZW5lcmF0b3ICEFpB/HKM/7evRk+DBz754wUwDQYJKoZIhvcNAQEBBQAEggEADPJwniDeIUXzxNrZCloitFdscQ59Bz1dj9DLBREAiM8jmxM0LLicTJDUv272Qm/4ZQgdqpFYBFjGab/9MX+Ih2x47FkVY1woBkckMaC/QOFv84gbboeQCmJYZC/rZJdh8rCMS+CEPq3uH1PVrvtSdZ9uxnaJ+E4exTPPviIiLIPtqWafNlzdbBt8HZjYaVw+SSe+CGzD2pAQeNttq3Rt/6NjCzrjG8ufKwvRoqnrInMs4x6nnN5/xvobKIBSv4/726usfk8Ug+9Q6Benvfpmre2+1M5PnGTfq78cO3o6mI3cPoBUjp5M0iJjAMGeMt81tyHkimZrEZm6pLa4NQMOEjArBgkqhkiG9w0BBwEwFAYIKoZIhvcNAwcECC5nVaiJaWt+gAhgeYvxUOYHXw==",
+        "protectedSettings": "MIIBsAYJKoZIhvcNAQcDoIIBoTCCAZ0CAQAxggFpMIIBZQIBADBNMDkxNzA1BgoJkiaJk/Microsoft.Azure.Security.Monitoring.AzureSecurityLinuxAgent==",
         "publicSettings": {"enableGenevaUpload":true}
       }
     }
@@ -135,7 +135,7 @@
     {
       "handlerSettings": {
         "protectedSettingsCertThumbprint": "59A10F50FFE2A0408D3F03FE336C8FD5716CF25C",
-        "protectedSettings": "*** REDACTED ***"
+        "protectedSettings": "MIIBsAYJKoZIhvcNAQcDoIIBoTCCAZ0CAQAxggFpddesZQewdDBgegkxNzA1BgoJkgergres/Microsoft.OSTCExtensions.VMAccessForLinux=="
       }
     }
   ]
diff --git a/tests/data/hostgaplugin/vm_settings.json b/tests/data/hostgaplugin/vm_settings.json
index 630976bc..7b402720 100644
--- a/tests/data/hostgaplugin/vm_settings.json
+++ b/tests/data/hostgaplugin/vm_settings.json
@@ -56,7 +56,7 @@
             "settings": [
                 {
                     "protectedSettingsCertThumbprint": "4C4F304667711036E64AF4894B76EB208A863BD4",
-                    "protectedSettings": "MIIBsAYJKoZIhvcNAQcDoIIBoTCCAZ0CAQAxggFpMIIBZQIBADBNMDkxNzA1BgoJkiaJk/IsZAEZFidXaW5kb3dzIEF6dXJlIENSUCBDZXJ0aWZpY2F0ZSBHZW5lcmF0b3ICEFpB/HKM/7evRk+DBz754wUwDQYJKoZIhvcNAQEBBQAEggEADPJwniDeIUXzxNrZCloitFdscQ59Bz1dj9DLBREAiM8jmxM0LLicTJDUv272Qm/4ZQgdqpFYBFjGab/9MX+Ih2x47FkVY1woBkckMaC/QOFv84gbboeQCmJYZC/rZJdh8rCMS+CEPq3uH1PVrvtSdZ9uxnaJ+E4exTPPviIiLIPtqWafNlzdbBt8HZjYaVw+SSe+CGzD2pAQeNttq3Rt/6NjCzrjG8ufKwvRoqnrInMs4x6nnN5/xvobKIBSv4/726usfk8Ug+9Q6Benvfpmre2+1M5PnGTfq78cO3o6mI3cPoBUjp5M0iJjAMGeMt81tyHkimZrEZm6pLa4NQMOEjArBgkqhkiG9w0BBwEwFAYIKoZIhvcNAwcECC5nVaiJaWt+gAhgeYvxUOYHXw==",
+                    "protectedSettings": "MIIBsAYJKoZIhvcNAQcDoIIBoTCCAZ0CAQAxggFpMIIBZQIBADBNMDkxNzA1BgoJkiaJk/Microsoft.Azure.Monitor.AzureMonitorLinuxAgent==",
                     "publicSettings": "{\"GCS_AUTO_CONFIG\":true}"
                 }
             ]
@@ -76,7 +76,7 @@
             "settings": [
                 {
                     "protectedSettingsCertThumbprint": "4C4F304667711036E64AF4894B76EB208A863BD4",
-                    "protectedSettings": "MIIBsAYJKoZIhvcNAQcDoIIBoTCCAZ0CAQAxggFpMIIBZQIBADBNMDkxNzA1BgoJkiaJk/IsZAEZFidXaW5kb3dzIEF6dXJlIENSUCBDZXJ0aWZpY2F0ZSBHZW5lcmF0b3ICEFpB/HKM/7evRk+DBz754wUwDQYJKoZIhvcNAQEBBQAEggEADPJwniDeIUXzxNrZCloitFdscQ59Bz1dj9DLBREAiM8jmxM0LLicTJDUv272Qm/4ZQgdqpFYBFjGab/9MX+Ih2x47FkVY1woBkckMaC/QOFv84gbboeQCmJYZC/rZJdh8rCMS+CEPq3uH1PVrvtSdZ9uxnaJ+E4exTPPviIiLIPtqWafNlzdbBt8HZjYaVw+SSe+CGzD2pAQeNttq3Rt/6NjCzrjG8ufKwvRoqnrInMs4x6nnN5/xvobKIBSv4/726usfk8Ug+9Q6Benvfpmre2+1M5PnGTfq78cO3o6mI3cPoBUjp5M0iJjAMGeMt81tyHkimZrEZm6pLa4NQMOEjArBgkqhkiG9w0BBwEwFAYIKoZIhvcNAwcECC5nVaiJaWt+gAhgeYvxUOYHXw==",
+                    "protectedSettings": "MIIBsAYJKoZIhvcNAQcDoIIBoTCCAZ0CAQAxggFpMIIBZQIBADBNMDkxNzA1BgoJkiaJk/Microsoft.Azure.Security.Monitoring.AzureSecurityLinuxAgent==",
                     "publicSettings": "{\"enableGenevaUpload\":true}"
                 }
             ]
@@ -192,7 +192,7 @@
             "settings": [
                 {
                     "protectedSettingsCertThumbprint": "59A10F50FFE2A0408D3F03FE336C8FD5716CF25C",
-                    "protectedSettings": "*** REDACTED ***"
+                    "protectedSettings": "MIIBsAYJKoZIhvcNAQcDoIIBoTCCAZ0CAQAxggFpddesZQewdDBgegkxNzA1BgoJkgergres/Microsoft.OSTCExtensions.VMAccessForLinux=="
                 }
             ]
         }
diff --git a/tests/data/wire/ext_conf_no_extensions-block_blob.xml b/tests/data/wire/ext_conf_no_extensions-block_blob.xml
new file mode 100644
index 00000000..3395b17a
--- /dev/null
+++ b/tests/data/wire/ext_conf_no_extensions-block_blob.xml
@@ -0,0 +1,12 @@
+<Extensions version="1.0.0.0" goalStateIncarnation="9">
+  <GuestAgentExtension xmlns:i="http://www.w3.org/2001/XMLSchema-instance">
+    <GAFamilies>
+    </GAFamilies>
+  </GuestAgentExtension>
+  <Plugins>
+  </Plugins>
+  <PluginSettings>
+  </PluginSettings>
+  <StatusUploadBlob statusBlobType="BlockBlob">http://foo</StatusUploadBlob>
+</Extensions>
+
diff --git a/tests/data/wire/ext_conf_no_extensions-no_status_blob.xml b/tests/data/wire/ext_conf_no_extensions-no_status_blob.xml
new file mode 100644
index 00000000..6632f352
--- /dev/null
+++ b/tests/data/wire/ext_conf_no_extensions-no_status_blob.xml
@@ -0,0 +1,11 @@
+<Extensions version="1.0.0.0" goalStateIncarnation="9">
+  <GuestAgentExtension xmlns:i="http://www.w3.org/2001/XMLSchema-instance">
+    <GAFamilies>
+    </GAFamilies>
+  </GuestAgentExtension>
+  <Plugins>
+  </Plugins>
+  <PluginSettings>
+  </PluginSettings>
+</Extensions>
+
diff --git a/tests/data/wire/ext_conf_no_extensions-page_blob.xml b/tests/data/wire/ext_conf_no_extensions-page_blob.xml
new file mode 100644
index 00000000..57724789
--- /dev/null
+++ b/tests/data/wire/ext_conf_no_extensions-page_blob.xml
@@ -0,0 +1,24 @@
+<Extensions version="1.0.0.0" goalStateIncarnation="9">
+  <GuestAgentExtension xmlns:i="http://www.w3.org/2001/XMLSchema-instance">
+    <GAFamilies>
+      <GAFamily>
+        <Name>Prod</Name>
+        <Uris>
+            <Uri>http://mock-goal-state/manifest_of_ga.xml</Uri>
+        </Uris>
+      </GAFamily>
+      <GAFamily>
+        <Name>Test</Name>
+        <Uris>
+            <Uri>http://mock-goal-state/manifest_of_ga.xml</Uri>
+          </Uris>
+      </GAFamily>
+    </GAFamilies>
+  </GuestAgentExtension>
+  <Plugins>
+  </Plugins>
+  <PluginSettings>
+  </PluginSettings>
+  <StatusUploadBlob statusBlobType="PageBlob">http://sas_url</StatusUploadBlob>
+</Extensions>
+
diff --git a/tests/ga/test_extension.py b/tests/ga/test_extension.py
index 754bcba2..9ad8297b 100644
--- a/tests/ga/test_extension.py
+++ b/tests/ga/test_extension.py
@@ -33,7 +33,7 @@ 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.archive import AGENT_STATUS_FILE
 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, \
@@ -46,7 +46,7 @@ from azurelinuxagent.common.protocol.wire import WireProtocol, InVMArtifactsProf
 from azurelinuxagent.common.utils.restutil import KNOWN_WIRESERVER_IP
 
 from azurelinuxagent.ga.exthandlers import ExtHandlerInstance, migrate_handler_state, \
-    get_exthandlers_handler, AGENT_STATUS_FILE, ExtCommandEnvVariable, HandlerManifest, NOT_RUN, \
+    get_exthandlers_handler, ExtCommandEnvVariable, HandlerManifest, NOT_RUN, \
     ExtensionStatusValue, HANDLER_COMPLETE_NAME_PATTERN, HandlerEnvironment, GoalStateStatus
 
 from tests.protocol import mockwiredata
@@ -56,7 +56,6 @@ 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
@@ -170,8 +169,11 @@ class TestExtensionCleanup(AgentTestCase):
             self.assertEqual(0, TestExtensionCleanup._count_extension_directories(), "All extension directories should be removed")
 
     def test_cleanup_removes_orphaned_packages(self):
+        data_file = mockwiredata.DATA_FILE_NO_EXT.copy()
+        data_file["ext_conf"] = "wire/ext_conf_no_extensions-no_status_blob.xml"
+
         no_of_orphaned_packages = 5
-        with self._setup_test_env(mockwiredata.DATA_FILE_NO_EXT) as (exthandlers_handler, protocol, no_of_exts):
+        with self._setup_test_env(data_file) as (exthandlers_handler, protocol, no_of_exts):
             self.assertEqual(no_of_exts, 0, "Test setup error - Extensions found in ExtConfig")
 
             # Create random extension directories
@@ -1704,7 +1706,7 @@ class TestExtension_Deprecated(TestExtensionBase):
                         datafile = mockwiredata.DATA_FILE
 
                 _, protocol = self._create_mock(mockwiredata.WireProtocolData(datafile), *args)  # pylint: disable=no-value-for-parameter
-                ext_handlers = protocol.client.get_extensions_goal_state().extensions
+                ext_handlers = protocol.get_goal_state().extensions_goal_state.extensions
                 self.assertEqual(1, len(ext_handlers))
                 ext_handler = ext_handlers[0]
                 self.assertEqual('OSTCExtensions.ExampleHandlerLinux', ext_handler.name)
@@ -2448,7 +2450,7 @@ class TestExtension_Deprecated(TestExtensionBase):
         test_data = mockwiredata.WireProtocolData(mockwiredata.DATA_FILE_REQUIRED_FEATURES)
         _, protocol = self._create_mock(test_data, mock_get, mock_crypt_util, *args)
 
-        required_features = protocol.get_extensions_goal_state().required_features
+        required_features = protocol.get_goal_state().extensions_goal_state.required_features
         self.assertEqual(3, len(required_features), "Incorrect features parsed")
         for i, feature in enumerate(required_features):
             self.assertEqual(feature, "TestRequiredFeature{0}".format(i+1), "Name mismatch")
@@ -2508,8 +2510,8 @@ class TestExtensionSequencing(AgentTestCase):
             for ext in handler.settings:
                 ext.dependencyLevel = level
 
-        exthandlers_handler.protocol.client.get_extensions_goal_state()._extensions *= 0
-        exthandlers_handler.protocol.client.get_extensions_goal_state().extensions.extend(all_handlers)
+        exthandlers_handler.protocol.get_goal_state().extensions_goal_state._extensions *= 0
+        exthandlers_handler.protocol.get_goal_state().extensions_goal_state.extensions.extend(all_handlers)
 
     def _validate_extension_sequence(self, expected_sequence, exthandlers_handler):
         installed_extensions = [a[0].ext_handler.name for a, _ in exthandlers_handler.handle_ext_handler.call_args_list]
@@ -3299,7 +3301,7 @@ class TestAdditionalLocationsExtensions(AgentTestCase):
         manifest_location_handler.num_times_called = 0
 
         with mock_wire_protocol(self.test_data, http_get_handler=manifest_location_handler) as protocol:
-            ext_handlers = protocol.client.get_extensions_goal_state().extensions
+            ext_handlers = protocol.get_goal_state().extensions_goal_state.extensions
 
             with self.assertRaises(ExtensionDownloadError):
                 protocol.client.fetch_manifest(ext_handlers[0].manifest_uris,
@@ -3350,124 +3352,94 @@ class TestExtension(TestExtensionBase, HttpRequestPredicates):
                 )
 
             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",
+                "__comment__": "The __status__ property is the actual status reported to CRP",
+                "__status__": {
+                    "version": "1.1",
+                    "timestampUTC": "1970-01-01T00:00:00Z",
+                    "aggregateStatus": {
+                        "guestAgentStatus": {
+                            "version": AGENT_VERSION,
                             "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
+                                "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,
+                                            "formattedMessage": {
+                                                "lang": "en-US",
+                                                "message": None
+                                            }
+                                        },
+                                        "version": 1.0,
+                                        "timestampUTC": "1970-01-01T00:00:00Z"
                                     },
-                                    "version": 1,
-                                    "timestampUTC": "1970-01-01T00:00:00Z"
+                                    "sequenceNumber": 0
+                                }
+                            }
+                        ],
+                        "vmArtifactsAggregateStatus": {
+                            "goalStateAggregateStatus": {
+                                "formattedMessage": {
+                                    "lang": "en-US",
+                                    "message": "GoalState executed successfully"
                                 },
-                                "sequenceNumber": 0
-                            },
-                            "supportsMultiConfig": False
+                                "timestampUTC": "1970-01-01T00:00:00Z",
+                                "inSvdSeqNo": "1",
+                                "status": "Success",
+                                "code": 0
+                            }
                         }
-                    ],
-                    "vmArtifactsAggregateStatus": {
-                        "goalStateAggregateStatus": {
-                            "formattedMessage": {
-                                "lang": "en-US",
-                                "message": "GoalState executed successfully"
-                            },
-                            "timestampUTC": "1970-01-01T00:00:00Z",
-                            "inSvdSeqNo": "1",
-                            "status": "Success",
-                            "code": 0
-                        }
-                    }
+                    },
+                    "guestOSInfo": None,
+                    "supportedFeatures": supported_features
                 },
-                "supportedFeatures": supported_features,
-                "_metadataNotSentToCRP": {
+                "__debug__": {
                     "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()]
-
+                    "extensionSupportedFeatures": [name for name, _ in get_agent_supported_features_list_for_extensions().items()],
+                    "supportsMultiConfig": {
+                        "OSTCExtensions.ExampleHandlerLinux": False
+                    }
                 }
-
             }
 
+
             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
+            # Don't compare the guestOSInfo
+            status_property = actual_status_json.get("__status__")
+            self.assertIsNotNone(status_property, "The status file is missing the __status__ property")
+            self.assertIsNotNone(status_property.get("guestOSInfo"), "The status file is missing the guestOSInfo property")
+            status_property["guestOSInfo"] = None
+
             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"))))
-
     def test_it_should_process_extensions_only_if_allowed(self):
         def assert_extensions_called(exthandlers_handler, expected_call_count=0):
             extension_name = 'OSTCExtensions.ExampleHandlerLinux'
diff --git a/tests/ga/test_multi_config_extension.py b/tests/ga/test_multi_config_extension.py
index 2e22affb..643fb718 100644
--- a/tests/ga/test_multi_config_extension.py
+++ b/tests/ga/test_multi_config_extension.py
@@ -51,7 +51,7 @@ class TestMultiConfigExtensionsConfigParsing(AgentTestCase):
 
     def _mock_and_assert_ext_handlers(self, expected_handlers):
         with mock_wire_protocol(self.test_data) as protocol:
-            ext_handlers = protocol.client.get_extensions_goal_state().extensions
+            ext_handlers = protocol.get_goal_state().extensions_goal_state.extensions
             for ext_handler in ext_handlers:
                 if ext_handler.name not in expected_handlers:
                     continue
diff --git a/tests/ga/test_update.py b/tests/ga/test_update.py
index b0f15b80..1d0fffc2 100644
--- a/tests/ga/test_update.py
+++ b/tests/ga/test_update.py
@@ -672,9 +672,7 @@ class TestGuestAgent(UpdateTestCase):
         host_uri = 'host_uri'
         api_uri = URI_FORMAT_GET_API_VERSIONS.format(host_uri, HOST_PLUGIN_PORT)
         art_uri = URI_FORMAT_GET_EXTENSION_ARTIFACT.format(host_uri, HOST_PLUGIN_PORT)
-        mock_host = HostPluginProtocol(host_uri,
-                                       'container_id',
-                                       'role_config')
+        mock_host = HostPluginProtocol(host_uri)
 
         pkg = ExtHandlerPackage(version=str(self._get_agent_version()))
         pkg.uris.append(ext_uri)
@@ -2853,7 +2851,8 @@ class ReportStatusTestCase(AgentTestCase):
         except IndexError:
             raise HttpError()
 
-    def test_update_handler_should_report_status_even_on_failed_goal_state_fetch(self):
+    @patch("azurelinuxagent.common.conf.get_enable_fast_track", return_value=True)
+    def test_update_handler_should_report_status_even_on_failed_goal_state_fetch(self, _):
 
         try:
             # Returning None forces the mock wire data to return the contents in the static
diff --git a/tests/protocol/mockwiredata.py b/tests/protocol/mockwiredata.py
index 3fd6c756..0267db21 100644
--- a/tests/protocol/mockwiredata.py
+++ b/tests/protocol/mockwiredata.py
@@ -53,8 +53,7 @@ DATA_FILE_INVALID_VM_META_DATA = DATA_FILE.copy()
 DATA_FILE_INVALID_VM_META_DATA["ext_conf"] = "wire/ext_conf_invalid_vm_metadata.xml"
 
 DATA_FILE_NO_EXT = DATA_FILE.copy()
-DATA_FILE_NO_EXT["goal_state"] = "wire/goal_state_no_ext.xml"
-DATA_FILE_NO_EXT["ext_conf"] = None
+DATA_FILE_NO_EXT["ext_conf"] = "wire/ext_conf_no_extensions-block_blob.xml"
 
 DATA_FILE_NOOP_GS = DATA_FILE.copy()
 DATA_FILE_NOOP_GS["goal_state"] = "wire/goal_state_noop.xml"
@@ -197,7 +196,7 @@ class WireProtocolData(object):
             self.in_vm_artifacts_profile = load_data(in_vm_artifacts_profile_file)
 
     def mock_http_get(self, url, *_, **kwargs):
-        content = None
+        content = ''
         response_headers = []
 
         resp = MagicMock()
diff --git a/tests/protocol/test_extensions_goal_state.py b/tests/protocol/test_extensions_goal_state.py
index 15c3f605..cc929938 100644
--- a/tests/protocol/test_extensions_goal_state.py
+++ b/tests/protocol/test_extensions_goal_state.py
@@ -4,9 +4,7 @@ import copy
 import re
 import sys
 
-from azurelinuxagent.common.protocol.extensions_goal_state import ExtensionsGoalState, GoalStateMismatchError
 from azurelinuxagent.common.protocol.extensions_goal_state_factory import ExtensionsGoalStateFactory
-from azurelinuxagent.common.utils import textutil
 from tests.protocol.mocks import mockwiredata, mock_wire_protocol
 from tests.tools import AgentTestCase, load_data
 
@@ -16,63 +14,9 @@ if sys.version_info[0] < 3 or sys.version_info[0] == 3 and sys.version_info[1] <
 
 
 class ExtensionsGoalStateTestCase(AgentTestCase):
-    def test_compare_should_succeed_when_extensions_config_and_vm_settings_are_equal(self):
-        with mock_wire_protocol(mockwiredata.DATA_FILE_VM_SETTINGS) as protocol:
-            from_extensions_config = protocol.client.get_extensions_goal_state()
-            from_vm_settings = protocol.client._cached_vm_settings
-
-            try:
-                ExtensionsGoalState.compare(from_extensions_config, from_vm_settings)
-            except Exception as exception:
-                self.fail("Compare goal state failed: {0}".format(textutil.format_exception(exception)))
-
-    def test_compare_should_report_mismatches_between_extensions_config_and_vm_settings(self):
-        with mock_wire_protocol(mockwiredata.DATA_FILE_VM_SETTINGS) as protocol:
-            from_extensions_config = protocol.client.get_extensions_goal_state()
-            from_vm_settings = protocol.client._cached_vm_settings
-
-            public_settings_mismatch = [""]
-
-            def assert_compare_raises(setup_copy, failing_attribute):
-                from_vm_settings_copy = copy.deepcopy(from_vm_settings)
-                setup_copy(from_vm_settings_copy)
-
-                with self.assertRaisesRegexCM(GoalStateMismatchError, re.escape("(Attribute: {0})".format(failing_attribute)), re.DOTALL) as context_manager:
-                    ExtensionsGoalState.compare(from_extensions_config, from_vm_settings_copy)
-                if context_manager.exception.attribute == 'publicSettings':
-                    public_settings_mismatch[0] = str(context_manager.exception)
-
-            assert_compare_raises(lambda c: setattr(c, "_activity_id",              'MOCK_ACTIVITY_ID'),        "activity_id")
-            assert_compare_raises(lambda c: setattr(c, "_correlation_id",           'MOCK_CORRELATION_ID'),     "correlation_id")
-            assert_compare_raises(lambda c: setattr(c, "_created_on_timestamp",     'MOCK_TIMESTAMP'),          "created_on_timestamp")
-            assert_compare_raises(lambda c: setattr(c, "_status_upload_blob",       'MOCK_UPLOAD_BLOB'),        "status_upload_blob")
-            assert_compare_raises(lambda c: setattr(c, "_status_upload_blob_type",  'MOCK_UPLOAD_BLOB_TYPE'),   "status_upload_blob_type")
-            assert_compare_raises(lambda c: setattr(c, "_required_features",        ['MOCK_REQUIRED_FEATURE']), "required_features")
-            assert_compare_raises(lambda c: setattr(c, "_on_hold",                  False),                     "on_hold")
-
-            assert_compare_raises(lambda c: setattr(c.agent_manifests[0], "family",  'MOCK_FAMILY'),  r"agent_manifests[0].family")
-            assert_compare_raises(lambda c: setattr(c.agent_manifests[0], "requested_version_string", 'MOCK_VERSION'), r"agent_manifests[0].requested_version_string")
-            assert_compare_raises(lambda c: setattr(c.agent_manifests[0], "uris",    ['MOCK_URI']),   r"agent_manifests[0].uris")
-
-            assert_compare_raises(lambda c: setattr(c.extensions[0], "version",  'MOCK_NAME'),         r"extensions[0].version")
-            assert_compare_raises(lambda c: setattr(c.extensions[0], "state",  'MOCK_STATE'),          r"extensions[0].state")
-            assert_compare_raises(lambda c: setattr(c.extensions[0], "manifest_uris",  ['MOCK_URI']),  r"extensions[0].manifest_uris")
-            assert_compare_raises(lambda c: setattr(c.extensions[0], "supports_multi_config",  True),  r"extensions[0].supports_multi_config")
-
-            # NOTE: protectedSettings are not compared, so we skip them below
-            assert_compare_raises(lambda c: setattr(c.extensions[0].settings[0], "name",                  'MOCK_NAME'),                 r"extensions[0].settings[0].name")
-            assert_compare_raises(lambda c: setattr(c.extensions[0].settings[0], "sequenceNumber",        98765),                       r"extensions[0].settings[0].sequenceNumber")
-            assert_compare_raises(lambda c: setattr(c.extensions[0].settings[0], "publicSettings",        {'MOCK_NAME': 'MOCK_VALUE'}), r"extensions[0].settings[0].publicSettings")
-            assert_compare_raises(lambda c: setattr(c.extensions[0].settings[0], "certificateThumbprint", 'MOCK_CERT'),                 r"extensions[0].settings[0].certificateThumbprint")
-            assert_compare_raises(lambda c: setattr(c.extensions[0].settings[0], "dependencyLevel",       56789),                       r"extensions[0].settings[0].dependencyLevel")
-            assert_compare_raises(lambda c: setattr(c.extensions[0].settings[0], "state",                 'MOCK_STATE'),                r"extensions[0].settings[0].state")
-
-            expected = r'^\[GoalStateMismatchError\] Mismatch in Goal States \[Incarnation 1\] != \[Etag: 1\]: \[REDACTED\] != \[REDACTED\] \(Attribute: .*\.publicSettings\)$'
-            self.assertRegex(public_settings_mismatch[0], expected, 'Expected the protected settings to be redacted. Got: "{0}"'.format(public_settings_mismatch[0]))
-
     def test_create_from_extensions_config_should_assume_block_when_blob_type_is_not_valid(self):
         data_file = mockwiredata.DATA_FILE.copy()
-        data_file["vm_settings"] = "hostgaplugin/ext_conf-invalid_blob_type.xml"
+        data_file["ext_conf"] = "hostgaplugin/ext_conf-invalid_blob_type.xml"
         with mock_wire_protocol(data_file) as protocol:
             extensions_goal_state = ExtensionsGoalStateFactory.create_from_extensions_config(123, load_data("hostgaplugin/ext_conf-invalid_blob_type.xml"), protocol)
             self.assertEqual("BlockBlob", extensions_goal_state.status_upload_blob_type, 'Expected BlockBob for an invalid statusBlobType')
@@ -81,24 +25,3 @@ class ExtensionsGoalStateTestCase(AgentTestCase):
         extensions_goal_state = ExtensionsGoalStateFactory.create_from_vm_settings(1234567890, load_data("hostgaplugin/vm_settings-invalid_blob_type.json"))
         self.assertEqual("BlockBlob", extensions_goal_state.status_upload_blob_type, 'Expected BlockBob for an invalid statusBlobType')
 
-    def test_extension_goal_state_should_parse_requested_version_properly(self):
-        with mock_wire_protocol(mockwiredata.DATA_FILE_VM_SETTINGS) as protocol:
-            fabric_manifests, _ = protocol.get_vmagent_manifests()
-            for manifest in fabric_manifests:
-                self.assertEqual(manifest.requested_version_string, "0.0.0.0", "Version should be None")
-
-            vm_settings_ga_manifests = protocol.client._cached_vm_settings.agent_manifests
-            for manifest in vm_settings_ga_manifests:
-                self.assertEqual(manifest.requested_version_string, "0.0.0.0", "Version should be None")
-
-        data_file = mockwiredata.DATA_FILE_VM_SETTINGS.copy()
-        data_file["vm_settings"] = "hostgaplugin/vm_settings-requested_version.json"
-        data_file["ext_conf"] = "hostgaplugin/ext_conf-requested_version.xml"
-        with mock_wire_protocol(data_file) as protocol:
-            fabric_manifests, _ = protocol.get_vmagent_manifests()
-            for manifest in fabric_manifests:
-                self.assertEqual(manifest.requested_version_string, "9.9.9.10", "Version should be 9.9.9.10")
-
-            vm_settings_ga_manifests = protocol.client._cached_vm_settings.agent_manifests
-            for manifest in vm_settings_ga_manifests:
-                self.assertEqual(manifest.requested_version_string, "9.9.9.9", "Version should be 9.9.9.9")
diff --git a/tests/protocol/test_extensions_goal_state_from_extensions_config.py b/tests/protocol/test_extensions_goal_state_from_extensions_config.py
index 0fb1ef13..3f2c4d0b 100644
--- a/tests/protocol/test_extensions_goal_state_from_extensions_config.py
+++ b/tests/protocol/test_extensions_goal_state_from_extensions_config.py
@@ -8,21 +8,21 @@ from tests.tools import AgentTestCase
 class ExtensionsGoalStateFromExtensionsConfigTestCase(AgentTestCase):
     def test_it_should_parse_in_vm_metadata(self):
         with mock_wire_protocol(mockwiredata.DATA_FILE_IN_VM_META_DATA) as protocol:
-            extensions_goal_state = protocol.get_extensions_goal_state()
+            extensions_goal_state = protocol.get_goal_state().extensions_goal_state
             self.assertEqual("555e551c-600e-4fb4-90ba-8ab8ec28eccc", extensions_goal_state.activity_id, "Incorrect activity Id")
             self.assertEqual("400de90b-522e-491f-9d89-ec944661f531", extensions_goal_state.correlation_id, "Incorrect correlation Id")
             self.assertEqual('2020-11-09T17:48:50.412125Z', extensions_goal_state.created_on_timestamp, "Incorrect GS Creation time")
 
     def test_it_should_use_default_values_when_in_vm_metadata_is_missing(self):
         with mock_wire_protocol(mockwiredata.DATA_FILE) as protocol:
-            extensions_goal_state = protocol.get_extensions_goal_state()
+            extensions_goal_state = protocol.get_goal_state().extensions_goal_state
             self.assertEqual(AgentGlobals.GUID_ZERO, extensions_goal_state.activity_id, "Incorrect activity Id")
             self.assertEqual(AgentGlobals.GUID_ZERO, extensions_goal_state.correlation_id, "Incorrect correlation Id")
             self.assertEqual('1900-01-01T00:00:00.000000Z', extensions_goal_state.created_on_timestamp, "Incorrect GS Creation time")
 
     def test_it_should_use_default_values_when_in_vm_metadata_is_invalid(self):
         with mock_wire_protocol(mockwiredata.DATA_FILE_INVALID_VM_META_DATA) as protocol:
-            extensions_goal_state = protocol.get_extensions_goal_state()
+            extensions_goal_state = protocol.get_goal_state().extensions_goal_state
             self.assertEqual(AgentGlobals.GUID_ZERO, extensions_goal_state.activity_id, "Incorrect activity Id")
             self.assertEqual(AgentGlobals.GUID_ZERO, extensions_goal_state.correlation_id, "Incorrect correlation Id")
             self.assertEqual('1900-01-01T00:00:00.000000Z', extensions_goal_state.created_on_timestamp, "Incorrect GS Creation time")
@@ -31,8 +31,7 @@ class ExtensionsGoalStateFromExtensionsConfigTestCase(AgentTestCase):
         data_file = mockwiredata.DATA_FILE_VM_SETTINGS.copy()
         data_file["ext_conf"] = "hostgaplugin/ext_conf-no_status_upload_blob.xml"
         with mock_wire_protocol(data_file) as protocol:
-            extensions_goal_state = protocol.get_extensions_goal_state()
+            extensions_goal_state = protocol.get_goal_state().extensions_goal_state
 
             self.assertIsNone(extensions_goal_state.status_upload_blob, "Expected status upload blob to be None")
             self.assertEqual("BlockBlob", extensions_goal_state.status_upload_blob_type, "Expected status upload blob to be Block")
-
diff --git a/tests/protocol/test_extensions_goal_state_from_vm_settings.py b/tests/protocol/test_extensions_goal_state_from_vm_settings.py
index fe30eff0..e34d4bcf 100644
--- a/tests/protocol/test_extensions_goal_state_from_vm_settings.py
+++ b/tests/protocol/test_extensions_goal_state_from_vm_settings.py
@@ -6,7 +6,8 @@ import os.path
 from azurelinuxagent.common.protocol.extensions_goal_state_factory import ExtensionsGoalStateFactory
 from azurelinuxagent.common.protocol.extensions_goal_state_from_vm_settings import _CaseFoldedDict
 from azurelinuxagent.common.utils import fileutil
-from tests.tools import AgentTestCase, data_dir
+from tests.protocol.mocks import mockwiredata, mock_wire_protocol
+from tests.tools import AgentTestCase, data_dir, patch
 
 
 class ExtensionsGoalStateFromVmSettingsTestCase(AgentTestCase):
@@ -47,6 +48,19 @@ class ExtensionsGoalStateFromVmSettingsTestCase(AgentTestCase):
         # dependency level (multi-config)
         self.assertEqual(1, vm_settings.extensions[3].settings[1].dependencyLevel, "Incorrect dependency level (multi-config)")
 
+    @patch("azurelinuxagent.common.conf.get_enable_fast_track", return_value=True)
+    def test_extension_goal_state_should_parse_requested_version_properly(self, _):
+        with mock_wire_protocol(mockwiredata.DATA_FILE_VM_SETTINGS) as protocol:
+            manifests, _ = protocol.get_vmagent_manifests()
+            for manifest in manifests:
+                self.assertEqual(manifest.requested_version_string, "0.0.0.0", "Version should be None")
+
+        data_file = mockwiredata.DATA_FILE_VM_SETTINGS.copy()
+        data_file["vm_settings"] = "hostgaplugin/vm_settings-requested_version.json"
+        with mock_wire_protocol(data_file) as protocol:
+            manifests, _ = protocol.get_vmagent_manifests()
+            for manifest in manifests:
+                self.assertEqual(manifest.requested_version_string, "9.9.9.9", "Version should be 9.9.9.9")
     def test_create_from_vm_settings_should_parse_missing_status_upload_blob_as_none(self):
         vm_settings_text = fileutil.read_file(os.path.join(data_dir, "hostgaplugin/vm_settings-no_status_upload_blob.json"))
         vm_settings = ExtensionsGoalStateFactory.create_from_vm_settings("123", vm_settings_text)
diff --git a/tests/protocol/test_goal_state.py b/tests/protocol/test_goal_state.py
index 492fddcc..fdd41303 100644
--- a/tests/protocol/test_goal_state.py
+++ b/tests/protocol/test_goal_state.py
@@ -1,8 +1,13 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the Apache License.
 
-from azurelinuxagent.common.exception import IncompleteGoalStateError
-from azurelinuxagent.common.protocol.goal_state import GoalState, _NUM_GS_FETCH_RETRIES
+import glob
+import os
+import re
+
+from azurelinuxagent.common.protocol.goal_state import GoalState, _GET_GOAL_STATE_MAX_ATTEMPTS
+from azurelinuxagent.common.exception import ProtocolError
+from azurelinuxagent.common.utils.archive import ARCHIVE_DIRECTORY_NAME
 from tests.protocol.mocks import mock_wire_protocol
 from tests.protocol import mockwiredata
 from tests.tools import AgentTestCase, patch
@@ -16,7 +21,64 @@ class GoalStateTestCase(AgentTestCase):
             protocol.mock_wire_data.set_incarnation(2)
 
             with patch('time.sleep') as mock_sleep:
-                with self.assertRaises(IncompleteGoalStateError):
+                with self.assertRaises(ProtocolError):
                     GoalState(protocol.client)
-                self.assertEqual(_NUM_GS_FETCH_RETRIES, mock_sleep.call_count, "Unexpected number of retries")
+                self.assertEqual(_GET_GOAL_STATE_MAX_ATTEMPTS, mock_sleep.call_count, "Unexpected number of retries")
+
+    @patch("azurelinuxagent.common.conf.get_enable_fast_track", return_value=True)
+    def test_fetch_full_goal_state_should_save_goal_state_to_history_directory(self, _):
+        with mock_wire_protocol(mockwiredata.DATA_FILE_VM_SETTINGS) as protocol:
+            # use a new goal state with a specific test incarnation and etag
+            protocol.mock_wire_data.set_incarnation(999)
+            protocol.mock_wire_data.set_etag(888)
+            goal_state = GoalState(protocol.client)
+
+            matches = glob.glob(os.path.join(self.tmp_dir, ARCHIVE_DIRECTORY_NAME, "*_999"))
+            self.assertTrue(len(matches) == 1, "Expected one history directory for incarnation 999. Got: {0}".format(matches))
+
+            history_directory = matches[0]
+            extensions_config_file = os.path.join(history_directory, "ExtensionsConfig.xml")
+            expected_files = [
+                os.path.join(history_directory, "GoalState.xml"),
+                os.path.join(history_directory, "SharedConfig.xml"),
+                os.path.join(history_directory, "HostingEnvironmentConfig.xml"),
+                extensions_config_file,
+            ]
+
+            matches = glob.glob(os.path.join(self.tmp_dir, ARCHIVE_DIRECTORY_NAME, "*_888"))
+            self.assertTrue(len(matches) == 1, "Expected one history directory for etag 888. Got: {0}".format(matches))
+
+            history_directory = matches[0]
+            vm_settings_file = os.path.join(history_directory, "VmSettings.json")
+            expected_files.append(vm_settings_file)
+
+            for f in expected_files:
+                self.assertTrue(os.path.exists(f), "{0} was not saved".format(f))
+
+            extensions_goal_state = goal_state.extensions_goal_state
+            protected_settings = []
+            for ext_handler in extensions_goal_state.extensions:
+                for extension in ext_handler.settings:
+                    if extension.protectedSettings is not None:
+                        protected_settings.append(extension.protectedSettings)
+            if len(protected_settings) == 0:
+                raise Exception("The test goal state does not include any protected settings")
+
+            for file_name in extensions_config_file, vm_settings_file:
+                with open(file_name, "r") as stream:
+                    file_contents = stream.read()
+
+                    for settings in protected_settings:
+                        self.assertNotIn(
+                            settings,
+                            file_contents,
+                            "The protectedSettings should not have been saved to {0}".format(file_name))
+
+                    matches = re.findall(r'"protectedSettings"\s*:\s*"\*\*\* REDACTED \*\*\*"', file_contents)
+                    self.assertEqual(
+                        len(matches),
+                        len(protected_settings),
+                        "Could not find the expected number of redacted settings in {0}.\nExpected {1}.\n{2}".format(file_name, len(protected_settings), file_contents))
+
+
 
diff --git a/tests/protocol/test_hostplugin.py b/tests/protocol/test_hostplugin.py
index 2a18e48a..c18627a0 100644
--- a/tests/protocol/test_hostplugin.py
+++ b/tests/protocol/test_hostplugin.py
@@ -26,16 +26,17 @@ import azurelinuxagent.common.protocol.hostplugin as hostplugin
 import azurelinuxagent.common.protocol.restapi as restapi
 import azurelinuxagent.common.protocol.wire as wire
 from azurelinuxagent.common.errorstate import ErrorState
-from azurelinuxagent.common.exception import HttpError, ResourceGoneError
+from azurelinuxagent.common.exception import HttpError, ResourceGoneError, ProtocolError
 from azurelinuxagent.common.future import ustr, httpclient
 from azurelinuxagent.common.osutil.default import UUID_PATTERN
-from azurelinuxagent.common.protocol.hostplugin import API_VERSION
+from azurelinuxagent.common.protocol.hostplugin import API_VERSION, _VmSettingsErrorReporter, VmSettingsNotSupported
+from azurelinuxagent.common.protocol.goal_state import GoalState
 from azurelinuxagent.common.utils import restutil
 from azurelinuxagent.common.version import AGENT_VERSION, AGENT_NAME
-from tests.protocol.mocks import mock_wire_protocol, MockHttpResponse
+from tests.protocol.mocks import mock_wire_protocol, mockwiredata, MockHttpResponse
 from tests.protocol.HttpRequestPredicates import HttpRequestPredicates
 from tests.protocol.mockwiredata import DATA_FILE, DATA_FILE_NO_EXT
-from tests.tools import AgentTestCase, PY_VERSION_MAJOR, Mock, PropertyMock, patch
+from tests.tools import AgentTestCase, PY_VERSION_MAJOR, Mock, patch
 
 
 hostplugin_status_url = "http://168.63.129.16:32526/status"
@@ -61,10 +62,8 @@ class TestHostPlugin(HttpRequestPredicates, AgentTestCase):
 
     def _init_host(self):
         with mock_wire_protocol(DATA_FILE) as protocol:
-            test_goal_state = protocol.client.get_goal_state()
-            host_plugin = wire.HostPluginProtocol(wireserver_url,
-                                                  test_goal_state.container_id,
-                                                  test_goal_state.role_config_name)
+            host_plugin = wire.HostPluginProtocol(wireserver_url)
+            GoalState.update_host_plugin_headers(protocol.client)
             self.assertTrue(host_plugin.health_service is not None)
             return host_plugin
 
@@ -150,13 +149,10 @@ class TestHostPlugin(HttpRequestPredicates, AgentTestCase):
     @staticmethod
     @contextlib.contextmanager
     def create_mock_protocol():
-        with mock_wire_protocol(DATA_FILE_NO_EXT) as protocol:
-            # These tests use mock wire data that don't have any extensions (extension config will be empty).
-            # Populate the upload blob and set an initial empty status before returning the protocol.
-            protocol.client._extensions_goal_state = Mock(wraps=protocol.client._extensions_goal_state)
-            type(protocol.client._extensions_goal_state).status_upload_blob = PropertyMock(return_value=sas_url)
-            type(protocol.client._extensions_goal_state).status_upload_blob_type = PropertyMock(return_value=page_blob_type)
+        data_file = DATA_FILE_NO_EXT.copy()
+        data_file["ext_conf"] = "wire/ext_conf_no_extensions-page_blob.xml"
 
+        with mock_wire_protocol(data_file) as protocol:
             status = restapi.VMStatus(status="Ready", message="Guest Agent is running")
             protocol.client.status_blob.set_vm_status(status)
 
@@ -174,7 +170,7 @@ class TestHostPlugin(HttpRequestPredicates, AgentTestCase):
                                   should_initialize,
                                   should_report_healthy):
 
-        host = hostplugin.HostPluginProtocol(endpoint='ws', container_id='cid', role_config_name='rcf')
+        host = hostplugin.HostPluginProtocol(endpoint='ws')
 
         host.is_initialized = False
         patch_http_get.return_value = MockResponse(body=response_body,
@@ -439,11 +435,8 @@ class TestHostPlugin(HttpRequestPredicates, AgentTestCase):
 
     def test_validate_block_blob(self):
         with mock_wire_protocol(DATA_FILE) as protocol:
-            test_goal_state = protocol.client._goal_state
+            host_client = protocol.client.get_host_plugin()
 
-            host_client = wire.HostPluginProtocol(wireserver_url,
-                                                  test_goal_state.container_id,
-                                                  test_goal_state.role_config_name)
             self.assertFalse(host_client.is_initialized)
             self.assertTrue(host_client.api_versions is None)
             self.assertTrue(host_client.health_service is not None)
@@ -472,7 +465,7 @@ class TestHostPlugin(HttpRequestPredicates, AgentTestCase):
                     # first call is to host plugin
                     self._validate_hostplugin_args(
                         patch_http.call_args_list[0],
-                        test_goal_state,
+                        protocol.get_goal_state(),
                         exp_method, exp_url, exp_data)
 
                     # second call is to health service
@@ -482,11 +475,9 @@ class TestHostPlugin(HttpRequestPredicates, AgentTestCase):
     def test_validate_page_blobs(self):
         """Validate correct set of data is sent for page blobs"""
         with mock_wire_protocol(DATA_FILE) as protocol:
-            test_goal_state = protocol.client._goal_state
+            test_goal_state = protocol.get_goal_state()
 
-            host_client = wire.HostPluginProtocol(wireserver_url,
-                                                  test_goal_state.container_id,
-                                                  test_goal_state.role_config_name)
+            host_client = protocol.client.get_host_plugin()
 
             self.assertFalse(host_client.is_initialized)
             self.assertTrue(host_client.api_versions is None)
@@ -548,7 +539,7 @@ class TestHostPlugin(HttpRequestPredicates, AgentTestCase):
         http_put_handler.args, http_put_handler.kwargs = [], {}
 
         with mock_wire_protocol(DATA_FILE, http_put_handler=http_put_handler) as protocol:
-            test_goal_state = protocol.client.get_goal_state()
+            test_goal_state = protocol.get_goal_state()
 
             expected_url = hostplugin.URI_FORMAT_PUT_LOG.format(wireserver_url, hostplugin.HOST_PLUGIN_PORT)
             expected_headers = {'x-ms-version': '2015-09-01',
@@ -557,9 +548,7 @@ class TestHostPlugin(HttpRequestPredicates, AgentTestCase):
                                 "x-ms-client-name": AGENT_NAME,
                                 "x-ms-client-version": AGENT_VERSION}
 
-            host_client = wire.HostPluginProtocol(wireserver_url,
-                                                  test_goal_state.container_id,
-                                                  test_goal_state.role_config_name)
+            host_client = protocol.client.get_host_plugin()
 
             self.assertFalse(host_client.is_initialized, "Host plugin should not be initialized!")
 
@@ -590,11 +579,9 @@ class TestHostPlugin(HttpRequestPredicates, AgentTestCase):
         http_put_handler.args, http_put_handler.kwargs = [], {}
 
         with mock_wire_protocol(DATA_FILE, http_put_handler=http_put_handler) as protocol:
-            test_goal_state = protocol.client.get_goal_state()
 
-            host_client = wire.HostPluginProtocol(wireserver_url,
-                                                  test_goal_state.container_id,
-                                                  test_goal_state.role_config_name)
+            host_client = wire.HostPluginProtocol(wireserver_url)
+            GoalState.update_host_plugin_headers(protocol.client)
 
             self.assertFalse(host_client.is_initialized, "Host plugin should not be initialized!")
 
@@ -608,7 +595,7 @@ class TestHostPlugin(HttpRequestPredicates, AgentTestCase):
 
     def test_validate_get_extension_artifacts(self):
         with mock_wire_protocol(DATA_FILE) as protocol:
-            test_goal_state = protocol.client._goal_state
+            test_goal_state = protocol.get_goal_state()
 
             expected_url = hostplugin.URI_FORMAT_GET_EXTENSION_ARTIFACT.format(wireserver_url, hostplugin.HOST_PLUGIN_PORT)
             expected_headers = {'x-ms-version': '2015-09-01',
@@ -616,9 +603,8 @@ class TestHostPlugin(HttpRequestPredicates, AgentTestCase):
                                 "x-ms-host-config-name": test_goal_state.role_config_name,
                                 "x-ms-artifact-location": sas_url}
 
-            host_client = wire.HostPluginProtocol(wireserver_url,
-                                                  test_goal_state.container_id,
-                                                  test_goal_state.role_config_name)
+            host_client = protocol.client.get_host_plugin()
+
             self.assertFalse(host_client.is_initialized)
             self.assertTrue(host_client.api_versions is None)
             self.assertTrue(host_client.health_service is not None)
@@ -862,6 +848,145 @@ class TestHostPlugin(HttpRequestPredicates, AgentTestCase):
         self.assertEqual(False, actual)
 
 
+class TestHostPluginVmSettings(HttpRequestPredicates, AgentTestCase):
+    def test_it_should_raise_protocol_error_when_the_vm_settings_request_fails(self):
+        def http_get_handler(url, *_, **__):
+            if self.is_host_plugin_vm_settings_request(url):
+                return MockHttpResponse(httpclient.INTERNAL_SERVER_ERROR, body="TEST ERROR")
+            return None
+
+        with mock_wire_protocol(mockwiredata.DATA_FILE_VM_SETTINGS) as protocol:
+            protocol.set_http_handlers(http_get_handler=http_get_handler)
+            with self.assertRaisesRegexCM(ProtocolError, r'GET vmSettings \[correlation ID: .* eTag: .*\]: \[HTTP Failed\] \[500: None].*TEST ERROR.*'):
+                protocol.client.get_host_plugin().fetch_vm_settings(False)
+
+    @staticmethod
+    def _fetch_vm_settings_ignoring_errors(protocol):
+        try:
+            protocol.client.get_host_plugin().fetch_vm_settings(False)
+        except (ProtocolError, VmSettingsNotSupported):
+            pass
+
+    @patch("azurelinuxagent.common.conf.get_enable_fast_track", return_value=True)
+    def test_it_should_keep_track_of_errors_in_vm_settings_requests(self, _):
+        mock_response = None
+
+        def http_get_handler(url, *_, **__):
+            if self.is_host_plugin_vm_settings_request(url):
+                if isinstance(mock_response, Exception):
+                    # E0702: Raising NoneType while only classes or instances are allowed (raising-bad-type) - Disabled: we never raise None
+                    raise mock_response  # pylint: disable=raising-bad-type
+                return mock_response
+            return None
+
+        with mock_wire_protocol(mockwiredata.DATA_FILE_VM_SETTINGS, http_get_handler=http_get_handler) as protocol:
+            mock_response = MockHttpResponse(httpclient.INTERNAL_SERVER_ERROR)
+            self._fetch_vm_settings_ignoring_errors(protocol)
+
+            mock_response = MockHttpResponse(httpclient.BAD_REQUEST)
+            self._fetch_vm_settings_ignoring_errors(protocol)
+            self._fetch_vm_settings_ignoring_errors(protocol)
+
+            mock_response = IOError("timed out")
+            self._fetch_vm_settings_ignoring_errors(protocol)
+
+            mock_response = httpclient.HTTPException()
+            self._fetch_vm_settings_ignoring_errors(protocol)
+            self._fetch_vm_settings_ignoring_errors(protocol)
+
+            # force the summary by resetting its period and calling update_goal_state
+            with patch("azurelinuxagent.common.protocol.hostplugin.add_event") as add_event:
+                mock_response = None  # stop producing errors
+                protocol.client._host_plugin._vm_settings_error_reporter._next_period = datetime.datetime.now()
+                self._fetch_vm_settings_ignoring_errors(protocol)
+            summary_text = [kwargs["message"] for _, kwargs in add_event.call_args_list if kwargs["op"] == "VmSettingsSummary"]
+
+            self.assertEqual(1, len(summary_text), "Exactly 1 summary should have been produced. Got: {0} ".format(summary_text))
+
+            summary = json.loads(summary_text[0])
+
+            expected = {
+                "requests":       6 + 2,  # two extra calls to update_goal_state (when creating the mock protocol and when forcing the summary)
+                "errors":         6,
+                "serverErrors":   1,
+                "clientErrors":   2,
+                "timeouts":       1,
+                "failedRequests": 2
+            }
+
+            self.assertEqual(expected, summary, "The count of errors is incorrect")
+
+    @patch("azurelinuxagent.common.conf.get_enable_fast_track", return_value=True)
+    def test_it_should_limit_the_number_of_errors_it_reports(self, _):
+        with mock_wire_protocol(mockwiredata.DATA_FILE_VM_SETTINGS) as protocol:
+            def http_get_handler(url, *_, **__):
+                if self.is_host_plugin_vm_settings_request(url):
+                    return MockHttpResponse(httpclient.BAD_GATEWAY)  # HostGAPlugin returns 502 for internal errors
+                return None
+            protocol.set_http_handlers(http_get_handler=http_get_handler)
+
+            def get_telemetry_messages():
+                return [kwargs["message"] for _, kwargs in add_event.call_args_list if kwargs["op"] == "VmSettings"]
+
+            def get_log_messages():
+                return [arg[0][0] for arg in logger_info.call_args_list   if "[VmSettings]" in arg[0][0]]
+
+            def fetch_vm_settings():
+                try:
+                    host_plugin.fetch_vm_settings(True)
+                except ProtocolError:
+                    pass  # All calls produce an error; ignore it
+
+            with patch("azurelinuxagent.common.protocol.hostplugin.add_event") as add_event:
+                with patch('azurelinuxagent.common.logger.info') as logger_info:
+                    host_plugin = protocol.client.get_host_plugin()
+                    for _ in range(_VmSettingsErrorReporter._MaxTelemetryErrors + 3):
+                        fetch_vm_settings()
+
+                    telemetry_messages = get_telemetry_messages()
+                    self.assertEqual(_VmSettingsErrorReporter._MaxTelemetryErrors, len(telemetry_messages), "The number of errors reported to telemetry is not the max allowed (got: {0})".format(telemetry_messages))
+
+                    log_messages = get_log_messages()
+                    self.assertEqual(_VmSettingsErrorReporter._MaxLogErrors, len(log_messages), "The number of errors reported to the local log is not the max allowed (got: {0})".format(telemetry_messages))
+
+            # Reset the error reporter and verify that additional errors are reported
+            host_plugin._vm_settings_error_reporter._next_period = datetime.datetime.now()
+            fetch_vm_settings()  # this triggers the reset
+
+            with patch("azurelinuxagent.common.protocol.hostplugin.add_event") as add_event:
+                fetch_vm_settings()
+
+                telemetry_messages = get_telemetry_messages()
+                self.assertEqual(1, len(telemetry_messages), "Expected additional errors to be reported to telemetry in the next period (got: {0})".format(telemetry_messages))
+
+                log_messages = get_log_messages()
+                self.assertEqual(1, len(log_messages), "Expected additional errors to be reported to the local log in the next period (got: {0})".format(telemetry_messages))
+
+    @patch("azurelinuxagent.common.conf.get_enable_fast_track", return_value=True)
+    def test_it_should_stop_issuing_vm_settings_requests_when_api_is_not_supported(self, _):
+        with mock_wire_protocol(mockwiredata.DATA_FILE_VM_SETTINGS) as protocol:
+            def http_get_handler(url, *_, **__):
+                if self.is_host_plugin_vm_settings_request(url):
+                    return MockHttpResponse(httpclient.NOT_FOUND)  # HostGAPlugin returns 404 if the API is not supported
+                return None
+            protocol.set_http_handlers(http_get_handler=http_get_handler)
+
+            def get_vm_settings_call_count():
+                return len([url for url in protocol.get_tracked_urls() if "vmSettings" in url])
+
+            self._fetch_vm_settings_ignoring_errors(protocol)
+            self.assertEqual(1, get_vm_settings_call_count(), "There should have been an initial call to vmSettings.")
+
+            protocol.client.update_goal_state()
+            protocol.client.update_goal_state()
+            self.assertEqual(1, get_vm_settings_call_count(), "Additional calls to update_goal_state should not have produced extra calls to vmSettings.")
+
+            # reset the vmSettings check period; this should restart the calls to the API
+            protocol.client._host_plugin._host_plugin_supports_vm_settings_next_check = datetime.datetime.now()
+            protocol.client.update_goal_state()
+            self.assertEqual(2, get_vm_settings_call_count(), "A second call to vmSettings was expecting after the check period has elapsed.")
+
+
 class MockResponse:
     def __init__(self, body, status_code, reason=''):
         self.body = body
diff --git a/tests/protocol/test_wire.py b/tests/protocol/test_wire.py
index 48e6c9b4..442e14de 100644
--- a/tests/protocol/test_wire.py
+++ b/tests/protocol/test_wire.py
@@ -19,14 +19,11 @@
 import contextlib
 import json
 import os
-import re
 import socket
 import time
 import unittest
 import uuid
-from datetime import datetime, timedelta
 
-from azurelinuxagent.common import conf
 from azurelinuxagent.common.agent_supported_feature import SupportedFeatureNames, get_supported_feature_by_name, \
     get_agent_supported_features_list_for_crp
 from azurelinuxagent.common.future import httpclient
@@ -34,16 +31,14 @@ from azurelinuxagent.common.event import WALAEventOperation
 from azurelinuxagent.common.exception import ResourceGoneError, ProtocolError, \
     ExtensionDownloadError, HttpError
 from azurelinuxagent.common.protocol import hostplugin
-from azurelinuxagent.common.protocol.extensions_goal_state import GoalStateMismatchError
-from azurelinuxagent.common.protocol.extensions_goal_state_factory import ExtensionsGoalStateFactory
 from azurelinuxagent.common.protocol.extensions_goal_state_from_extensions_config import ExtensionsGoalStateFromExtensionsConfig
 from azurelinuxagent.common.protocol.extensions_goal_state_from_vm_settings import ExtensionsGoalStateFromVmSettings
 from azurelinuxagent.common.protocol.hostplugin import HostPluginProtocol
 from azurelinuxagent.common.protocol.wire import WireProtocol, WireClient, \
-    StatusBlob, VMStatus, EXT_CONF_FILE_NAME, _VmSettingsErrorReporter
+    StatusBlob, VMStatus
 from azurelinuxagent.common.telemetryevent import GuestAgentExtensionEventsSchema, \
     TelemetryEventParam, TelemetryEvent
-from azurelinuxagent.common.utils import restutil, textutil
+from azurelinuxagent.common.utils import restutil
 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
@@ -52,7 +47,7 @@ from tests.protocol.mocks import mock_wire_protocol, MockHttpResponse
 from tests.protocol.HttpRequestPredicates import HttpRequestPredicates
 from tests.protocol.mockwiredata import DATA_FILE_NO_EXT, DATA_FILE
 from tests.protocol.mockwiredata import WireProtocolData
-from tests.tools import Mock, PropertyMock, patch, AgentTestCase, load_bin_data, mock_sleep
+from tests.tools import patch, AgentTestCase
 
 data_with_bom = b'\xef\xbb\xbfhehe'
 testurl = 'http://foo'
@@ -75,14 +70,8 @@ def get_event(message, duration=30000, evt_type="", is_internal=False, is_succes
 
 
 @contextlib.contextmanager
-def create_mock_protocol(status_upload_blob=None, status_upload_blob_type=None):
+def create_mock_protocol():
     with mock_wire_protocol(DATA_FILE_NO_EXT) as protocol:
-        # These tests use mock wire data that dont have any extensions (extension config will be empty).
-        # Mock the upload blob and artifacts profile blob.
-        protocol.client._extensions_goal_state = Mock(wraps=protocol.client._extensions_goal_state)
-        type(protocol.client._extensions_goal_state).status_upload_blob = PropertyMock(return_value=status_upload_blob)
-        type(protocol.client._extensions_goal_state).status_upload_blob_type = PropertyMock(return_value=status_upload_blob_type)
-
         yield protocol
 
 
@@ -103,7 +92,7 @@ class TestWireProtocol(AgentTestCase, HttpRequestPredicates):
             protocol.detect()
             protocol.get_vminfo()
             protocol.get_certs()
-            ext_handlers = protocol.client.get_extensions_goal_state().extensions
+            ext_handlers = protocol.get_goal_state().extensions_goal_state.extensions
             for ext_handler in ext_handlers:
                 protocol.get_ext_handler_pkgs(ext_handler)
 
@@ -216,13 +205,13 @@ class TestWireProtocol(AgentTestCase, HttpRequestPredicates):
 
     def test_status_blob_parsing(self, *args):  # pylint: disable=unused-argument
         with mock_wire_protocol(mockwiredata.DATA_FILE) as protocol:
-            extensions_goal_state = protocol.client.get_extensions_goal_state()
+            extensions_goal_state = protocol.get_goal_state().extensions_goal_state
             self.assertIsInstance(extensions_goal_state, ExtensionsGoalStateFromExtensionsConfig)
             self.assertEqual(extensions_goal_state.status_upload_blob,
                              'https://test.blob.core.windows.net/vhds/test-cs12.test-cs12.test-cs12.status?'
                              'sr=b&sp=rw&se=9999-01-01&sk=key1&sv=2014-02-14&'
                              'sig=hfRh7gzUE7sUtYwke78IOlZOrTRCYvkec4hGZ9zZzXo')
-            self.assertEqual(protocol.client.get_extensions_goal_state().status_upload_blob_type, u'BlockBlob')
+            self.assertEqual(protocol.get_goal_state().extensions_goal_state.status_upload_blob_type, u'BlockBlob')
 
     def test_get_host_ga_plugin(self, *args):  # pylint: disable=unused-argument
         with mock_wire_protocol(mockwiredata.DATA_FILE) as protocol:
@@ -246,7 +235,7 @@ class TestWireProtocol(AgentTestCase, HttpRequestPredicates):
             self.assertEqual(len(urls), 1, 'Expected one post request to the host: [{0}]'.format(urls))
 
     def test_upload_status_blob_host_ga_plugin(self, *_):
-        with create_mock_protocol(status_upload_blob=testurl, status_upload_blob_type=testtype) as protocol:
+        with create_mock_protocol() as protocol:
             protocol.client.status_blob.vm_status = VMStatus(message="Ready", status="Ready")
 
             with patch.object(HostPluginProtocol, "ensure_initialized", return_value=True):
@@ -259,7 +248,7 @@ class TestWireProtocol(AgentTestCase, HttpRequestPredicates):
                         self.assertFalse(HostPluginProtocol.is_default_channel)
 
     def test_upload_status_blob_reports_prepare_error(self, *_):
-        with create_mock_protocol(status_upload_blob=testurl, status_upload_blob_type=testtype) as protocol:
+        with create_mock_protocol() as protocol:
             protocol.client.status_blob.vm_status = VMStatus(message="Ready", status="Ready")
 
             with patch.object(StatusBlob, "prepare", side_effect=Exception) as mock_prepare:
@@ -271,11 +260,11 @@ class TestWireProtocol(AgentTestCase, HttpRequestPredicates):
         data_file["ext_conf"] = "wire/ext_conf_in_vm_empty_artifacts_profile.xml"
 
         with mock_wire_protocol(data_file) as protocol:
-            self.assertFalse(protocol.get_extensions_goal_state().on_hold)
+            self.assertFalse(protocol.get_goal_state().extensions_goal_state.on_hold)
 
     def test_it_should_set_on_hold_to_false_when_the_in_vm_artifacts_profile_is_not_valid(self, *_):
         with mock_wire_protocol(mockwiredata.DATA_FILE_IN_VM_ARTIFACTS_PROFILE) as protocol:
-            extensions_on_hold = protocol.get_extensions_goal_state().on_hold
+            extensions_on_hold = protocol.get_goal_state().extensions_goal_state.on_hold
             self.assertTrue(extensions_on_hold, "Extensions should be on hold in the test data")
 
             def http_get_handler(url, *_, **kwargs):
@@ -286,24 +275,24 @@ class TestWireProtocol(AgentTestCase, HttpRequestPredicates):
 
             mock_response = MockHttpResponse(200, body=None)
             protocol.client.update_goal_state(force_update=True)
-            extensions_on_hold = protocol.get_extensions_goal_state().on_hold
+            extensions_on_hold = protocol.get_goal_state().extensions_goal_state.on_hold
             self.assertFalse(extensions_on_hold, "Extensions should not be on hold when the in-vm artifacts profile response body is None")
 
             mock_response = MockHttpResponse(200, '   '.encode('utf-8'))
             protocol.client.update_goal_state(force_update=True)
-            extensions_on_hold = protocol.get_extensions_goal_state().on_hold
+            extensions_on_hold = protocol.get_goal_state().extensions_goal_state.on_hold
             self.assertFalse(extensions_on_hold, "Extensions should not be on hold when the in-vm artifacts profile response is an empty string")
 
             mock_response = MockHttpResponse(200, '{ }'.encode('utf-8'))
             protocol.client.update_goal_state(force_update=True)
-            extensions_on_hold = protocol.get_extensions_goal_state().on_hold
+            extensions_on_hold = protocol.get_goal_state().extensions_goal_state.on_hold
             self.assertFalse(extensions_on_hold, "Extensions should not be on hold when the in-vm artifacts profile response is an empty json object")
 
             with patch("azurelinuxagent.common.protocol.extensions_goal_state_from_extensions_config.add_event") as add_event:
                 mock_response = MockHttpResponse(200, 'invalid json'.encode('utf-8'))
                 protocol.client.update_goal_state(force_update=True)
 
-                extensions_on_hold = protocol.get_extensions_goal_state().on_hold
+                extensions_on_hold = protocol.get_goal_state().extensions_goal_state.on_hold
                 self.assertFalse(extensions_on_hold, "Extensions should not be on hold when the in-vm artifacts profile response is not valid json")
 
                 events = [kwargs for _, kwargs in add_event.call_args_list if kwargs['op'] == WALAEventOperation.ArtifactsProfileBlob]
@@ -476,11 +465,11 @@ class TestWireProtocol(AgentTestCase, HttpRequestPredicates):
 
 class TestWireClient(HttpRequestPredicates, AgentTestCase):
     def test_get_ext_conf_without_extensions_should_retrieve_vmagent_manifests_info(self, *args):  # pylint: disable=unused-argument
-        # Basic test for get_extensions_goal_state() when extensions are not present in the config. The test verifies that
-        # get_extensions_goal_state() fetches the correct data by comparing the returned data with the test data provided the
+        # Basic test for extensions_goal_state when extensions are not present in the config. The test verifies that
+        # extensions_goal_state fetches the correct data by comparing the returned data with the test data provided the
         # mock_wire_protocol.
         with mock_wire_protocol(mockwiredata.DATA_FILE_NO_EXT) as protocol:
-            extensions_goal_state = protocol.client.get_extensions_goal_state()
+            extensions_goal_state = protocol.get_goal_state().extensions_goal_state
 
             ext_handlers_names = [ext_handler.name for ext_handler in extensions_goal_state.extensions]
             self.assertEqual(0, len(extensions_goal_state.extensions),
@@ -488,19 +477,14 @@ class TestWireClient(HttpRequestPredicates, AgentTestCase):
             vmagent_manifests = [manifest.family for manifest in extensions_goal_state.agent_manifests]
             self.assertEqual(0, len(extensions_goal_state.agent_manifests),
                              "Unexpected number of vmagent manifests in the extension config: [{0}]".format(vmagent_manifests))
-            self.assertIsNone(extensions_goal_state.status_upload_blob,
-                              "Status upload blob in the extension config is expected to be None")
-            self.assertIsNone(extensions_goal_state.status_upload_blob_type,
-                              "Type of status upload blob in the extension config is expected to be None")
             self.assertFalse(extensions_goal_state.on_hold,
                               "Extensions On Hold is expected to be False")
 
     def test_get_ext_conf_with_extensions_should_retrieve_ext_handlers_and_vmagent_manifests_info(self):
-        # Basic test for get_extensions_goal_state() when extensions are present in the config. The test verifies that get_extensions_goal_state()
+        # Basic test for extensions_goal_state when extensions are present in the config. The test verifies that extensions_goal_state
         # fetches the correct data by comparing the returned data with the test data provided the mock_wire_protocol.
         with mock_wire_protocol(mockwiredata.DATA_FILE) as protocol:
-            wire_protocol_client = protocol.client
-            extensions_goal_state = wire_protocol_client.get_extensions_goal_state()
+            extensions_goal_state = protocol.get_goal_state().extensions_goal_state
 
             ext_handlers_names = [ext_handler.name for ext_handler in extensions_goal_state.extensions]
             self.assertEqual(1, len(extensions_goal_state.extensions),
@@ -1055,7 +1039,7 @@ class UpdateGoalStateTestCase(HttpRequestPredicates, AgentTestCase):
                 else:
                     protocol.client.update_goal_state()
 
-                sequence_number = protocol.client.get_extensions_goal_state().extensions[0].settings[0].sequenceNumber
+                sequence_number = protocol.get_goal_state().extensions_goal_state.extensions[0].settings[0].sequenceNumber
 
                 self.assertEqual(protocol.client.get_goal_state().incarnation, new_incarnation)
                 self.assertEqual(protocol.client.get_hosting_env().deployment_name, new_hosting_env_deployment_name)
@@ -1072,8 +1056,8 @@ class UpdateGoalStateTestCase(HttpRequestPredicates, AgentTestCase):
 
             # The container id, role config name and shared config can change without the incarnation changing; capture the initial
             # goal state and then change those fields.
-            goal_state = protocol.client.get_goal_state().xml_text
-            shared_conf = protocol.client.get_shared_conf().xml_text
+            container_id = protocol.client.get_goal_state().container_id
+            role_config_name = protocol.client.get_goal_state().role_config_name
 
             new_container_id = str(uuid.uuid4())
             new_role_config_name = str(uuid.uuid4())
@@ -1084,8 +1068,8 @@ class UpdateGoalStateTestCase(HttpRequestPredicates, AgentTestCase):
 
             protocol.client.update_goal_state()
 
-            self.assertEqual(protocol.client.get_goal_state().xml_text, goal_state)
-            self.assertEqual(protocol.client.get_shared_conf().xml_text, shared_conf)
+            self.assertEqual(protocol.client.get_goal_state().container_id, container_id)
+            self.assertEqual(protocol.client.get_goal_state().role_config_name, role_config_name)
 
             self.assertEqual(protocol.client.get_host_plugin().container_id, new_container_id)
             self.assertEqual(protocol.client.get_host_plugin().role_config_name, new_role_config_name)
@@ -1113,108 +1097,8 @@ class UpdateGoalStateTestCase(HttpRequestPredicates, AgentTestCase):
             self.assertEqual(protocol.client.get_host_plugin().container_id, new_container_id)
             self.assertEqual(protocol.client.get_host_plugin().role_config_name, new_role_config_name)
 
-    def test_update_goal_state_should_archive_last_goal_state(self):
-        # We use the last modified timestamp of the goal state to be archived to determine the archive's name.
-        mock_mtime = os.path.getmtime(self.tmp_dir)
-        with patch("azurelinuxagent.common.utils.archive.os.path.getmtime") as patch_mtime:
-            first_gs_ms = mock_mtime + timedelta(minutes=5).seconds
-            second_gs_ms = mock_mtime + timedelta(minutes=10).seconds
-            third_gs_ms = mock_mtime + timedelta(minutes=15).seconds
-
-            patch_mtime.side_effect = [first_gs_ms, second_gs_ms, third_gs_ms]
-
-            # The first goal state is created when we instantiate the protocol
-            with mock_wire_protocol(mockwiredata.DATA_FILE) as protocol:
-                history_dir = os.path.join(conf.get_lib_dir(), "history")
-                archives = os.listdir(history_dir)
-                self.assertEqual(len(archives), 0, "The goal state archive should have been empty since this is the first goal state")
-
-                # Create the second new goal state, so the initial one should be archived
-                protocol.mock_wire_data.set_incarnation("2")
-                protocol.client.update_goal_state()
-
-                # The initial goal state should be in the archive
-                first_archive_name = datetime.utcfromtimestamp(first_gs_ms).isoformat() + "_incarnation_1"
-                archives = os.listdir(history_dir)
-                self.assertEqual(len(archives), 1, "Only one goal state should have been archived")
-                self.assertEqual(archives[0], first_archive_name, "The name of goal state archive should match the first goal state timestamp and incarnation")
-
-                # Create the third goal state, so the second one should be archived too
-                protocol.mock_wire_data.set_incarnation("3")
-                protocol.client.update_goal_state()
-
-                # The second goal state should be in the archive
-                second_archive_name = datetime.utcfromtimestamp(second_gs_ms).isoformat() + "_incarnation_2"
-                archives = os.listdir(history_dir)
-                archives.sort()
-                self.assertEqual(len(archives), 2, "Two goal states should have been archived")
-                self.assertEqual(archives[1], second_archive_name, "The name of goal state archive should match the second goal state timestamp and incarnation")
-
-    def test_update_goal_state_should_not_persist_the_protected_settings(self):
-        with mock_wire_protocol(mockwiredata.DATA_FILE_MULTIPLE_EXT) as protocol:
-            # instantiating the protocol fetches the goal state, so there is no need to do another call to update_goal_state()
-            goal_state = protocol.client.get_goal_state()
-            extensions_goal_state = protocol.client.get_extensions_goal_state()
-
-            protected_settings = []
-            for ext_handler in extensions_goal_state.extensions:
-                for extension in ext_handler.settings:
-                    if extension.protectedSettings is not None:
-                        protected_settings.append(extension.protectedSettings)
-            if len(protected_settings) == 0:
-                raise Exception("The test goal state does not include any protected settings")
-
-            extensions_config_file = os.path.join(conf.get_lib_dir(), EXT_CONF_FILE_NAME.format(goal_state.incarnation))
-            if not os.path.exists(extensions_config_file):
-                raise Exception("Cannot find {0}".format(extensions_config_file))
-
-            with open(extensions_config_file, "r") as stream:
-                extensions_config = stream.read()
-
-                for settings in protected_settings:
-                    self.assertNotIn(settings, extensions_config, "The protectedSettings should not have been saved to {0}".format(extensions_config_file))
-
-                matches = re.findall(r'"protectedSettings"\s*:\s*"\*\*\* REDACTED \*\*\*"', extensions_config)
-                self.assertEqual(
-                    len(matches),
-                    len(protected_settings),
-                    "Could not find the expected number of redacted settings. Expected {0}.\n{1}".format(len(protected_settings), extensions_config))
-
-    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_goal_state = ExtensionsGoalStateFactory.create_from_extensions_config(123, file_.read(), protocol)
-        self.assertEqual(5, len(extensions_goal_state.extensions), "Incorrect number of extensions in ExtensionsConfig")
-        for e in extensions_goal_state.extensions:
-            if e.name in ("Microsoft.Azure.Monitor.AzureMonitorLinuxAgent", "Microsoft.Azure.Security.Monitoring.AzureSecurityLinuxAgent"):
-                self.assertEqual(e.settings[0].protectedSettings, "*** REDACTED ***", "The protected settings for {0} were not redacted".format(e.name))
-
-        with open(vm_settings_file, "r") as file_:
-            extensions_goal_state = ExtensionsGoalStateFactory.create_from_vm_settings(None, file_.read())
-        self.assertEqual(5, len(extensions_goal_state.extensions), "Incorrect number of extensions in vmSettings")
-        for e in extensions_goal_state.extensions:
-            if e.name in ("Microsoft.Azure.Monitor.AzureMonitorLinuxAgent", "Microsoft.Azure.Security.Monitoring.AzureSecurityLinuxAgent"):
-                self.assertEqual(e.settings[0].protectedSettings, "*** REDACTED ***", "The protected settings for {0} were not redacted".format(e.name))
-
-    def test_it_should_retry_get_vm_settings_on_resource_gone_error(self):
+    @patch("azurelinuxagent.common.conf.get_enable_fast_track", return_value=True)
+    def test_it_should_retry_get_vm_settings_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:
@@ -1242,88 +1126,10 @@ class UpdateGoalStateTestCase(HttpRequestPredicates, AgentTestCase):
             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")
 
-    def test_it_should_not_be_interrupted_by_errors_on_vm_settings(self):
-        def assert_no_exception(test_case, test_function, expected_error):
-            try:
-                with patch("azurelinuxagent.common.protocol.wire.add_event") as add_event:
-                    test_function()
-                    messages = [kwargs["message"] for _, kwargs in add_event.call_args_list]
-                    self.assertTrue(any(expected_error in m for m in messages), "The expected error [{0}] did not occur. Got: {1}".format(expected_error, messages))
-            except Exception as e:
-                self.fail("Error [{0}] produced an unexpected exception: {1}".format(test_case, textutil.format_exception(e)))
-
-        def test_error_in_http_request(test_case, mock_response, expected_error):
-            def do_mock_request():
-                def http_get_handler(url, *_, **__):
-                    if self.is_host_plugin_vm_settings_request(url):
-                        if isinstance(mock_response, Exception):
-                            raise mock_response
-                        return mock_response
-                    return None
-
-                with mock_wire_protocol(mockwiredata.DATA_FILE_VM_SETTINGS, http_get_handler=http_get_handler) as protocol:
-                    protocol.client.update_goal_state()
-
-            assert_no_exception(test_case, do_mock_request, expected_error)
-        #
-        # We test errors different kind of errors; none of them should make update_protocol raise an exception, but all of them should be reported
-        #
-        test_error_in_http_request("Internal error in the HostGAPlugin", MockHttpResponse(httpclient.BAD_GATEWAY), "[Internal error in HostGAPlugin] [HTTP Failed] [502: None]")
-        test_error_in_http_request("Arbitrary error in the request (BAD_REQUEST)", MockHttpResponse(httpclient.BAD_REQUEST), "[HTTP Failed] [400: None]")
-        test_error_in_http_request("Generic error in the request", Exception("GENERIC REQUEST ERROR"), "GENERIC REQUEST ERROR")
-        test_error_in_http_request("Response headers with no Etag", MockHttpResponse(200, b""), "The vmSettings response does not include an Etag header")
-        test_error_in_http_request("Invalid response (bad json)", MockHttpResponse(200, b"{ INVALID JSON ]", headers=[("Etag", 123)]), "Error parsing vmSettings")
-
-        # Lastly, test the goal state comparison
-        def fail_compare():
-            error = GoalStateMismatchError("TEST COMPARE FAILED", "dummy_attribute")
-            with patch("azurelinuxagent.common.protocol.extensions_goal_state.ExtensionsGoalState.compare", side_effect=error):
-                with mock_wire_protocol(mockwiredata.DATA_FILE_VM_SETTINGS) as protocol:
-                    protocol.client.update_goal_state()
-
-        assert_no_exception("Goal state mismatch", fail_compare, "TEST COMPARE FAILED")
-
-    def test_it_should_limit_the_number_of_errors_it_reports(self):
-        with mock_wire_protocol(mockwiredata.DATA_FILE_VM_SETTINGS) as protocol:
-            def http_get_handler(url, *_, **__):
-                if self.is_host_plugin_vm_settings_request(url):
-                    return MockHttpResponse(httpclient.BAD_GATEWAY)  # HostGAPlugin returns 502 for internal errors
-                return None
-            protocol.set_http_handlers(http_get_handler=http_get_handler)
-
-            def get_telemetry_messages():
-                return [kwargs["message"] for _, kwargs in add_event.call_args_list if kwargs["op"] == "VmSettings"]
-
-            def get_log_messages():
-                return [arg[0][0] for arg in logger_info.call_args_list   if "[VmSettings]" in arg[0][0]]
-
-            with patch("azurelinuxagent.common.protocol.wire.add_event") as add_event:
-                with patch('azurelinuxagent.common.logger.info') as logger_info:
-                    for _ in range(_VmSettingsErrorReporter._MaxTelemetryErrors + 3):
-                        protocol.client.update_goal_state()
-
-                    telemetry_messages = get_telemetry_messages()
-                    self.assertEqual(_VmSettingsErrorReporter._MaxTelemetryErrors, len(telemetry_messages), "The number of errors reported to telemetry is not the max allowed (got: {0})".format(telemetry_messages))
-
-                    log_messages = get_log_messages()
-                    self.assertEqual(_VmSettingsErrorReporter._MaxLogErrors, len(log_messages), "The number of errors reported to the local log is not the max allowed (got: {0})".format(telemetry_messages))
-
-            # Reset the error reporter and verify that additional errors are reported
-            protocol.client._vm_settings_error_reporter._next_period = datetime.now()
-            protocol.client.update_goal_state()  # this triggers the reset
-
-            with patch("azurelinuxagent.common.protocol.wire.add_event") as add_event:
-                protocol.client.update_goal_state()
-
-                telemetry_messages = get_telemetry_messages()
-                self.assertEqual(1, len(telemetry_messages), "Expected additional errors to be reported to telemetry in the next period (got: {0})".format(telemetry_messages))
-
-                log_messages = get_log_messages()
-                self.assertEqual(1, len(log_messages), "Expected additional errors to be reported to the local log in the next period (got: {0})".format(telemetry_messages))
-
-    def test_it_should_use_vm_settings_by_default(self):
+    @patch("azurelinuxagent.common.conf.get_enable_fast_track", return_value=True)
+    def test_it_should_use_vm_settings_by_default(self, _):
         with mock_wire_protocol(mockwiredata.DATA_FILE_VM_SETTINGS) as protocol:
-            extensions_goal_state = protocol.get_extensions_goal_state()
+            extensions_goal_state = protocol.get_goal_state().extensions_goal_state
             self.assertTrue(
                 isinstance(extensions_goal_state, ExtensionsGoalStateFromVmSettings),
                 'The extensions goal state should have been created from the vmSettings (got: {0})'.format(type(extensions_goal_state)))
@@ -1336,7 +1142,7 @@ class UpdateGoalStateTestCase(HttpRequestPredicates, AgentTestCase):
     def test_it_should_use_extensions_config_when_fast_track_is_disabled(self):
         with patch("azurelinuxagent.common.conf.get_enable_fast_track", return_value=False):
             with mock_wire_protocol(mockwiredata.DATA_FILE_VM_SETTINGS) as protocol:
-                self._assert_is_extensions_goal_state_from_extensions_config(protocol.get_extensions_goal_state())
+                self._assert_is_extensions_goal_state_from_extensions_config(protocol.get_goal_state().extensions_goal_state)
 
     def test_it_should_use_extensions_config_when_fast_track_is_not_supported(self):
         def http_get_handler(url, *_, **__):
@@ -1345,153 +1151,14 @@ class UpdateGoalStateTestCase(HttpRequestPredicates, AgentTestCase):
             return None
 
         with mock_wire_protocol(mockwiredata.DATA_FILE_VM_SETTINGS, http_get_handler=http_get_handler) as protocol:
-            self._assert_is_extensions_goal_state_from_extensions_config(protocol.get_extensions_goal_state())
-
-    def test_it_should_use_extensions_config_when_the_vm_settings_request_fails(self):
-        def http_get_handler(url, *_, **__):
-            if self.is_host_plugin_vm_settings_request(url):
-                return MockHttpResponse(httpclient.INTERNAL_SERVER_ERROR)
-            return None
-
-        with mock_wire_protocol(mockwiredata.DATA_FILE_VM_SETTINGS, http_get_handler=http_get_handler) as protocol:
-            self._assert_is_extensions_goal_state_from_extensions_config(protocol.get_extensions_goal_state())
+            self._assert_is_extensions_goal_state_from_extensions_config(protocol.get_goal_state().extensions_goal_state)
 
     def test_it_should_use_extensions_config_when_the_host_ga_plugin_version_is_not_supported(self):
         data_file = mockwiredata.DATA_FILE_VM_SETTINGS.copy()
         data_file["vm_settings"] = "hostgaplugin/vm_settings-unsupported_version.json"
 
         with mock_wire_protocol(data_file) as protocol:
-            self._assert_is_extensions_goal_state_from_extensions_config(protocol.get_extensions_goal_state())
-
-    def test_it_should_use_extensions_config_when_vm_settings_can_not_be_parsed(self):
-        data_file = mockwiredata.DATA_FILE_VM_SETTINGS.copy()
-        data_file["vm_settings"] = "hostgaplugin/vm_settings-parse_error.json"
-
-        with mock_wire_protocol(data_file) as protocol:
-            self._assert_is_extensions_goal_state_from_extensions_config(protocol.get_extensions_goal_state())
-
-    def test_it_should_use_extensions_config_when_vm_settings_do_not_match_extensions_config(self):
-        data_file = mockwiredata.DATA_FILE_VM_SETTINGS.copy()
-        data_file["vm_settings"] = "hostgaplugin/vm_settings-difference_in_required_features.json"
-
-        with patch('azurelinuxagent.common.event.EventLogger.add_event') as add_event_patcher:
-            with mock_wire_protocol(data_file) as protocol:
-                self._assert_is_extensions_goal_state_from_extensions_config(protocol.get_extensions_goal_state())
-
-                reported = [kwargs for _, kwargs in add_event_patcher.call_args_list if kwargs['op'] == "VmSettings" and "GoalStateMismatchError" in kwargs['message']]
-                self.assertEqual(1, len(reported), "The goal state mismatch should have been reported exactly once; got: {0}".format([kwargs['message'] for _, kwargs in add_event_patcher.call_args_list]))
-
-    def test_it_should_retry_vm_settings_and_extensions_config_do_not_match(self):
-        def http_get_handler(url, *_, **__):
-            if self.is_host_plugin_vm_settings_request(url):
-                response = MockHttpResponse(httpclient.OK)
-                response.body = load_bin_data("hostgaplugin/vm_settings-out-of-sync.json")
-                response.headers = [('ETag', "0123456789")]
-                return response
-            return None
-
-        data_file = mockwiredata.DATA_FILE_VM_SETTINGS.copy()
-        with mock_wire_protocol(data_file) as protocol:
-            protocol.set_http_handlers(http_get_handler=http_get_handler)
-
-            with patch('time.sleep', side_effect=lambda _: mock_sleep()):  # avoid the sleep during retry
-                with patch("azurelinuxagent.common.protocol.wire.add_event") as add_event:
-                    protocol.client.update_goal_state()
-
-            vm_settings_call_count = len([url for url in protocol.get_tracked_urls() if "vmSettings" in url])
-            self.assertEqual(2, vm_settings_call_count, "Expected 2 calls to vmSettings (original and retry)")
-
-            errors = [kwargs["message"] for _, kwargs in add_event.call_args_list if kwargs["op"] == "VmSettings"]
-            self.assertTrue(any("[GoalStateMismatchError]" in e for e in errors), "Expected GoalStateMismatchError to have been reported. Got: {0}".format(errors))
-
-    def test_it_should_compare_goal_states_when_vm_settings_change(self):
-        with mock_wire_protocol(mockwiredata.DATA_FILE_VM_SETTINGS) as protocol:
-            protocol.mock_wire_data.set_etag("aNewEtag")
-
-            with patch('azurelinuxagent.common.protocol.extensions_goal_state.ExtensionsGoalState.compare') as compare_patcher:
-                protocol.update_goal_state()
-
-            self.assertEqual(1, compare_patcher.call_count, "ExtensionsGoalState.compare() should have been called exactly once")
-
-    def test_it_should_compare_goal_states_when_extensions_config_change(self):
-        with mock_wire_protocol(mockwiredata.DATA_FILE_VM_SETTINGS) as protocol:
-            protocol.mock_wire_data.set_incarnation(468753)
-
-            with patch('azurelinuxagent.common.protocol.extensions_goal_state.ExtensionsGoalState.compare') as compare_patcher:
-                protocol.update_goal_state()
-
-            self.assertEqual(1, compare_patcher.call_count, "ExtensionsGoalState.compare() should have been called exactly once")
-
-    def test_it_should_keep_track_of_errors_in_vm_settings_requests(self):
-        mock_response = None
-
-        def http_get_handler(url, *_, **__):
-            if self.is_host_plugin_vm_settings_request(url):
-                if isinstance(mock_response, Exception):
-                    # E0702: Raising NoneType while only classes or instances are allowed (raising-bad-type) - Disabled: we never raise None
-                    raise mock_response  # pylint: disable=raising-bad-type
-                return mock_response
-            return None
-
-        with mock_wire_protocol(mockwiredata.DATA_FILE_VM_SETTINGS, http_get_handler=http_get_handler) as protocol:
-            mock_response = MockHttpResponse(httpclient.INTERNAL_SERVER_ERROR)
-            protocol.client.update_goal_state()
-
-            mock_response = MockHttpResponse(httpclient.BAD_REQUEST)
-            protocol.client.update_goal_state()
-            protocol.client.update_goal_state()
-
-            mock_response = IOError("timed out")
-            protocol.client.update_goal_state()
-
-            mock_response = httpclient.HTTPException()
-            protocol.client.update_goal_state()
-            protocol.client.update_goal_state()
-
-            # force the summary by resetting its period and calling update_goal_state
-            with patch("azurelinuxagent.common.protocol.wire.add_event") as add_event:
-                mock_response = None  # stop producing errors
-                protocol.client._vm_settings_error_reporter._next_period = datetime.now()
-                protocol.client.update_goal_state()
-            summary_text = [kwargs["message"] for _, kwargs in add_event.call_args_list if kwargs["op"] == "VmSettingsSummary"]
-
-            self.assertEqual(1, len(summary_text), "Exactly 1 summary should have been produced. Got: {0} ".format(summary_text))
-
-            summary = json.loads(summary_text[0])
-
-            expected = {
-                "requests":       6 + 2,  # two extra calls to update_goal_state (when creating the mock protocol and when forcing the summary)
-                "errors":         6,
-                "serverErrors":   1,
-                "clientErrors":   2,
-                "timeouts":       1,
-                "failedRequests": 2
-            }
-
-            self.assertEqual(expected, summary, "The count of errors is incorrect")
-
-    def test_it_should_stop_issuing_vm_settings_requests_when_api_is_not_supported(self):
-        with mock_wire_protocol(mockwiredata.DATA_FILE_VM_SETTINGS) as protocol:
-            def http_get_handler(url, *_, **__):
-                if self.is_host_plugin_vm_settings_request(url):
-                    return MockHttpResponse(httpclient.NOT_FOUND)  # HostGAPlugin returns 404 if the API is not supported
-                return None
-            protocol.set_http_handlers(http_get_handler=http_get_handler)
-
-            def get_vm_settings_call_count():
-                return len([url for url in protocol.get_tracked_urls() if "vmSettings" in url])
-
-            protocol.client.update_goal_state()
-            self.assertEqual(1, get_vm_settings_call_count(), "There should have been an initial call to vmSettings.")
-
-            protocol.client.update_goal_state()
-            protocol.client.update_goal_state()
-            self.assertEqual(1, get_vm_settings_call_count(), "Additional calls to update_goal_state should not have produced extra calls to vmSettings.")
-
-            # reset the vmSettings check period; this should restart the calls to the API
-            protocol.client._host_plugin_supports_vm_settings_next_check = datetime.now()
-            protocol.client.update_goal_state()
-            self.assertEqual(2, get_vm_settings_call_count(), "A second call to vmSettings was expecting after the check period has elapsed.")
+            self._assert_is_extensions_goal_state_from_extensions_config(protocol.get_goal_state().extensions_goal_state)
 
 
 class UpdateHostPluginFromGoalStateTestCase(AgentTestCase):
@@ -1509,9 +1176,6 @@ class UpdateHostPluginFromGoalStateTestCase(AgentTestCase):
                 new_container_id = str(uuid.uuid4())
                 new_role_config_name = str(uuid.uuid4())
 
-                goal_state_xml_text = protocol.mock_wire_data.goal_state
-                shared_conf_xml_text = protocol.mock_wire_data.shared_config
-
                 if incarnation_change:
                     protocol.mock_wire_data.set_incarnation(str(uuid.uuid4()))
 
@@ -1525,10 +1189,6 @@ class UpdateHostPluginFromGoalStateTestCase(AgentTestCase):
                 self.assertEqual(protocol.client.get_host_plugin().container_id, new_container_id)
                 self.assertEqual(protocol.client.get_host_plugin().role_config_name, new_role_config_name)
 
-                # it should not update the goal state
-                self.assertEqual(protocol.client.get_goal_state().xml_text, goal_state_xml_text)
-                self.assertEqual(protocol.client.get_shared_conf().xml_text, shared_conf_xml_text)
-
 
 if __name__ == '__main__':
     unittest.main()
diff --git a/tests/test_agent.py b/tests/test_agent.py
index e5a37dfd..b35b1424 100644
--- a/tests/test_agent.py
+++ b/tests/test_agent.py
@@ -38,7 +38,7 @@ Debug.CgroupDisableOnQuotaCheckFailure = True
 Debug.CgroupLogMetrics = False
 Debug.CgroupMonitorExpiryTime = 2022-01-31
 Debug.CgroupMonitorExtensionName = Microsoft.Azure.Monitor.AzureMonitorLinuxAgent
-Debug.EnableFastTrack = True
+Debug.EnableFastTrack = False
 Debug.EnableGAVersioning = False
 Debug.EtpCollectionPeriod = 300
 Debug.FirewallRulesLogPeriod = 86400
diff --git a/tests/utils/test_archive.py b/tests/utils/test_archive.py
index 2e468487..7834a7d2 100644
--- a/tests/utils/test_archive.py
+++ b/tests/utils/test_archive.py
@@ -8,7 +8,7 @@ from datetime import datetime, timedelta
 
 import azurelinuxagent.common.logger as logger
 from azurelinuxagent.common.utils import fileutil
-from azurelinuxagent.common.utils.archive import StateFlusher, StateArchiver, _MAX_ARCHIVED_STATES
+from azurelinuxagent.common.utils.archive import StateArchiver, _MAX_ARCHIVED_STATES
 from tests.tools import AgentTestCase, patch
 
 debug = False
@@ -53,41 +53,6 @@ class TestArchive(AgentTestCase):
         incarnation_no_ext = os.path.splitext(incarnation_ext)[0]
         return timestamp_str, incarnation_no_ext
 
-    def test_archive00(self):
-        """
-        StateFlusher should move all 'goal state' files to a new directory
-        under the history folder that is timestamped.
-        """
-        temp_files = [
-            'GoalState.0.xml',
-            'Prod.0.manifest.xml',
-            'Prod.0.agentsManifest',
-            'Microsoft.Azure.Extensions.CustomScript.0.xml'
-        ]
-
-        for temp_file in temp_files:
-            self._write_file(temp_file)
-
-        test_subject = StateFlusher(self.tmp_dir)
-        test_subject.flush()
-
-        self.assertTrue(os.path.exists(self.history_dir))
-        self.assertTrue(os.path.isdir(self.history_dir))
-
-        timestamp_dirs = os.listdir(self.history_dir)
-        self.assertEqual(1, len(timestamp_dirs))
-
-        timestamp_str, incarnation = self._parse_archive_name(timestamp_dirs[0])
-        self.assert_is_iso8601(timestamp_str)
-        timestamp = self.parse_isoformat(timestamp_str)
-        self.assert_datetime_close_to(timestamp, datetime.utcnow(), timedelta(seconds=30))
-        self.assertEqual("0", incarnation)
-
-        for temp_file in temp_files:
-            history_path = os.path.join(self.history_dir, timestamp_dirs[0], temp_file)
-            msg = "expected the temp file {0} to exist".format(history_path)
-            self.assertTrue(os.path.exists(history_path), msg)
-
     def test_archive01(self):
         """
         StateArchiver should archive all history directories by
@@ -97,17 +62,17 @@ class TestArchive(AgentTestCase):
           2. Deleting the timestamped directory
         """
         temp_files = [
-            'GoalState.0.xml',
-            'Prod.0.manifest.xml',
-            'Prod.0.agentsManifest',
-            'Microsoft.Azure.Extensions.CustomScript.0.xml'
+            'GoalState.xml',
+            'Prod.manifest.xml',
+            'Prod.agentsManifest',
+            'Microsoft.Azure.Extensions.CustomScript.xml'
         ]
 
-        for current_file in temp_files:
-            self._write_file(current_file)
+        # this directory matches the pattern that StateArchiver.archive() searches for
+        temp_directory = os.path.join(self.history_dir, datetime.utcnow().isoformat() + "_incarnation_0")
 
-        flusher = StateFlusher(self.tmp_dir)
-        flusher.flush()
+        for current_file in temp_files:
+            self._write_file(os.path.join(temp_directory, current_file))
 
         test_subject = StateArchiver(self.tmp_dir)
         test_subject.archive()
@@ -147,9 +112,9 @@ class TestArchive(AgentTestCase):
             timestamps.append(timestamp)
 
             if i % 2 == 0:
-                filename = os.path.join('history', "{0}_incarnation_0".format(timestamp.isoformat()), 'Prod.0.manifest.xml')
+                filename = os.path.join('history', "{0}_0".format(timestamp.isoformat()), 'Prod.manifest.xml')
             else:
-                filename = os.path.join('history', "{0}_incarnation_0.zip".format(timestamp.isoformat()))
+                filename = os.path.join('history', "{0}_0.zip".format(timestamp.isoformat()))
 
             self._write_file(filename)
 
@@ -166,18 +131,19 @@ class TestArchive(AgentTestCase):
         for i in range(0, _MAX_ARCHIVED_STATES):
             timestamp = timestamps[i + count].isoformat()
             if i % 2 == 0:
-                filename = "{0}_incarnation_0".format(timestamp)
+                filename = "{0}_0".format(timestamp)
             else:
-                filename = "{0}_incarnation_0.zip".format(timestamp)
+                filename = "{0}_0.zip".format(timestamp)
             self.assertTrue(filename in archived_entries, "'{0}' is not in the list of unpurged entires".format(filename))
 
     def test_archive03(self):
         """
-        All archives should be purged, both with the new naming (with incarnation number) and with the old naming.
+        All archives should be purged, both with the legacy naming (with incarnation number) and with the new naming.
         """
         start = datetime.now()
         timestamp1 = start + timedelta(seconds=5)
         timestamp2 = start + timedelta(seconds=10)
+        timestamp3 = start + timedelta(seconds=10)
 
         dir_old = timestamp1.isoformat()
         dir_new = "{0}_incarnation_1".format(timestamp2.isoformat())
@@ -185,12 +151,15 @@ class TestArchive(AgentTestCase):
         archive_old = "{0}.zip".format(timestamp1.isoformat())
         archive_new = "{0}_incarnation_1.zip".format(timestamp2.isoformat())
 
-        self._write_file(os.path.join("history", dir_old, "Prod.0.manifest.xml"))
-        self._write_file(os.path.join("history", dir_new, "Prod.1.manifest.xml"))
+        status = "{0}.zip".format(timestamp3.isoformat())
+
+        self._write_file(os.path.join("history", dir_old, "Prod.manifest.xml"))
+        self._write_file(os.path.join("history", dir_new, "Prod.manifest.xml"))
         self._write_file(os.path.join("history", archive_old))
         self._write_file(os.path.join("history", archive_new))
+        self._write_file(os.path.join("history", status))
 
-        self.assertEqual(4, len(os.listdir(self.history_dir)), "Not all entries were archived!")
+        self.assertEqual(5, len(os.listdir(self.history_dir)), "Not all entries were archived!")
 
         test_subject = StateArchiver(self.tmp_dir)
         with patch("azurelinuxagent.common.utils.archive._MAX_ARCHIVED_STATES", 0):