Import patches-unapplied version 2.2.16-0ubuntu1 to ubuntu/artful-proposed

Imported using git-ubuntu import.

Changelog parent: 43bdf9debe5377216aed0086bff2aad864f6ba82

New changelog entries:
  * New upstream release (LP: #1714299).

1 files changed, 204 insertions, 139 deletions

diff --git a/azurelinuxagent/common/protocol/wire.py b/azurelinuxagent/common/protocol/wire.py
index d731e11..4f3b7e0 100644
--- a/azurelinuxagent/common/protocol/wire.py
+++ b/azurelinuxagent/common/protocol/wire.py
@@ -26,7 +26,8 @@ import azurelinuxagent.common.conf as conf
 import azurelinuxagent.common.utils.fileutil as fileutil
 import azurelinuxagent.common.utils.textutil as textutil
 
-from azurelinuxagent.common.exception import ProtocolNotFoundError
+from azurelinuxagent.common.exception import ProtocolNotFoundError, \
+                                            ResourceGoneError
 from azurelinuxagent.common.future import httpclient, bytebuffer
 from azurelinuxagent.common.protocol.hostplugin import HostPluginProtocol
 from azurelinuxagent.common.protocol.restapi import *
@@ -96,7 +97,10 @@ class WireProtocol(Protocol):
         cryptutil = CryptUtil(conf.get_openssl_cmd())
         cryptutil.gen_transport_cert(trans_prv_file, trans_cert_file)
 
-        self.client.update_goal_state(forced=True)
+        self.update_goal_state(forced=True)
+
+    def update_goal_state(self, forced=False, max_retry=3):
+        self.client.update_goal_state(forced=forced, max_retry=max_retry)
 
     def get_vminfo(self):
         goal_state = self.client.get_goal_state()
@@ -117,7 +121,7 @@ class WireProtocol(Protocol):
 
     def get_vmagent_manifests(self):
         # Update goal state to get latest extensions config
-        self.client.update_goal_state()
+        self.update_goal_state()
         goal_state = self.client.get_goal_state()
         ext_conf = self.client.get_ext_conf()
         return ext_conf.vmagent_manifests, goal_state.incarnation
@@ -130,7 +134,7 @@ class WireProtocol(Protocol):
     def get_ext_handlers(self):
         logger.verbose("Get extension handler config")
         # Update goal state to get latest extensions config
-        self.client.update_goal_state()
+        self.update_goal_state()
         goal_state = self.client.get_goal_state()
         ext_conf = self.client.get_ext_conf()
         # In wire protocol, incarnation is equivalent to ETag
@@ -533,29 +537,27 @@ class WireClient(object):
             self.req_count = 0
 
     def call_wireserver(self, http_req, *args, **kwargs):
-        """
-        Call wire server; handle throttling (403), resource gone (410) and
-        service unavailable (503).
-        """
         self.prevent_throttling()
-        for retry in range(0, 3):
+
+        try:
+            # Never use the HTTP proxy for wireserver
+            kwargs['use_proxy'] = False
             resp = http_req(*args, **kwargs)
-            if resp.status == httpclient.FORBIDDEN:
-                logger.warn("Sending too many requests to wire server. ")
-                logger.info("Sleeping {0}s to avoid throttling.",
-                            LONG_WAITING_INTERVAL)
-                time.sleep(LONG_WAITING_INTERVAL)
-            elif resp.status == httpclient.SERVICE_UNAVAILABLE:
-                logger.warn("Service temporarily unavailable, sleeping {0}s "
-                            "before retrying.", LONG_WAITING_INTERVAL)
-                time.sleep(LONG_WAITING_INTERVAL)
-            elif resp.status == httpclient.GONE:
-                msg = args[0] if len(args) > 0 else ""
-                raise WireProtocolResourceGone(msg)
-            else:
-                return resp
-        raise ProtocolError(("Calling wire server failed: "
-                             "{0}").format(resp.status))
+        except Exception as e:
+            raise ProtocolError("[Wireserver Exception] {0}".format(
+                ustr(e)))
+
+        if resp is not None and resp.status == httpclient.GONE:
+            msg = args[0] if len(args) > 0 else ""
+            raise WireProtocolResourceGone(msg)
+
+        elif restutil.request_failed(resp):
+            msg = "[Wireserver Failed] URI {0} ".format(args[0])
+            if resp is not None:
+                msg += " [HTTP Failed] Status Code {0}".format(resp.status)
+            raise ProtocolError(msg)
+
+        return resp
 
     def decode_config(self, data):
         if data is None:
@@ -565,16 +567,9 @@ class WireClient(object):
         return xml_text
 
     def fetch_config(self, uri, headers):
-        try:
-            resp = self.call_wireserver(restutil.http_get,
-                                        uri,
-                                        headers=headers)
-        except HttpError as e:
-            raise ProtocolError(ustr(e))
-
-        if resp.status != httpclient.OK:
-            raise ProtocolError("{0} - {1}".format(resp.status, uri))
-
+        resp = self.call_wireserver(restutil.http_get,
+                                    uri,
+                                    headers=headers)
         return self.decode_config(resp.read())
 
     def fetch_cache(self, local_file):
@@ -589,29 +584,17 @@ class WireClient(object):
         try:
             fileutil.write_file(local_file, data)
         except IOError as e:
+            fileutil.clean_ioerror(e,
+                paths=[local_file])
             raise ProtocolError("Failed to write cache: {0}".format(e))
 
     @staticmethod
     def call_storage_service(http_req, *args, **kwargs):
-        """ 
-        Call storage service, handle SERVICE_UNAVAILABLE(503)
-        """
-
         # Default to use the configured HTTP proxy
-        if not 'chk_proxy' in kwargs or kwargs['chk_proxy'] is None:
-            kwargs['chk_proxy'] = True
+        if not 'use_proxy' in kwargs or kwargs['use_proxy'] is None:
+            kwargs['use_proxy'] = True
 
-        for retry in range(0, 3):
-            resp = http_req(*args, **kwargs)
-            if resp.status == httpclient.SERVICE_UNAVAILABLE:
-                logger.warn("Storage service is temporarily unavailable. ")
-                logger.info("Will retry in {0} seconds. ",
-                            LONG_WAITING_INTERVAL)
-                time.sleep(LONG_WAITING_INTERVAL)
-            else:
-                return resp
-        raise ProtocolError(("Calling storage endpoint failed: "
-                             "{0}").format(resp.status))
+        return http_req(*args, **kwargs)
 
     def fetch_manifest(self, version_uris):
         logger.verbose("Fetch manifest")
@@ -619,47 +602,61 @@ class WireClient(object):
             response = None
             if not HostPluginProtocol.is_default_channel():
                 response = self.fetch(version.uri)
+
             if not response:
                 if HostPluginProtocol.is_default_channel():
                     logger.verbose("Using host plugin as default channel")
                 else:
-                    logger.verbose("Manifest could not be downloaded, falling back to host plugin")
-                host = self.get_host_plugin()
-                uri, headers = host.get_artifact_request(version.uri)
-                response = self.fetch(uri, headers, chk_proxy=False)
-                if not response:
-                    host = self.get_host_plugin(force_update=True)
-                    logger.info("Retry fetch in {0} seconds",
-                                SHORT_WAITING_INTERVAL)
-                    time.sleep(SHORT_WAITING_INTERVAL)
-                else:
-                    host.manifest_uri = version.uri
-                    logger.verbose("Manifest downloaded successfully from host plugin")
-                    if not HostPluginProtocol.is_default_channel():
-                        logger.info("Setting host plugin as default channel")
-                        HostPluginProtocol.set_default_channel(True)
+                    logger.verbose("Failed to download manifest, "
+                                        "switching to host plugin")
+
+                try:
+                    host = self.get_host_plugin()
+                    uri, headers = host.get_artifact_request(version.uri)
+                    response = self.fetch(uri, headers, use_proxy=False)
+
+                # If the HostPlugin rejects the request,
+                # let the error continue, but set to use the HostPlugin
+                except ResourceGoneError:
+                    HostPluginProtocol.set_default_channel(True)
+                    raise
+
+                host.manifest_uri = version.uri
+                logger.verbose("Manifest downloaded successfully from host plugin")
+                if not HostPluginProtocol.is_default_channel():
+                    logger.info("Setting host plugin as default channel")
+                    HostPluginProtocol.set_default_channel(True)
+
             if response:
                 return response
+
         raise ProtocolError("Failed to fetch manifest from all sources")
 
-    def fetch(self, uri, headers=None, chk_proxy=None):
+    def fetch(self, uri, headers=None, use_proxy=None):
         logger.verbose("Fetch [{0}] with headers [{1}]", uri, headers)
-        return_value = None
         try:
             resp = self.call_storage_service(
-                restutil.http_get,
-                uri,
-                headers,
-                chk_proxy=chk_proxy)
-            if resp.status == httpclient.OK:
-                return_value = self.decode_config(resp.read())
-            else:
-                logger.warn("Could not fetch {0} [{1}]",
-                            uri,
-                            HostPluginProtocol.read_response_error(resp))
+                        restutil.http_get,
+                        uri,
+                        headers=headers,
+                        use_proxy=use_proxy)
+
+            if restutil.request_failed(resp):
+                msg = "[Storage Failed] URI {0} ".format(uri)
+                if resp is not None:
+                    msg += restutil.read_response_error(resp)
+                logger.warn(msg)
+                raise ProtocolError(msg)
+
+            return self.decode_config(resp.read())
+
         except (HttpError, ProtocolError) as e:
             logger.verbose("Fetch failed from [{0}]: {1}", uri, e)
