def get_dns_name_and_fallback_mount_target_ip_address()

in src/mount_efs/__init__.py [0:0]

• 84 lines of code
• 18 McCabe index (conditional complexity)

def get_dns_name_and_fallback_mount_target_ip_address(config, fs_id, options):
    def _validate_replacement_field_count(format_str, expected_ct):
        if format_str.count("{") != expected_ct or format_str.count("}") != expected_ct:
            raise ValueError(
                "DNS name format has an incorrect number of replacement fields"
            )

    # mounts using crossaccount have a predetermined dns name format
    if options and "crossaccount" in options:
        try:
            az_id = get_az_id_from_instance_metadata(config, options)
            region = get_target_region(config, options)
            dns_name_suffix = get_dns_name_suffix(config, region)
            dns_name = "%s.%s.efs.%s.%s" % (az_id, fs_id, region, dns_name_suffix)
        except RuntimeError:
            err_msg = "Cannot retrieve AZ-ID from metadata service. This is required for the crossaccount mount option."
            fatal_error(err_msg)
    else:
        dns_name_format = config.get(CONFIG_SECTION, "dns_name_format")

        if "{fs_id}" not in dns_name_format:
            raise ValueError("DNS name format must include {fs_id}")

        format_args = {"fs_id": fs_id}

        expected_replacement_field_ct = 1

        if "{az}" in dns_name_format:
            az = options.get("az")
            if az:
                expected_replacement_field_ct += 1
                format_args["az"] = az
            else:
                dns_name_format = dns_name_format.replace("{az}.", "")

        region = None
        if "{region}" in dns_name_format:
            region = get_target_region(config, options)
            expected_replacement_field_ct += 1
            format_args["region"] = region

        if "{dns_name_suffix}" in dns_name_format:
            expected_replacement_field_ct += 1
            region = region or get_target_region(config, options)
            dns_name_suffix = get_dns_name_suffix(config, region)
            format_args["dns_name_suffix"] = dns_name_suffix
            logging.debug("Using dns_name_suffix %s", dns_name_suffix)

        _validate_replacement_field_count(
            dns_name_format, expected_replacement_field_ct
        )
        dns_name = dns_name_format.format(**format_args)

    if "mounttargetip" in options:
        if "crossaccount" in options:
            fatal_error(
                "mounttargetip option is incompatible with crossaccount option."
            )
        ip_address = options.get("mounttargetip")
        logging.info(
            "Use the mount target ip address %s provided in the mount options to mount."
            % ip_address
        )
        try:
            mount_target_ip_address_can_be_resolved(
                ip_address,
                passed_via_options=True,
                network_namespace=options.get("netns") if "netns" in options else None,
            )
            return dns_name, options.get("mounttargetip")
        except FallbackException as e:
            fallback_message = e.message
            throw_ip_address_connect_failure_with_fallback_message(
                ip_address=ip_address, fallback_message=fallback_message
            )

    if dns_name_can_be_resolved(dns_name):
        return dns_name, None

    logging.info(
        "Failed to resolve %s, attempting to lookup mount target ip address using botocore.",
        dns_name,
    )

    try:
        fallback_mount_target_ip_address = get_fallback_mount_target_ip_address(
            config, options, fs_id, dns_name
        )
        logging.info(
            "Found fall back mount target ip address %s for file system %s",
            fallback_mount_target_ip_address,
            fs_id,
        )
        return dns_name, fallback_mount_target_ip_address
    except FallbackException as e:
        fallback_message = e.message

    throw_dns_resolve_failure_with_fallback_message(dns_name, fallback_message)