Fix: GCE _get_data crashes if DHCP lease fails

This commit addresses issue #5997 which reported crashes in init-local
when cloud-init was examining GCELocal as a potential datasource. When
all NICs failed at DHCP discovery cloud-init attempts to log the events
by dereferencing a value that was never assigned.

This commit modifies the _get_data function of DataSourceGCE.py by
adding an empty dictionary definition for the ret variable at the
top level of the function and some debugging logs when a candidate NIC
fails to obtain a DHCP lease. At the same time, the commit replaces the
direct key access operator on ret with the safe lookup method get(). This
commit also adds a unit test that mocks the observed situation

cloudinit/sources/DataSourceGCE.py

@@ -87,6 +87,7 @@ def __init__(self, sys_cfg, distro, paths):
 
     def _get_data(self):
         url_params = self.get_url_params()
+        ret = {}
         if self.perform_dhcp_setup:
             candidate_nics = net.find_candidate_nics()
             if DEFAULT_PRIMARY_INTERFACE in candidate_nics:
@@ -116,6 +117,9 @@ def _get_data(self):
                             )
                             continue
                 except NoDHCPLeaseError:
+                    LOG.debug(
+                        "Unable to obtain a DHCP lease for %s", candidate_nic
+                    )
                     continue
                 if ret["success"]:
                     self.distro.fallback_interface = candidate_nic
@@ -128,14 +132,14 @@ def _get_data(self):
         else:
             ret = read_md(address=self.metadata_address, url_params=url_params)
 
-        if not ret["success"]:
-            if ret["platform_reports_gce"]:
-                LOG.warning(ret["reason"])
+        if not ret.get("success"):
+            if ret.get("platform_reports_gce"):
+                LOG.warning(ret.get("reason"))
             else:
-                LOG.debug(ret["reason"])
+                LOG.debug(ret.get("reason"))
             return False
-        self.metadata = ret["meta-data"]
-        self.userdata_raw = ret["user-data"]
+        self.metadata = ret.get("meta-data")
+        self.userdata_raw = ret.get("user-data")
         return True
 
     @property

tests/unittests/sources/test_gce.py

@@ -488,3 +488,42 @@ def test_datasource_doesnt_use_ephemeral_dhcp(self, m_dhcp):
         ds = DataSourceGCE.DataSourceGCE(sys_cfg={}, distro=None, paths=None)
         ds._get_data()
         assert m_dhcp.call_count == 0
+
+    @mock.patch(
+        M_PATH + "EphemeralDHCPv4",
+        autospec=True,
+    )
+    @mock.patch(M_PATH + "net.find_candidate_nics")
+    def test_datasource_on_dhcp_lease_failure(
+        self, m_find_candidate_nics, m_dhcp
+    ):
+        self._set_mock_metadata()
+        distro = mock.MagicMock()
+        distro.get_tmp_exec_path = self.tmp_dir
+        ds = DataSourceGCE.DataSourceGCELocal(
+            sys_cfg={}, distro=distro, paths=None
+        )
+        m_find_candidate_nics.return_value = [
+            "ens0p4",
+            "ens0p5",
+        ]
+        m_dhcp.return_value.__enter__.side_effect = (
+            NoDHCPLeaseError,
+            NoDHCPLeaseError,
+        )
+        assert ds._get_data() is False
+        assert m_dhcp.call_args_list == [
+            mock.call(distro, iface="ens0p4"),
+            mock.call(distro, iface="ens0p5"),
+        ]
+
+        expected_logs = (
+            "Looking for the primary NIC in: ['ens0p4', 'ens0p5']",
+            "Unable to obtain a DHCP lease for ens0p4",
+            "Unable to obtain a DHCP lease for ens0p5",
+        )
+        for msg in expected_logs:
+            self.assertIn(
+                msg,
+                self.logs.getvalue(),
+            )