-        return return_value
+
+            if isinstance(e, ResourceGoneError):
+                raise
+
+        return None
 
     def update_hosting_env(self, goal_state):
         if goal_state.hosting_env_uri is None:
@@ -734,6 +731,12 @@ class WireClient(object):
                     self.host_plugin.container_id = goal_state.container_id
                     self.host_plugin.role_config_name = goal_state.role_config_name
                 return
+
+            except ProtocolError:
+                if retry < max_retry-1:
+                    continue
+                raise
+
             except WireProtocolResourceGone:
                 logger.info("Incarnation is out of date. Update goalstate.")
                 xml_text = self.fetch_config(uri, self.get_header())
@@ -791,20 +794,45 @@ class WireClient(object):
         return self.ext_conf
 
     def get_ext_manifest(self, ext_handler, goal_state):
-        local_file = MANIFEST_FILE_NAME.format(ext_handler.name,
-                                               goal_state.incarnation)
-        local_file = os.path.join(conf.get_lib_dir(), local_file)
-        xml_text = self.fetch_manifest(ext_handler.versionUris)
-        self.save_cache(local_file, xml_text)
-        return ExtensionManifest(xml_text)
+        for update_goal_state in [False, True]:
+            try:
+                if update_goal_state:
+                    self.update_goal_state(forced=True)
+                    goal_state = self.get_goal_state()
+
+                local_file = MANIFEST_FILE_NAME.format(
+                                ext_handler.name,
+                                goal_state.incarnation)
+                local_file = os.path.join(conf.get_lib_dir(), local_file)
+                xml_text = self.fetch_manifest(ext_handler.versionUris)
+                self.save_cache(local_file, xml_text)
+                return ExtensionManifest(xml_text)
+
+            except ResourceGoneError:
+                continue
+
+        raise ProtocolError("Failed to retrieve extension manifest")
 
     def get_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)
