try: add new 'netplan try' command

Makefile

@@ -36,6 +36,7 @@ clean:
 check: default linting
 	tests/generate.py
 	tests/cli.py
+	nosetests3 -v --with-coverage
 
 linting:
 	$(PYFLAKES3) $(PYCODE)

debian/control

@@ -14,6 +14,7 @@ Build-Depends: debhelper (>= 11),
  systemd,
  pyflakes3,
  pycodestyle | pep8,
+ python3-nose,
  pandoc,
 Vcs-Git: https://github.com/CanonicalLtd/netplan
 Vcs-Browser: https://github.com/CanonicalLtd/netplan

netplan/cli/commands/__init__.py

@@ -19,10 +19,12 @@
 from netplan.cli.commands.generate import NetplanGenerate
 from netplan.cli.commands.ip import NetplanIp
 from netplan.cli.commands.migrate import NetplanMigrate
+from netplan.cli.commands.try_command import NetplanTry
 
 __all__ = [
     'NetplanApply',
     'NetplanGenerate',
     'NetplanIp',
     'NetplanMigrate',
+    'NetplanTry',
 ]

netplan/cli/commands/apply.py

@@ -24,6 +24,7 @@
 import subprocess
 
 import netplan.cli.utils as utils
+from netplan.configmanager import ConfigurationError
 
 
 class NetplanApply(utils.NetplanCommand):
@@ -34,14 +35,18 @@ def __init__(self):
                          leaf=True)
 
     def run(self):  # pragma: nocover (covered in autopkgtest)
-        self.func = self.command_apply
+        self.func = NetplanApply.command_apply
 
         self.parse_args()
         self.run_command()
 
-    def command_apply(self):  # pragma: nocover (covered in autopkgtest)
-        if subprocess.call([utils.get_generator_path()]) != 0:
-            sys.exit(1)
+    @staticmethod
+    def command_apply(run_generate=True, sync=False, exit_on_error=True):  # pragma: nocover (covered in autopkgtest)
+        if run_generate and subprocess.call([utils.get_generator_path()]) != 0:
+            if exit_on_error:
+                sys.exit(os.EX_CONFIG)
+            else:
+                raise ConfigurationError("the configuration could not be generated")
 
         devices = os.listdir('/sys/class/net')
 
@@ -51,7 +56,7 @@ def command_apply(self):  # pragma: nocover (covered in autopkgtest)
         # stop backends
         if restart_networkd:
             logging.debug('netplan generated networkd configuration exists, restarting networkd')
-            subprocess.check_call(['systemctl', 'stop', '--no-block', 'systemd-networkd.service', 'netplan-wpa@*.service'])
+            utils.systemctl_networkd('stop', sync=sync, extra_services=['netplan-wpa@*.service'])
         else:
             logging.debug('no netplan generated networkd configuration exists')
 
@@ -66,7 +71,7 @@ def command_apply(self):  # pragma: nocover (covered in autopkgtest)
                     except subprocess.CalledProcessError:
                         pass
 
-                utils.systemctl_network_manager('stop')
+                utils.systemctl_network_manager('stop', sync=sync)
         else:
             logging.debug('no netplan generated NM configuration exists')
 
@@ -75,7 +80,7 @@ def command_apply(self):  # pragma: nocover (covered in autopkgtest)
         for device in devices:
             if not os.path.islink('/sys/class/net/' + device):
                 continue
-            if self.replug(device):
+            if NetplanApply.replug(device):
                 any_replug = True
             else:
                 # if the interface is up, we can still apply .link file changes
@@ -90,12 +95,13 @@ def command_apply(self):  # pragma: nocover (covered in autopkgtest)
 
         # (re)start backends
         if restart_networkd:
-            subprocess.check_call(['systemctl', 'start', '--no-block', 'systemd-networkd.service'] +
-                                  [os.path.basename(f) for f in glob.glob('/run/systemd/system/*.wants/netplan-wpa@*.service')])
+            netplan_wpa = [os.path.basename(f) for f in glob.glob('/run/systemd/system/*.wants/netplan-wpa@*.service')]
+            utils.systemctl_networkd('start', sync=sync, extra_services=netplan_wpa)
         if restart_nm:
