summaryrefslogtreecommitdiff
path: root/azurelinuxagent/common/protocol
diff options
context:
space:
mode:
Diffstat (limited to 'azurelinuxagent/common/protocol')
-rw-r--r--azurelinuxagent/common/protocol/hostplugin.py96
-rw-r--r--azurelinuxagent/common/protocol/metadata.py249
-rw-r--r--azurelinuxagent/common/protocol/restapi.py74
-rw-r--r--azurelinuxagent/common/protocol/wire.py392
4 files changed, 620 insertions, 191 deletions
diff --git a/azurelinuxagent/common/protocol/hostplugin.py b/azurelinuxagent/common/protocol/hostplugin.py
index 6569604..e83dd4b 100644
--- a/azurelinuxagent/common/protocol/hostplugin.py
+++ b/azurelinuxagent/common/protocol/hostplugin.py
@@ -22,19 +22,27 @@ from azurelinuxagent.common.utils import textutil
HOST_PLUGIN_PORT = 32526
URI_FORMAT_GET_API_VERSIONS = "http://{0}:{1}/versions"
+URI_FORMAT_GET_EXTENSION_ARTIFACT = "http://{0}:{1}/extensionArtifact"
URI_FORMAT_PUT_VM_STATUS = "http://{0}:{1}/status"
URI_FORMAT_PUT_LOG = "http://{0}:{1}/vmAgentLog"
API_VERSION = "2015-09-01"
-
+HEADER_CONTAINER_ID = "x-ms-containerid"
+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"
class HostPluginProtocol(object):
- def __init__(self, endpoint):
+ def __init__(self, endpoint, container_id, role_config_name):
if endpoint is None:
raise ProtocolError("Host plugin endpoint not provided")
self.is_initialized = False
self.is_available = False
self.api_versions = None
self.endpoint = endpoint
+ self.container_id = container_id
+ self.role_config_name = role_config_name
+ self.manifest_uri = None
def ensure_initialized(self):
if not self.is_initialized:
@@ -46,23 +54,48 @@ class HostPluginProtocol(object):
def get_api_versions(self):
url = URI_FORMAT_GET_API_VERSIONS.format(self.endpoint,
HOST_PLUGIN_PORT)
- logger.info("getting API versions at [{0}]".format(url))
+ logger.verbose("getting API versions at [{0}]".format(url))
+ return_val = []
try:
- response = restutil.http_get(url)
+ 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))
- return []
- return response.read()
+ 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))
- return []
- def put_vm_status(self, status_blob, sas_url):
+ 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
+ if textutil.is_str_none_or_whitespace(artifact_url):
+ logger.error("no extension artifact url was provided")
+ return
+
+ url = URI_FORMAT_GET_EXTENSION_ARTIFACT.format(self.endpoint,
+ HOST_PLUGIN_PORT)
+ headers = {HEADER_VERSION: API_VERSION,
+ HEADER_CONTAINER_ID: self.container_id,
+ HEADER_HOST_CONFIG_NAME: self.role_config_name,
+ HEADER_ARTIFACT_LOCATION: artifact_url}
+
+ if artifact_manifest_url is not None:
+ headers[HEADER_ARTIFACT_MANIFEST_LOCATION] = artifact_manifest_url
+
+ 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():
@@ -71,25 +104,30 @@ class HostPluginProtocol(object):
if status_blob is None or status_blob.vm_status is None:
logger.error("no status data was provided")
return
- url = URI_FORMAT_PUT_VM_STATUS.format(self.endpoint, HOST_PLUGIN_PORT)
- status = textutil.b64encode(status_blob.vm_status)
- headers = {"x-ms-version": API_VERSION}
- blob_headers = [{'headerName': 'x-ms-version',
- 'headerValue': status_blob.__storage_version__},
- {'headerName': 'x-ms-blob-type',
- 'headerValue': status_blob.type}]
- data = json.dumps({'requestUri': sas_url, 'headers': blob_headers,
- 'content': status}, sort_keys=True)
- logger.info("put VM status at [{0}]".format(url))
try:
- response = restutil.http_put(url, data, headers)
+ 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.error("put VM status returned status code [{0}]".format(
- response.status))
- except HttpError as e:
- logger.error("put VM status failed with [{0}]".format(e))
+ logger.error("PUT failed [{0}]", response.status)
+ 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, container_id, deployment_id):
+ def put_vm_log(self, content):
"""
Try to upload the given content to the host plugin
:param deployment_id: the deployment id, which is obtained from the
@@ -102,18 +140,18 @@ class HostPluginProtocol(object):
if not self.ensure_initialized():
logger.error("host plugin channel is not available")
return
- if content is None or container_id is None or deployment_id is None:
+ if content is None or self.goal_state.container_id is None or self.goal_state.deployment_id is None:
logger.error(
"invalid arguments passed: "
"[{0}], [{1}], [{2}]".format(
content,
- container_id,
- deployment_id))
+ self.goal_state.container_id,
+ self.goal_state.deployment_id))
return
url = URI_FORMAT_PUT_LOG.format(self.endpoint, HOST_PLUGIN_PORT)
- headers = {"x-ms-vmagentlog-deploymentid": deployment_id,
- "x-ms-vmagentlog-containerid": container_id}
+ headers = {"x-ms-vmagentlog-deploymentid": self.goal_state.deployment_id,
+ "x-ms-vmagentlog-containerid": self.goal_state.container_id}
logger.info("put VM log at [{0}]".format(url))
try:
response = restutil.http_put(url, content, headers)
diff --git a/azurelinuxagent/common/protocol/metadata.py b/azurelinuxagent/common/protocol/metadata.py
index f86f72f..c61e373 100644
--- a/azurelinuxagent/common/protocol/metadata.py
+++ b/azurelinuxagent/common/protocol/metadata.py
@@ -16,39 +16,42 @@
#
# Requires Python 2.4+ and Openssl 1.0+
+import base64
import json
-import shutil
import os
-import time
-from azurelinuxagent.common.exception import ProtocolError, HttpError
-from azurelinuxagent.common.future import httpclient, ustr
+import shutil
+import re
import azurelinuxagent.common.conf as conf
-import azurelinuxagent.common.logger as logger
-import azurelinuxagent.common.utils.restutil as restutil
-import azurelinuxagent.common.utils.textutil as textutil
import azurelinuxagent.common.utils.fileutil as fileutil
-from azurelinuxagent.common.utils.cryptutil import CryptUtil
+import azurelinuxagent.common.utils.shellutil as shellutil
+import azurelinuxagent.common.utils.textutil as textutil
+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'
+METADATA_ENDPOINT = '169.254.169.254'
+APIVERSION = '2015-05-01-preview'
BASE_URI = "http://{0}/Microsoft.Compute/{1}?api-version={2}{3}"
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"
-#TODO remote workarround for azure stack
+# TODO remote workaround for azure stack
MAX_PING = 30
RETRY_PING_INTERVAL = 10
+
def _add_content_type(headers):
if headers is None:
headers = {}
headers["content-type"] = "application/json"
return headers
-class MetadataProtocol(Protocol):
+class MetadataProtocol(Protocol):
def __init__(self, apiversion=APIVERSION, endpoint=METADATA_ENDPOINT):
self.apiversion = apiversion
self.endpoint = endpoint
@@ -65,11 +68,12 @@ class MetadataProtocol(Protocol):
self.apiversion, "")
self.vm_status_uri = BASE_URI.format(self.endpoint, "status/vmagent",
self.apiversion, "")
- self.ext_status_uri = BASE_URI.format(self.endpoint,
+ 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
def _get_data(self, url, headers=None):
try:
@@ -82,13 +86,12 @@ class MetadataProtocol(Protocol):
data = resp.read()
etag = resp.getheader('ETag')
- if data is None:
- return None
- data = json.loads(ustr(data, encoding="utf-8"))
+ 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)
+ headers = _add_content_type(headers)
try:
resp = restutil.http_put(url, json.dumps(data), headers=headers)
except HttpError as e:
@@ -97,16 +100,16 @@ class MetadataProtocol(Protocol):
raise ProtocolError("{0} - PUT: {1}".format(resp.status, url))
def _post_data(self, url, data, headers=None):
- headers = _add_content_type(headers)
+ 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:
raise ProtocolError("{0} - POST: {1}".format(resp.status, url))
-
+
def _get_trans_cert(self):
- trans_crt_file = os.path.join(conf.get_lib_dir(),
+ 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))
@@ -115,22 +118,22 @@ class MetadataProtocol(Protocol):
def detect(self):
self.get_vminfo()
- trans_prv_file = os.path.join(conf.get_lib_dir(),
+ trans_prv_file = os.path.join(conf.get_lib_dir(),
TRANSPORT_PRV_FILE_NAME)
- trans_cert_file = os.path.join(conf.get_lib_dir(),
+ 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
+ # "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(),
+ prv_file = os.path.join(conf.get_lib_dir(),
"{0}.prv".format(thumbprint))
- crt_file = os.path.join(conf.get_lib_dir(),
+ 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_goal_state(forced=True)
def get_vminfo(self):
vminfo = VMInfo()
@@ -139,18 +142,42 @@ class MetadataProtocol(Protocol):
return vminfo
def get_certs(self):
- #TODO download and save certs
- return CertList()
+ certlist = CertList()
+ certificatedata = CertificateData()
+ data, etag = self._get_data(self.cert_uri)
+
+ 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_vmagent_manifests(self, last_etag=None):
manifests = VMAgentManifestList()
+ self.update_goal_state()
data, etag = self._get_data(self.vmagent_uri)
- if last_etag == None or last_etag < etag:
- set_properties("vmAgentManifests", manifests.vmAgentManifests, data)
+ if last_etag is None or last_etag < etag:
+ set_properties("vmAgentManifests",
+ manifests.vmAgentManifests,
+ data)
return manifests, etag
def get_vmagent_pkgs(self, vmagent_manifest):
- #Agent package is the same with extension handler
+ # Agent package is the same with extension handler
vmagent_pkgs = ExtHandlerPackageList()
data = None
for manifest_uri in vmagent_manifest.versionsManifestUris:
@@ -168,27 +195,35 @@ class MetadataProtocol(Protocol):
return vmagent_pkgs
def get_ext_handlers(self, last_etag=None):
+ self.update_goal_state()
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 == None or last_etag < etag:
+ 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):
- ext_handler_pkgs = ExtHandlerPackageList()
- data = None
+ logger.info("Get extension handler packages")
+ pkg_list = ExtHandlerPackageList()
+
+ manifest = None
for version_uri in ext_handler.versionUris:
try:
- data, etag = self._get_data(version_uri.uri)
+ manifest, etag = self._get_data(version_uri.uri)
+ logger.info("Successfully downloaded manifest")
break
except ProtocolError as e:
- logger.warn("Failed to get version uris: {0}", e)
- logger.info("Retry getting version uris")
- set_properties("extensionPackages", ext_handler_pkgs, data)
- return ext_handler_pkgs
+ 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)
@@ -198,7 +233,8 @@ class MetadataProtocol(Protocol):
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
+ # TODO code field is not implemented for metadata protocol yet.
+ # Remove it
handler_statuses = data['vmAgent']['extensionHandlers']
for handler_status in handler_statuses:
try:
@@ -215,9 +251,134 @@ class MetadataProtocol(Protocol):
self._put_data(uri, data)
def report_event(self, events):
- #TODO disable telemetry for azure stack test
- #validate_param('events', events, TelemetryEventList)
- #data = get_properties(events)
- #self._post_data(self.event_uri, data)
+ # TODO disable telemetry for azure stack test
+ # validate_param('events', events, TelemetryEventList)
+ # data = get_properties(events)
+ # self._post_data(self.event_uri, data)
pass
+ def update_certs(self):
+ certificates = self.get_certs()
+ return certificates.cert_list
+
+ def update_goal_state(self, forced=False, max_retry=3):
+ # Start updating goalstate, retry on 410
+ for retry in range(0, max_retry):
+ try:
+ self.update_certs()
+ return
+ except:
+ logger.verbose("Incarnation is out of date. Update goalstate.")
+ raise ProtocolError("Exceeded max retry updating goal state")
+
+
+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/restapi.py b/azurelinuxagent/common/protocol/restapi.py
index 7f00488..5f71cf2 100644
--- a/azurelinuxagent/common/protocol/restapi.py
+++ b/azurelinuxagent/common/protocol/restapi.py
@@ -16,15 +16,12 @@
#
# Requires Python 2.4+ and Openssl 1.0+
#
-import os
-import copy
-import re
-import json
-import xml.dom.minidom
+
import azurelinuxagent.common.logger as logger
+import azurelinuxagent.common.utils.restutil as restutil
from azurelinuxagent.common.exception import ProtocolError, HttpError
from azurelinuxagent.common.future import ustr
-import azurelinuxagent.common.utils.restutil as restutil
+
def validate_param(name, val, expected_type):
if val is None:
@@ -33,13 +30,14 @@ def validate_param(name, val, expected_type):
raise ProtocolError(("{0} type should be {1} not {2}"
"").format(name, expected_type, type(val)))
+
def set_properties(name, obj, data):
if isinstance(obj, DataContract):
validate_param("Property '{0}'".format(name), data, dict)
for prob_name, prob_val in data.items():
prob_full_name = "{0}.{1}".format(name, prob_name)
try:
- prob = getattr(obj, prob_name)
+ prob = getattr(obj, prob_name)
except AttributeError:
logger.warn("Unknown property: {0}", prob_full_name)
continue
@@ -56,6 +54,7 @@ def set_properties(name, obj, data):
else:
return data
+
def get_properties(obj):
if isinstance(obj, DataContract):
data = {}
@@ -72,16 +71,21 @@ def get_properties(obj):
else:
return obj
+
class DataContract(object):
pass
+
class DataContractList(list):
def __init__(self, item_cls):
self.item_cls = item_cls
+
"""
Data contract between guest and host
"""
+
+
class VMInfo(DataContract):
def __init__(self, subscriptionId=None, vmName=None, containerId=None,
roleName=None, roleInstanceName=None, tenantName=None):
@@ -92,30 +96,40 @@ class VMInfo(DataContract):
self.roleInstanceName = roleInstanceName
self.tenantName = tenantName
+class CertificateData(DataContract):
+ def __init__(self, certificateData=None):
+ self.certificateData = certificateData
+
class Cert(DataContract):
- def __init__(self, name=None, thumbprint=None, certificateDataUri=None):
+ def __init__(self, name=None, thumbprint=None, certificateDataUri=None, storeName=None, storeLocation=None):
self.name = name
self.thumbprint = thumbprint
self.certificateDataUri = certificateDataUri
+ self.storeLocation = storeLocation
+ self.storeName = storeName
class CertList(DataContract):
def __init__(self):
self.certificates = DataContractList(Cert)
-#TODO: confirm vmagent manifest schema
+
+# TODO: confirm vmagent manifest schema
class VMAgentManifestUri(DataContract):
def __init__(self, uri=None):
self.uri = uri
+
class VMAgentManifest(DataContract):
def __init__(self, family=None):
self.family = family
self.versionsManifestUris = DataContractList(VMAgentManifestUri)
+
class VMAgentManifestList(DataContract):
def __init__(self):
self.vmAgentManifests = DataContractList(VMAgentManifest)
+
class Extension(DataContract):
def __init__(self, name=None, sequenceNumber=None, publicSettings=None,
protectedSettings=None, certificateThumbprint=None):
@@ -125,6 +139,7 @@ class Extension(DataContract):
self.protectedSettings = protectedSettings
self.certificateThumbprint = certificateThumbprint
+
class ExtHandlerProperties(DataContract):
def __init__(self):
self.version = None
@@ -132,40 +147,49 @@ class ExtHandlerProperties(DataContract):
self.state = None
self.extensions = DataContractList(Extension)
+
class ExtHandlerVersionUri(DataContract):
def __init__(self):
self.uri = None
+
class ExtHandler(DataContract):
def __init__(self, name=None):
self.name = name
self.properties = ExtHandlerProperties()
self.versionUris = DataContractList(ExtHandlerVersionUri)
+
class ExtHandlerList(DataContract):
def __init__(self):
self.extHandlers = DataContractList(ExtHandler)
+
class ExtHandlerPackageUri(DataContract):
def __init__(self, uri=None):
self.uri = uri
+
class ExtHandlerPackage(DataContract):
- def __init__(self, version = None):
+ def __init__(self, version=None):
self.version = version
self.uris = DataContractList(ExtHandlerPackageUri)
# TODO update the naming to align with metadata protocol
self.isinternal = False
+ self.disallow_major_upgrade = False
+
class ExtHandlerPackageList(DataContract):
def __init__(self):
self.versions = DataContractList(ExtHandlerPackage)
+
class VMProperties(DataContract):
def __init__(self, certificateThumbprint=None):
- #TODO need to confirm the property name
+ # TODO need to confirm the property name
self.certificateThumbprint = certificateThumbprint
+
class ProvisionStatus(DataContract):
def __init__(self, status=None, subStatus=None, description=None):
self.status = status
@@ -173,6 +197,7 @@ class ProvisionStatus(DataContract):
self.description = description
self.properties = VMProperties()
+
class ExtensionSubStatus(DataContract):
def __init__(self, name=None, status=None, code=None, message=None):
self.name = name
@@ -180,6 +205,7 @@ class ExtensionSubStatus(DataContract):
self.code = code
self.message = message
+
class ExtensionStatus(DataContract):
def __init__(self, configurationAppliedTime=None, operation=None,
status=None, seq_no=None, code=None, message=None):
@@ -191,8 +217,9 @@ class ExtensionStatus(DataContract):
self.message = message
self.substatusList = DataContractList(ExtensionSubStatus)
+
class ExtHandlerStatus(DataContract):
- def __init__(self, name=None, version=None, status=None, code=0,
+ def __init__(self, name=None, version=None, status=None, code=0,
message=None):
self.name = name
self.version = version
@@ -201,6 +228,7 @@ class ExtHandlerStatus(DataContract):
self.message = message
self.extensions = DataContractList(ustr)
+
class VMAgentStatus(DataContract):
def __init__(self, version=None, status=None, message=None):
self.version = version
@@ -208,27 +236,31 @@ class VMAgentStatus(DataContract):
self.message = message
self.extensionHandlers = DataContractList(ExtHandlerStatus)
+
class VMStatus(DataContract):
def __init__(self):
self.vmAgent = VMAgentStatus()
+
class TelemetryEventParam(DataContract):
def __init__(self, name=None, value=None):
self.name = name
self.value = value
+
class TelemetryEvent(DataContract):
def __init__(self, eventId=None, providerId=None):
self.eventId = eventId
self.providerId = providerId
self.parameters = DataContractList(TelemetryEventParam)
+
class TelemetryEventList(DataContract):
def __init__(self):
self.events = DataContractList(TelemetryEvent)
-class Protocol(DataContract):
+class Protocol(DataContract):
def detect(self):
raise NotImplementedError()
@@ -240,8 +272,8 @@ class Protocol(DataContract):
def get_vmagent_manifests(self):
raise NotImplementedError()
-
- def get_vmagent_pkgs(self):
+
+ def get_vmagent_pkgs(self, manifest):
raise NotImplementedError()
def get_ext_handlers(self):
@@ -250,13 +282,16 @@ class Protocol(DataContract):
def get_ext_handler_pkgs(self, extension):
raise NotImplementedError()
- def download_ext_handler_pkg(self, uri):
+ def get_artifacts_profile(self):
+ raise NotImplementedError()
+
+ def download_ext_handler_pkg(self, uri, headers=None):
try:
- resp = restutil.http_get(uri, chk_proxy=True)
+ resp = restutil.http_get(uri, chk_proxy=True, headers=headers)
if resp.status == restutil.httpclient.OK:
return resp.read()
- except HttpError as e:
- raise ProtocolError("Failed to download from: {0}".format(uri), e)
+ except Exception as e:
+ logger.warn("Failed to download from: {0}".format(uri), e)
def report_provision_status(self, provision_status):
raise NotImplementedError()
@@ -269,4 +304,3 @@ class Protocol(DataContract):
def report_event(self, event):
raise NotImplementedError()
-
diff --git a/azurelinuxagent/common/protocol/wire.py b/azurelinuxagent/common/protocol/wire.py
index 29a1663..9f634e9 100644
--- a/azurelinuxagent/common/protocol/wire.py
+++ b/azurelinuxagent/common/protocol/wire.py
@@ -16,14 +16,18 @@
#
# Requires Python 2.4+ and Openssl 1.0+
+import json
+import os
+import re
import time
import xml.sax.saxutils as saxutils
import azurelinuxagent.common.conf as conf
from azurelinuxagent.common.exception import ProtocolNotFoundError
from azurelinuxagent.common.future import httpclient, bytebuffer
-from azurelinuxagent.common.utils.textutil import parse_doc, findall, find, findtext, \
- getattrib, gettext, remove_bom, get_bytes_from_pem
+from azurelinuxagent.common.utils.textutil import parse_doc, findall, find, \
+ findtext, getattrib, gettext, remove_bom, get_bytes_from_pem, parse_json
import azurelinuxagent.common.utils.fileutil as fileutil
+import azurelinuxagent.common.utils.textutil as textutil
from azurelinuxagent.common.utils.cryptutil import CryptUtil
from azurelinuxagent.common.protocol.restapi import *
from azurelinuxagent.common.protocol.hostplugin import HostPluginProtocol
@@ -66,10 +70,13 @@ class WireProtocol(Protocol):
"""Slim layer to adapt wire protocol data to metadata protocol interface"""
# TODO: Clean-up goal state processing
- # At present, some methods magically update GoalState (e.g., get_vmagent_manifests), others (e.g., get_vmagent_pkgs)
- # assume its presence. A better approach would make an explicit update call that returns the incarnation number and
- # establishes that number the "context" for all other calls (either by updating the internal state of the protocol or
- # by having callers pass the incarnation number to the method).
+ # At present, some methods magically update GoalState (e.g.,
+ # get_vmagent_manifests), others (e.g., get_vmagent_pkgs)
+ # assume its presence. A better approach would make an explicit update
+ # call that returns the incarnation number and
+ # establishes that number the "context" for all other calls (either by
+ # updating the internal state of the protocol or
+ # by having callers pass the incarnation number to the method).
def __init__(self, endpoint):
if endpoint is None:
@@ -133,6 +140,22 @@ class WireProtocol(Protocol):
man = self.client.get_ext_manifest(ext_handler, goal_state)
return man.pkg_list
+ def get_artifacts_profile(self):
+ logger.verbose("Get In-VM Artifacts Profile")
+ return self.client.get_artifacts_profile()
+
+ def download_ext_handler_pkg(self, uri, headers=None):
+ package = super(WireProtocol, self).download_ext_handler_pkg(uri)
+
+ if package is not None:
+ return package
+ else:
+ logger.warn("Download did not succeed, falling back to host plugin")
+ host = self.client.get_host_plugin()
+ uri, headers = host.get_artifact_request(uri, host.manifest_uri)
+ package = super(WireProtocol, self).download_ext_handler_pkg(uri, headers=headers)
+ return package
+
def report_provision_status(self, provision_status):
validate_param("provision_status", provision_status, ProvisionStatus)
@@ -269,7 +292,7 @@ def ext_status_to_v1(ext_name, ext_status):
"timestampUTC": timestamp
}
if len(v1_sub_status) != 0:
- v1_ext_status['substatus'] = v1_sub_status
+ v1_ext_status['status']['substatus'] = v1_sub_status
return v1_ext_status
@@ -348,8 +371,8 @@ class StatusBlob(object):
# TODO upload extension only if content has changed
logger.verbose("Upload status blob")
upload_successful = False
- self.type = self.get_blob_type(url)
self.data = self.to_json()
+ self.type = self.get_blob_type(url)
try:
if self.type == "BlockBlob":
self.put_block_blob(url, self.data)
@@ -365,41 +388,45 @@ class StatusBlob(object):
return upload_successful
def get_blob_type(self, url):
- # Check blob type
- logger.verbose("Check blob type.")
+ logger.verbose("Get blob type")
timestamp = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
try:
- resp = self.client.call_storage_service(restutil.http_head, url, {
- "x-ms-date": timestamp,
- 'x-ms-version': self.__class__.__storage_version__
- })
+ resp = self.client.call_storage_service(
+ restutil.http_head,
+ url,
+ {
+ "x-ms-date": timestamp,
+ "x-ms-version": self.__class__.__storage_version__
+ })
except HttpError as e:
- raise ProtocolError((u"Failed to get status blob type: {0}"
- u"").format(e))
+ raise ProtocolError("Failed to get status blob type: {0}", e)
+
if resp is None or resp.status != httpclient.OK:
- raise ProtocolError(("Failed to get status blob type: {0}"
- "").format(resp.status))
+ raise ProtocolError("Failed to get status blob type")
blob_type = resp.getheader("x-ms-blob-type")
- logger.verbose("Blob type={0}".format(blob_type))
+ logger.verbose("Blob type: [{0}]", blob_type)
return blob_type
def put_block_blob(self, url, data):
- logger.verbose("Upload block blob")
+ 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__
- })
+ 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__
+ })
if resp.status != httpclient.CREATED:
raise UploadError(
"Failed to upload block blob: {0}".format(resp.status))
def put_page_blob(self, url, data):
- logger.verbose("Replace old page blob")
+ logger.verbose("Put page blob")
# Convert string into bytes
data = bytearray(data, encoding='utf-8')
@@ -407,14 +434,17 @@ class StatusBlob(object):
# 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__
- })
+ 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__
+ })
if resp.status != httpclient.CREATED:
raise UploadError(
"Failed to clean up page blob: {0}".format(resp.status))
@@ -437,7 +467,9 @@ class StatusBlob(object):
buf = bytearray(buf_size)
buf[0: content_size] = data[start: end]
resp = self.client.call_storage_service(
- restutil.http_put, url, bytebuffer(buf),
+ restutil.http_put,
+ url,
+ bytebuffer(buf),
{
"x-ms-date": timestamp,
"x-ms-range": "bytes={0}-{1}".format(start, page_end - 1),
@@ -465,7 +497,8 @@ def event_param_to_v1(param):
attr_type = 'mt:bool'
elif param_type is float:
attr_type = 'mt:float64'
- return param_format.format(param.name, saxutils.quoteattr(ustr(param.value)),
+ return param_format.format(param.name,
+ saxutils.quoteattr(ustr(param.value)),
attr_type)
@@ -491,8 +524,9 @@ class WireClient(object):
self.ext_conf = None
self.last_request = 0
self.req_count = 0
+ self.host_plugin = None
self.status_blob = StatusBlob(self)
- self.host_plugin = HostPluginProtocol(self.endpoint)
+ self.status_blob_type_reported = False
def prevent_throttling(self):
"""
@@ -501,37 +535,42 @@ class WireClient(object):
now = time.time()
if now - self.last_request < 1:
logger.verbose("Last request issued less than 1 second ago")
- logger.verbose("Sleep {0} second to avoid throttling.",
- SHORT_WAITING_INTERVAL)
+ logger.verbose("Sleep {0} second to avoid throttling.",
+ SHORT_WAITING_INTERVAL)
time.sleep(SHORT_WAITING_INTERVAL)
self.last_request = now
self.req_count += 1
if self.req_count % 3 == 0:
- logger.verbose("Sleep {0} second to avoid throttling.",
- SHORT_WAITING_INTERVAL)
+ logger.verbose("Sleep {0} second to avoid throttling.",
+ SHORT_WAITING_INTERVAL)
time.sleep(SHORT_WAITING_INTERVAL)
self.req_count = 0
def call_wireserver(self, http_req, *args, **kwargs):
"""
- Call wire server. Handle throttling(403) and Resource Gone(410)
+ Call wire server; handle throttling (403), resource gone (410) and
+ service unavailable (503).
"""
self.prevent_throttling()
for retry in range(0, 3):
resp = http_req(*args, **kwargs)
if resp.status == httpclient.FORBIDDEN:
- logger.warn("Sending too much request to wire server")
- logger.info("Sleep {0} second to avoid throttling.",
+ 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))
+ raise ProtocolError(("Calling wire server failed: "
+ "{0}").format(resp.status))
def decode_config(self, data):
if data is None:
@@ -542,12 +581,13 @@ class WireClient(object):
def fetch_config(self, uri, headers):
try:
- resp = self.call_wireserver(restutil.http_get, uri,
+ resp = self.call_wireserver(restutil.http_get,
+ uri,
headers=headers)
except HttpError as e:
raise ProtocolError(ustr(e))
- if (resp.status != httpclient.OK):
+ if resp.status != httpclient.OK:
raise ProtocolError("{0} - {1}".format(resp.status, uri))
return self.decode_config(resp.read())
@@ -566,41 +606,65 @@ class WireClient(object):
except IOError as e:
raise ProtocolError("Failed to write cache: {0}".format(e))
- def call_storage_service(self, http_req, *args, **kwargs):
+ @staticmethod
+ def call_storage_service(http_req, *args, **kwargs):
"""
Call storage service, handle SERVICE_UNAVAILABLE(503)
"""
+
+ # force the chk_proxy arg to True, since all calls to storage should
+ # use a configured proxy
+ kwargs['chk_proxy'] = True
+
for retry in range(0, 3):
resp = http_req(*args, **kwargs)
if resp.status == httpclient.SERVICE_UNAVAILABLE:
- logger.warn("Storage service is not avaible temporaryly")
- logger.info("Will retry later, in {0} seconds",
+ 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))
+ raise ProtocolError(("Calling storage endpoint failed: "
+ "{0}").format(resp.status))
def fetch_manifest(self, version_uris):
- for version_uri in version_uris:
- logger.verbose("Fetch ext handler manifest: {0}", version_uri.uri)
- try:
- resp = self.call_storage_service(restutil.http_get,
- version_uri.uri, None,
- chk_proxy=True)
- except HttpError as e:
- raise ProtocolError(ustr(e))
-
+ logger.verbose("Fetch manifest")
+ for version in version_uris:
+ response = self.fetch(version.uri)
+ if not response:
+ 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)
+ if not response:
+ logger.info("Retry fetch in {0} seconds",
+ LONG_WAITING_INTERVAL)
+ time.sleep(LONG_WAITING_INTERVAL)
+ else:
+ host.manifest_uri = version.uri
+ logger.verbose("Manifest downloaded successfully from host plugin")
+ if response:
+ return response
+ raise ProtocolError("Failed to fetch manifest from all sources")
+
+ def fetch(self, uri, headers=None):
+ logger.verbose("Fetch [{0}] with headers [{1}]", uri, headers)
+ return_value = None
+ try:
+ resp = self.call_storage_service(
+ restutil.http_get,
+ uri,
+ headers)
if resp.status == httpclient.OK:
- return self.decode_config(resp.read())
- logger.warn("Failed to fetch ExtensionManifest: {0}, {1}",
- resp.status, version_uri.uri)
- logger.info("Will retry later, in {0} seconds",
- LONG_WAITING_INTERVAL)
- time.sleep(LONG_WAITING_INTERVAL)
- raise ProtocolError(("Failed to fetch ExtensionManifest from "
- "all sources"))
+ return_value = self.decode_config(resp.read())
+ else:
+ logger.warn("Could not fetch {0} [{1}]",
+ uri,
+ resp.status)
+ except (HttpError, ProtocolError) as e:
+ logger.verbose("Fetch failed from [{0}]", uri)
+ return return_value
def update_hosting_env(self, goal_state):
if goal_state.hosting_env_uri is None:
@@ -640,6 +704,7 @@ class WireClient(object):
xml_text = self.fetch_config(goal_state.ext_uri, self.get_header())
self.save_cache(local_file, xml_text)
self.ext_conf = ExtensionsConfig(xml_text)
+ self.status_blob_type_reported = False
def update_goal_state(self, forced=False, max_retry=3):
uri = GOAL_STATE_URI.format(self.endpoint)
@@ -671,6 +736,9 @@ class WireClient(object):
self.update_shared_conf(goal_state)
self.update_certs(goal_state)
self.update_ext_conf(goal_state)
+ if self.host_plugin is not None:
+ self.host_plugin.container_id = goal_state.container_id
+ self.host_plugin.role_config_name = goal_state.role_config_name
return
except WireProtocolResourceGone:
logger.info("Incarnation is out of date. Update goalstate.")
@@ -680,7 +748,7 @@ class WireClient(object):
raise ProtocolError("Exceeded max retry updating goal state")
def get_goal_state(self):
- if (self.goal_state is None):
+ if self.goal_state is None:
incarnation_file = os.path.join(conf.get_lib_dir(),
INCARNATION_FILE_NAME)
incarnation = self.fetch_cache(incarnation_file)
@@ -693,14 +761,16 @@ class WireClient(object):
def get_hosting_env(self):
if (self.hosting_env is None):
- local_file = os.path.join(conf.get_lib_dir(), HOSTING_ENV_FILE_NAME)
+ local_file = os.path.join(conf.get_lib_dir(),
+ HOSTING_ENV_FILE_NAME)
xml_text = self.fetch_cache(local_file)
self.hosting_env = HostingEnv(xml_text)
return self.hosting_env
def get_shared_conf(self):
if (self.shared_conf is None):
- local_file = os.path.join(conf.get_lib_dir(), SHARED_CONF_FILE_NAME)
+ local_file = os.path.join(conf.get_lib_dir(),
+ SHARED_CONF_FILE_NAME)
xml_text = self.fetch_cache(local_file)
self.shared_conf = SharedConfig(xml_text)
return self.shared_conf
@@ -715,7 +785,7 @@ class WireClient(object):
return self.certs
def get_ext_conf(self):
- if (self.ext_conf is None):
+ if self.ext_conf is None:
goal_state = self.get_goal_state()
if goal_state.ext_uri is None:
self.ext_conf = ExtensionsConfig(None)
@@ -724,6 +794,7 @@ class WireClient(object):
local_file = os.path.join(conf.get_lib_dir(), local_file)
xml_text = self.fetch_cache(local_file)
self.ext_conf = ExtensionsConfig(xml_text)
+ self.status_blob_type_reported = False
return self.ext_conf
def get_ext_manifest(self, ext_handler, goal_state):
@@ -761,9 +832,42 @@ class WireClient(object):
def upload_status_blob(self):
ext_conf = self.get_ext_conf()
if ext_conf.status_upload_blob is not None:
- if not self.status_blob.upload(ext_conf.status_upload_blob):
- self.host_plugin.put_vm_status(self.status_blob,
- ext_conf.status_upload_blob)
+ 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:
+ # errors have already been logged
+ 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)
+
+ """
+ Emit an event to determine if the type in the extension config
+ matches the actual type from the HTTP HEAD request.
+ """
+ def report_blob_type(self, head_type, config_type):
+ if head_type and config_type:
+ is_match = head_type == config_type
+ if self.status_blob_type_reported is False:
+ message = \
+ 'Blob type match [{0}]'.format(head_type) if is_match else \
+ 'Blob type mismatch [HEAD {0}], [CONFIG {1}]'.format(
+ head_type,
+ config_type)
+
+ from azurelinuxagent.common.event import add_event, WALAEventOperation
+ from azurelinuxagent.common.version import AGENT_NAME, CURRENT_VERSION
+ add_event(AGENT_NAME,
+ version=CURRENT_VERSION,
+ is_success=is_match,
+ message=message,
+ op=WALAEventOperation.HealthCheck)
+ self.status_blob_type_reported = True
def report_role_prop(self, thumbprint):
goal_state = self.get_goal_state()
@@ -774,14 +878,17 @@ class WireClient(object):
role_prop_uri = ROLE_PROP_URI.format(self.endpoint)
headers = self.get_header_for_xml_content()
try:
- resp = self.call_wireserver(restutil.http_post, role_prop_uri,
- role_prop, headers=headers)
+ resp = self.call_wireserver(restutil.http_post,
+ role_prop_uri,
+ role_prop,
+ headers=headers)
except HttpError as e:
- raise ProtocolError((u"Failed to send role properties: {0}"
- u"").format(e))
+ raise ProtocolError((u"Failed to send role properties: "
+ u"{0}").format(e))
if resp.status != httpclient.ACCEPTED:
- raise ProtocolError((u"Failed to send role properties: {0}"
- u", {1}").format(resp.status, resp.read()))
+ raise ProtocolError((u"Failed to send role properties: "
+ u",{0}: {1}").format(resp.status,
+ resp.read()))
def report_health(self, status, substatus, description):
goal_state = self.get_goal_state()
@@ -795,14 +902,21 @@ class WireClient(object):
health_report_uri = HEALTH_REPORT_URI.format(self.endpoint)
headers = self.get_header_for_xml_content()
try:
- resp = self.call_wireserver(restutil.http_post, health_report_uri,
- health_report, headers=headers, max_retry=8)
+ # 30 retries with 10s sleep gives ~5min for wireserver updates;
+ # this is retried 3 times with 15s sleep before throwing a
+ # ProtocolError, for a total of ~15min.
+ resp = self.call_wireserver(restutil.http_post,
+ health_report_uri,
+ health_report,
+ headers=headers,
+ max_retry=30)
except HttpError as e:
- raise ProtocolError((u"Failed to send provision status: {0}"
- u"").format(e))
+ raise ProtocolError((u"Failed to send provision status: "
+ u"{0}").format(e))
if resp.status != httpclient.OK:
- raise ProtocolError((u"Failed to send provision status: {0}"
- u", {1}").format(resp.status, resp.read()))
+ raise ProtocolError((u"Failed to send provision status: "
+ u",{0}: {1}").format(resp.status,
+ resp.read()))
def send_event(self, provider_id, event_str):
uri = TELEMETRY_URI.format(self.endpoint)
@@ -820,7 +934,8 @@ class WireClient(object):
if resp.status != httpclient.OK:
logger.verbose(resp.read())
- raise ProtocolError("Failed to send events:{0}".format(resp.status))
+ raise ProtocolError(
+ "Failed to send events:{0}".format(resp.status))
def report_event(self, event_list):
buf = {}
@@ -867,6 +982,38 @@ class WireClient(object):
"x-ms-guest-agent-public-x509-cert": cert
}
+ 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,
+ goal_state.role_config_name)
+ return self.host_plugin
+
+ def has_artifacts_profile_blob(self):
+ return self.ext_conf and not \
+ textutil.is_str_none_or_whitespace(self.ext_conf.artifacts_profile_blob)
+
+ def get_artifacts_profile(self):
+ artifacts_profile = None
+ if self.has_artifacts_profile_blob():
+ blob = self.ext_conf.artifacts_profile_blob
+ logger.info("Getting the artifacts profile")
+ profile = self.fetch(blob)
+
+ 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))
+
+ if not textutil.is_str_none_or_whitespace(profile):
+ logger.info("Artifacts profile downloaded successfully")
+ artifacts_profile = InVMArtifactsProfile(profile)
+
+ return artifacts_profile
+
+
class VersionInfo(object):
def __init__(self, xml_text):
"""
@@ -880,14 +1027,16 @@ class VersionInfo(object):
xml_doc = parse_doc(xml_text)
preferred = find(xml_doc, "Preferred")
self.preferred = findtext(preferred, "Version")
- logger.info("Fabric preferred wire protocol version:{0}", self.preferred)
+ logger.info("Fabric preferred wire protocol version:{0}",
+ self.preferred)
self.supported = []
supported = find(xml_doc, "Supported")
supported_version = findall(supported, "Version")
for node in supported_version:
version = gettext(node)
- logger.verbose("Fabric supported wire protocol version:{0}", version)
+ logger.verbose("Fabric supported wire protocol version:{0}",
+ version)
self.supported.append(version)
def get_preferred(self):
@@ -909,8 +1058,10 @@ class GoalState(object):
self.certs_uri = None
self.ext_uri = None
self.role_instance_id = None
+ self.role_config_name = None
self.container_id = None
self.load_balancer_probe_port = None
+ self.xml_text = None
self.parse(xml_text)
def parse(self, xml_text):
@@ -927,6 +1078,8 @@ class GoalState(object):
self.ext_uri = findtext(xml_doc, "ExtensionsConfig")
role_instance = find(xml_doc, "RoleInstance")
self.role_instance_id = findtext(role_instance, "InstanceId")
+ role_config = find(role_instance, "Configuration")
+ self.role_config_name = findtext(role_config, "ConfigName")
container = find(xml_doc, "Container")
self.container_id = findtext(container, "ContainerId")
lbprobe_ports = find(xml_doc, "LBProbePorts")
@@ -947,6 +1100,7 @@ class HostingEnv(object):
self.vm_name = None
self.role_name = None
self.deployment_name = None
+ self.xml_text = None
self.parse(xml_text)
def parse(self, xml_text):
@@ -980,6 +1134,7 @@ class SharedConfig(object):
# Not used currently
return self
+
class Certificates(object):
"""
Object containing certificates of host and provisioned user.
@@ -1089,6 +1244,8 @@ class ExtensionsConfig(object):
self.ext_handlers = ExtHandlerList()
self.vmagent_manifests = VMAgentManifestList()
self.status_upload_blob = None
+ self.status_upload_blob_type = None
+ self.artifacts_profile_blob = None
if xml_text is not None:
self.parse(xml_text)
@@ -1123,6 +1280,13 @@ class ExtensionsConfig(object):
self.parse_plugin_settings(ext_handler, plugin_settings)
self.status_upload_blob = findtext(xml_doc, "StatusUploadBlob")
+ self.artifacts_profile_blob = findtext(xml_doc, "InVMArtifactsProfileBlob")
+
+ status_upload_node = find(xml_doc, "StatusUploadBlob")
+ self.status_upload_blob_type = getattrib(status_upload_node,
+ "statusBlobType")
+ logger.verbose("Extension config shows status blob type as [{0}]",
+ self.status_upload_blob_type)
def parse_plugin(self, plugin):
ext_handler = ExtHandler()
@@ -1176,7 +1340,8 @@ class ExtensionsConfig(object):
ext.sequenceNumber = seqNo
ext.publicSettings = handler_settings.get("publicSettings")
ext.protectedSettings = handler_settings.get("protectedSettings")
- thumbprint = handler_settings.get("protectedSettingsCertThumbprint")
+ thumbprint = handler_settings.get(
+ "protectedSettingsCertThumbprint")
ext.certificateThumbprint = thumbprint
ext_handler.properties.extensions.append(ext)
@@ -1191,14 +1356,21 @@ class ExtensionManifest(object):
def parse(self, xml_text):
xml_doc = parse_doc(xml_text)
- self._handle_packages(findall(find(xml_doc, "Plugins"), "Plugin"), False)
- self._handle_packages(findall(find(xml_doc, "InternalPlugins"), "Plugin"), True)
+ self._handle_packages(findall(find(xml_doc,
+ "Plugins"),
+ "Plugin"),
+ False)
+ self._handle_packages(findall(find(xml_doc,
+ "InternalPlugins"),
+ "Plugin"),
+ True)
def _handle_packages(self, packages, isinternal):
for package in packages:
version = findtext(package, "Version")
- disallow_major_upgrade = findtext(package, "DisallowMajorVersionUpgrade")
+ disallow_major_upgrade = findtext(package,
+ "DisallowMajorVersionUpgrade")
if disallow_major_upgrade is None:
disallow_major_upgrade = ''
disallow_major_upgrade = disallow_major_upgrade.lower() == "true"
@@ -1216,3 +1388,27 @@ class ExtensionManifest(object):
pkg.isinternal = isinternal
self.pkg_list.versions.append(pkg)
+
+
+# Do not extend this class
+class InVMArtifactsProfile(object):
+ """
+ deserialized json string of InVMArtifactsProfile.
+ It is expected to contain the following fields:
+ * inVMArtifactsProfileBlobSeqNo
+ * profileId (optional)
+ * onHold (optional)
+ * certificateThumbprint (optional)
+ * encryptedHealthChecks (optional)
+ * encryptedApplicationProfile (optional)
+ """
+
+ def __init__(self, artifacts_profile):
+ if not textutil.is_str_none_or_whitespace(artifacts_profile):
+ self.__dict__.update(parse_json(artifacts_profile))
+
+ def is_on_hold(self):
+ # hasattr() is not available in Python 2.6
+ if 'onHold' in self.__dict__:
+ return self.onHold.lower() == 'true'
+ return False
generated by cgit v1.2.3 (git 2.39.1) at 2025-07-04 12:31:13 +0000