net: in netplan renderer delete known image-builtin content.

When rendering network configuration to netplan, remove known
"builtin" configurations.  The specific example here is Ubuntu Core
that has netplan configuration in etc/netplan/00-snapd-config.yaml.

We also delete the derived files since netplan will have created
these derived files in its generator that runs well before cloud-init.

LP: #1675576

2 files changed, 106 insertions, 1 deletions

diff --git a/cloudinit/net/netplan.py b/cloudinit/net/netplan.py
index 7444ae33..825fe831 100644
--- a/cloudinit/net/netplan.py
+++ b/cloudinit/net/netplan.py
@@ -10,6 +10,21 @@ from cloudinit import log as logging
 from cloudinit import util
 from cloudinit.net import SYS_CLASS_NET, get_devicelist
 
+KNOWN_SNAPD_CONFIG = b"""\
+# This is the initial network config.
+# It can be overwritten by cloud-init or console-conf.
+network:
+    version: 2
+    ethernets:
+        all-en:
+            match:
+                name: "en*"
+            dhcp4: true
+        all-eth:
+            match:
+                name: "eth*"
+            dhcp4: true
+"""
 
 LOG = logging.getLogger(__name__)
 NET_CONFIG_TO_V2 = {
@@ -154,6 +169,28 @@ def _extract_bond_slaves_by_name(interfaces, entry, bond_master):
         entry.update({'interfaces': bond_slave_names})
 
 
+def _clean_default(target=None):
+    # clean out any known default files and derived files in target
+    # LP: #1675576
+    tpath = util.target_path(target, "etc/netplan/00-snapd-config.yaml")
+    if not os.path.isfile(tpath):
+        return
+    content = util.load_file(tpath, decode=False)
+    if content != KNOWN_SNAPD_CONFIG:
+        return
+
+    derived = [util.target_path(target, f) for f in (
+               'run/systemd/network/10-netplan-all-en.network',
+               'run/systemd/network/10-netplan-all-eth.network',
+               'run/systemd/generator/netplan.stamp')]
+    existing = [f for f in derived if os.path.isfile(f)]
+    LOG.debug("removing known config '%s' and derived existing files: %s",
+              tpath, existing)
+
+    for f in [tpath] + existing:
+        os.unlink(f)
+
+
 class Renderer(renderer.Renderer):
     """Renders network information in a /etc/netplan/network.yaml format."""
 
@@ -166,6 +203,7 @@ class Renderer(renderer.Renderer):
                                        'etc/netplan/50-cloud-init.yaml')
         self.netplan_header = config.get('netplan_header', None)
         self._postcmds = config.get('postcmds', False)
+        self.clean_default = config.get('clean_default', True)
 
     def render_network_state(self, target, network_state):
         # check network state for version
@@ -182,6 +220,8 @@ class Renderer(renderer.Renderer):
             header += "\n"
         util.write_file(fpnplan, header + content)
 
+        if self.clean_default:
+            _clean_default(target=target)
         self._netplan_generate(run=self._postcmds)
         self._net_setup_link(run=self._postcmds)
 
diff --git a/tests/unittests/test_net.py b/tests/unittests/test_net.py
index 4f07d804..bfd04ba0 100644
--- a/tests/unittests/test_net.py
+++ b/tests/unittests/test_net.py
@@ -974,12 +974,14 @@ iface eth0 inet dhcp
 
 class TestNetplanNetRendering(CiTestCase):
 
+    @mock.patch("cloudinit.net.netplan._clean_default")
     @mock.patch("cloudinit.net.sys_dev_path")
     @mock.patch("cloudinit.net.read_sys_net")
     @mock.patch("cloudinit.net.get_devicelist")
     def test_default_generation(self, mock_get_devicelist,
                                 mock_read_sys_net,
-                                mock_sys_dev_path):
+                                mock_sys_dev_path,
+                                mock_clean_default):
         tmp_dir = self.tmp_dir()
         _setup_test(tmp_dir, mock_get_devicelist,
                     mock_read_sys_net, mock_sys_dev_path)
@@ -1013,6 +1015,69 @@ network:
             set-name: eth1000
 """
         self.assertEqual(expected.lstrip(), contents.lstrip())
+        self.assertEqual(1, mock_clean_default.call_count)
+
+
+class TestNetplanCleanDefault(CiTestCase):
+    snapd_known_path = 'etc/netplan/00-snapd-config.yaml'
+    snapd_known_content = textwrap.dedent("""\
+        # This is the initial network config.
+        # It can be overwritten by cloud-init or console-conf.
+        network:
+            version: 2
+            ethernets:
+                all-en:
+                    match:
+                        name: "en*"
+                    dhcp4: true
+                all-eth:
+                    match:
+                        name: "eth*"
+                    dhcp4: true
+        """)
+    stub_known = {
+        'run/systemd/network/10-netplan-all-en.network': 'foo-en',
+        'run/systemd/network/10-netplan-all-eth.network': 'foo-eth',
+        'run/systemd/generator/netplan.stamp': 'stamp',
+    }
+
+    def test_clean_known_config_cleaned(self):
+        content = {self.snapd_known_path: self.snapd_known_content, }
+        content.update(self.stub_known)
+        tmpd = self.tmp_dir()
+        files = sorted(populate_dir(tmpd, content))
+        netplan._clean_default(target=tmpd)
+        found = [t for t in files if os.path.exists(t)]
+        self.assertEqual([], found)
+
+    def test_clean_unknown_config_not_cleaned(self):
+        content = {self.snapd_known_path: self.snapd_known_content, }
+        content.update(self.stub_known)
+        content[self.snapd_known_path] += "# user put a comment\n"
+        tmpd = self.tmp_dir()
+        files = sorted(populate_dir(tmpd, content))
+        netplan._clean_default(target=tmpd)
+        found = [t for t in files if os.path.exists(t)]
+        self.assertEqual(files, found)
+
+    def test_clean_known_config_cleans_only_expected(self):
+        astamp = "run/systemd/generator/another.stamp"
+        anet = "run/systemd/network/10-netplan-all-lo.network"
+        ayaml = "etc/netplan/01-foo-config.yaml"
+        content = {
+            self.snapd_known_path: self.snapd_known_content,
+            astamp: "stamp",
+            anet: "network",
+            ayaml: "yaml",
+        }
+        content.update(self.stub_known)
+
+        tmpd = self.tmp_dir()
+        files = sorted(populate_dir(tmpd, content))
+        netplan._clean_default(target=tmpd)
+        found = [t for t in files if os.path.exists(t)]
+        expected = [util.target_path(tmpd, f) for f in (astamp, anet, ayaml)]
+        self.assertEqual(sorted(expected), found)
 
 
 class TestNetplanPostcommands(CiTestCase):