diff options
Diffstat (limited to 'vicn')
| -rw-r--r-- | vicn/core/resource_mgr.py | 62 | ||||
| -rw-r--r-- | vicn/core/state.py | 2 | ||||
| -rw-r--r-- | vicn/core/task.py | 38 | ||||
| -rw-r--r-- | vicn/helpers/__init__.py | 0 | ||||
| -rw-r--r-- | vicn/resource/central.py | 91 | ||||
| -rw-r--r-- | vicn/resource/linux/certificate.py | 9 | ||||
| -rw-r--r-- | vicn/resource/linux/keypair.py | 71 | ||||
| -rw-r--r-- | vicn/resource/linux/link.py | 75 | ||||
| -rw-r--r-- | vicn/resource/linux/net_device.py | 3 | ||||
| -rw-r--r-- | vicn/resource/linux/physical.py | 40 | ||||
| -rw-r--r-- | vicn/resource/lxd/lxd_hypervisor.py | 12 | ||||
| -rw-r--r-- | vicn/resource/node.py | 8 | ||||
| -rw-r--r-- | vicn/resource/ns3/emulated_channel.py | 12 | ||||
| -rw-r--r-- | vicn/resource/ns3/emulated_lte_channel.py | 10 |
14 files changed, 265 insertions, 168 deletions
diff --git a/vicn/core/resource_mgr.py b/vicn/core/resource_mgr.py index f6082488..57dcafef 100644 --- a/vicn/core/resource_mgr.py +++ b/vicn/core/resource_mgr.py @@ -44,6 +44,8 @@ from vicn.core.task import EmptyTask, BashTask log = logging.getLogger(__name__) +# NOTE: Do not fully reinitialize a resource after a step fails since it will +# call initialize several times, and might created spurious resources. ENABLE_LXD_WORKAROUND = True # Monitoring queries @@ -148,6 +150,9 @@ class ResourceManager(metaclass=Singleton): # For debug self._committed = set() + self._num = 0 + self._num_clean = 0 + def terminate(self): self._router.terminate() @@ -317,6 +322,7 @@ class ResourceManager(metaclass=Singleton): Committing a resource creates an asyncio function implementing a state management automaton. """ + self._num += 1 asyncio.ensure_future(self._process_resource(resource)) def commit(self): @@ -496,10 +502,10 @@ class ResourceManager(metaclass=Singleton): # Task management #-------------------------------------------------------------------------- - def schedule(self, task): + def schedule(self, task, resource = None): if task is None or isinstance(task, EmptyTask): return - self._task_mgr.schedule(task) + self._task_mgr.schedule(task, resource) #-------------------------------------------------------------------------- # Asynchronous resource API @@ -785,8 +791,10 @@ class ResourceManager(metaclass=Singleton): self.attr_log(resource, attribute, 'Current state is {}'.format(state)) + # AttributeState.ERROR if resource._state.attr_change_success == False: - log.error('Attribute error') + log.error('Attribute error {} for resource {}'.format( + resource.get_uuid(), attribute.name)) e = resource._state.attr_change_value[attribute.name] import traceback; traceback.print_tb(e.__traceback__) raise NotImplementedError @@ -794,7 +802,7 @@ class ResourceManager(metaclass=Singleton): # Signal update errors to the parent resource resource._state.attr_change_event[attribute.name].set() - elif state == AttributeState.UNINITIALIZED: + if state == AttributeState.UNINITIALIZED: pending_state = AttributeState.PENDING_INIT elif state in AttributeState.INITIALIZED: pending_state = AttributeState.PENDING_UPDATE @@ -898,10 +906,10 @@ class ResourceManager(metaclass=Singleton): new_state = AttributeState.CLEAN else: - log.error('Attribute error') + log.error('Attribute error {} for resource {}'.format( + resource.get_uuid(), attribute.name)) e = resource._state.attr_change_value[attribute.name] - import traceback; traceback.print_tb(e.__traceback__) - raise NotImplementedError + new_state = AttributeState.ERROR else: raise RuntimeError @@ -1046,13 +1054,20 @@ class ResourceManager(metaclass=Singleton): It is important to centralize state change since some states are associated with Events(). """ + prev_state = resource._state.state resource._state.state = state if state == ResourceState.CLEAN: # Monitoring hook self._monitor(resource) resource._state.clean.set() + if prev_state != ResourceState.CLEAN: + self._num_clean += 1 + log.info("Resource {} is marked as CLEAN ({}/{})".format( + resource.get_uuid(), self._num_clean, self._num)) else: resource._state.clean.clear() + if prev_state == ResourceState.CLEAN: + self._num_clean -= 1 if state == ResourceState.INITIALIZED: resource._state.init.set() @@ -1211,12 +1226,7 @@ class ResourceManager(metaclass=Singleton): state = resource._state.state self.log(resource, 'Current state is {}'.format(state)) - if resource._state.change_success == False: - e = resource._state.change_value - import traceback; traceback.print_tb(e.__traceback__) - raise NotImplementedError - - elif state == ResourceState.UNINITIALIZED: + if state == ResourceState.UNINITIALIZED: pending_state = ResourceState.PENDING_DEPS elif state == ResourceState.DEPS_OK: pending_state = ResourceState.PENDING_INIT @@ -1296,9 +1306,10 @@ class ResourceManager(metaclass=Singleton): raise RuntimeError if task is not None and not isinstance(task, EmptyTask): + resource._state.change_success = None # undetermined state state_change = functools.partial(self._trigger_state_change, resource) task.add_done_callback(state_change) - self.schedule(task) + self.schedule(task, resource) self.log(resource, 'Trigger {} -> {}. Waiting task completion'.format( state, pending_state)) @@ -1321,8 +1332,8 @@ class ResourceManager(metaclass=Singleton): new_state = ResourceState.INITIALIZED else: e = resource._state.change_value - import traceback; traceback.print_tb(e.__traceback__) - raise NotImplementedError + log.error('Cannot setup resource {} : {}'.format( + resource.get_uuid(), e)) elif pending_state == ResourceState.PENDING_GET: if resource._state.change_success == True: @@ -1339,13 +1350,13 @@ class ResourceManager(metaclass=Singleton): # does not exists. anyways the bug should only occur # with container.execute(), not container.get() log.error('LXD Fix (not found). Reset resource') - new_state = ResourceState.UNINITIALIZED + new_state = ResourceState.INITIALIZED elif ENABLE_LXD_WORKAROUND and isinstance(e, LXDAPIException): # "not found" is the normal exception when the container # does not exists. anyways the bug should only occur # with container.execute(), not container.get() log.error('LXD Fix (API error). Reset resource') - new_state = ResourceState.UNINITIALIZED + new_state = ResourceState.INITIALIZED elif isinstance(e, ResourceNotFound): # The resource does not exist self.log(resource, S_GET_DONE.format( @@ -1354,8 +1365,9 @@ class ResourceManager(metaclass=Singleton): resource._state.change_value = None else: e = resource._state.change_value - import traceback; traceback.print_tb(e.__traceback__) - raise NotImplementedError + log.error('Cannot get resource state {} : {}'.format( + resource.get_uuid(), e)) + new_state = ResourceState.ERROR resource._state.change_success = True elif pending_state == ResourceState.PENDING_KEYS: @@ -1372,8 +1384,8 @@ class ResourceManager(metaclass=Singleton): resource._state.change_success = True else: e = resource._state.change_value - import traceback; traceback.print_tb(e.__traceback__) - raise NotImplementedError + log.error('Cannot create resource {} : {}'.format( + resource.get_uuid(), e)) elif pending_state == ResourceState.PENDING_CREATE: if resource._state.change_success == True: @@ -1386,19 +1398,19 @@ class ResourceManager(metaclass=Singleton): if ENABLE_LXD_WORKAROUND and isinstance(e, LxdNotFound): log.error('LXD Fix (not found). Reset resource') - new_state = ResourceState.UNINITIALIZED + new_state = ResourceState.INITIALIZED resource._state.change_success = True elif ENABLE_LXD_WORKAROUND and \ isinstance(e, LXDAPIException): log.error('LXD Fix (API error). Reset resource') - new_state = ResourceState.UNINITIALIZED + new_state = ResourceState.INITIALIZED resource._state.change_success = True elif 'File exists' in str(e): new_state = ResourceState.CREATED resource._state.change_success = True elif 'dpkg --configure -a' in str(e): resource._dpkg_configure_a = True - new_state = ResourceState.UNINITIALIZED + new_state = ResourceState.INITIALIZED resource._state.change_success = True else: self.log(resource, 'CREATE failed: {}'.format(e)) diff --git a/vicn/core/state.py b/vicn/core/state.py index d5069b24..c32b8237 100644 --- a/vicn/core/state.py +++ b/vicn/core/state.py @@ -46,6 +46,7 @@ class ResourceState: PENDING_UPDATE = 'PENDING_UPDATE' PENDING_DELETE = 'PENDING_DELETE' DELETED = 'DELETED' + ERROR = 'ERROR' class AttributeState: UNINITIALIZED = 'UNINITIALIZED' @@ -54,6 +55,7 @@ class AttributeState: PENDING_INIT = 'PENDING_INIT' PENDING_UPDATE = 'PENDING_UPDATE' CLEAN = 'CLEAN' + ERROR = 'ERROR' class Operations: SET = 'set' diff --git a/vicn/core/task.py b/vicn/core/task.py index 2e9bc275..5aecb40b 100644 --- a/vicn/core/task.py +++ b/vicn/core/task.py @@ -34,7 +34,7 @@ log = logging.getLogger(__name__) EXECUTOR=concurrent.futures.ThreadPoolExecutor #EXECUTOR=concurrent.futures.ProcessPoolExecutor -MAX_WORKERS=50 # None +MAX_WORKERS = 50 # None class BaseTask: """Base class for all tasks @@ -171,6 +171,27 @@ def get_attributes_task(resource, attribute_names): return {attribute_name: ret} return func() +def _get_func_desc(f): + """ + Returns a string representation of a function for logging purposes. + + Todo: args and keywords (including from partial) + """ + partial = isinstance(f, functools.partial) + if partial: + f = f.func + + s = '' + if hasattr(f, '__name__'): + s += f.__name__ + if hasattr(f, '__doc__') and f.__doc__: + if s: + s += ' : ' + s += f.__doc__ + + return 'partial<{}>'.format(s) if partial else s + + class PythonTask(Task): def __init__(self, func, *args, **kwargs): super().__init__() @@ -197,8 +218,8 @@ class PythonTask(Task): fut.add_done_callback(self._done_callback) def __repr__(self): - return '<Task[py] {} / {} {}>'.format(self._func, self._args, - self._kwargs) + s = _get_func_desc(self._func) + return '<Task[py] {}>'.format(s) if s else '<Task[py]>' class PythonAsyncTask(PythonTask): async def execute(self, *args, **kwargs): @@ -213,7 +234,8 @@ class PythonAsyncTask(PythonTask): fut.add_done_callback(self._done_callback) def __repr__(self): - return '<Task[apy]>' + s = _get_func_desc(self._func) + return '<Task[apy] {}>'.format(s) if s else '<Task[apy]>' class PythonInlineTask(PythonTask): async def execute(self, *args, **kwargs): @@ -229,6 +251,10 @@ class PythonInlineTask(PythonTask): self._future.set_exception(e) return self._future + def __repr__(self): + s = _get_func_desc(self._func) + return '<Task[ipy] {}>'.format(s) if s else '<Task[ipy]>' + class BashTask(Task): def __init__(self, node, cmd, parameters=None, parse=None, as_root=False, output=False, pre=None, post=None, lock=None): @@ -339,12 +365,14 @@ class TaskManager: loop = asyncio.get_event_loop() loop.set_default_executor(executor) - def schedule(self, task): + def schedule(self, task, resource = None): """All instances of BaseTask can be scheduled Here we might decide to do more advanced scheduling, like merging bash tasks, etc. thanks to the task algebra. """ + uuid = resource.get_uuid() if resource else '(unknown)' + log.info('Scheduling task {} for resource {}'.format(task, uuid)) asyncio.ensure_future(task.execute()) @task diff --git a/vicn/helpers/__init__.py b/vicn/helpers/__init__.py new file mode 100644 index 00000000..e69de29b --- /dev/null +++ b/vicn/helpers/__init__.py diff --git a/vicn/resource/central.py b/vicn/resource/central.py index 73a43478..6f3c680b 100644 --- a/vicn/resource/central.py +++ b/vicn/resource/central.py @@ -16,8 +16,9 @@ # limitations under the License. # -import networkx as nx import logging +import networkx as nx +import os from netmodel.model.type import String from netmodel.util.misc import pairwise @@ -54,7 +55,7 @@ CMD_NAT = '\n'.join([ # For containers CMD_IP_FORWARD = 'echo 1 > /proc/sys/net/ipv4/ip_forward' -HOST_FILE_PATH = "/etc/hosts.vicn" +HOST_FILE = "hosts.vicn" #------------------------------------------------------------------------------ # Routing strategies @@ -162,29 +163,29 @@ def _get_l2_graph(manager, with_managed = False): if G.has_edge(src.node._state.uuid, dst.node._state.uuid): continue - map_node_interface = { src.node._state.uuid : src._state.uuid, + map_node_interface = { src.node._state.uuid : src._state.uuid, dst.node._state.uuid: dst._state.uuid} - G.add_edge(src.node._state.uuid, dst.node._state.uuid, + G.add_edge(src.node._state.uuid, dst.node._state.uuid, map_node_interface = map_node_interface) else: # This is for a normal Channel for src_it in range(0,len(channel.interfaces)): src = channel.interfaces[src_it] - # Iterate over the remaining interface to create all the + # Iterate over the remaining interface to create all the # possible combination for dst_it in range(src_it+1,len(channel.interfaces)): dst = channel.interfaces[dst_it] - - if not with_managed and (not src.managed or + + if not with_managed and (not src.managed or not dst.managed): continue if G.has_edge(src.node._state.uuid, dst.node._state.uuid): continue - map_node_interface = { - src.node._state.uuid : src._state.uuid, + map_node_interface = { + src.node._state.uuid : src._state.uuid, dst.node._state.uuid: dst._state.uuid} - G.add_edge(src.node._state.uuid, dst.node._state.uuid, + G.add_edge(src.node._state.uuid, dst.node._state.uuid, map_node_interface = map_node_interface) return G @@ -198,9 +199,9 @@ def _get_icn_graph(manager): other_node = other_face.node if G.has_edge(node._state.uuid, other_node._state.uuid): continue - map_node_face = { node._state.uuid: face._state.uuid, + map_node_face = { node._state.uuid: face._state.uuid, other_node._state.uuid: other_face._state.uuid } - G.add_edge(node._state.uuid, other_node._state.uuid, + G.add_edge(node._state.uuid, other_node._state.uuid, map_node_face = map_node_face) return G @@ -224,7 +225,9 @@ class IPAssignment(Resource): raise ResourceNotFound def __subresources__(self): - self.host_file = TextFile(node = None, filename = HOST_FILE_PATH, + basedir = os.path.dirname(self._state.manager._base) + self.host_file = TextFile(node = None, + filename = os.path.join(basedir, HOST_FILE), overwrite = True) return self.host_file @@ -241,13 +244,13 @@ class IPAssignment(Resource): # We sort nodes by names for IP assignment. This code ensures that # interfaces on the same channel get consecutive IP addresses. That # way, we can assign /31 on p2p channels. - channels = sorted(self._state.manager.by_type(Channel), + channels = sorted(self._state.manager.by_type(Channel), key = lambda x : x.get_sortable_name()) - channels.extend(sorted(self._state.manager.by_type(Node), + channels.extend(sorted(self._state.manager.by_type(Node), key = lambda node : node.name)) host_file_content = "" - + # Dummy code to start IP addressing on an even number for /31 ip = AddressManager().get_ip(None) if int(ip[-1]) % 2 == 0: @@ -256,7 +259,7 @@ class IPAssignment(Resource): for channel in channels: # Sort interfaces in a deterministic order to ensure consistent # addressing across restarts of the tool - interfaces = sorted(channel.interfaces, + interfaces = sorted(channel.interfaces, key = lambda x : x.device_name) for interface in interfaces: @@ -279,7 +282,7 @@ class IPAssignment(Resource): host_file_content += '# {} {} {}\n'.format( interface.node.name, interface.device_name, ip) if interface == interface.node.host_interface: - host_file_content += '{} {}\n'.format(ip, + host_file_content += '{} {}\n'.format(ip, interface.node.name) self.host_file.content = host_file_content @@ -339,7 +342,7 @@ class IPRoutes(Resource): G = _get_l2_graph(self._state.manager) origins = self._get_ip_origins() - + # node -> list(origins for which we have routes) ip_routes = dict() @@ -347,7 +350,7 @@ class IPRoutes(Resource): routes = list() for src, prefix, dst in strategy(G, origins): data = G.get_edge_data(src, dst) - + map_ = data['map_node_interface'] next_hop_interface = map_[src] @@ -356,7 +359,7 @@ class IPRoutes(Resource): src_node = self._state.manager.by_uuid(src) mac_addr = None - if ((hasattr(next_hop_ingress, 'vpp') and + if ((hasattr(next_hop_ingress, 'vpp') and next_hop_ingress.vpp is not None) or (hasattr(src_node, 'vpp') and src_node.vpp is not None)): mac_addr = next_hop_ingress.mac_address @@ -366,15 +369,15 @@ class IPRoutes(Resource): ip_routes[src_node] = list() if prefix in ip_routes[src_node]: continue - + if prefix == next_hop_ingress.ip_address: # Direct route on src_node.name : # route add [prefix] dev [next_hop_interface_.device_name] - route = IPRoute(node = src_node, + route = IPRoute(node = src_node, managed = False, owner = self, ip_address = prefix, - mac_address = mac_addr, + mac_address = mac_addr, interface = next_hop_interface) else: # We need to be sure we have a route to the gw from the node @@ -387,11 +390,11 @@ class IPRoutes(Resource): interface = next_hop_interface) ip_routes[src_node].append(next_hop_ingress.ip_address) pre_routes.append(pre_route) - - # Route on src_node.name: + + # Route on src_node.name: # route add [prefix] dev [next_hop_interface_.device_name] # via [next_hop_ingress.ip_address] - route = IPRoute(node = src_node, + route = IPRoute(node = src_node, managed = False, owner = self, ip_address = prefix, @@ -407,7 +410,7 @@ class IPRoutes(Resource): IP routing strategy : direct routes only """ routes = list() - G = _get_l2_graph(self._state.manager) + G = _get_l2_graph(self._state.manager) for src_node_uuid, dst_node_uuid, data in G.edges_iter(data = True): src_node = self._state.manager.by_uuid(src_node_uuid) dst_node = self._state.manager.by_uuid(dst_node_uuid) @@ -423,7 +426,7 @@ class IPRoutes(Resource): dst_node.name, dst.device_name, dst.ip_address, src_node.name, src.device_name, src.ip_address)) - route = IPRoute(node = src_node, + route = IPRoute(node = src_node, managed = False, owner = self, ip_address = dst.ip_address, @@ -431,7 +434,7 @@ class IPRoutes(Resource): interface = src) routes.append(route) - route = IPRoute(node = dst_node, + route = IPRoute(node = dst_node, managed = False, owner = self, ip_address = src.ip_address, @@ -498,14 +501,14 @@ class ICNFaces(Resource): owner = self, protocol = protocol, src_nic = src, - dst_mac = dst.mac_address) + dst_mac = dst.mac_address) dst_face = L2Face(node = dst_node, owner = self, protocol = protocol, src_nic = dst, dst_mac = src.mac_address) - elif protocol in (FaceProtocol.tcp4, FaceProtocol.tcp6, + elif protocol in (FaceProtocol.tcp4, FaceProtocol.tcp6, FaceProtocol.udp4, FaceProtocol.udp6): src_face = L4Face(node = src_node, owner = self, @@ -513,14 +516,14 @@ class ICNFaces(Resource): src_ip = src.ip_address, dst_ip = dst.ip_address, src_port = TMP_DEFAULT_PORT, - dst_port = TMP_DEFAULT_PORT) + dst_port = TMP_DEFAULT_PORT) dst_face = L4Face(node = dst_node, owner = self, protocol = protocol, src_ip = dst.ip_address, dst_ip = src.ip_address, src_port = TMP_DEFAULT_PORT, - dst_port = TMP_DEFAULT_PORT) + dst_port = TMP_DEFAULT_PORT) else: raise NotImplementedError @@ -583,7 +586,7 @@ class ICNRoutes(Resource): routes = list() for src, prefix, dst in strategy(G, origins): data = G.get_edge_data(src, dst) - + map_ = data['map_node_face'] next_hop_face = map_[src] @@ -675,9 +678,9 @@ class ContainerSetup(Resource): route_gw.node.routing_table.routes << route_gw # c) dns - dns_server_entry = DnsServerEntry(node = self.container, + dns_server_entry = DnsServerEntry(node = self.container, owner = self, - ip_address = self.container.node.bridge.ip_address, + ip_address = self.container.node.bridge.ip_address, interface_name = self.container.host_interface.device_name) return dns_server_entry @@ -710,7 +713,7 @@ class ContainersSetup(Resource): if len(containers) == 0: return None - container_resources = [ContainerSetup(owner = self, container = c) + container_resources = [ContainerSetup(owner = self, container = c) for c in containers] return Resource.__concurrent__(*container_resources) @@ -737,7 +740,7 @@ class CentralIP(Resource): def __subresources__(self): ip_assign = IPAssignment(owner=self) containers_setup = ContainersSetup(owner=self) - ip_routes = IPRoutes(owner = self, + ip_routes = IPRoutes(owner = self, routing_strategy = self.ip_routing_strategy) return ip_assign > (ip_routes | containers_setup) @@ -758,10 +761,10 @@ class CentralICN(Resource): """ # Choices: spt, max_flow - icn_routing_strategy = Attribute(String, + icn_routing_strategy = Attribute(String, description = 'ICN routing strategy', - default = 'spt') - face_protocol = Attribute(String, + default = 'spt') + face_protocol = Attribute(String, description = 'Protocol used to create faces', default = 'ether') @@ -777,8 +780,8 @@ class CentralICN(Resource): return ('CentralIP',) def __subresources__(self): - icn_faces = ICNFaces(owner = self, protocol_name = self.face_protocol) - icn_routes = ICNRoutes(owner = self, + icn_faces = ICNFaces(owner = self, protocol_name = self.face_protocol) + icn_routes = ICNRoutes(owner = self, routing_strategy = self.icn_routing_strategy) return icn_faces > icn_routes diff --git a/vicn/resource/linux/certificate.py b/vicn/resource/linux/certificate.py index e8750dff..7f9b8a74 100644 --- a/vicn/resource/linux/certificate.py +++ b/vicn/resource/linux/certificate.py @@ -31,6 +31,8 @@ DEFAULT_SUBJECT = '/CN=www.cisco.com/L=Paris/O=Cisco/C=FR' CMD_CREATE='\n'.join([ '# Generate a new certificate', + 'mkdir -p $(dirname {self.key})', + 'mkdir -p $(dirname {self.cert})', 'openssl req -x509 -newkey rsa:' + DEFAULT_RSA_LENGTH + ' -keyout {self.key} -out {self.cert} -subj ' + DEFAULT_SUBJECT + ' -nodes' ]) @@ -40,9 +42,6 @@ class Certificate(Resource): Resource: Certificate Implements a SSL certificate. - - Todo: - - ideally, this should be implemented as a pair of tightly coupled files. """ node = Attribute(Node, description = 'Node on which the certificate is created', @@ -53,6 +52,10 @@ class Certificate(Resource): key = Attribute(String, description = 'Key path', mandatory = True) + #-------------------------------------------------------------------------- + # Resource lifecycle + #-------------------------------------------------------------------------- + @inline_task def __initialize__(self): self._cert_file = File(node = Reference(self, 'node'), diff --git a/vicn/resource/linux/keypair.py b/vicn/resource/linux/keypair.py new file mode 100644 index 00000000..a81a40d4 --- /dev/null +++ b/vicn/resource/linux/keypair.py @@ -0,0 +1,71 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +# +# Copyright (c) 2017 Cisco and/or its affiliates. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at: +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +import os.path + +from netmodel.model.type import String +from vicn.core.attribute import Attribute, Multiplicity, Reference +from vicn.core.exception import ResourceNotFound +from vicn.core.resource import Resource +from vicn.core.task import task, inline_task, BashTask +from vicn.resource.linux.file import File +from vicn.resource.node import Node + +CMD_CREATE=''' +mkdir -p {dirname} +ssh-keygen -t rsa -N "" -f {self.key} +''' + +class Keypair(Resource): + """ + Resource: Keypair + + Implements a SSH keypair + """ + node = Attribute(Node, + description = 'Node on which the certificate is created', + mandatory = True, + multiplicity = Multiplicity.ManyToOne) + key = Attribute(String, description = 'Key path', + mandatory = True) + + #-------------------------------------------------------------------------- + # Resource lifecycle + #-------------------------------------------------------------------------- + + @inline_task + def __initialize__(self): + self._pubkey_file = File(node = Reference(self, 'node'), + filename = self.key + '.pub', + managed = False) + self._key_file = File(node = Reference(self, 'node'), + filename = self.key, + managed = False) + + def __get__(self): + return self._pubkey_file.__get__() | self._key_file.__get__() + + def __create__(self): + return BashTask(None, CMD_CREATE, { + 'dirname': os.path.dirname(self.key), + 'self': self}) + + def __delete__(self): + return self._pubkey_file.__delete__() | self._key_file.__delete__() + + diff --git a/vicn/resource/linux/link.py b/vicn/resource/linux/link.py index 4304a948..a4771f9c 100644 --- a/vicn/resource/linux/link.py +++ b/vicn/resource/linux/link.py @@ -42,8 +42,8 @@ CMD_DELETE_IF_EXISTS='ip link show {interface.device_name} && ' \ CMD_CREATE=''' # Create veth pair in the host node ip link add name {tmp_src} type veth peer name {tmp_dst} -ip link set dev {tmp_src} netns {pid[0]} name {interface.src.device_name} -ip link set dev {tmp_dst} netns {pid[1]} name {interface.dst.device_name} +ip link set dev {tmp_src} netns {pid[0]} name {interface._src.device_name} +ip link set dev {tmp_dst} netns {pid[1]} name {interface._dst.device_name} ''' CMD_UP=''' ip link set dev {interface.device_name} up @@ -61,9 +61,6 @@ class Link(Channel): the current implementation. """ - src = Attribute(Interface, description = 'Source interface') - dst = Attribute(Interface, description = 'Destination interface') - capacity = Attribute(Integer, description = 'Link capacity (Mb/s)') delay = Attribute(String, description = 'Link propagation delay') @@ -73,29 +70,29 @@ class Link(Channel): mandatory = True) def __init__(self, *args, **kwargs): - assert 'src' not in kwargs and 'dst' not in kwargs assert 'src_node' in kwargs and 'dst_node' in kwargs + self._src = None + self._dst = None super().__init__(*args, **kwargs) @inline_task def __initialize__(self): - # We create two managed net devices that are pre-setup # but the resource manager has to take over for IP addresses etc. # Being done in initialize, those attributes won't be considered as # dependencies and will thus not block the resource state machine. - self.src = NonTapBaseNetDevice(node = self.src_node, + self._src = NonTapBaseNetDevice(node = self.src_node, device_name = self.dst_node.name, channel = self, capacity = self.capacity, - owner = self.owner) - self.dst = NonTapBaseNetDevice(node = self.dst_node, + owner = self) + self._dst = NonTapBaseNetDevice(node = self.dst_node, device_name = self.src_node.name, channel = self, capacity = self.capacity, - owner = self.owner) - self.dst.remote = self.src - self.src.remote = self.dst + owner = self) + self._dst.remote = self._src + self._src.remote = self._dst #-------------------------------------------------------------------------- # Internal methods @@ -104,21 +101,8 @@ class Link(Channel): async def _commit(self): manager = self._state.manager - # We mark the src and dst interfaces created because we are pre-setting - # them up in __create__ using a VethPair - # We go through both INITIALIZED and CREATED stats to raise the proper - # events and satisfy any eventual wait_* command. - await manager._set_resource_state(self.src, ResourceState.INITIALIZED) - await manager._set_resource_state(self.dst, ResourceState.INITIALIZED) - await manager._set_resource_state(self.src, ResourceState.CREATED) - await manager._set_resource_state(self.dst, ResourceState.CREATED) - - # We mark the attribute clean so that it is not updated - await manager._set_attribute_state(self, 'src', AttributeState.CLEAN) - await manager._set_attribute_state(self, 'dst', AttributeState.CLEAN) - - manager.commit_resource(self.src) - manager.commit_resource(self.dst) + manager.commit_resource(self._src) + manager.commit_resource(self._dst) # Disable rp_filtering # self.src.rp_filter = False @@ -143,36 +127,23 @@ class Link(Channel): # Resource lifecycle #-------------------------------------------------------------------------- - @async_task - async def __get__(self): - manager = self._state.manager + def __get__(self): + return (self._src.__get__() | self._dst.__get__()) > async_task(self._commit)() - try: - await run_task(self.src.__get__(), manager) - await run_task(self.dst.__get__(), manager) - except ResourceNotFound: - # This is raised if any of the two side of the VethPair is missing - raise ResourceNotFound - - # We always need to commit the two endpoints so that their attributes - # are correctly updated - await self._commit() - def __create__(self): assert self.src_node.get_type() == 'lxccontainer' assert self.dst_node.get_type() == 'lxccontainer' src_host = self.src_node.node dst_host = self.dst_node.node - assert src_host == dst_host host = src_host # Sometimes a down interface persists on one side delif_src = BashTask(self.src_node, CMD_DELETE_IF_EXISTS, - {'interface': self.src}) + {'interface': self._src}) delif_dst = BashTask(self.dst_node, CMD_DELETE_IF_EXISTS, - {'interface': self.dst}) + {'interface': self._dst}) pid_src = get_attributes_task(self.src_node, ['pid']) pid_dst = get_attributes_task(self.dst_node, ['pid']) @@ -185,17 +156,13 @@ class Link(Channel): create = BashTask(host, CMD_CREATE, {'interface': self, 'tmp_src': tmp_src, 'tmp_dst': tmp_dst}) - up_src = BashTask(self.src_node, CMD_UP, {'interface': self.src}) - up_dst = BashTask(self.dst_node, CMD_UP, {'interface': self.dst}) - - @async_task - async def set_state(): - # We always need to commit the two endpoints so that their attributes - # are correctly updated - await self._commit() + up_src = BashTask(self.src_node, CMD_UP, {'interface': self._src}) + up_dst = BashTask(self.dst_node, CMD_UP, {'interface': self._dst}) delif = delif_src | delif_dst up = up_src | up_dst pid = pid_src | pid_dst - return ((delif > (pid @ create)) > up) > set_state() + return ((delif > (pid @ create)) > up) > async_task(self._commit)() + def __delete__(self): + return self._src.__delete__() | self._dst.__delete__() diff --git a/vicn/resource/linux/net_device.py b/vicn/resource/linux/net_device.py index f0a08991..84a946a4 100644 --- a/vicn/resource/linux/net_device.py +++ b/vicn/resource/linux/net_device.py @@ -464,6 +464,9 @@ class NonTapBaseNetDevice(BaseNetDevice): # Attributes #-------------------------------------------------------------------------- + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + def _get_offload(self): return BashTask(self.node, CMD_GET_OFFLOAD, {'netdevice': self}, parse = lambda rv : rv.stdout.strip() == 'on') diff --git a/vicn/resource/linux/physical.py b/vicn/resource/linux/physical.py index e5eba2d3..8decb517 100644 --- a/vicn/resource/linux/physical.py +++ b/vicn/resource/linux/physical.py @@ -22,16 +22,17 @@ import logging import subprocess import shlex -from netmodel.model.type import String, Integer -from netmodel.util.misc import is_local_host -from netmodel.util.socket import check_port -from vicn.core.attribute import Attribute -from vicn.core.commands import Command, ReturnValue -from vicn.core.exception import ResourceNotFound -from vicn.core.task import Task, task -from vicn.resource.node import Node, DEFAULT_USERNAME -from vicn.resource.node import DEFAULT_SSH_PUBLIC_KEY -from vicn.resource.node import DEFAULT_SSH_PRIVATE_KEY +from netmodel.model.type import String, Integer +from netmodel.util.misc import is_local_host +from netmodel.util.socket import check_port +from vicn.core.attribute import Attribute +from vicn.core.commands import Command, ReturnValue +from vicn.core.exception import ResourceNotFound, VICNException +from vicn.core.task import Task, task +from vicn.resource.linux.keypair import Keypair +from vicn.resource.node import Node, DEFAULT_USERNAME +from vicn.resource.node import DEFAULT_SSH_PUBLIC_KEY +from vicn.resource.node import DEFAULT_SSH_PRIVATE_KEY log = logging.getLogger(__name__) @@ -42,6 +43,9 @@ CMD_SSH = 'ssh {ssh_options} -i {private_key} -p {port} ' \ CMD_SSH_NF = 'ssh -n -f {ssh_options} -i {private_key} -p {port} ' \ '{user}@{host} {command}' +FN_KEY = os.path.expanduser(os.path.join( + '~', '.vicn', 'ssh_client_cert', 'ssh_client_key')) + class Physical(Node): """ Resource: Physical @@ -67,12 +71,20 @@ class Physical(Node): # Resource lifecycle #-------------------------------------------------------------------------- - @task - def __get__(self, attributes=None): + def __subresources__(self): + """ + Require a SSH keypair to be present for authentication on nodes + """ + return Keypair(node = None, key = FN_KEY) + + def __initialize__(self): + """ + Initialization require the ssh port to be open on the node, and the ssh + public key to be copied on the remote node. + """ if not check_port(self.hostname, self.ssh_port): - raise ResourceNotFound + raise VICNException - def __create__(self): tasks = list() modes = (True, False) if DEFAULT_USERNAME != 'root' else (True,) for as_root in modes: diff --git a/vicn/resource/lxd/lxd_hypervisor.py b/vicn/resource/lxd/lxd_hypervisor.py index 328f3fdf..b6e1c9ff 100644 --- a/vicn/resource/lxd/lxd_hypervisor.py +++ b/vicn/resource/lxd/lxd_hypervisor.py @@ -44,11 +44,10 @@ logging.getLogger("requests").setLevel(logging.WARNING) logging.getLogger("urllib3").setLevel(logging.WARNING) log = logging.getLogger(__name__) -# FIXME use system-wide files -DEFAULT_CERT_PATH = os.path.join(os.path.dirname(__file__), - '..', '..', '..', 'config', 'lxd_client_cert', 'client_cert.pem') -DEFAULT_KEY_PATH = os.path.join(os.path.dirname(__file__), - '..', '..', '..', 'config', 'lxd_client_cert', 'client_key.pem') +DEFAULT_CERT_PATH = os.path.expanduser(os.path.join( + '~', '.vicn', 'lxd_client_cert', 'client_cert.pem')) +DEFAULT_KEY_PATH = os.path.expanduser(os.path.join( + '~', '.vicn', 'lxd_client_cert', 'client_key.pem')) # FIXME hardcoded password for LXD server DEFAULT_TRUST_PASSWORD = 'vicn' @@ -191,8 +190,7 @@ class LxdHypervisor(Service): cert = DEFAULT_CERT_PATH, key = DEFAULT_KEY_PATH, owner = self) - lxd_cert_install = LxdInstallCert(node = Reference(self, 'node'), - certificate = lxd_local_cert, + lxd_cert_install = LxdInstallCert(certificate = lxd_local_cert, owner = self) return (lxd_init | lxd_local_cert) > lxd_cert_install diff --git a/vicn/resource/node.py b/vicn/resource/node.py index bfb2f9ec..ad519666 100644 --- a/vicn/resource/node.py +++ b/vicn/resource/node.py @@ -27,10 +27,10 @@ from vicn.core.resource import Resource log = logging.getLogger(__name__) DEFAULT_USERNAME = 'root' -DEFAULT_SSH_PRIVATE_KEY = os.path.join(os.path.dirname(__file__), - '..', '..', 'config', 'ssh_client_cert', 'ssh_client_key') -DEFAULT_SSH_PUBLIC_KEY = os.path.join(os.path.dirname(__file__), - '..', '..', 'config', 'ssh_client_cert', 'ssh_client_key.pub') +DEFAULT_SSH_PRIVATE_KEY = os.path.expanduser(os.path.join( + '~', '.vicn', 'ssh_client_cert', 'ssh_client_key')) +DEFAULT_SSH_PUBLIC_KEY = os.path.expanduser(os.path.join( + '~', '.vicn', 'ssh_client_cert', 'ssh_client_key.pub')) class Node(Resource): """ diff --git a/vicn/resource/ns3/emulated_channel.py b/vicn/resource/ns3/emulated_channel.py index 08d7a14b..5f61960c 100644 --- a/vicn/resource/ns3/emulated_channel.py +++ b/vicn/resource/ns3/emulated_channel.py @@ -69,7 +69,7 @@ class EmulatedChannel(Channel, Application): ap = Attribute(Node, description = 'AP', key = True) stations = Attribute(Node, description = 'List of stations', multiplicity = Multiplicity.OneToMany, key = True) - control_port = Attribute(Integer, + control_port = Attribute(Integer, description = 'Control port for the simulation') # Overloaded attributes @@ -141,7 +141,7 @@ class EmulatedChannel(Channel, Application): managed = False) self._ap_if = VethPair(node = self.ap, name = 'vh-' + ap.name + '-' + self.name, - device_name = 'vh-' + ap.name + '-' + self.name, + device_name = 'vh-' + ap.name + '-' + self.name, host = host, owner = self) self._ap_bridged = self._ap_if.host @@ -152,7 +152,7 @@ class EmulatedChannel(Channel, Application): interfaces.append(self._ap_if) # Add a tap interface for the AP... - self._ap_tap = TapDevice(node = self.node, + self._ap_tap = TapDevice(node = self.node, owner = self, device_name = 'tap-' + ap.name + '-' + self.name, up = True, @@ -165,11 +165,11 @@ class EmulatedChannel(Channel, Application): await wait_resources(interfaces) # NOTE: only set channel after the resource is created or it might - # create loops which, at this time, are not handled + # create loops which, at this time, are not handled self._ap_if.set('channel', self) # Add interfaces to bridge - vlan = AddressManager().get('vlan', self, tag='ap') + vlan = AddressManager().get('vlan', self, tag='ap') # AS the container has created the VethPair already without Vlan, we # need to delete and recreate it @@ -181,8 +181,6 @@ class EmulatedChannel(Channel, Application): task = self.node.bridge._add_interface(self._ap_tap, vlan = vlan) await run_task(task, self._state.manager) - print('/!\ pass information to the running simulation') - @inline_task def _get_ap(self): return {'ap': None} diff --git a/vicn/resource/ns3/emulated_lte_channel.py b/vicn/resource/ns3/emulated_lte_channel.py index 8c7382cb..39370228 100644 --- a/vicn/resource/ns3/emulated_lte_channel.py +++ b/vicn/resource/ns3/emulated_lte_channel.py @@ -61,7 +61,7 @@ class EmulatedLteChannel(EmulatedChannel): # ... and each station if not station.managed: sta_if = None - else: + else: if isinstance(station, LxcContainer): host = NetDevice(node = station.node, device_name='vhh-' + station.name + '-' + self.name, @@ -103,8 +103,8 @@ class EmulatedLteChannel(EmulatedChannel): task = self.node.bridge._remove_interface(bridged_sta) await run_task(task, self._state.manager) - - task = self.node.bridge._add_interface(bridged_sta, + + task = self.node.bridge._add_interface(bridged_sta, vlan = vlan) await run_task(task, self._state.manager) @@ -178,8 +178,8 @@ class EmulatedLteChannel(EmulatedChannel): # Coma-separated list of stations' IP/netmask len 'sta-ips' : ','.join(sta_ips), # Base station IP/netmask len - 'bs-ip' : AddressManager().get_ip(self._ap_if) + - DEFAULT_NETMASK, + 'bs-ip' : AddressManager().get_ip(self._ap_if) + '/' + + str(DEFAULT_NETMASK), 'txBuffer' : '800000', 'isFading' : 'true' if DEFAULT_FADING_ENABLED else 'false', } |