diff options
Diffstat (limited to 'azurelinuxagent/common/protocol')
| -rw-r--r-- | azurelinuxagent/common/protocol/hostplugin.py | 246 | ||||
| -rw-r--r-- | azurelinuxagent/common/protocol/wire.py | 112 |
2 files changed, 251 insertions, 107 deletions
diff --git a/azurelinuxagent/common/protocol/hostplugin.py b/azurelinuxagent/common/protocol/hostplugin.py index bdae56e..70bf8b4 100644 --- a/azurelinuxagent/common/protocol/hostplugin.py +++ b/azurelinuxagent/common/protocol/hostplugin.py @@ -16,8 +16,17 @@ # # Requires Python 2.4+ and Openssl 1.0+ # -from azurelinuxagent.common.protocol.wire import * + +import base64 +import json + +from azurelinuxagent.common import logger +from azurelinuxagent.common.exception import ProtocolError, HttpError +from azurelinuxagent.common.future import ustr, httpclient +from azurelinuxagent.common.utils import restutil from azurelinuxagent.common.utils import textutil +from azurelinuxagent.common.utils.textutil import remove_bom +from azurelinuxagent.common.version import PY_VERSION_MAJOR HOST_PLUGIN_PORT = 32526 URI_FORMAT_GET_API_VERSIONS = "http://{0}:{1}/versions" @@ -30,12 +39,15 @@ HEADER_VERSION = "x-ms-version" HEADER_HOST_CONFIG_NAME = "x-ms-host-config-name" HEADER_ARTIFACT_LOCATION = "x-ms-artifact-location" HEADER_ARTIFACT_MANIFEST_LOCATION = "x-ms-artifact-manifest-location" +MAXIMUM_PAGEBLOB_PAGE_SIZE = 4 * 1024 * 1024 # Max page size: 4MB class HostPluginProtocol(object): + _is_default_channel = False + def __init__(self, endpoint, container_id, role_config_name): if endpoint is None: - raise ProtocolError("Host plugin endpoint not provided") + raise ProtocolError("HostGAPlugin: Endpoint not provided") self.is_initialized = False self.is_available = False self.api_versions = None @@ -45,45 +57,51 @@ class HostPluginProtocol(object): self.role_config_name = role_config_name self.manifest_uri = None + @staticmethod + def is_default_channel(): + return HostPluginProtocol._is_default_channel + + @staticmethod + def set_default_channel(is_default): + HostPluginProtocol._is_default_channel = is_default + def ensure_initialized(self): if not self.is_initialized: self.api_versions = self.get_api_versions() self.is_available = API_VERSION in self.api_versions self.is_initialized = True - - from azurelinuxagent.common.event import add_event, WALAEventOperation - add_event(name="WALA", - op=WALAEventOperation.InitializeHostPlugin, - is_success=self.is_available) + from azurelinuxagent.common.event import WALAEventOperation, report_event + report_event(WALAEventOperation.InitializeHostPlugin, + is_success=self.is_available) return self.is_available def get_api_versions(self): url = URI_FORMAT_GET_API_VERSIONS.format(self.endpoint, HOST_PLUGIN_PORT) - logger.verbose("getting API versions at [{0}]".format(url)) + logger.verbose("HostGAPlugin: Getting API versions at [{0}]".format( + url)) return_val = [] try: headers = {HEADER_CONTAINER_ID: self.container_id} response = restutil.http_get(url, headers) if response.status != httpclient.OK: logger.error( - "get API versions returned status code [{0}]".format( - response.status)) + "HostGAPlugin: Failed Get API versions: {0}".format( + self.read_response_error(response))) else: return_val = ustr(remove_bom(response.read()), encoding='utf-8') except HttpError as e: - logger.error("get API versions failed with [{0}]".format(e)) + logger.error("HostGAPlugin: Exception Get API versions: {0}".format(e)) return return_val def get_artifact_request(self, artifact_url, artifact_manifest_url=None): if not self.ensure_initialized(): - logger.error("host plugin channel is not available") - return None, None + raise ProtocolError("HostGAPlugin: Host plugin channel is not available") + if textutil.is_str_none_or_whitespace(artifact_url): - logger.error("no extension artifact url was provided") - return None, None + raise ProtocolError("HostGAPlugin: No extension artifact url was provided") url = URI_FORMAT_GET_EXTENSION_ARTIFACT.format(self.endpoint, HOST_PLUGIN_PORT) @@ -97,45 +115,6 @@ class HostPluginProtocol(object): return url, headers - def put_vm_status(self, status_blob, sas_url, config_blob_type=None): - """ - Try to upload the VM status via the host plugin /status channel - :param sas_url: the blob SAS url to pass to the host plugin - :param config_blob_type: the blob type from the extension config - :type status_blob: StatusBlob - """ - if not self.ensure_initialized(): - logger.error("host plugin channel is not available") - return - if status_blob is None or status_blob.vm_status is None: - logger.error("no status data was provided") - return - try: - url = URI_FORMAT_PUT_VM_STATUS.format(self.endpoint, HOST_PLUGIN_PORT) - logger.verbose("Posting VM status to host plugin") - status = textutil.b64encode(status_blob.data) - blob_type = status_blob.type if status_blob.type else config_blob_type - headers = {HEADER_VERSION: API_VERSION, - "Content-type": "application/json", - HEADER_CONTAINER_ID: self.container_id, - HEADER_HOST_CONFIG_NAME: self.role_config_name} - blob_headers = [{'headerName': 'x-ms-version', - 'headerValue': status_blob.__storage_version__}, - {'headerName': 'x-ms-blob-type', - 'headerValue': blob_type}] - data = json.dumps({'requestUri': sas_url, 'headers': blob_headers, - 'content': status}, sort_keys=True) - response = restutil.http_put(url, data=data, headers=headers) - if response.status != httpclient.OK: - logger.warn("PUT {0} [{1}: {2}]", - url, - response.status, - response.reason) - else: - logger.verbose("Successfully uploaded status to host plugin") - except Exception as e: - logger.error("Put VM status failed [{0}]", e) - def put_vm_log(self, content): """ Try to upload the given content to the host plugin @@ -147,13 +126,13 @@ class HostPluginProtocol(object): :return: """ if not self.ensure_initialized(): - logger.error("host plugin channel is not available") - return + raise ProtocolError("HostGAPlugin: Host plugin channel is not available") + if content is None \ or self.container_id is None \ or self.deployment_id is None: logger.error( - "invalid arguments passed: " + "HostGAPlugin: Invalid arguments passed: " "[{0}], [{1}], [{2}]".format( content, self.container_id, @@ -163,11 +142,158 @@ class HostPluginProtocol(object): headers = {"x-ms-vmagentlog-deploymentid": self.deployment_id, "x-ms-vmagentlog-containerid": self.container_id} - logger.info("put VM log at [{0}]".format(url)) + logger.info("HostGAPlugin: Put VM log to [{0}]".format(url)) try: response = restutil.http_put(url, content, headers) if response.status != httpclient.OK: - logger.error("put log returned status code [{0}]".format( + logger.error("HostGAPlugin: Put log failed: Code {0}".format( response.status)) except HttpError as e: - logger.error("put log failed with [{0}]".format(e)) + logger.error("HostGAPlugin: Put log exception: {0}".format(e)) + + def put_vm_status(self, status_blob, sas_url, config_blob_type=None): + """ + Try to upload the VM status via the host plugin /status channel + :param sas_url: the blob SAS url to pass to the host plugin + :param config_blob_type: the blob type from the extension config + :type status_blob: StatusBlob + """ + if not self.ensure_initialized(): + raise ProtocolError("HostGAPlugin: HostGAPlugin is not available") + + if status_blob is None or status_blob.vm_status is None: + raise ProtocolError("HostGAPlugin: Status blob was not provided") + + logger.verbose("HostGAPlugin: Posting VM status") + try: + blob_type = status_blob.type if status_blob.type else config_blob_type + + if blob_type == "BlockBlob": + self._put_block_blob_status(sas_url, status_blob) + else: + self._put_page_blob_status(sas_url, status_blob) + + if not HostPluginProtocol.is_default_channel(): + logger.info("HostGAPlugin: Setting host plugin as default channel") + HostPluginProtocol.set_default_channel(True) + except Exception as e: + message = "HostGAPlugin: Exception Put VM status: {0}".format(e) + logger.error(message) + from azurelinuxagent.common.event import WALAEventOperation, report_event + report_event(op=WALAEventOperation.ReportStatus, + is_success=False, + message=message) + logger.warn("HostGAPlugin: resetting default channel") + HostPluginProtocol.set_default_channel(False) + + def _put_block_blob_status(self, sas_url, status_blob): + url = URI_FORMAT_PUT_VM_STATUS.format(self.endpoint, HOST_PLUGIN_PORT) + + response = restutil.http_put(url, + data=self._build_status_data( + sas_url, + status_blob.get_block_blob_headers(len(status_blob.data)), + bytearray(status_blob.data, encoding='utf-8')), + headers=self._build_status_headers()) + + if response.status != httpclient.OK: + raise HttpError("HostGAPlugin: Put BlockBlob failed: {0}".format( + self.read_response_error(response))) + else: + logger.verbose("HostGAPlugin: Put BlockBlob status succeeded") + + def _put_page_blob_status(self, sas_url, status_blob): + url = URI_FORMAT_PUT_VM_STATUS.format(self.endpoint, HOST_PLUGIN_PORT) + + # Convert the status into a blank-padded string whose length is modulo 512 + status = bytearray(status_blob.data, encoding='utf-8') + status_size = int((len(status) + 511) / 512) * 512 + status = bytearray(status_blob.data.ljust(status_size), encoding='utf-8') + + # First, initialize an empty blob + response = restutil.http_put(url, + data=self._build_status_data( + sas_url, + status_blob.get_page_blob_create_headers(status_size)), + headers=self._build_status_headers()) + + if response.status != httpclient.OK: + raise HttpError( + "HostGAPlugin: Failed PageBlob clean-up: {0}".format( + self.read_response_error(response))) + else: + logger.verbose("HostGAPlugin: PageBlob clean-up succeeded") + + # Then, upload the blob in pages + if sas_url.count("?") <= 0: + sas_url = "{0}?comp=page".format(sas_url) + else: + sas_url = "{0}&comp=page".format(sas_url) + + start = 0 + end = 0 + while start < len(status): + # Create the next page + end = start + min(len(status) - start, MAXIMUM_PAGEBLOB_PAGE_SIZE) + page_size = int((end - start + 511) / 512) * 512 + buf = bytearray(page_size) + buf[0: end - start] = status[start: end] + + # Send the page + response = restutil.http_put(url, + data=self._build_status_data( + sas_url, + status_blob.get_page_blob_page_headers(start, end), + buf), + headers=self._build_status_headers()) + + if response.status != httpclient.OK: + raise HttpError( + "HostGAPlugin Error: Put PageBlob bytes [{0},{1}]: " \ + "{2}".format( + start, end, self.read_response_error(response))) + + # Advance to the next page (if any) + start = end + + def _build_status_data(self, sas_url, blob_headers, content=None): + headers = [] + for name in iter(blob_headers.keys()): + headers.append({ + 'headerName': name, + 'headerValue': blob_headers[name] + }) + + data = { + 'requestUri': sas_url, + 'headers': headers + } + if not content is None: + data['content'] = self._base64_encode(content) + return json.dumps(data, sort_keys=True) + + def _build_status_headers(self): + return { + HEADER_VERSION: API_VERSION, + "Content-type": "application/json", + HEADER_CONTAINER_ID: self.container_id, + HEADER_HOST_CONFIG_NAME: self.role_config_name + } + + def _base64_encode(self, data): + s = base64.b64encode(bytes(data)) + if PY_VERSION_MAJOR > 2: + return s.decode('utf-8') + return s + + @staticmethod + def read_response_error(response): + if response is None: + return '' + body = remove_bom(response.read()) + if PY_VERSION_MAJOR < 3 and body is not None: + body = ustr(body, encoding='utf-8') + return "{0}, {1}, {2}".format( + response.status, + response.reason, + body) diff --git a/azurelinuxagent/common/protocol/wire.py b/azurelinuxagent/common/protocol/wire.py index 71c3e37..265e2dd 100644 --- a/azurelinuxagent/common/protocol/wire.py +++ b/azurelinuxagent/common/protocol/wire.py @@ -370,11 +370,14 @@ class StatusBlob(object): __storage_version__ = "2014-02-14" + def prepare(self, blob_type): + logger.verbose("Prepare status blob") + self.data = self.to_json() + self.type = blob_type + def upload(self, url): # TODO upload extension only if content has changed - logger.verbose("Upload status blob") upload_successful = False - self.data = self.to_json() self.type = self.get_blob_type(url) try: if self.type == "BlockBlob": @@ -384,7 +387,12 @@ class StatusBlob(object): else: raise ProtocolError("Unknown blob type: {0}".format(self.type)) except HttpError as e: - logger.warn("Initial upload failed [{0}]".format(e)) + message = "Initial upload failed [{0}]".format(e) + logger.warn(message) + from azurelinuxagent.common.event import WALAEventOperation, report_event + report_event(op=WALAEventOperation.ReportStatus, + is_success=False, + message=message) else: logger.verbose("Uploading status blob succeeded") upload_successful = True @@ -411,48 +419,54 @@ class StatusBlob(object): logger.verbose("Blob type: [{0}]", blob_type) return blob_type + def get_block_blob_headers(self, blob_size): + return { + "Content-Length": ustr(blob_size), + "x-ms-blob-type": "BlockBlob", + "x-ms-date": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + "x-ms-version": self.__class__.__storage_version__ + } + def put_block_blob(self, url, data): logger.verbose("Put block blob") - timestamp = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) - resp = self.client.call_storage_service( - restutil.http_put, - url, - data, - { - "x-ms-date": timestamp, - "x-ms-blob-type": "BlockBlob", - "Content-Length": ustr(len(data)), - "x-ms-version": self.__class__.__storage_version__ - }) + headers = self.get_block_blob_headers(len(data)) + resp = self.client.call_storage_service(restutil.http_put, url, data, headers) if resp.status != httpclient.CREATED: raise UploadError( "Failed to upload block blob: {0}".format(resp.status)) + def get_page_blob_create_headers(self, blob_size): + return { + "Content-Length": "0", + "x-ms-blob-content-length": ustr(blob_size), + "x-ms-blob-type": "PageBlob", + "x-ms-date": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + "x-ms-version": self.__class__.__storage_version__ + } + + def get_page_blob_page_headers(self, start, end): + return { + "Content-Length": ustr(end - start), + "x-ms-date": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + "x-ms-range": "bytes={0}-{1}".format(start, end - 1), + "x-ms-page-write": "update", + "x-ms-version": self.__class__.__storage_version__ + } + def put_page_blob(self, url, data): logger.verbose("Put page blob") - # Convert string into bytes + # Convert string into bytes and align to 512 bytes data = bytearray(data, encoding='utf-8') - timestamp = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) - - # Align to 512 bytes page_blob_size = int((len(data) + 511) / 512) * 512 - resp = self.client.call_storage_service( - restutil.http_put, - url, - "", - { - "x-ms-date": timestamp, - "x-ms-blob-type": "PageBlob", - "Content-Length": "0", - "x-ms-blob-content-length": ustr(page_blob_size), - "x-ms-version": self.__class__.__storage_version__ - }) + + headers = self.get_page_blob_create_headers(page_blob_size) + resp = self.client.call_storage_service(restutil.http_put, url, "", headers) if resp.status != httpclient.CREATED: raise UploadError( "Failed to clean up page blob: {0}".format(resp.status)) - if url.count("?") < 0: + if url.count("?") <= 0: url = "{0}?comp=page".format(url) else: url = "{0}&comp=page".format(url) @@ -469,17 +483,12 @@ class StatusBlob(object): buf_size = page_end - start buf = bytearray(buf_size) buf[0: content_size] = data[start: end] + headers = self.get_page_blob_page_headers(start, page_end) resp = self.client.call_storage_service( restutil.http_put, url, bytebuffer(buf), - { - "x-ms-date": timestamp, - "x-ms-range": "bytes={0}-{1}".format(start, page_end - 1), - "x-ms-page-write": "update", - "x-ms-version": self.__class__.__storage_version__, - "Content-Length": ustr(page_end - start) - }) + headers) if resp is None or resp.status != httpclient.CREATED: raise UploadError( "Failed to upload page blob: {0}".format(resp.status)) @@ -634,9 +643,14 @@ class WireClient(object): def fetch_manifest(self, version_uris): logger.verbose("Fetch manifest") for version in version_uris: - response = self.fetch(version.uri) + response = None + if not HostPluginProtocol.is_default_channel(): + response = self.fetch(version.uri) if not response: - logger.verbose("Manifest could not be downloaded, falling back to host plugin") + 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) @@ -648,6 +662,9 @@ class WireClient(object): 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) if response: return response raise ProtocolError("Failed to fetch manifest from all sources") @@ -663,12 +680,11 @@ class WireClient(object): if resp.status == httpclient.OK: return_value = self.decode_config(resp.read()) else: - logger.warn("Could not fetch {0} [{1}: {2}]", + logger.warn("Could not fetch {0} [{1}]", uri, - resp.status, - resp.reason) + HostPluginProtocol.read_response_error(resp)) except (HttpError, ProtocolError) as e: - logger.verbose("Fetch failed from [{0}]", uri) + logger.verbose("Fetch failed from [{0}]: {1}", uri, e) return return_value def update_hosting_env(self, goal_state): @@ -839,10 +855,12 @@ class WireClient(object): if ext_conf.status_upload_blob is not None: uploaded = False try: - uploaded = self.status_blob.upload(ext_conf.status_upload_blob) - self.report_blob_type(self.status_blob.type, - ext_conf.status_upload_blob_type) - except (HttpError, ProtocolError) as e: + self.status_blob.prepare(ext_conf.status_upload_blob_type) + if not HostPluginProtocol.is_default_channel(): + uploaded = self.status_blob.upload(ext_conf.status_upload_blob) + self.report_blob_type(self.status_blob.type, + ext_conf.status_upload_blob_type) + except (HttpError, ProtocolError): # errors have already been logged pass if not uploaded: |