| Server IP : 74.208.127.88 / Your IP : 216.73.216.199 Web Server : Apache/2.4.41 (Ubuntu) System : Linux ubuntu 5.4.0-163-generic #180-Ubuntu SMP Tue Sep 5 13:21:23 UTC 2023 x86_64 User : www-data ( 33) PHP Version : 7.4.3-4ubuntu2.29 Disable Function : pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wifcontinued,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_get_handler,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,pcntl_async_signals,pcntl_unshare, MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : OFF | Sudo : ON | Pkexec : ON Directory : /snap/core20/2501/lib/python3/dist-packages/cloudinit/sources/azure/ |
Upload File : |
# Copyright (C) 2022 Microsoft Corporation.
#
# This file is part of cloud-init. See LICENSE file for license information.
import logging
from datetime import datetime, timezone
from typing import Optional
from cloudinit import version
from cloudinit.reporting import handlers, instantiated_handler_registry
from cloudinit.sources.azure import errors, identity
LOG = logging.getLogger(__name__)
def get_kvp_handler() -> Optional[handlers.HyperVKvpReportingHandler]:
"""Get instantiated KVP telemetry handler."""
kvp_handler = instantiated_handler_registry.registered_items.get(
"telemetry"
)
if not isinstance(kvp_handler, handlers.HyperVKvpReportingHandler):
return None
return kvp_handler
def report_via_kvp(report: str) -> bool:
"""Report to host via PROVISIONING_REPORT KVP key."""
kvp_handler = get_kvp_handler()
if kvp_handler is None:
LOG.debug("KVP handler not enabled, skipping host report.")
return False
kvp_handler.write_key("PROVISIONING_REPORT", report)
return True
def report_failure_to_host(error: errors.ReportableError) -> bool:
return report_via_kvp(error.as_encoded_report())
def report_success_to_host() -> bool:
try:
vm_id = identity.query_vm_id()
except Exception as id_error:
vm_id = f"failed to read vm id: {id_error!r}"
report = errors.encode_report(
[
"result=success",
f"agent=Cloud-Init/{version.version_string()}",
f"timestamp={datetime.now(timezone.utc).isoformat()}",
f"vm_id={vm_id}",
]
)
return report_via_kvp(report)