Repository: Azure/WALinuxAgent
Pull request #1796 (2020-03-04 17:47:09)

Adding Metadata protocol back


<!-- Reviewable:start -->
This change is [<img src="https://reviewable.io/review_button.svg" height="34" align="absmiddle" alt="Reviewable"/>](https://reviewable.io/reviews/azure/walinuxagent/1796)
<!-- Reviewable:end -->


codecov[bot]: # [Codecov](https://codecov.io/gh/Azure/WALinuxAgent/pull/1796?src=pr&el=h1) Report
> Merging [#1796](https://codecov.io/gh/Azure/WALinuxAgent/pull/1796?src=pr&el=desc) into [develop](https://codecov.io/gh/Azure/WALinuxAgent/commit/1230836e5456d3fb06fecfe4acfd22fee4010621&el=desc) will **decrease** coverage by `0.29%`.
> The diff coverage is `n/a`.

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

```diff
@@             Coverage Diff             @@
##           develop    #1796      +/-   ##
===========================================
- Coverage    68.87%   68.57%   -0.30%     
===========================================
  Files           81       82       +1     
  Lines        11361    11721     +360     
  Branches      1601     1642      +41     
===========================================
+ Hits          7825     8038     +213     
- Misses        3204     3335     +131     
- Partials       332      348      +16     
```


| [Impacted Files](https://codecov.io/gh/Azure/WALinuxAgent/pull/1796?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [azurelinuxagent/common/protocol/goal\_state.py](https://codecov.io/gh/Azure/WALinuxAgent/pull/1796/diff?src=pr&el=tree#diff-YXp1cmVsaW51eGFnZW50L2NvbW1vbi9wcm90b2NvbC9nb2FsX3N0YXRlLnB5) | `95.01% <0.00%> (ø)` | |
| [azurelinuxagent/common/protocol/metadata.py](https://codecov.io/gh/Azure/WALinuxAgent/pull/1796/diff?src=pr&el=tree#diff-YXp1cmVsaW51eGFnZW50L2NvbW1vbi9wcm90b2NvbC9tZXRhZGF0YS5weQ==) | `53.31% <0.00%> (ø)` | |
| [azurelinuxagent/common/utils/restutil.py](https://codecov.io/gh/Azure/WALinuxAgent/pull/1796/diff?src=pr&el=tree#diff-YXp1cmVsaW51eGFnZW50L2NvbW1vbi91dGlscy9yZXN0dXRpbC5weQ==) | `91.47% <0.00%> (+0.38%)` | :arrow_up: |
| [azurelinuxagent/common/protocol/restapi.py](https://codecov.io/gh/Azure/WALinuxAgent/pull/1796/diff?src=pr&el=tree#diff-YXp1cmVsaW51eGFnZW50L2NvbW1vbi9wcm90b2NvbC9yZXN0YXBpLnB5) | `87.42% <0.00%> (+0.59%)` | :arrow_up: |
| [azurelinuxagent/common/utils/cryptutil.py](https://codecov.io/gh/Azure/WALinuxAgent/pull/1796/diff?src=pr&el=tree#diff-YXp1cmVsaW51eGFnZW50L2NvbW1vbi91dGlscy9jcnlwdHV0aWwucHk=) | `50.00% <0.00%> (+1.92%)` | :arrow_up: |
| [azurelinuxagent/common/protocol/util.py](https://codecov.io/gh/Azure/WALinuxAgent/pull/1796/diff?src=pr&el=tree#diff-YXp1cmVsaW51eGFnZW50L2NvbW1vbi9wcm90b2NvbC91dGlsLnB5) | `83.41% <0.00%> (+3.64%)` | :arrow_up: |

------

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

narrieta: @adepue - could you check this PR? Thanks
adepue: Looks good to me 

Patch:
diff --git a/azurelinuxagent/common/protocol/goal_state.py b/azurelinuxagent/common/protocol/goal_state.py
index 83717d08..8787c09b 100644
--- a/azurelinuxagent/common/protocol/goal_state.py
+++ b/azurelinuxagent/common/protocol/goal_state.py
@@ -74,7 +74,15 @@ class GoalState(object):
 
         GoalState.ContainerID = self.container_id
 
-        if not (full_goal_state or base_incarnation is not None and self.incarnation != base_incarnation):
+        fetch_full_goal_state = False
+        if full_goal_state:
+            fetch_full_goal_state = True
+            reason = 'force update'
+        elif base_incarnation is not None and self.incarnation != base_incarnation:
+            fetch_full_goal_state = True
+            reason = 'new incarnation'
+
+        if not fetch_full_goal_state:
             self.hosting_env = None
             self.shared_conf = None
             self.certs = None
@@ -82,36 +90,42 @@ class GoalState(object):
             self.remote_access = None
             return
 
-        logger.info('Fetching new goal state [incarnation {0}]', self.incarnation)
+        logger.info('Fetching new goal state [incarnation {0} ({1})]', self.incarnation, reason)
 
-        uri = findtext(xml_doc, "HostingEnvironmentConfig")
-        xml_text = wire_client.fetch_config(uri, wire_client.get_header())
-        self.hosting_env = HostingEnv(xml_text)
-
-        uri = findtext(xml_doc, "SharedConfig")
-        xml_text = wire_client.fetch_config(uri, wire_client.get_header())
-        self.shared_conf = SharedConfig(xml_text)
+        try:
+            uri = findtext(xml_doc, "HostingEnvironmentConfig")
+            xml_text = wire_client.fetch_config(uri, wire_client.get_header())
+            self.hosting_env = HostingEnv(xml_text)
 
-        uri = findtext(xml_doc, "Certificates")
-        if uri is None:
-            self.certs = None
-        else:
-            xml_text = wire_client.fetch_config(uri, wire_client.get_header_for_cert())
-            self.certs = Certificates(xml_text)
-
-        uri = findtext(xml_doc, "ExtensionsConfig")
-        if uri is None:
-            self.ext_conf = ExtensionsConfig(None)
-        else:
+            uri = findtext(xml_doc, "SharedConfig")
             xml_text = wire_client.fetch_config(uri, wire_client.get_header())
-            self.ext_conf = ExtensionsConfig(xml_text)
+            self.shared_conf = SharedConfig(xml_text)
 
-        uri = findtext(container, "RemoteAccessInfo")
-        if uri is None:
-            self.remote_access = None
-        else:
-            xml_text = wire_client.fetch_config(uri, wire_client.get_header_for_cert())
-            self.remote_access = RemoteAccess(xml_text)
+            uri = findtext(xml_doc, "Certificates")
+            if uri is None:
+                self.certs = None
+            else:
+                xml_text = wire_client.fetch_config(uri, wire_client.get_header_for_cert())
+                self.certs = Certificates(xml_text)
+
+            uri = findtext(xml_doc, "ExtensionsConfig")
+            if uri is None:
+                self.ext_conf = ExtensionsConfig(None)
+            else:
+                xml_text = wire_client.fetch_config(uri, wire_client.get_header())
+                self.ext_conf = ExtensionsConfig(xml_text)
+
+            uri = findtext(container, "RemoteAccessInfo")
+            if uri is None:
+                self.remote_access = None
+            else:
+                xml_text = wire_client.fetch_config(uri, wire_client.get_header_for_cert())
+                self.remote_access = RemoteAccess(xml_text)
+        except Exception as e:
+            logger.warn("Fetching the goal state failed: {0}", ustr(e))
+            raise
+        finally:
+            logger.info('Fetch goal state completed')
 
     @staticmethod
     def fetch_goal_state(wire_client):
diff --git a/azurelinuxagent/common/protocol/metadata.py b/azurelinuxagent/common/protocol/metadata.py
new file mode 100644
index 00000000..1956311d
--- /dev/null
+++ b/azurelinuxagent/common/protocol/metadata.py
@@ -0,0 +1,479 @@
+# Microsoft Azure Linux Agent
+#
+# Copyright 2018 Microsoft Corporation
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# Requires Python 2.6+ and Openssl 1.0+
+
+import base64
+import json
+import os
+import shutil
+import re
+import sys
+import traceback
+
+import azurelinuxagent.common.conf as conf
+from azurelinuxagent.common.datacontract import get_properties, set_properties, validate_param
+from azurelinuxagent.common.exception import HttpError, ProtocolError
+import azurelinuxagent.common.logger as logger
+from azurelinuxagent.common.utils import restutil
+import azurelinuxagent.common.utils.fileutil as fileutil
+import azurelinuxagent.common.utils.shellutil as shellutil
+import azurelinuxagent.common.utils.textutil as textutil
+from azurelinuxagent.common.telemetryevent import TelemetryEventList
+
+from azurelinuxagent.common.future import httpclient
+from azurelinuxagent.common.protocol.restapi import *
+from azurelinuxagent.common.utils.cryptutil import CryptUtil
+
+METADATA_ENDPOINT = '169.254.169.254'
+APIVERSION = '2015-05-01-preview'
+BASE_URI = "http://{0}/Microsoft.Compute/{1}?api-version={2}"
+
+TRANSPORT_PRV_FILE_NAME = "V2TransportPrivate.pem"
+TRANSPORT_CERT_FILE_NAME = "V2TransportCert.pem"
+P7M_FILE_NAME = "Certificates.p7m"
+P7B_FILE_NAME = "Certificates.p7b"
+PEM_FILE_NAME = "Certificates.pem"
+
+IF_NONE_MATCH_HEADER = "If-None-Match"
+
+KEY_AGENT_VERSION_URIS = "versionsManifestUris"
+KEY_URI = "uri"
+
+# TODO remote workaround for azure stack
+MAX_PING = 30
+RETRY_PING_INTERVAL = 10
+
+
+def get_traceback(e):
+    if sys.version_info[0] == 3:
+        return e.__traceback__
+    elif sys.version_info[0] == 2:
+        ex_type, ex, tb = sys.exc_info()
+        return tb
+
+
+def _add_content_type(headers):
+    if headers is None:
+        headers = {}
+    headers["content-type"] = "application/json"
+    return headers
+
+
+class MetadataProtocol(Protocol):
+    def __init__(self, apiversion=APIVERSION, endpoint=METADATA_ENDPOINT):
+        self.apiversion = apiversion
+        self.endpoint = endpoint
+        self.identity_uri = BASE_URI.format(self.endpoint, "identity",
+                                            self.apiversion)
+        self.cert_uri = BASE_URI.format(self.endpoint, "certificates",
+                                        self.apiversion)
+        self.ext_uri = BASE_URI.format(self.endpoint, "extensionHandlers",
+                                       self.apiversion)
+        self.vmagent_uri = BASE_URI.format(self.endpoint, "vmAgentVersions",
+                                           self.apiversion)
+        self.provision_status_uri = BASE_URI.format(self.endpoint,
+                                                    "provisioningStatus",
+                                                    self.apiversion, "")
+        self.vm_status_uri = BASE_URI.format(self.endpoint, "status/vmagent",
+                                             self.apiversion, "")
+        self.ext_status_uri = BASE_URI.format(self.endpoint,
+                                              "status/extensions/{0}",
+                                              self.apiversion, "")
+        self.event_uri = BASE_URI.format(self.endpoint, "status/telemetry",
+                                         self.apiversion, "")
+        self.certs = None
+        self.agent_manifests = None
+        self.agent_etag = None
+        self.cert_etag = None
+
+    def _get_data(self, url, headers=None):
+        try:
+            resp = restutil.http_get(url, headers=headers)
+        except HttpError as e:
+            raise ProtocolError(ustr(e))
+
+        # NOT_MODIFIED (304) response means the call was successful, so allow that to proceed.
+        is_not_modified = restutil.request_not_modified(resp)
+        if restutil.request_failed(resp) and not is_not_modified:
+            raise ProtocolError("{0} - GET: {1}".format(resp.status, url))
+
+        data = resp.read()
+        etag = resp.getheader('ETag')
+
+        # If the response was 304, then explicilty set data to None
+        if is_not_modified:
+            data = None
+
+        if data is not None:
+            data = json.loads(ustr(data, encoding="utf-8"))
+
+        return data, etag
+
+    def _put_data(self, url, data, headers=None):
+        headers = _add_content_type(headers)
+        try:
+            resp = restutil.http_put(url, json.dumps(data), headers=headers)
+        except HttpError as e:
+            raise ProtocolError(ustr(e))
+        if restutil.request_failed(resp):
+            raise ProtocolError("{0} - PUT: {1}".format(resp.status, url))
+
+    def _post_data(self, url, data, headers=None):
+        headers = _add_content_type(headers)
+        try:
+            resp = restutil.http_post(url, json.dumps(data), headers=headers)
+        except HttpError as e:
+            raise ProtocolError(ustr(e))
+        if resp.status != httpclient.CREATED:
+            logger.warn("{0} for POST {1}".format(resp.status, url))
+
+    def _get_trans_cert(self):
+        trans_crt_file = os.path.join(conf.get_lib_dir(),
+                                      TRANSPORT_CERT_FILE_NAME)
+        if not os.path.isfile(trans_crt_file):
+            raise ProtocolError("{0} is missing.".format(trans_crt_file))
+        content = fileutil.read_file(trans_crt_file)
+        return textutil.get_bytes_from_pem(content)
+
+    def supports_overprovisioning(self):
+        # Metadata protocol does not support overprovisioning
+        return False
+
+    def detect(self):
+        self.get_vminfo()
+        trans_prv_file = os.path.join(conf.get_lib_dir(),
+                                      TRANSPORT_PRV_FILE_NAME)
+        trans_cert_file = os.path.join(conf.get_lib_dir(),
+                                       TRANSPORT_CERT_FILE_NAME)
+        cryptutil = CryptUtil(conf.get_openssl_cmd())
+        cryptutil.gen_transport_cert(trans_prv_file, trans_cert_file)
+
+        # "Install" the cert and private key to /var/lib/waagent
+        thumbprint = cryptutil.get_thumbprint_from_crt(trans_cert_file)
+        prv_file = os.path.join(conf.get_lib_dir(),
+                                "{0}.prv".format(thumbprint))
+        crt_file = os.path.join(conf.get_lib_dir(),
+                                "{0}.crt".format(thumbprint))
+        shutil.copyfile(trans_prv_file, prv_file)
+        shutil.copyfile(trans_cert_file, crt_file)
+        self._update_certs_with_retry()
+
+    def update_host_plugin_from_goal_state(self):
+        # Metadata doesn't cache the goal state; nothing to do here...
+        pass
+
+    def update_goal_state(self):
+        # Metadata doesn't cache the goal state; nothing to do here...
+        pass
+
+    def get_endpoint(self):
+        return self.endpoint
+
+    def get_vminfo(self):
+        vminfo = VMInfo()
+        data, etag = self._get_data(self.identity_uri)
+        set_properties("vminfo", vminfo, data)
+        return vminfo
+
+    def get_certs(self):
+        certlist = CertList()
+        certificatedata = CertificateData()
+        headers = None if self.cert_etag is None else {IF_NONE_MATCH_HEADER: self.cert_etag}
+        data, etag = self._get_data(self.cert_uri, headers=headers)
+
+        if self.cert_etag is None or self.cert_etag != etag:
+            self.cert_etag = etag
+
+            set_properties("certlist", certlist, data)
+
+            cert_list = get_properties(certlist)
+
+            headers = {
+                "x-ms-vmagent-public-x509-cert": self._get_trans_cert()
+            }
+
+            for cert_i in cert_list["certificates"]:
+                certificate_data_uri = cert_i['certificateDataUri']
+                data, etag = self._get_data(certificate_data_uri, headers=headers)
+                set_properties("certificatedata", certificatedata, data)
+                json_certificate_data = get_properties(certificatedata)
+
+                self.certs = Certificates(self, json_certificate_data)
+
+        if self.certs is None:
+            return None
+        return self.certs
+
+    def get_incarnation(self):
+        # Always return 0 since Azure Stack does not maintain goal state
+        # incarnation identifiers
+        return 0
+
+    def get_vmagent_manifests(self):
+        self._update_certs_with_retry()
+
+        headers = None if self.agent_etag is None else {IF_NONE_MATCH_HEADER: self.agent_etag}
+
+        data, etag = self._get_data(self.vmagent_uri, headers=headers)
+        if self.agent_etag is None or self.agent_etag != etag:
+            self.agent_etag = etag
+
+            # Create a list with a single manifest
+            # -- The protocol lacks "family," use the configured family
+            self.agent_manifests = VMAgentManifestList()
+
+            manifest = VMAgentManifest()
+            manifest.family = family=conf.get_autoupdate_gafamily()
+            
+            if not KEY_AGENT_VERSION_URIS in data:
+                raise ProtocolError(
+                    "Agent versions missing '{0}': {1}".format(
+                        KEY_AGENT_VERSION_URIS, data))
+
+            for version in data[KEY_AGENT_VERSION_URIS]:
+                if not KEY_URI in version:
+                    raise ProtocolError(
+                        "Agent versions missing '{0': {1}".format(
+                            KEY_URI, data))
+                manifest_uri = VMAgentManifestUri(uri=version[KEY_URI])
+                manifest.versionsManifestUris.append(manifest_uri)
+        
+            self.agent_manifests.vmAgentManifests.append(manifest)
+        
+        return self.agent_manifests, self.agent_etag
+
+    def get_vmagent_pkgs(self, vmagent_manifest):
+        data = None
+        etag = None
+        for manifest_uri in vmagent_manifest.versionsManifestUris:
+            try:
+                data, etag = self._get_data(manifest_uri.uri)
+                break
+            except ProtocolError as e:
+                logger.verbose(
+                    "Error retrieving agent package from {0}: {1}".format(
+                        manifest_uri, e))
+
+        if data is None:
+            raise ProtocolError(
+                "Failed retrieving agent package from all URIs")
+
+        vmagent_pkgs = ExtHandlerPackageList()
+        set_properties("vmAgentVersions", vmagent_pkgs, data)
+        return vmagent_pkgs
+
+    def get_ext_handlers(self, last_etag=None):
+        self._update_certs_with_retry()
+        headers = {
+            "x-ms-vmagent-public-x509-cert": self._get_trans_cert()
+        }
+        ext_list = ExtHandlerList()
+        data, etag = self._get_data(self.ext_uri, headers=headers)
+        if last_etag is None or last_etag != etag:
+            set_properties("extensionHandlers", ext_list.extHandlers, data)
+        return ext_list, etag
+
+    def get_ext_handler_pkgs(self, ext_handler):
+        logger.verbose("Get extension handler packages")
+        pkg_list = ExtHandlerPackageList()
+
+        manifest = None
+        for version_uri in ext_handler.versionUris:
+            try:
+                manifest, etag = self._get_data(version_uri.uri)
+                logger.verbose("Successfully downloaded manifest")
+                break
+            except ProtocolError as e:
+                logger.warn("Failed to fetch manifest: {0}", e)
+
+        if manifest is None:
+            raise ValueError("Extension manifest is empty")
+
+        set_properties("extensionPackages", pkg_list, manifest)
+
+        return pkg_list
+
+    def report_provision_status(self, provision_status):
+        validate_param('provisionStatus', provision_status, ProvisionStatus)
+        data = get_properties(provision_status)
+        self._put_data(self.provision_status_uri, data)
+
+    def report_vm_status(self, vm_status):
+        validate_param('vmStatus', vm_status, VMStatus)
+        data = get_properties(vm_status)
+        # TODO code field is not implemented for metadata protocol yet.
+        # Remove it
+        handler_statuses = data['vmAgent']['extensionHandlers']
+        for handler_status in handler_statuses:
+            try:
+                handler_status.pop('code', None)
+            except KeyError:
+                pass
+
+        self._put_data(self.vm_status_uri, data)
+
+    def report_ext_status(self, ext_handler_name, ext_name, ext_status):
+        validate_param('extensionStatus', ext_status, ExtensionStatus)
+        data = get_properties(ext_status)
+        uri = self.ext_status_uri.format(ext_name)
+        self._put_data(uri, data)
+
+    def report_event(self, events):
+        validate_param('events', events, TelemetryEventList)
+        data = get_properties(events)
+        self._post_data(self.event_uri, data)
+
+    def update_certs(self):
+        certificates = self.get_certs()
+        return certificates.cert_list
+
+    def _update_certs_with_retry(self, max_retry=3):
+        # Start updating goalstate, retry on 410
+        for retry in range(0, max_retry):
+            try:
+                self.update_certs()
+                return
+            except Exception as e:
+                logger.verbose("Incarnation is out of date. Update goalstate.")
+                msg = u"Exception updating certs: {0}".format(ustr(e))
+                logger.warn(msg)
+                detailed_msg = '{0} {1}'.format(msg, traceback.extract_tb(get_traceback(e)))
+                logger.verbose(detailed_msg)
+        raise ProtocolError("Exceeded max retry updating goal state")
+
+    def download_ext_handler_pkg(self, uri, destination, headers=None, use_proxy=True):
+        success = False
+        try:
+            resp = restutil.http_get(uri, headers=headers, use_proxy=use_proxy)
+            if restutil.request_succeeded(resp):
+                fileutil.write_file(destination, bytearray(resp.read()), asbin=True)
+                success = True
+        except Exception as e:
+            logger.warn("Failed to download from: {0}".format(uri), e)
+        return success
+
+
+class Certificates(object):
+    """
+    Object containing certificates of host and provisioned user.
+    """
+
+    def __init__(self, client, json_text):
+        self.cert_list = CertList()
+        self.parse(json_text)
+
+    def parse(self, json_text):
+        """
+        Parse multiple certificates into seperate files.
+        """
+
+        data = json_text["certificateData"]
+        if data is None:
+            logger.verbose("No data in json_text received!")
+            return
+
+        cryptutil = CryptUtil(conf.get_openssl_cmd())
+        p7b_file = os.path.join(conf.get_lib_dir(), P7B_FILE_NAME)
+
+        # Wrapping the certificate lines.
+        # decode and save the result into p7b_file
+        fileutil.write_file(p7b_file, base64.b64decode(data), asbin=True) 
+
+        ssl_cmd = "openssl pkcs7 -text -in {0} -inform der | grep -v '^-----' "
+        ret, data = shellutil.run_get_output(ssl_cmd.format(p7b_file))
+
+        p7m_file = os.path.join(conf.get_lib_dir(), P7M_FILE_NAME)
+        p7m = ("MIME-Version:1.0\n"
+               "Content-Disposition: attachment; filename=\"{0}\"\n"
+               "Content-Type: application/x-pkcs7-mime; name=\"{1}\"\n"
+               "Content-Transfer-Encoding: base64\n"
+               "\n"
+               "{2}").format(p7m_file, p7m_file, data)
+
+        self.save_cache(p7m_file, p7m)
+
+        trans_prv_file = os.path.join(conf.get_lib_dir(),
+                                      TRANSPORT_PRV_FILE_NAME)
+        trans_cert_file = os.path.join(conf.get_lib_dir(),
+                                       TRANSPORT_CERT_FILE_NAME)
+        pem_file = os.path.join(conf.get_lib_dir(), PEM_FILE_NAME)
+        # decrypt certificates
+        cryptutil.decrypt_p7m(p7m_file, trans_prv_file, trans_cert_file,
+                              pem_file)
+
+        # The parsing process use public key to match prv and crt.
+        buf = []
+        begin_crt = False
+        begin_prv = False
+        prvs = {}
+        thumbprints = {}
+        index = 0
+        v1_cert_list = []
+        with open(pem_file) as pem:
+            for line in pem.readlines():
+                buf.append(line)
+                if re.match(r'[-]+BEGIN.*KEY[-]+', line):
+                    begin_prv = True
+                elif re.match(r'[-]+BEGIN.*CERTIFICATE[-]+', line):
+                    begin_crt = True
+                elif re.match(r'[-]+END.*KEY[-]+', line):
+                    tmp_file = self.write_to_tmp_file(index, 'prv', buf)
+                    pub = cryptutil.get_pubkey_from_prv(tmp_file)
+                    prvs[pub] = tmp_file
+                    buf = []
+                    index += 1
+                    begin_prv = False
+                elif re.match(r'[-]+END.*CERTIFICATE[-]+', line):
+                    tmp_file = self.write_to_tmp_file(index, 'crt', buf)
+                    pub = cryptutil.get_pubkey_from_crt(tmp_file)
+                    thumbprint = cryptutil.get_thumbprint_from_crt(tmp_file)
+                    thumbprints[pub] = thumbprint
+                    # Rename crt with thumbprint as the file name
+                    crt = "{0}.crt".format(thumbprint)
+                    v1_cert_list.append({
+                        "name": None,
+                        "thumbprint": thumbprint
+                    })
+                    os.rename(tmp_file, os.path.join(conf.get_lib_dir(), crt))
+                    buf = []
+                    index += 1
+                    begin_crt = False
+
+        # Rename prv key with thumbprint as the file name
+        for pubkey in prvs:
+            thumbprint = thumbprints[pubkey]
+            if thumbprint:
+                tmp_file = prvs[pubkey]
+                prv = "{0}.prv".format(thumbprint)
+                os.rename(tmp_file, os.path.join(conf.get_lib_dir(), prv))
+
+        for v1_cert in v1_cert_list:
+            cert = Cert()
+            set_properties("certs", cert, v1_cert)
+            self.cert_list.certificates.append(cert)
+
+    def save_cache(self, local_file, data):
+        try:
+            fileutil.write_file(local_file, data)
+        except IOError as e:
+            raise ProtocolError("Failed to write cache: {0}".format(e))
+
+    def write_to_tmp_file(self, index, suffix, buf):
+        file_name = os.path.join(conf.get_lib_dir(),
+                                 "{0}.{1}".format(index, suffix))
+        self.save_cache(file_name, "".join(buf))
+        return file_name
diff --git a/azurelinuxagent/common/protocol/util.py b/azurelinuxagent/common/protocol/util.py
index 9cfd788c..803caa5f 100644
--- a/azurelinuxagent/common/protocol/util.py
+++ b/azurelinuxagent/common/protocol/util.py
@@ -20,6 +20,8 @@
 import errno
 import os
 import re
+import shutil
+import threading
 import time
 
 import azurelinuxagent.common.conf as conf
@@ -34,16 +36,28 @@ from azurelinuxagent.common.osutil import get_osutil
 from azurelinuxagent.common.dhcp import get_dhcp_handler
 from azurelinuxagent.common.protocol.ovfenv import OvfEnv
 from azurelinuxagent.common.protocol.wire import WireProtocol
-from azurelinuxagent.common.utils.restutil import KNOWN_WIRESERVER_IP, IOErrorCounter
+from azurelinuxagent.common.protocol.metadata import MetadataProtocol
+from azurelinuxagent.common.utils.restutil import KNOWN_WIRESERVER_IP, \
+                                                  IOErrorCounter
 
 OVF_FILE_NAME = "ovf-env.xml"
+TAG_FILE_NAME = "useMetadataEndpoint.tag"
 PROTOCOL_FILE_NAME = "Protocol"
 MAX_RETRY = 360
 PROBE_INTERVAL = 10
 ENDPOINT_FILE_NAME = "WireServerEndpoint"
 PASSWORD_PATTERN = "<UserPassword>.*?<"
 PASSWORD_REPLACEMENT = "<UserPassword>*<"
-WIRE_PROTOCOL_NAME = "WireProtocol"
+
+
+class _nameset(set):
+    def __getattr__(self, name):
+        if name in self:
+            return name
+        raise AttributeError("%s not a valid value" % name)
+
+
+prots = _nameset(("WireProtocol", "MetadataProtocol"))
 
 
 def get_protocol_util():
@@ -72,7 +86,9 @@ class ProtocolUtil(SingletonPerThread):
         """
         dvd_mount_point = conf.get_dvd_mount_point()
         ovf_file_path_on_dvd = os.path.join(dvd_mount_point, OVF_FILE_NAME)
+        tag_file_path_on_dvd = os.path.join(dvd_mount_point, TAG_FILE_NAME)
         ovf_file_path = os.path.join(conf.get_lib_dir(), OVF_FILE_NAME)
+        tag_file_path = self._get_tag_file_path()
 
         try:
             self.osutil.mount_dvd()
@@ -98,6 +114,15 @@ class ProtocolUtil(SingletonPerThread):
                                 "{0}: {1}".format(ovf_file_path,
                                                   ustr(e)))
 
+        try:
+            if os.path.isfile(tag_file_path_on_dvd):
+                logger.info("Found {0} in provisioning ISO", TAG_FILE_NAME)
+                shutil.copyfile(tag_file_path_on_dvd, tag_file_path)
+        except (IOError, OSError) as e:
+            raise ProtocolError("[CopyOvfEnv] Error copying file "
+                                "{0} to {1}: {2}".format(tag_file_path,
+                                                         tag_file_path,
+                                                         ustr(e)))
         self._cleanup_ovf_dvd()
 
         return ovfenv
@@ -122,13 +147,20 @@ class ProtocolUtil(SingletonPerThread):
                 "ovf-env.xml is missing from {0}".format(ovf_file_path))
 
     def _get_protocol_file_path(self):
-        return os.path.join(conf.get_lib_dir(), PROTOCOL_FILE_NAME)
+        return os.path.join(
+            conf.get_lib_dir(),
+            PROTOCOL_FILE_NAME)
 
     def _get_wireserver_endpoint_file_path(self):
         return os.path.join(
             conf.get_lib_dir(),
             ENDPOINT_FILE_NAME)
 
+    def _get_tag_file_path(self):
+        return os.path.join(
+            conf.get_lib_dir(),
+            TAG_FILE_NAME)
+
     def get_wireserver_endpoint(self):
 
         if self.endpoint:
@@ -180,46 +212,61 @@ class ProtocolUtil(SingletonPerThread):
                 return
             logger.error("Failed to clear wiresever endpoint: {0}", e)
 
-    def _detect_protocol(self):
+    def _detect_wire_protocol(self):
+        endpoint = self.dhcp_handler.endpoint
+        if endpoint is None:
+            '''
+            Check if DHCP can be used to get the wire protocol endpoint
+            '''
+            dhcp_available = self.osutil.is_dhcp_available()
+            if dhcp_available:
+                logger.info("WireServer endpoint is not found. Rerun dhcp handler")
+                try:
+                    self.dhcp_handler.run()
+                except DhcpError as e:
+                    raise ProtocolError(ustr(e))
+                endpoint = self.dhcp_handler.endpoint
+            else:
+                logger.info("_detect_wire_protocol: DHCP not available")
+                endpoint = self.get_wireserver_endpoint()
+
+        try:
+            protocol = WireProtocol(endpoint)
+            protocol.detect()
+            self._set_wireserver_endpoint(endpoint)
+            return protocol
+        except ProtocolError as e:
+            logger.info("WireServer is not responding. Reset dhcp endpoint")
+            self.dhcp_handler.endpoint = None
+            self.dhcp_handler.skip_cache = True
+            raise e
+
+    def _detect_metadata_protocol(self):
+        protocol = MetadataProtocol()
+        protocol.detect()
+        return protocol
+
+    def _detect_protocol(self, protocols):
         """
         Probe protocol endpoints in turn.
         """
         self.clear_protocol()
 
         for retry in range(0, MAX_RETRY):
-            try:
-                endpoint = self.dhcp_handler.endpoint
-                if endpoint is None:
-                    '''
-                    Check if DHCP can be used to get the wire protocol endpoint
-                    '''
-                    dhcp_available = self.osutil.is_dhcp_available()
-                    if dhcp_available:
-                        logger.info("WireServer endpoint is not found. Rerun dhcp handler")
-                        try:
-                            self.dhcp_handler.run()
-                        except DhcpError as e:
-                            raise ProtocolError(ustr(e))
-                        endpoint = self.dhcp_handler.endpoint
-                    else:
-                        logger.info("_detect_protocol: DHCP not available")
-                        endpoint = self.get_wireserver_endpoint()
-
+            for protocol_name in protocols:
                 try:
-                    protocol = WireProtocol(endpoint)
-                    protocol.detect()
-                    self._set_wireserver_endpoint(endpoint)
-                    return protocol
+                    protocol = self._detect_wire_protocol() \
+                                if protocol_name == prots.WireProtocol \
+                                else self._detect_metadata_protocol()
+
+                    return (protocol_name, protocol)
+
                 except ProtocolError as e:
-                    logger.info("WireServer is not responding. Reset dhcp endpoint")
-                    self.dhcp_handler.endpoint = None
-                    self.dhcp_handler.skip_cache = True
-                    raise e
-            except ProtocolError as e:
-                logger.info("Protocol endpoint not found: {0}", e)
+                    logger.info("Protocol endpoint not found: {0}, {1}",
+                                protocol_name, e)
 
             if retry < MAX_RETRY - 1:
-                logger.info("Retry detect protocol: retry={0}", retry)
+                logger.info("Retry detect protocols: retry={0}", retry)
                 time.sleep(PROBE_INTERVAL)
         raise ProtocolNotFoundError("No protocol found.")
 
@@ -231,8 +278,15 @@ class ProtocolUtil(SingletonPerThread):
         if not os.path.isfile(protocol_file_path):
             raise ProtocolNotFoundError("No protocol found")
 
-        endpoint = self.get_wireserver_endpoint()
-        return WireProtocol(endpoint)
+        protocol_name = fileutil.read_file(protocol_file_path)
+        if protocol_name == prots.WireProtocol:
+            endpoint = self.get_wireserver_endpoint()
+            return WireProtocol(endpoint)
+        elif protocol_name == prots.MetadataProtocol:
+            return MetadataProtocol()
+        else:
+            raise ProtocolNotFoundError(("Unknown protocol: {0}"
+                                         "").format(protocol_name))
 
     def _save_protocol(self, protocol_name):
         """
@@ -263,9 +317,10 @@ class ProtocolUtil(SingletonPerThread):
                 return
             logger.error("Failed to clear protocol endpoint: {0}", e)
 
-    def get_protocol(self):
+    def get_protocol(self, by_file=False):
         """
-        Detect protocol by endpoint.
+        Detect protocol by endpoints, if by_file is True,
+        detect MetadataProtocol in priority.
         :returns: protocol instance
         """
 
@@ -277,13 +332,19 @@ class ProtocolUtil(SingletonPerThread):
             return self._protocol
         except ProtocolNotFoundError:
             pass
+        logger.info("Detect protocol endpoints")
+        protocols = [prots.WireProtocol]
 
-        logger.info("Detect protocol endpoint")
-
-        protocol = self._detect_protocol()
+        if by_file:
+            tag_file_path = self._get_tag_file_path()
+            if os.path.isfile(tag_file_path):
+                protocols.insert(0, prots.MetadataProtocol)
+        else:
+            protocols.append(prots.MetadataProtocol)
+        protocol_name, protocol = self._detect_protocol(protocols)
 
         IOErrorCounter.set_protocol_endpoint(endpoint=protocol.get_endpoint())
-        self._save_protocol(WIRE_PROTOCOL_NAME)
+        self._save_protocol(protocol_name)
 
         self._protocol = protocol
         return self._protocol
diff --git a/azurelinuxagent/common/protocol/wire.py b/azurelinuxagent/common/protocol/wire.py
index bd202e62..7ad96f02 100644
--- a/azurelinuxagent/common/protocol/wire.py
+++ b/azurelinuxagent/common/protocol/wire.py
@@ -89,6 +89,7 @@ class WireProtocol(Protocol):
         cryptutil.gen_transport_cert(trans_prv_file, trans_cert_file)
 
         # Set the initial goal state
+        logger.info('Initializing goal state during protocol detection')
         self.client.update_goal_state(forced=True)
 
     def update_goal_state(self):
diff --git a/azurelinuxagent/daemon/main.py b/azurelinuxagent/daemon/main.py
index db4dd185..264c3268 100644
--- a/azurelinuxagent/daemon/main.py
+++ b/azurelinuxagent/daemon/main.py
@@ -31,6 +31,7 @@ from azurelinuxagent.common.event import add_event, WALAEventOperation, initiali
 from azurelinuxagent.common.future import ustr
 from azurelinuxagent.common.osutil import get_osutil
 from azurelinuxagent.common.protocol.util import get_protocol_util
+from azurelinuxagent.common.protocol.wire import WireProtocol
 from azurelinuxagent.common.rdma import setup_rdma_device
 from azurelinuxagent.common.version import AGENT_NAME, AGENT_LONG_NAME, \
     AGENT_VERSION, \
@@ -162,7 +163,11 @@ class DaemonHandler(object):
                 # - Changes to RDMA state may not increment the goal state
                 #   incarnation number. A forced update ensures the most
                 #   current values.
-                self.protocol_util.get_protocol().client.update_goal_state(forced=True)
+                protocol = self.protocol_util.get_protocol()
+                if type(protocol) is not WireProtocol:
+                    raise Exception("Attempt to setup RDMA without Wireserver")
+
+                protocol.client.update_goal_state(forced=True)
 
                 setup_rdma_device(nd_version, protocol.client.get_shared_conf())
             except Exception as e:
diff --git a/azurelinuxagent/ga/update.py b/azurelinuxagent/ga/update.py
index 38992057..f9fb4a8c 100644
--- a/azurelinuxagent/ga/update.py
+++ b/azurelinuxagent/ga/update.py
@@ -48,6 +48,7 @@ from azurelinuxagent.common.future import ustr
 from azurelinuxagent.common.osutil import get_osutil
 from azurelinuxagent.common.protocol.util import get_protocol_util
 from azurelinuxagent.common.protocol.hostplugin import HostPluginProtocol
+from azurelinuxagent.common.protocol.wire import WireProtocol
 from azurelinuxagent.common.utils.flexible_version import FlexibleVersion
 from azurelinuxagent.common.version import AGENT_NAME, AGENT_VERSION, AGENT_DIR_PATTERN, \
                                             CURRENT_AGENT, CURRENT_VERSION, DISTRO_NAME, DISTRO_VERSION, \
@@ -526,8 +527,9 @@ class UpdateHandler(object):
             logger.warn(u"Exception occurred loading available agents: {0}", ustr(e))
         return
 
-    def _get_host_plugin(self, protocol):
-        return protocol.client.get_host_plugin() if protocol and protocol.client else None
+    def _get_host_plugin(self, protocol=None):
+        return protocol.client.get_host_plugin() \
+            if protocol and type(protocol) is WireProtocol and protocol.client else None
 
     def _get_pid_parts(self):
         pid_file = conf.get_agent_pid_file_path()
diff --git a/azurelinuxagent/pa/provision/default.py b/azurelinuxagent/pa/provision/default.py
index 7b066dcc..85b19223 100644
--- a/azurelinuxagent/pa/provision/default.py
+++ b/azurelinuxagent/pa/provision/default.py
@@ -77,7 +77,7 @@ class ProvisionHandler(object):
             logger.info("Copying ovf-env.xml")
             ovf_env = self.protocol_util.copy_ovf_env()
 
-            self.protocol_util.get_protocol()
+            self.protocol_util.get_protocol(by_file=True)
             self.report_not_ready("Provisioning", "Starting")
             logger.info("Starting provisioning")
 


Test patch:
diff --git a/tests/ga/test_update.py b/tests/ga/test_update.py
index cdacc36f..d1e9fde6 100644
--- a/tests/ga/test_update.py
+++ b/tests/ga/test_update.py
@@ -11,6 +11,7 @@ from azurelinuxagent.ga.env import EnvHandler
 
 from azurelinuxagent.common.protocol.goal_state import ExtensionsConfig
 from azurelinuxagent.common.protocol.hostplugin import *
+from azurelinuxagent.common.protocol.metadata import *
 from azurelinuxagent.common.protocol.util import ProtocolUtil
 from azurelinuxagent.common.protocol.wire import *
 from azurelinuxagent.common.version import AGENT_PKG_GLOB, AGENT_DIR_GLOB
@@ -895,6 +896,13 @@ class TestUpdate(UpdateTestCase):
         self.assertEqual(1, mock_get_host.call_count)
         self.assertEqual("faux host", host)
 
+    @patch('azurelinuxagent.common.protocol.wire.WireClient.get_host_plugin')
+    def test_get_host_plugin_returns_none_otherwise(self, mock_get_host):
+        protocol = MetadataProtocol()
+        host = self.update_handler._get_host_plugin(protocol=protocol)
+        mock_get_host.assert_not_called()
+        self.assertEqual(None, host)
+
     def test_get_latest_agent(self):
         latest_version = self.prepare_agents()
 
@@ -1464,7 +1472,6 @@ class TestUpdate(UpdateTestCase):
             "The heartbeat was not written to the agent's log"
         )
 
-
 class MonitorThreadTest(AgentTestCase):
     def setUp(self):
         AgentTestCase.setUp(self)
diff --git a/tests/protocol/mockmetadata.py b/tests/protocol/mockmetadata.py
new file mode 100644
index 00000000..6f6de003
--- /dev/null
+++ b/tests/protocol/mockmetadata.py
@@ -0,0 +1,66 @@
+# Copyright 2018 Microsoft Corporation
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# Requires Python 2.6+ and Openssl 1.0+
+#
+
+from tests.tools import load_data, MagicMock, Mock
+from azurelinuxagent.common.future import httpclient
+
+DATA_FILE = {
+        "identity": "metadata/identity.json",
+        "certificates": "metadata/certificates.json",
+        "certificates_data": "metadata/certificates_data.json",
+        "ext_handlers": "metadata/ext_handlers.json",
+        "ext_handler_pkgs": "metadata/ext_handler_pkgs.json",
+        "trans_prv": "metadata/trans_prv",
+        "trans_cert": "metadata/trans_cert",        
+}
+
+DATA_FILE_NO_EXT = DATA_FILE.copy()
+DATA_FILE_NO_EXT["ext_handlers"] = "metadata/ext_handlers_no_ext.json"
+
+
+class MetadataProtocolData(object):
+    def __init__(self, data_files):
+        self.identity = load_data(data_files.get("identity"))
+        self.certificates = load_data(data_files.get("certificates"))
+        self.certificates_data = load_data(data_files.get("certificates_data"))
+        self.ext_handlers = load_data(data_files.get("ext_handlers"))
+        self.ext_handler_pkgs = load_data(data_files.get("ext_handler_pkgs"))
+        self.trans_prv = load_data(data_files.get("trans_prv"))
+        self.trans_cert = load_data(data_files.get("trans_cert"))
+        
+    def mock_http_get(self, url, *args, **kwargs):
+        content = None
+        if url.count(u"identity?") > 0:
+            content = self.identity
+        elif url.count(u"certificates") > 0:
+            content = self.certificates
+        elif url.count(u"certificates_data") > 0:
+            content = self.certificates_data
+        elif url.count(u"extensionHandlers") > 0:
+            content = self.ext_handlers
+        elif url.count(u"versionUri") > 0:
+            content = self.ext_handler_pkgs
+        else:
+            raise Exception("Bad url {0}".format(url))
+        resp = MagicMock()
+        resp.status = httpclient.OK
+        if content is None:
+            resp.read = Mock(return_value=None)
+        else:
+            resp.read = Mock(return_value=content.encode("utf-8"))
+        return resp
+
diff --git a/tests/protocol/test_metadata.py b/tests/protocol/test_metadata.py
new file mode 100644
index 00000000..272c6b04
--- /dev/null
+++ b/tests/protocol/test_metadata.py
@@ -0,0 +1,154 @@
+# Copyright 2018 Microsoft Corporation
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# Requires Python 2.6+ and Openssl 1.0+
+#
+
+from azurelinuxagent.common.protocol.metadata import *
+from azurelinuxagent.common.protocol.restapi import *
+from azurelinuxagent.common.telemetryevent import TelemetryEventList, TelemetryEvent
+from azurelinuxagent.common.utils import restutil
+
+from tests.protocol.mockmetadata import *
+from tests.tools import AgentTestCase, patch
+
+
+class TestMetadataProtocolGetters(AgentTestCase):
+    def load_json(self, path):
+        return json.loads(ustr(load_data(path)), encoding="utf-8")
+
+    @patch("time.sleep")
+    def _test_getters(self, test_data ,_):
+        with patch.object(restutil, 'http_get', test_data.mock_http_get):
+            protocol = MetadataProtocol()
+            protocol.detect()
+            protocol.get_vminfo()
+            protocol.get_certs()
+            ext_handlers, etag = protocol.get_ext_handlers()
+            for ext_handler in ext_handlers.extHandlers:
+                protocol.get_ext_handler_pkgs(ext_handler)
+
+    def test_getters(self, *args):
+        test_data = MetadataProtocolData(DATA_FILE)
+        self._test_getters(test_data, *args)
+
+    def test_getters_no(self, *args):
+        test_data = MetadataProtocolData(DATA_FILE_NO_EXT)
+        self._test_getters(test_data, *args)
+
+    @patch("azurelinuxagent.common.protocol.metadata.MetadataProtocol._update_certs_with_retry")
+    @patch("azurelinuxagent.common.protocol.metadata.MetadataProtocol._get_data")
+    def test_get_vmagents_manifests(self, mock_get, mock_update):
+        data = self.load_json("metadata/vmagent_manifests.json")
+        mock_get.return_value = data, 42
+
+        protocol = MetadataProtocol()
+        manifests, etag = protocol.get_vmagent_manifests()
+
+        self.assertEqual(mock_update.call_count, 1)
+        self.assertEqual(mock_get.call_count, 1)
+
+        manifests_uri = BASE_URI.format(
+            METADATA_ENDPOINT,
+            "vmAgentVersions",
+            APIVERSION)
+        self.assertEqual(mock_get.call_args[0][0], manifests_uri)
+
+        self.assertEqual(etag, 42)
+        self.assertNotEqual(None, manifests)
+        self.assertEqual(len(manifests.vmAgentManifests), 1)
+
+        manifest = manifests.vmAgentManifests[0]
+        self.assertEqual(manifest.family, conf.get_autoupdate_gafamily())
+        self.assertEqual(len(manifest.versionsManifestUris), 2)
+
+        # Same etag returns the same data
+        data = self.load_json("metadata/vmagent_manifests_invalid1.json")
+        mock_get.return_value = data, 42
+        next_manifests, etag = protocol.get_vmagent_manifests()
+
+        self.assertEqual(etag, 42)
+        self.assertEqual(manifests, next_manifests)
+
+        # New etag returns new data
+        mock_get.return_value = data, 43
+        self.assertRaises(ProtocolError, protocol.get_vmagent_manifests)
+
+    @patch("azurelinuxagent.common.protocol.metadata.MetadataProtocol._update_certs_with_retry")
+    @patch("azurelinuxagent.common.protocol.metadata.MetadataProtocol._get_data")
+    def test_get_vmagents_manifests_raises(self, mock_get, mock_update):
+        data = self.load_json("metadata/vmagent_manifests_invalid1.json")
+        mock_get.return_value = data, 42
+
+        protocol = MetadataProtocol()
+        self.assertRaises(ProtocolError, protocol.get_vmagent_manifests)
+
+        data = self.load_json("metadata/vmagent_manifests_invalid2.json")
+        mock_get.return_value = data, 43
+        self.assertRaises(ProtocolError, protocol.get_vmagent_manifests)
+
+    @patch("azurelinuxagent.common.protocol.metadata.MetadataProtocol._update_certs_with_retry")
+    @patch("azurelinuxagent.common.protocol.metadata.MetadataProtocol._get_data")
+    def test_get_vmagent_pkgs(self, mock_get, mock_update):
+        data = self.load_json("metadata/vmagent_manifests.json")
+        mock_get.return_value = data, 42
+
+        protocol = MetadataProtocol()
+        manifests, etag = protocol.get_vmagent_manifests()
+        manifest = manifests.vmAgentManifests[0]
+
+        data = self.load_json("metadata/vmagent_manifest1.json")
+        mock_get.return_value = data, 42
+        pkgs = protocol.get_vmagent_pkgs(manifest)
+
+        self.assertNotEqual(None, pkgs)
+        self.assertEqual(len(pkgs.versions), 2)
+
+        for pkg in pkgs.versions:
+            self.assertNotEqual(None, pkg.version)
+            self.assertTrue(len(pkg.uris) > 0)
+
+            for uri in pkg.uris:
+                self.assertTrue(uri.uri.endswith("zip"))
+
+    @patch("azurelinuxagent.common.protocol.metadata.MetadataProtocol._post_data")
+    def test_report_event(self, mock_post):
+        events = TelemetryEventList()
+
+        data = self.load_json("events/1478123456789000.tld")
+        event = TelemetryEvent()
+        set_properties("event", event, data)
+        events.events.append(event)
+
+        data = self.load_json("events/1478123456789001.tld")
+        event = TelemetryEvent()
+        set_properties("event", event, data)
+        events.events.append(event)
+
+        data = self.load_json("events/1479766858966718.tld")
+        event = TelemetryEvent()
+        set_properties("event", event, data)
+        events.events.append(event)
+
+        protocol = MetadataProtocol()
+        protocol.report_event(events)
+
+        events_uri = BASE_URI.format(
+            METADATA_ENDPOINT,
+            "status/telemetry",
+            APIVERSION)
+
+        self.assertEqual(mock_post.call_count, 1)
+        self.assertEqual(mock_post.call_args[0][0], events_uri)
+        self.assertEqual(mock_post.call_args[0][1], get_properties(events))
diff --git a/tests/protocol/test_protocol_util.py b/tests/protocol/test_protocol_util.py
index 1107c280..108f6c59 100644
--- a/tests/protocol/test_protocol_util.py
+++ b/tests/protocol/test_protocol_util.py
@@ -16,12 +16,13 @@
 #
 
 import unittest
+import os
 from multiprocessing import Queue
 from threading import Thread
 
 from tests.tools import AgentTestCase, MagicMock, Mock, patch, clear_singleton_instances
 from azurelinuxagent.common.exception import *
-from azurelinuxagent.common.protocol.util import get_protocol_util
+from azurelinuxagent.common.protocol.util import get_protocol_util, TAG_FILE_NAME
 from azurelinuxagent.common.utils.restutil import KNOWN_WIRESERVER_IP
 from azurelinuxagent.common.protocol.util import ProtocolUtil
 from errno import ENOENT
@@ -67,9 +68,11 @@ class TestProtocolUtil(AgentTestCase):
         self.assertEqual(2, queue.qsize())  # Assert that there are 2 objects in the queue
         self.assertNotEqual(queue.get(), queue.get())
     
+    @patch("azurelinuxagent.common.protocol.util.MetadataProtocol")
     @patch("azurelinuxagent.common.protocol.util.WireProtocol")
-    def test_detect_protocol(self, WireProtocol, _):
+    def test_detect_protocol(self, WireProtocol, MetadataProtocol, _):
         WireProtocol.return_value = MagicMock()
+        MetadataProtocol.return_value = MagicMock()
 
         protocol_util = get_protocol_util()
         
@@ -84,11 +87,42 @@ class TestProtocolUtil(AgentTestCase):
         protocol_util.clear_protocol()
         WireProtocol.return_value.detect.side_effect = ProtocolError()
 
+        protocol = protocol_util.get_protocol()
+        self.assertEquals(MetadataProtocol.return_value, protocol)
+
+        # Test no protocol is available
+        protocol_util.clear_protocol()
+        WireProtocol.return_value.detect.side_effect = ProtocolError()
+        MetadataProtocol.return_value.detect.side_effect = ProtocolError()
+
         self.assertRaises(ProtocolError, protocol_util.get_protocol)
 
+    def test_detect_protocol_by_file(self, _):
+        protocol_util = get_protocol_util()
+        protocol_util._detect_wire_protocol = Mock()
+        protocol_util._detect_metadata_protocol = Mock()
+
+        tag_file = os.path.join(self.tmp_dir, TAG_FILE_NAME)
+
+        # Test tag file doesn't exist
+        protocol_util.get_protocol(by_file=True)
+        protocol_util._detect_wire_protocol.assert_any_call()
+        protocol_util._detect_metadata_protocol.assert_not_called()
+
+        # Test tag file exists
+        protocol_util.clear_protocol()
+        protocol_util._detect_wire_protocol.reset_mock()
+        protocol_util._detect_metadata_protocol.reset_mock()
+        with open(tag_file, "w+") as tag_fd:
+            tag_fd.write("")
+
+        protocol_util.get_protocol(by_file=True)
+        protocol_util._detect_metadata_protocol.assert_any_call()
+        protocol_util._detect_wire_protocol.assert_not_called()
+
     @patch("azurelinuxagent.common.conf.get_lib_dir")
     @patch("azurelinuxagent.common.protocol.util.WireProtocol")
-    def test_detect_protocol_no_dhcp(self, WireProtocol, mock_get_lib_dir, _):
+    def test_detect_wire_protocol_no_dhcp(self, WireProtocol, mock_get_lib_dir, _):
         WireProtocol.return_value.detect = Mock()
         mock_get_lib_dir.return_value = self.tmp_dir
 
@@ -104,28 +138,65 @@ class TestProtocolUtil(AgentTestCase):
         endpoint_file = protocol_util._get_wireserver_endpoint_file_path()
 
         # Test wire protocol when no endpoint file has been written
-        protocol_util._detect_protocol()
+        protocol_util._detect_wire_protocol()
         self.assertEqual(KNOWN_WIRESERVER_IP, protocol_util.get_wireserver_endpoint())
 
+        # Test wire protocol when endpoint was previously detected
+        protocol_util.clear_protocol()
+        with open(endpoint_file, "w+") as endpoint_fd:
+            endpoint_fd.write("baz.qux")
+
+        protocol_util._detect_wire_protocol()
+        self.assertEqual("baz.qux", protocol_util.get_wireserver_endpoint())
+
         # Test wire protocol on dhcp failure
+        protocol_util.clear_protocol()
         protocol_util.osutil.is_dhcp_available.return_value = True
         protocol_util.dhcp_handler.run.side_effect = DhcpError()
 
-        self.assertRaises(ProtocolError, protocol_util._detect_protocol)
+        self.assertRaises(ProtocolError, protocol_util._detect_wire_protocol)
 
+    @patch("azurelinuxagent.common.protocol.util.MetadataProtocol")
     @patch("azurelinuxagent.common.protocol.util.WireProtocol")
-    def test_get_protocol(self, WireProtocol, _):
+    def test_get_protocol(self, WireProtocol, MetadataProtocol, _):
         WireProtocol.return_value = MagicMock()
+        MetadataProtocol.return_value = MagicMock()
+
         protocol_util = get_protocol_util()
         protocol_util.get_wireserver_endpoint = Mock()
         protocol_util._detect_protocol = MagicMock()
+
+        # Test for wire protocol
         protocol_util._save_protocol("WireProtocol")
 
         protocol = protocol_util.get_protocol()
-
         self.assertEquals(WireProtocol.return_value, protocol)
         protocol_util.get_wireserver_endpoint.assert_any_call()
 
+        # Test to ensure protocol persists
+        protocol_util.get_wireserver_endpoint.reset_mock()
+        protocol_util._save_protocol("MetadataProtocol")
+
+        protocol = protocol_util.get_protocol()
+        self.assertEquals(WireProtocol.return_value, protocol)
+        protocol_util.get_wireserver_endpoint.assert_not_called()
+
+        # Test for metadata protocol
+        protocol_util.clear_protocol()
+        protocol_util._save_protocol("MetadataProtocol")
+
+        protocol = protocol_util.get_protocol()
+        self.assertEquals(MetadataProtocol.return_value, protocol)
+        protocol_util.get_wireserver_endpoint.assert_not_called()
+
+        # Test for unknown protocol
+        protocol_util.clear_protocol()
+        protocol_util._save_protocol("Not_a_Protocol")
+        protocol_util._detect_protocol.side_effect = NotImplementedError()
+
+        self.assertRaises(NotImplementedError, protocol_util.get_protocol)
+        protocol_util.get_wireserver_endpoint.assert_not_called()
+
     @patch("azurelinuxagent.common.utils.fileutil")
     @patch("azurelinuxagent.common.conf.get_lib_dir")
     def test_endpoint_file_states(self, mock_get_lib_dir, mock_fileutil, _):