-        xml_text = self.fetch_manifest(vmagent_manifest.versionsManifestUris)
-        fileutil.write_file(local_file, xml_text)
-        return ExtensionManifest(xml_text)
+        for update_goal_state in [False, True]:
+            try:
+                if update_goal_state:
+                    self.update_goal_state(forced=True)
+                    goal_state = self.get_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)
+                xml_text = self.fetch_manifest(
+                            vmagent_manifest.versionsManifestUris)
+                fileutil.write_file(local_file, xml_text)
+                return ExtensionManifest(xml_text)
+
+            except ResourceGoneError:
+                continue
+
+        raise ProtocolError("Failed to retrieve GAFamily manifest")
 
     def check_wire_protocol_version(self):
         uri = VERSION_INFO_URI.format(self.endpoint)
@@ -823,39 +851,55 @@ class WireClient(object):
             raise ProtocolNotFoundError(error)
 
     def upload_status_blob(self):
-        ext_conf = self.get_ext_conf()
+        for update_goal_state in [False, True]:
+            try:
+                if update_goal_state:
+                    self.update_goal_state(forced=True)
 
-        blob_uri = ext_conf.status_upload_blob
-        blob_type = ext_conf.status_upload_blob_type
+                ext_conf = self.get_ext_conf()
 
-        if blob_uri is not None:
+                blob_uri = ext_conf.status_upload_blob
+                blob_type = ext_conf.status_upload_blob_type
 
-            if not blob_type in ["BlockBlob", "PageBlob"]:
-                blob_type = "BlockBlob"
-                logger.verbose("Status Blob type is unspecified "
-                    "-- assuming it is a BlockBlob")
+                if blob_uri is not None:
+
+                    if not blob_type in ["BlockBlob", "PageBlob"]:
+                        blob_type = "BlockBlob"
+                        logger.verbose("Status Blob type is unspecified "
+                            "-- assuming it is a BlockBlob")
+
+                    try:
+                        self.status_blob.prepare(blob_type)
+                    except Exception as e:
+                        self.report_status_event(
+                            "Exception creating status blob: {0}", ustr(e))
+                        return
+
+                    if not HostPluginProtocol.is_default_channel():
+                        try:
+                            if self.status_blob.upload(blob_uri):
+                                return
+                        except HttpError as e:
+                            pass
+
+                    host = self.get_host_plugin()
+                    host.put_vm_status(self.status_blob,
+                                    ext_conf.status_upload_blob,
+                                    ext_conf.status_upload_blob_type)
+                    HostPluginProtocol.set_default_channel(True)
+                    return
 
-            try:
-                self.status_blob.prepare(blob_type)
             except Exception as e:
