diff options
Diffstat (limited to 'azurelinuxagent/common/protocol/hostplugin.py')
| -rw-r--r-- | azurelinuxagent/common/protocol/hostplugin.py | 246 |
1 files changed, 186 insertions, 60 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) |