-            utils.systemctl_network_manager('start')
+            utils.systemctl_network_manager('start', sync=sync)
 
-    def replug(self, device):  # pragma: nocover (covered in autopkgtest)
+    @staticmethod
+    def replug(device):  # pragma: nocover (covered in autopkgtest)
         '''Unbind and rebind device if it is down'''
 
         devdir = os.path.join('/sys/class/net', device)

netplan/cli/commands/try_command.py

@@ -0,0 +1,169 @@
+#!/usr/bin/python3
+#
+# Copyright (C) 2018 Canonical, Ltd.
+# Author: Mathieu Trudel-Lapierre <mathieu.trudel-lapierre@canonical.com>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; version 3.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+'''netplan try command line'''
+
+import os
+import time
+import signal
+import sys
+import logging
+import subprocess
+
+from netplan.configmanager import ConfigManager
+import netplan.cli.utils as utils
+from netplan.cli.commands.apply import NetplanApply
+import netplan.terminal
+
+# Keep a timeout long enough to allow the network to converge, 60 seconds may
+# be slightly short given some complex configs, i.e. if STP must reconverge.
+DEFAULT_INPUT_TIMEOUT = 120
+
+
+class NetplanTry(utils.NetplanCommand):
+
+    def __init__(self):
+        super().__init__(command_id='try',
+                         description='Try to apply a new netplan config to running '
+                                     'system, with automatic rollback',
+                         leaf=True)
+        self.configuration_changed = False
+        self.new_interfaces = None
+        self.config_manager = ConfigManager()
+
+    def run(self):  # pragma: nocover (requires user input)
+        self.parser.add_argument('--config-file',
+                                 help='Apply the config file in argument in addition to current configuration.')
+        self.parser.add_argument('--timeout',
+                                 type=int, default=DEFAULT_INPUT_TIMEOUT,
+                                 help="Maximum number of seconds to wait for the user's confirmation")
+
+        self.func = self.command_try
+
+        self.parse_args()
+        self.run_command()
+
+    def command_try(self):  # pragma: nocover (requires user input)
+        if not self.is_revertable():
+            sys.exit(os.EX_CONFIG)
+
+        try:
+            fd = sys.stdin.fileno()
+            t = netplan.terminal.Terminal(fd)
+
+            # we really don't want to be interrupted while doing backup/revert operations
+            signal.signal(signal.SIGINT, self._signal_handler)
+
+            self.backup()
+            self.setup()
+
+            NetplanApply.command_apply(run_generate=True, sync=True, exit_on_error=False)
+
+            t.get_confirmation_input(timeout=self.timeout)
+        except netplan.terminal.InputRejected:
+            print("\nReverting.")
+            self.revert()
+        except netplan.terminal.InputAccepted:
+            print("\nConfiguration accepted.")
+        except Exception as e:
+            print("\nAn error occured: %s" % e)
+            print("\nReverting.")
+            self.revert()
+        finally:
+            self.cleanup()
+
+    def backup(self):  # pragma: nocover (requires user input)
+        backup_config_dir = False
+        if self.config_file:
+            backup_config_dir = True
+        self.config_manager.backup(backup_config_dir=backup_config_dir)
+
+    def setup(self):  # pragma: nocover (requires user input)
+        if self.config_file:
+            dest_dir = os.path.join("/", "etc", "netplan")
+            dest_name = os.path.basename(self.config_file).rstrip('.yaml')
+            dest_suffix = time.time()
+            dest_path = os.path.join(dest_dir, "{}.{}.yaml".format(dest_name, dest_suffix))
+            self.config_manager.add({self.config_file: dest_path})
+        self.configuration_changed = True
+
+    def revert(self):  # pragma: nocover (requires user input)
+        self.config_manager.revert()
+        NetplanApply.command_apply(run_generate=False, sync=True, exit_on_error=False)
+        for ifname in self.new_interfaces:
+            if ifname not in self.config_manager.bonds and \
+               ifname not in self.config_manager.bridges and \
+               ifname not in self.config_manager.vlans:
+                logging.debug("{} will not be removed: not a virtual interface".format(ifname))
+                continue
+            try:
+                cmd = ['ip', 'link', 'del', ifname]
+                subprocess.check_call(cmd)
+            except subprocess.CalledProcessError:
+                logging.warn("Could not revert (remove) new interface '{}'".format(ifname))
+
+    def cleanup(self):  # pragma: nocover (requires user input)
+        self.config_manager.cleanup()
+
+    def is_revertable(self):  # pragma: nocover (requires user input)
+        '''
+        Check if the configuration is revertable, if it doesn't contain bits
+        that we know are likely to render the system unstable if we apply it,
+        or if we revert.
+
+        Returns True if the parsed config is "revertable", meaning that we
+        can actually rely on backends to re-apply /all/ of the relevant
+        configuration to interfaces when their config changes.
+
+        Returns False if the parsed config contains options that are known
+        to not cleanly revert via the backend.
+        '''
+
+        # Parse; including any new config file passed on the command-line:
+        # new config might include things we can't revert.
+        extra_config = []
+        if self.config_file:
+            extra_config.append(self.config_file)
+        self.config_manager.parse(extra_config=extra_config)
+        self.new_interfaces = self.config_manager.new_interfaces
+
+        logging.debug("New interfaces: {}".format(self.new_interfaces))
+
+        revert_unsupported = []
+
+        # Bridges and bonds are special. They typically include (or could include)
+        # more than one device in them, and they can be set with special parameters
+        # to tweak their behavior, which are really hard to "revert", especially
+        # as systemd-networkd doesn't necessarily touch them when config changes.
+        multi_iface = {}
+        multi_iface.update(self.config_manager.bridges)
+        multi_iface.update(self.config_manager.bonds)
+        for ifname, settings in multi_iface.items():
+            if settings and 'parameters' in settings:
+                reason = "reverting custom parameters for bridges and bonds is not supported"
+                revert_unsupported.append((ifname, reason))
+
+        if revert_unsupported:
+            for ifname, reason in revert_unsupported:
+                print("{}: {}".format(ifname, reason))
+            print("\nPlease carefully review the configuration and use 'netplan apply' directly.")
+            return False
+        return True
+
+    def _signal_handler(self, signal, frame):  # pragma: nocover (requires user input)
+        if self.configuration_changed:
+            raise netplan.terminal.InputRejected()

netplan/cli/utils.py

@@ -51,15 +51,35 @@ def nm_running():  # pragma: nocover (covered in autopkgtest)
         return False
 
 
-def systemctl_network_manager(action):  # pragma: nocover (covered in autopkgtest)
+def systemctl_network_manager(action, sync=False):  # pragma: nocover (covered in autopkgtest)
     service_name = NM_SERVICE_NAME
 
+    command = ['systemctl', action]
+    if not sync:
+        command.append('--no-block')
+
     # If the network-manager snap is installed use its service
     # name rather than the one of the deb packaged NetworkManager
     if is_nm_snap_enabled():
         service_name = NM_SNAP_SERVICE_NAME
 
-    subprocess.check_call(['systemctl', action, '--no-block', service_name])
+    command.append(service_name)
+
+    subprocess.check_call(command)
+
+
+def systemctl_networkd(action, sync=False, extra_services=[]):  # pragma: nocover (covered in autopkgtest)
+    command = ['systemctl', action]
+
+    if not sync:
+        command.append('--no-block')
+
+    command.append('systemd-networkd.service')
+
+    for service in extra_services:
+        command.append(service)
+
+    subprocess.check_call(command)
 
 
 class NetplanCommand(argparse.Namespace):