ACIL FM

Nama	Ukuran	Aksi
azure	-	Open
helpers	-	Open
__pycache__	-	Open
DataSourceAkamai.py	12.66 MB	View DL Edit
DataSourceAliYun.py	15.21 MB	View DL Edit
DataSourceAltCloud.py	8.42 MB	View DL Edit
DataSourceAzure.py	75.51 MB	View DL Edit
DataSourceBigstep.py	1.9 MB	View DL Edit
DataSourceCloudCIX.py	5.19 MB	View DL Edit
DataSourceCloudSigma.py	3.86 MB	View DL Edit
DataSourceCloudStack.py	11.21 MB	View DL Edit
DataSourceConfigDrive.py	11.23 MB	View DL Edit
DataSourceDigitalOcean.py	4.2 MB	View DL Edit
DataSourceEc2.py	41.76 MB	View DL Edit
DataSourceExoscale.py	8.55 MB	View DL Edit
DataSourceGCE.py	13.33 MB	View DL Edit
DataSourceHetzner.py	5.39 MB	View DL Edit
DataSourceIBMCloud.py	14.65 MB	View DL Edit
DataSourceLXD.py	17.2 MB	View DL Edit
DataSourceMAAS.py	14.85 MB	View DL Edit
DataSourceNoCloud.py	15.98 MB	View DL Edit
DataSourceNone.py	1.27 MB	View DL Edit
DataSourceNWCS.py	4.41 MB	View DL Edit
DataSourceOpenNebula.py	15.67 MB	View DL Edit
DataSourceOpenStack.py	10.1 MB	View DL Edit
DataSourceOracle.py	21.07 MB	View DL Edit
DataSourceOVF.py	12.7 MB	View DL Edit
DataSourceRbxCloud.py	7.88 MB	View DL Edit
DataSourceScaleway.py	14.73 MB	View DL Edit
DataSourceSmartOS.py	34.26 MB	View DL Edit
DataSourceUpCloud.py	5.2 MB	View DL Edit
DataSourceVMware.py	35.31 MB	View DL Edit
DataSourceVultr.py	4.51 MB	View DL Edit
DataSourceWSL.py	14.37 MB	View DL Edit
__init__.py	43.63 MB	View DL Edit

Edit file: /usr/lib/python3.9/site-packages/cloudinit/sources/DataSourceHetzner.py

# Author: Jonas Keidel <jonas.keidel@hetzner.com>
# Author: Markus Schade <markus.schade@hetzner.com>
#
# This file is part of cloud-init. See LICENSE file for license information.
#
"""Hetzner Cloud API Documentation
https://docs.hetzner.cloud/"""

import logging

import cloudinit.sources.helpers.hetzner as hc_helper
from cloudinit import dmi, net, sources, util
from cloudinit.net.dhcp import NoDHCPLeaseError
from cloudinit.net.ephemeral import EphemeralDHCPv4

LOG = logging.getLogger(__name__)

BASE_URL_V1 = "http://169.254.169.254/hetzner/v1"

BUILTIN_DS_CONFIG = {
    "metadata_url": BASE_URL_V1 + "/metadata",
    "userdata_url": BASE_URL_V1 + "/userdata",
}

MD_RETRIES = 60
MD_TIMEOUT = 2
MD_WAIT_RETRY = 2

class DataSourceHetzner(sources.DataSource):

dsname = "Hetzner"

def __init__(self, sys_cfg, distro, paths):
        sources.DataSource.__init__(self, sys_cfg, distro, paths)
        self.distro = distro
        self.metadata = dict()
        self.ds_cfg = util.mergemanydict(
            [
                util.get_cfg_by_path(sys_cfg, ["datasource", "Hetzner"], {}),
                BUILTIN_DS_CONFIG,
            ]
        )
        self.metadata_address = self.ds_cfg["metadata_url"]
        self.userdata_address = self.ds_cfg["userdata_url"]
        self.retries = self.ds_cfg.get("retries", MD_RETRIES)
        self.timeout = self.ds_cfg.get("timeout", MD_TIMEOUT)
        self.wait_retry = self.ds_cfg.get("wait_retry", MD_WAIT_RETRY)
        self._network_config = sources.UNSET
        self.dsmode = sources.DSMODE_NETWORK
        self.metadata_full = None

def _get_data(self):
        (on_hetzner, serial) = get_hcloud_data()

if not on_hetzner:
            return False

try:
            with EphemeralDHCPv4(
                self.distro,
                iface=net.find_fallback_nic(),
                connectivity_urls_data=[
                    {
                        "url": BASE_URL_V1 + "/metadata/instance-id",
                    }
                ],
            ):
                md = hc_helper.read_metadata(
                    self.metadata_address,
                    timeout=self.timeout,
                    sec_between=self.wait_retry,
                    retries=self.retries,
                )
                ud = hc_helper.read_userdata(
                    self.userdata_address,
                    timeout=self.timeout,
                    sec_between=self.wait_retry,
                    retries=self.retries,
                )
        except NoDHCPLeaseError as e:
            LOG.error("Bailing, DHCP Exception: %s", e)
            raise

# Hetzner cloud does not support binary user-data. So here, do a
        # base64 decode of the data if we can. The end result being that a
        # user can provide base64 encoded (possibly gzipped) data as user-data.
        #
        # The fallout is that in the event of b64 encoded user-data,
        # /var/lib/cloud-init/cloud-config.txt will not be identical to the
        # user-data provided.  It will be decoded.
        self.userdata_raw = util.maybe_b64decode(ud)
        self.metadata_full = md

# hostname is name provided by user at launch.  The API enforces it is
        # a valid hostname, but it is not guaranteed to be resolvable in dns or
        # fully qualified.
        self.metadata["instance-id"] = md["instance-id"]
        self.metadata["local-hostname"] = md["hostname"]
        self.metadata["network-config"] = md.get("network-config", None)
        self.metadata["public-keys"] = md.get("public-keys", None)
        self.vendordata_raw = md.get("vendor_data", None)

# instance-id and serial from SMBIOS should be identical
        if self.get_instance_id() != serial:
            raise RuntimeError(
                "SMBIOS serial does not match instance ID from metadata"
            )

return True

def check_instance_id(self, sys_cfg):
        return sources.instance_id_matches_system_uuid(
            self.get_instance_id(), "system-serial-number"
        )

@property
    def network_config(self):
        """Configure the networking. This needs to be done each boot, since
        the IP information may have changed due to snapshot and/or
        migration.
        """

if self._network_config is None:
            LOG.warning(
                "Found None as cached _network_config. Resetting to %s",
                sources.UNSET,
            )
            self._network_config = sources.UNSET

if self._network_config != sources.UNSET:
            return self._network_config

_net_config = self.metadata["network-config"]
        if not _net_config:
            raise RuntimeError("Unable to get meta-data from server....")

self._network_config = _net_config

return self._network_config

def get_hcloud_data():
    vendor_name = dmi.read_dmi_data("system-manufacturer")
    if vendor_name != "Hetzner":
        return (False, None)

serial = dmi.read_dmi_data("system-serial-number")
    if serial:
        LOG.debug("Running on Hetzner Cloud: serial=%s", serial)
    else:
        raise RuntimeError("Hetzner Cloud detected, but no serial found")

return (True, serial)

# Used to match classes to dependencies
datasources = [
    (DataSourceHetzner, (sources.DEP_FILESYSTEM,)),
]

# Return a list of data sources that match this set of dependencies
def get_datasource_list(depends):
    return sources.list_from_depends(depends, datasources)

Batal