summaryrefslogtreecommitdiff
path: root/cloudinit/net/activators.py
diff options
context:
space:
mode:
Diffstat (limited to 'cloudinit/net/activators.py')
-rw-r--r--cloudinit/net/activators.py156
1 files changed, 156 insertions, 0 deletions
diff --git a/cloudinit/net/activators.py b/cloudinit/net/activators.py
new file mode 100644
index 00000000..34fee3bf
--- /dev/null
+++ b/cloudinit/net/activators.py
@@ -0,0 +1,156 @@
+# This file is part of cloud-init. See LICENSE file for license information.
+import logging
+import os
+from abc import ABC, abstractmethod
+from typing import Iterable, List, Type
+
+from cloudinit import subp
+from cloudinit import util
+from cloudinit.net.eni import available as eni_available
+from cloudinit.net.netplan import available as netplan_available
+from cloudinit.net.network_state import NetworkState
+from cloudinit.net.sysconfig import NM_CFG_FILE
+
+
+LOG = logging.getLogger(__name__)
+
+
+class NetworkActivator(ABC):
+ @staticmethod
+ @abstractmethod
+ def available() -> bool:
+ raise NotImplementedError()
+
+ @staticmethod
+ @abstractmethod
+ def bring_up_interface(device_name: str) -> bool:
+ raise NotImplementedError()
+
+ @classmethod
+ def bring_up_interfaces(cls, device_names: Iterable[str]) -> bool:
+ all_succeeded = True
+ for device in device_names:
+ if not cls.bring_up_interface(device):
+ all_succeeded = False
+ return all_succeeded
+
+ @classmethod
+ def bring_up_all_interfaces(cls, network_state: NetworkState) -> bool:
+ return cls.bring_up_interfaces(
+ [i['name'] for i in network_state.iter_interfaces()]
+ )
+
+
+class IfUpDownActivator(NetworkActivator):
+ # Note that we're not overriding bring_up_interfaces to pass something
+ # like ifup --all because it isn't supported everywhere.
+ # E.g., NetworkManager has a ifupdown plugin that requires the name
+ # of a specific connection.
+ @staticmethod
+ def available(target=None) -> bool:
+ """Return true if ifupdown can be used on this system."""
+ return eni_available(target=target)
+
+ @staticmethod
+ def bring_up_interface(device_name: str) -> bool:
+ """Bring up interface using ifup."""
+ cmd = ['ifup', device_name]
+ LOG.debug("Attempting to run bring up interface %s using command %s",
+ device_name, cmd)
+ try:
+ (_out, err) = subp.subp(cmd)
+ if len(err):
+ LOG.warning("Running %s resulted in stderr output: %s",
+ cmd, err)
+ return True
+ except subp.ProcessExecutionError:
+ util.logexc(LOG, "Running interface command %s failed", cmd)
+ return False
+
+
+class NetworkManagerActivator(NetworkActivator):
+ @staticmethod
+ def available(target=None) -> bool:
+ config_present = os.path.isfile(
+ subp.target_path(target, path=NM_CFG_FILE)
+ )
+ nmcli_present = subp.which('nmcli', target=target)
+ return config_present and bool(nmcli_present)
+
+ @staticmethod
+ def bring_up_interface(device_name: str) -> bool:
+ try:
+ subp.subp(['nmcli', 'connection', 'up', device_name])
+ except subp.ProcessExecutionError:
+ util.logexc(LOG, "nmcli failed to bring up {}".format(device_name))
+ return False
+ return True
+
+
+class NetplanActivator(NetworkActivator):
+ @staticmethod
+ def available(target=None) -> bool:
+ return netplan_available(target=target)
+
+ @staticmethod
+ def _apply_netplan():
+ LOG.debug('Applying current netplan config')
+ try:
+ subp.subp(['netplan', 'apply'], capture=True)
+ except subp.ProcessExecutionError:
+ util.logexc(LOG, "netplan apply failed")
+ return False
+ return True
+
+ @staticmethod
+ def bring_up_interface(device_name: str) -> bool:
+ LOG.debug("Calling 'netplan apply' rather than "
+ "bringing up individual interfaces")
+ return NetplanActivator._apply_netplan()
+
+ @staticmethod
+ def bring_up_interfaces(device_names: Iterable[str]) -> bool:
+ LOG.debug("Calling 'netplan apply' rather than "
+ "bringing up individual interfaces")
+ return NetplanActivator._apply_netplan()
+
+ @staticmethod
+ def bring_up_all_interfaces(network_state: NetworkState) -> bool:
+ return NetplanActivator._apply_netplan()
+
+
+# This section is mostly copied and pasted from renderers.py. An abstract
+# version to encompass both seems overkill at this point
+DEFAULT_PRIORITY = [
+ IfUpDownActivator,
+ NetworkManagerActivator,
+ NetplanActivator,
+]
+
+
+def search_activator(
+ priority=None, target=None
+) -> List[Type[NetworkActivator]]:
+ if priority is None:
+ priority = DEFAULT_PRIORITY
+
+ unknown = [i for i in priority if i not in DEFAULT_PRIORITY]
+ if unknown:
+ raise ValueError(
+ "Unknown activators provided in priority list: %s" % unknown)
+
+ return [activator for activator in priority if activator.available(target)]
+
+
+def select_activator(priority=None, target=None) -> Type[NetworkActivator]:
+ found = search_activator(priority, target)
+ if not found:
+ if priority is None:
+ priority = DEFAULT_PRIORITY
+ tmsg = ""
+ if target and target != "/":
+ tmsg = " in target=%s" % target
+ raise RuntimeError(
+ "No available network activators found%s. Searched "
+ "through list: %s" % (tmsg, priority))
+ return found[0]