+                # If the HostPlugin rejects the request,
+                # let the error continue, but set to use the HostPlugin
+                if isinstance(e, ResourceGoneError):
+                    HostPluginProtocol.set_default_channel(True)
+                    continue
+
                 self.report_status_event(
-                    "Exception creating status blob: {0}",
-                    e)
+                    "Exception uploading status blob: {0}", ustr(e))
                 return
 
-            uploaded = False
-            if not HostPluginProtocol.is_default_channel():
-                try:
-                    uploaded = self.status_blob.upload(blob_uri)
-                except HttpError as e:
-                    pass
-
-            if not uploaded:
-                host = self.get_host_plugin()
-                host.put_vm_status(self.status_blob,
-                                   ext_conf.status_upload_blob,
-                                   ext_conf.status_upload_blob_type)
-
     def report_role_prop(self, thumbprint):
         goal_state = self.get_goal_state()
         role_prop = _build_role_properties(goal_state.container_id,
@@ -896,11 +940,12 @@ class WireClient(object):
                                         health_report_uri,
                                         health_report,
                                         headers=headers,
-                                        max_retry=30)
+                                        max_retry=30,
+                                        retry_delay=15)
         except HttpError as e:
             raise ProtocolError((u"Failed to send provision status: "
                                  u"{0}").format(e))
-        if resp.status != httpclient.OK:
+        if restutil.request_failed(resp):
             raise ProtocolError((u"Failed to send provision status: "
                                  u",{0}: {1}").format(resp.status,
                                                       resp.read()))
@@ -919,7 +964,7 @@ class WireClient(object):
         except HttpError as e:
             raise ProtocolError("Failed to send events:{0}".format(e))
 
-        if resp.status != httpclient.OK:
+        if restutil.request_failed(resp):
             logger.verbose(resp.read())
             raise ProtocolError(
                 "Failed to send events:{0}".format(resp.status))
@@ -979,12 +1024,8 @@ class WireClient(object):
             "x-ms-guest-agent-public-x509-cert": cert
         }
 
-    def get_host_plugin(self, force_update=False):
-        if self.host_plugin is None or force_update:
-            if force_update:
-                logger.warn("Forcing update of goal state")
-                self.goal_state = None
-                self.update_goal_state(forced=True)
+    def get_host_plugin(self):
+        if self.host_plugin is None:
             goal_state = self.get_goal_state()
             self.host_plugin = HostPluginProtocol(self.endpoint,
                                                   goal_state.container_id,
@@ -997,23 +1038,47 @@ class WireClient(object):
 
     def get_artifacts_profile(self):
         artifacts_profile = None
-        if self.has_artifacts_profile_blob():
-            blob = self.ext_conf.artifacts_profile_blob
-            logger.verbose("Getting the artifacts profile")
-            profile = self.fetch(blob)
+        for update_goal_state in [False, True]:
+            try:
+                if update_goal_state:
+                    self.update_goal_state(forced=True)
 
-            if profile is None:
-                logger.warn("Download failed, falling back to host plugin")
-                host = self.get_host_plugin()
-                uri, headers = host.get_artifact_request(blob)
-                profile = self.decode_config(self.fetch(uri, headers, chk_proxy=False))
+                if self.has_artifacts_profile_blob():
+                    blob = self.ext_conf.artifacts_profile_blob
 
-            if not textutil.is_str_none_or_whitespace(profile):
-                    logger.verbose("Artifacts profile downloaded successfully")
-                    artifacts_profile = InVMArtifactsProfile(profile)
+                    profile = None
+                    if not HostPluginProtocol.is_default_channel():
+                        logger.verbose("Retrieving the artifacts profile")
+                        profile = self.fetch(blob)
 
-        return artifacts_profile
+                    if profile is None:
+                        if HostPluginProtocol.is_default_channel():
+                            logger.verbose("Using host plugin as default channel")
+                        else:
+                            logger.verbose("Failed to download artifacts profile, "
+                                                "switching to host plugin")
 
+                        host = self.get_host_plugin()
+                        uri, headers = host.get_artifact_request(blob)
+                        config = self.fetch(uri, headers, use_proxy=False)
+                        profile = self.decode_config(config)
+
+                    if not textutil.is_str_none_or_whitespace(profile):
+                        logger.verbose("Artifacts profile downloaded")
+                        artifacts_profile = InVMArtifactsProfile(profile)
+
+                return artifacts_profile
+
+            except ResourceGoneError:
+                HostPluginProtocol.set_default_channel(True)
+                continue
+
+            except Exception as e:
+                logger.warn(
+                    "Exception retrieving artifacts profile: {0}".format(
+                        ustr(e)))
+
+        return None
 
 class VersionInfo(object):
     def __init__(self, xml_text):