fix(cloudstack): Improve domain-name DHCP lease lookup (Cloudstack) (#6554)

* Adding case-insensitive options for systemd-networkd leases ("DOMAINNAME", "Domain", "domain-name").
* Falling back gracefully from systemd leases to ISC dhclient leases.
* Including dhcpcd ephemeral leases as an additional fallback.
* Returning an empty string when no domain name found instead of None for non-fatal missing cases.

Author: CodeBleu

cloudinit/sources/DataSourceCloudStack.py

@@ -87,46 +87,47 @@ def __init__(self, sys_cfg, distro, paths):
         self.vr_addr = None
 
     def _get_domainname(self):
+        """Try obtaining a "domain-name" DHCP lease parameter:
+        - From systemd-networkd lease (case-insensitive)
+        - From ISC dhclient
+        - From dhcpcd (ephemeral)
+        - Return empty string if not found (non-fatal)
         """
-        Try obtaining a "domain-name" DHCP lease parameter:
-        - From systemd-networkd lease
-        - From dhclient lease
-        """
+
         LOG.debug("Try obtaining domain name from networkd leases")
-        domainname = dhcp.networkd_get_option_from_leases("DOMAINNAME")
-        if domainname:
-            return domainname
+        for key in ["DOMAINNAME", "Domain", "domain-name"]:
+            domainname = dhcp.networkd_get_option_from_leases(key)
+            if domainname:
+                return domainname.strip()
+
         LOG.debug(
-            "Could not obtain FQDN from networkd leases. "
-            "Falling back to ISC dhclient"
+            "Could not obtain FQDN from networkd leases. Falling back to "
+            "ISC dhclient"
         )
-
-        # some distros might use isc-dhclient for network setup via their
-        # network manager. If this happens, the lease is more recent than the
-        # ephemeral lease, so use it first.
         with suppress(dhcp.NoDHCPLeaseMissingDhclientError):
             domain_name = dhcp.IscDhclient().get_key_from_latest_lease(
                 self.distro, "domain-name"
             )
             if domain_name:
-                return domain_name
+                return domain_name.strip()
 
         LOG.debug(
-            "Could not obtain FQDN from ISC dhclient leases. "
-            "Falling back to %s",
+            "Could not obtain FQDN from ISC dhclient leases. Falling back to "
+            "%s",
             self.distro.dhcp_client.client_name,
         )
-
-        # If no distro leases were found, check the ephemeral lease that
-        # cloud-init set up.
-        with suppress(FileNotFoundError):
+        try:
             latest_lease = self.distro.dhcp_client.get_newest_lease(
                 self.distro.fallback_interface
             )
-            domain_name = latest_lease.get("domain-name") or None
-            return domain_name
-        LOG.debug("No dhcp leases found")
-        return None
+            domain_name = latest_lease.get("domain-name")
+            if domain_name:
+                return domain_name.strip()
+        except (NoDHCPLeaseError, FileNotFoundError, AttributeError):
+            pass
+
+        LOG.debug("No domain name found in any DHCP lease; returning empty")
+        return ""
 
     def get_hostname(
         self,

tests/unittests/sources/test_cloudstack.py

@@ -2,6 +2,7 @@
 # pylint: disable=attribute-defined-outside-init
 from socket import gaierror
 from textwrap import dedent
+from unittest.mock import patch
 
 import pytest
 
@@ -220,6 +221,32 @@ def test_get_hostname_fqdn_fallback(self, cloudstack_ds, mocker):
             result = cloudstack_ds.get_hostname(fqdn=True)
             assert expected == result
 
+    @pytest.mark.parametrize(
+        "lease_key,expected_domain",
+        [
+            ("DOMAINNAME", "example.com"),
+            ("Domain", "example.com"),
+            ("domain-name", "example.com"),
+        ],
+    )
+    def test__get_domainname_supports_all_casing_variants(
+        self, lease_key, expected_domain
+    ):
+        """Ensure _get_domainname works with DOMAINNAME, Domain and
+        domain-name."""
+        # Mock the helper to return the domain only when the exact key is asked
+        with patch(
+            "cloudinit.net.dhcp.networkd_get_option_from_leases"
+        ) as m_get:
+            m_get.side_effect = lambda key, extra_keys=None: (
+                "example.com   " if key == lease_key else None
+            )
+
+            ds = DataSourceCloudStack(
+                {}, distro=MockDistro(), paths=helpers.Paths({})
+            )
+            assert ds._get_domainname() == expected_domain
+
 
 class TestGetDataServer:
     @pytest.mark.parametrize(