diff options
Diffstat (limited to 'vicn/core')
| -rw-r--r-- | vicn/core/api.py | 11 | ||||
| -rw-r--r-- | vicn/core/attribute.py | 136 | ||||
| -rw-r--r-- | vicn/core/collection.py | 23 | ||||
| -rw-r--r-- | vicn/core/commands.py | 2 | ||||
| -rw-r--r-- | vicn/core/exception.py | 2 | ||||
| -rw-r--r-- | vicn/core/resource.py | 341 | ||||
| -rw-r--r-- | vicn/core/resource_mgr.py | 357 | ||||
| -rw-r--r-- | vicn/core/sa_collections.py | 264 | ||||
| -rw-r--r-- | vicn/core/sa_compat.py | 270 | ||||
| -rw-r--r-- | vicn/core/state.py | 37 | ||||
| -rw-r--r-- | vicn/core/task.py | 96 |
11 files changed, 563 insertions, 976 deletions
diff --git a/vicn/core/api.py b/vicn/core/api.py index 708e2581..ccbb9b24 100644 --- a/vicn/core/api.py +++ b/vicn/core/api.py @@ -19,6 +19,7 @@ import asyncio import json import logging +import os import resource as ulimit import sys @@ -51,6 +52,7 @@ class API(metaclass = Singleton): def terminate(self): # XXX not valid if nothing has been initialized ResourceManager().terminate() + os._exit(0) def parse_topology_file(self, topology_fn, resources, settings): log.info("Parsing topology file %(topology_fn)s" % locals()) @@ -87,7 +89,7 @@ class API(metaclass = Singleton): if nofile is not None and nofile > 0: if nofile < 1024: log.error('Too few allowed open files for the process') - import os; os._exit(1) + os._exit(1) log.info('Setting open file descriptor limit to {}'.format( nofile)) @@ -95,15 +97,18 @@ class API(metaclass = Singleton): ulimit.RLIMIT_NOFILE, (nofile, nofile)) - ResourceManager(base=scenario[-1], settings=settings) + base = os.path.dirname(scenario_list[-1]) + base = os.path.abspath(base) + ResourceManager(base = base, settings = settings) for resource in resources: try: ResourceManager().create_from_dict(**resource) except Exception as e: + import traceback; traceback.print_exc() log.error("Could not create resource '%r': %r" % \ (resource, e,)) - import os; os._exit(1) + os._exit(1) self._configured = True diff --git a/vicn/core/attribute.py b/vicn/core/attribute.py index 3afe0d6e..02520cbd 100644 --- a/vicn/core/attribute.py +++ b/vicn/core/attribute.py @@ -22,109 +22,34 @@ import logging import operator import types -from netmodel.model.mapper import ObjectSpecification -from netmodel.model.type import Type, Self -from netmodel.util.meta import inheritors -from netmodel.util.misc import is_iterable -from vicn.core.exception import VICNListException -from vicn.core.requirement import Requirement, RequirementList -from vicn.core.sa_collections import InstrumentedList -from vicn.core.state import UUID, NEVER_SET, Operations +from netmodel.model.attribute import Attribute as BaseAttribute, NEVER_SET +from netmodel.model.attribute import Multiplicity, DEFAULT +from netmodel.model.type import Self +from netmodel.model.uuid import UUID +from netmodel.util.misc import is_iterable +from vicn.core.requirement import Requirement, RequirementList +from vicn.core.state import Operations log = logging.getLogger(__name__) #------------------------------------------------------------------------------ -# Attribute Multiplicity -#------------------------------------------------------------------------------ - -class Multiplicity: - OneToOne = '1_1' - OneToMany = '1_N' - ManyToOne = 'N_1' - ManyToMany = 'N_N' - - @staticmethod - def reverse(value): - reverse_map = { - Multiplicity.OneToOne: Multiplicity.OneToOne, - Multiplicity.OneToMany: Multiplicity.ManyToOne, - Multiplicity.ManyToOne: Multiplicity.OneToMany, - Multiplicity.ManyToMany: Multiplicity.ManyToMany, - } - return reverse_map[value] - - -# Default attribute properties values (default to None) -DEFAULT = { - 'multiplicity' : Multiplicity.OneToOne, - 'mandatory' : False, -} - -#------------------------------------------------------------------------------ # Attribute #------------------------------------------------------------------------------ -class Attribute(abc.ABC, ObjectSpecification): - properties = [ - 'name', - 'type', - 'key', - 'description', - 'default', - 'choices', - 'mandatory', - 'multiplicity', - 'ro', - 'auto', - 'func', +class Attribute(BaseAttribute): + properties = BaseAttribute.properties + properties.extend([ 'requirements', - 'reverse_name', - 'reverse_description', - 'reverse_auto' - ] + 'remote_default' + ]) def __init__(self, *args, **kwargs): - for key in Attribute.properties: - value = kwargs.pop(key, NEVER_SET) - setattr(self, key, value) - - if len(args) == 1: - self.type, = args - elif len(args) == 2: - self.name, self.type = args - - # self.type is optional since the type can be inherited. Although we - # will have to verify the attribute is complete at some point - if self.type: - if isinstance(self.type, str): - self.type = Type.from_string(self.type) - assert self.type is Self or Type.exists(self.type) + super().__init__(*args, **kwargs) # Post processing attribute properties if self.requirements is not NEVER_SET: self.requirements = RequirementList(self.requirements) - self.is_aggregate = False - - self._reverse_attributes = list() - - #-------------------------------------------------------------------------- - # Display - #-------------------------------------------------------------------------- - - def __repr__(self): - return '<Attribute {}>'.format(self.name) - - __str__ = __repr__ - - # The following functions are required to allow comparing attributes, and - # using them as dict keys - - def __eq__(self, other): - return self.name == other.name - - def __hash__(self): - return hash(self.name) #-------------------------------------------------------------------------- # Descriptor protocol @@ -132,6 +57,8 @@ class Attribute(abc.ABC, ObjectSpecification): # see. https://docs.python.org/3/howto/descriptor.html #-------------------------------------------------------------------------- + # XXX Overloaded & simpler + def __get__(self, instance, owner=None): if not instance: return self @@ -144,9 +71,6 @@ class Attribute(abc.ABC, ObjectSpecification): instance.set(self.name, value, blocking=False) - def __delete__(self, instance): - raise NotImplementedError - #-------------------------------------------------------------------------- def do_list_add(self, instance, value): @@ -170,7 +94,7 @@ class Attribute(abc.ABC, ObjectSpecification): value, cur_value) # prevent instrumented list to perform operation - raise VICNListException + raise InstrumentedListException def do_list_remove(self, instance, value): if instance.is_local_attribute(self.name): @@ -187,7 +111,7 @@ class Attribute(abc.ABC, ObjectSpecification): value, cur_value) # prevent instrumented list to perform operation - raise VICNListException + raise InstrumentedListException def do_list_clear(self, instance): if instance.is_local_attribute(self.name): @@ -201,7 +125,7 @@ class Attribute(abc.ABC, ObjectSpecification): value, cur_value) # prevent instrumented list to perform operation - raise VICNListException + raise InstrumentedListException def handle_getitem(self, instance, item): if isinstance(item, UUID): @@ -210,29 +134,6 @@ class Attribute(abc.ABC, ObjectSpecification): return item #-------------------------------------------------------------------------- - # Accessors - #-------------------------------------------------------------------------- - - def __getattribute__(self, name): - value = super().__getattribute__(name) - if value is NEVER_SET: - if name == 'default': - return list() if self.is_collection else None - return DEFAULT.get(name, None) - return value - - def has_reverse_attribute(self): - return self.reverse_name and self.multiplicity - - @property - def is_collection(self): - return self.multiplicity in (Multiplicity.OneToMany, - Multiplicity.ManyToMany) - - def is_set(self, instance): - return instance.is_set(self.name) - - #-------------------------------------------------------------------------- # Operations #-------------------------------------------------------------------------- @@ -256,6 +157,7 @@ class Attribute(abc.ABC, ObjectSpecification): #------------------------------------------------------------------------------ +# XXX Move this to object, be careful of access to self._reference ! class Reference: """ Value reference. diff --git a/vicn/core/collection.py b/vicn/core/collection.py deleted file mode 100644 index fb222891..00000000 --- a/vicn/core/collection.py +++ /dev/null @@ -1,23 +0,0 @@ -#!/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. -# - -from vicn.core.sa_collections import InstrumentedList -from netmodel.model.collection import Collection - -class Collection(InstrumentedList, Collection): - pass diff --git a/vicn/core/commands.py b/vicn/core/commands.py index 41c06bf5..c04ab264 100644 --- a/vicn/core/commands.py +++ b/vicn/core/commands.py @@ -65,7 +65,7 @@ class ReturnValue: def _clean(self, value): if value is None or isinstance(value, str): return value - return value.decode('utf-8') + return value.decode('utf-8').strip() def _set_stdout(self, value): self._stdout = self._clean(value) diff --git a/vicn/core/exception.py b/vicn/core/exception.py index 977fc8ad..4389531f 100644 --- a/vicn/core/exception.py +++ b/vicn/core/exception.py @@ -34,8 +34,6 @@ class InitializeException(VICNException): pass class CheckException(VICNException): pass class SetupException(VICNException): pass -class VICNListException(VICNException): pass - class ResourceNotFound(VICNException): pass class VICNWouldBlock(VICNException): diff --git a/vicn/core/resource.py b/vicn/core/resource.py index f92e1255..878a8108 100644 --- a/vicn/core/resource.py +++ b/vicn/core/resource.py @@ -32,14 +32,17 @@ from threading import Event as ThreadEvent # LXD workaround from pylxd.exceptions import NotFound as LXDAPIException +from netmodel.model.collection import Collection +from netmodel.model.key import Key from netmodel.model.mapper import ObjectSpecification +from netmodel.model.object import Object from netmodel.model.type import String, Bool, Integer, Dict from netmodel.model.type import BaseType, Self +from netmodel.model.uuid import UUID from netmodel.util.deprecated import deprecated from netmodel.util.singleton import Singleton from vicn.core.attribute import Attribute, Multiplicity, Reference from vicn.core.attribute import NEVER_SET -from vicn.core.collection import Collection from vicn.core.commands import ReturnValue from vicn.core.event import Event, AttributeChangedEvent from vicn.core.exception import VICNException, ResourceNotFound @@ -47,7 +50,7 @@ from vicn.core.exception import VICNWouldBlock from vicn.core.resource_factory import ResourceFactory from vicn.core.requirement import Requirement, Property from vicn.core.scheduling_algebra import SchedulingAlgebra -from vicn.core.state import ResourceState, UUID +from vicn.core.state import ResourceState from vicn.core.state import Operations, InstanceState from vicn.core.task import run_task, BashTask @@ -73,29 +76,29 @@ class TopLevelResource: pass class FactoryResource(TopLevelResource): pass class CategoryResource(TopLevelResource): pass -#------------------------------------------------------------------------------ - -class ResourceMetaclass(ABCMeta): - def __init__(cls, class_name, parents, attrs): - """ - Args: - cls: The class type we're registering. - class_name: A String containing the class_name. - parents: The parent class types of 'cls'. - attrs: The attribute (members) of 'cls'. - """ - super().__init__(class_name, parents, attrs) - - # We use the metaclass to create attributes for instance, even before - # the Resource Factory is called. They are needed both for initializing - # attributes and reverse attributes, in whatever order. Only class - # creation allow us to clear _attributes, otherwise, we will just add - # those from the parent, siblings, etc... - cls._sanitize() +##------------------------------------------------------------------------------ +# +#class ResourceMetaclass(ABCMeta, ObjectSpecification): +# def __init__(cls, class_name, parents, attrs): +# """ +# Args: +# cls: The class type we're registering. +# class_name: A String containing the class_name. +# parents: The parent class types of 'cls'. +# attrs: The attribute (members) of 'cls'. +# """ +# super().__init__(class_name, parents, attrs) +# +# # We use the metaclass to create attributes for instance, even before +# # the Resource Factory is called. They are needed both for initializing +# # attributes and reverse attributes, in whatever order. Only class +# # creation allow us to clear _attributes, otherwise, we will just add +# # those from the parent, siblings, etc... +# cls._sanitize() #------------------------------------------------------------------------------ -class BaseResource(BaseType, ABC, metaclass=ResourceMetaclass): +class BaseResource(Object): #, ABC, metaclass=ResourceMetaclass): """Base Resource class The base Resource class implements all the logic related to resource @@ -409,6 +412,12 @@ class BaseResource(BaseType, ABC, metaclass=ResourceMetaclass): if isinstance(value, UUID): value = self.from_uuid(value) + # XXX XXX quick fix + from netmodel.model.type import InetAddress + if issubclass(attribute.type, InetAddress) and value is not None \ + and not isinstance(value, InetAddress) and not isinstance(value, Reference): + value = attribute.type(value) + if set_reverse and attribute.reverse_name: for base in self.__class__.mro(): if not hasattr(base, '_reverse_attributes'): @@ -420,7 +429,7 @@ class BaseResource(BaseType, ABC, metaclass=ResourceMetaclass): value.set(ra.name, self, set_reverse = False) elif ra.multiplicity == Multiplicity.ManyToOne: for element in value: - value.set(ra.name, self, set_reverse = False) + element.set(ra.name, self, set_reverse = False) elif ra.multiplicity == Multiplicity.OneToMany: if value is not None: collection = value.get(ra.name) @@ -445,7 +454,7 @@ class BaseResource(BaseType, ABC, metaclass=ResourceMetaclass): return value def set(self, attribute_name, value, current=False, set_reverse=True, - blocking = True): + blocking = None): value = self._set(attribute_name, value, current=current, set_reverse=set_reverse) if self.is_local_attribute(attribute_name) or current: @@ -455,11 +464,10 @@ class BaseResource(BaseType, ABC, metaclass=ResourceMetaclass): vars(self)[attribute_name] = value else: - fut = self._state.manager.attribute_set(self, attribute_name, value) - asyncio.ensure_future(fut) + self._state.manager.attribute_set(self, attribute_name, value) async def async_set(self, attribute_name, value, current=False, - set_reverse=True, blocking=True): + set_reverse=True, blocking=None): """ Example: - setting the ip address on a node's interface @@ -470,8 +478,7 @@ class BaseResource(BaseType, ABC, metaclass=ResourceMetaclass): """ value = self._set(attribute_name, value, current=current, set_reverse=set_reverse) - await self._state.manager.attribute_set(self, attribute_name, value, - blocking=blocking) + await self._state.manager.attribute_set_async(self, attribute_name, value) def set_many(self, attribute_dict, current=False): if not attribute_dict: @@ -541,101 +548,101 @@ class BaseResource(BaseType, ABC, metaclass=ResourceMetaclass): except AttributeError: return None - @classmethod - def _sanitize(cls): - """ - This methods performs sanitization of the object declaration. - - More specifically: - - it goes over all attributes and sets their name based on the python - object attribute name. - - it establishes mutual object relationships through reverse attributes. - - """ - cls._reverse_attributes = dict() - cur_reverse_attributes = dict() - for name, obj in vars(cls).items(): - if not isinstance(obj, ObjectSpecification): - continue - - # XXX it seems obj should always be an attribute, confirm ! - if isinstance(obj, Attribute): - obj.name = name - - # Remember whether a reverse_name is defined before loading - # inherited properties from parent - has_reverse = bool(obj.reverse_name) - - # Handle overloaded attributes - # By recursion, it is sufficient to look into the parent - for base in cls.__bases__: - if hasattr(base, name): - parent_attribute = getattr(base, name) - obj.merge(parent_attribute) - assert obj.type - - # Handle reverse attribute - # - # NOTE: we need to do this after merging to be sure we get all - # properties inherited from parent (eg. multiplicity) - # - # See "Reverse attributes" section in BaseResource docstring. - # - # Continueing with the same example, let's detail how it is handled: - # - # Original declaration: - # >>> - # class Group(Resource): - # resources = Attribute(Resource, description = 'Resources belonging to the group', - # multiplicity = Multiplicity.ManyToMany, - # default = [], - # reverse_name = 'groups', - # reverse_description = 'Groups to which the resource belongs') - # <<< - # - # Local variables: - # cls = <class 'vicn.resource.group.Group'> - # obj = <Attribute resources> - # obj.type = <class 'vicn.core.Resource'> - # reverse_attribute = <Attribute groups> - # - # Result: - # 1) Group._reverse_attributes = - # { <Attribute resources> : [<Attribute groups>, ...], ...} - # 2) Add attribute <Attribute groups> to class Resource - # 3) Resource._reverse_attributes = - # { <Attribute groups> : [<Attribute resources], ...], ...} - # - if has_reverse: - a = { - 'name' : obj.reverse_name, - 'description' : obj.reverse_description, - 'multiplicity' : Multiplicity.reverse(obj.multiplicity), - 'reverse_name' : obj.name, - 'reverse_description' : obj.description, - 'auto' : obj.reverse_auto, - } - reverse_attribute = Attribute(cls, **a) - reverse_attribute.is_aggregate = True - - # 1) Store the reverse attributes to be later inserted in the - # remote class, at the end of the function - # TODO : clarify the reasons to perform this in two steps - cur_reverse_attributes[obj.type] = reverse_attribute - - # 2) - if not obj in cls._reverse_attributes: - cls._reverse_attributes[obj] = list() - cls._reverse_attributes[obj].append(reverse_attribute) - - # 3) - if not reverse_attribute in obj.type._reverse_attributes: - obj.type._reverse_attributes[reverse_attribute] = list() - obj.type._reverse_attributes[reverse_attribute].append(obj) - - # Insert newly created reverse attributes in the remote class - for kls, a in cur_reverse_attributes.items(): - setattr(kls, a.name, a) +####### @classmethod +####### def _sanitize(cls): +####### """ +####### This methods performs sanitization of the object declaration. +####### +####### More specifically: +####### - it goes over all attributes and sets their name based on the python +####### object attribute name. +####### - it establishes mutual object relationships through reverse attributes. +####### +####### """ +####### cls._reverse_attributes = dict() +####### cur_reverse_attributes = dict() +####### for name, obj in vars(cls).items(): +####### if not isinstance(obj, ObjectSpecification): +####### continue +####### +####### # XXX it seems obj should always be an attribute, confirm ! +####### if isinstance(obj, Attribute): +####### obj.name = name +####### +####### # Remember whether a reverse_name is defined before loading +####### # inherited properties from parent +####### has_reverse = bool(obj.reverse_name) +####### +####### # Handle overloaded attributes +####### # By recursion, it is sufficient to look into the parent +####### for base in cls.__bases__: +####### if hasattr(base, name): +####### parent_attribute = getattr(base, name) +####### obj.merge(parent_attribute) +####### assert obj.type +####### +####### # Handle reverse attribute +####### # +####### # NOTE: we need to do this after merging to be sure we get all +####### # properties inherited from parent (eg. multiplicity) +####### # +####### # See "Reverse attributes" section in BaseResource docstring. +####### # +####### # Continueing with the same example, let's detail how it is handled: +####### # +####### # Original declaration: +####### # >>> +####### # class Group(Resource): +####### # resources = Attribute(Resource, description = 'Resources belonging to the group', +####### # multiplicity = Multiplicity.ManyToMany, +####### # default = [], +####### # reverse_name = 'groups', +####### # reverse_description = 'Groups to which the resource belongs') +####### # <<< +####### # +####### # Local variables: +####### # cls = <class 'vicn.resource.group.Group'> +####### # obj = <Attribute resources> +####### # obj.type = <class 'vicn.core.Resource'> +####### # reverse_attribute = <Attribute groups> +####### # +####### # Result: +####### # 1) Group._reverse_attributes = +####### # { <Attribute resources> : [<Attribute groups>, ...], ...} +####### # 2) Add attribute <Attribute groups> to class Resource +####### # 3) Resource._reverse_attributes = +####### # { <Attribute groups> : [<Attribute resources], ...], ...} +####### # +####### if has_reverse: +####### a = { +####### 'name' : obj.reverse_name, +####### 'description' : obj.reverse_description, +####### 'multiplicity' : Multiplicity.reverse(obj.multiplicity), +####### 'reverse_name' : obj.name, +####### 'reverse_description' : obj.description, +####### 'auto' : obj.reverse_auto, +####### } +####### reverse_attribute = Attribute(cls, **a) +####### reverse_attribute.is_aggregate = True +####### +####### # 1) Store the reverse attributes to be later inserted in the +####### # remote class, at the end of the function +####### # TODO : clarify the reasons to perform this in two steps +####### cur_reverse_attributes[obj.type] = reverse_attribute +####### +####### # 2) +####### if not obj in cls._reverse_attributes: +####### cls._reverse_attributes[obj] = list() +####### cls._reverse_attributes[obj].append(reverse_attribute) +####### +####### # 3) +####### if not reverse_attribute in obj.type._reverse_attributes: +####### obj.type._reverse_attributes[reverse_attribute] = list() +####### obj.type._reverse_attributes[reverse_attribute].append(obj) +####### +####### # Insert newly created reverse attributes in the remote class +####### for kls, a in cur_reverse_attributes.items(): +####### setattr(kls, a.name, a) @classmethod def iter_attributes(cls, aggregates = False): @@ -648,13 +655,54 @@ class BaseResource(BaseType, ABC, metaclass=ResourceMetaclass): yield attribute - def iter_keys(self): - for attribute in self.iter_attributes(): - if attribute.key == True: - yield attribute + @classmethod + def iter_keys(cls): + for name in dir(cls): + key = getattr(cls, name) + if not isinstance(key, Key): + continue + yield key + + def get_attributes(self, aggregates = False): + return list(self.iter_attributes(aggregates = aggregates)) + + @classmethod + def has_attribute(cls, name): + return name in [a.name for a in cls.attributes()] + + def get_attribute_names(self, aggregates = False): + return set(a.name + for a in self.iter_attributes(aggregates = aggregates)) + + def get_attribute_dict(self, field_names = None, aggregates = False, + uuid = True): + assert not field_names or field_names.is_star() + attributes = self.get_attributes(aggregates = aggregates) + + ret = dict() + for a in attributes: + if not a.is_set(self): + continue + value = getattr(self, a.name) + if a.is_collection: + ret[a.name] = list() + for x in value: + if uuid and isinstance(x, Resource): + x = x._state.uuid._uuid + ret[a.name].append(x) + else: + if uuid and isinstance(value, Resource): + value = value._state.uuid._uuid + ret[a.name] = value + return ret + + @classmethod + def get_keys(cls): + return list(cls.iter_keys()) - def get_keys(self): - return list(self.iter_keys()) + @classmethod + def has_key_attribute(cls, attribute): + return any(attribute in key for key in cls.iter_keys()) def auto_instanciate(self, attribute): if self.managed is False: @@ -694,35 +742,6 @@ class BaseResource(BaseType, ABC, metaclass=ResourceMetaclass): self._state.manager.commit_resource(resource) return resource - def get_attributes(self, aggregates = False): - return list(self.iter_attributes(aggregates = aggregates)) - - def get_attribute_names(self, aggregates = False): - return set(a.name - for a in self.iter_attributes(aggregates = aggregates)) - - def get_attribute_dict(self, field_names = None, aggregates = False, - uuid = True): - assert not field_names or field_names.is_star() - attributes = self.get_attributes(aggregates = aggregates) - - ret = dict() - for a in attributes: - if not a.is_set(self): - continue - value = getattr(self, a.name) - if a.is_collection: - ret[a.name] = list() - for x in value: - if uuid and isinstance(x, Resource): - x = x._state.uuid._uuid - ret[a.name].append(x) - else: - if uuid and isinstance(value, Resource): - value = value._state.uuid._uuid - ret[a.name] = value - return ret - def get_tuple(self): return (self.__class__, self._get_attribute_dict()) @@ -877,10 +896,6 @@ class BaseResource(BaseType, ABC, metaclass=ResourceMetaclass): # Accessors #--------------------------------------------------------------------------- - @classmethod - def has_attribute(cls, name): - return name in [a.name for a in cls.attributes()] - def has_callback(self, action, attribute): return hasattr(self, '_{}_{}'.format(action, attribute.name)) diff --git a/vicn/core/resource_mgr.py b/vicn/core/resource_mgr.py index 84deb7b4..8db7f04d 100644 --- a/vicn/core/resource_mgr.py +++ b/vicn/core/resource_mgr.py @@ -22,6 +22,7 @@ import time # LXD workaround from pylxd.exceptions import NotFound as LxdNotFound, LXDAPIException +from netmodel.model.collection import Collection from netmodel.model.filter import Filter from netmodel.model.query import Query, ACTION_SELECT, ACTION_INSERT from netmodel.model.query import ACTION_UPDATE, ACTION_SUBSCRIBE @@ -33,10 +34,10 @@ from netmodel.util.meta import inheritors from netmodel.util.singleton import Singleton from netmodel.util.misc import is_iterable from vicn.core.attribute import NEVER_SET +from vicn.core.commands import ReturnValue from vicn.core.exception import VICNException, ResourceNotFound from vicn.core.resource_factory import ResourceFactory from vicn.core.resource import Resource, FactoryResource, EmptyResource -from vicn.core.sa_collections import InstrumentedList from vicn.core.state import InstanceState, ResourceState from vicn.core.state import AttributeState, Operations, PendingValue from vicn.core.task import TaskManager, wait_task, task, async_task @@ -46,12 +47,14 @@ 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 +ENABLE_LXD_WORKAROUND = False +DEFAULT_QTPLAYER_PORT = 8999 # Monitoring queries Q_SUB_VPP = 'SUBSCRIBE SUM(*) FROM interface WHERE device_name INCLUDED [{}]' Q_SUB_IF = 'SUBSCRIBE * FROM interface WHERE device_name == "{}"' +Q_SUB_VPP_IF = 'SUBSCRIBE * FROM vpp_interface WHERE device_name == "{}"' Q_SUB_STATS = 'SUBSCRIBE * FROM stats' Q_SUB_EMULATOR_IF = 'SUBSCRIBE * FROM interface WHERE id == "{}"' Q_SUB_EMULATOR = 'SUBSCRIBE * FROM interface WHERE device_name == "{}"' @@ -146,6 +149,8 @@ class ResourceManager(metaclass=Singleton): # Monitoring self._monitored = set() self._pending_monitoring = set() + self._map_ip_interface = dict() + self._monitored_channels = set() # For debug self._committed = set() @@ -200,6 +205,13 @@ class ResourceManager(metaclass=Singleton): def _broadcast_packet(self, packet): self._broadcast(packet.to_query()) + def _on_qtplayer_packet(self, name, packet): + query = packet.to_query() + query.params['name'] = name + query.reply = True + self._ws.execute(query) + return None + def _on_ns_record(self, packet): query = packet.to_query() @@ -240,6 +252,35 @@ class ResourceManager(metaclass=Singleton): return None return None return None + elif query.object_name == 'vpp_interface': + device_names = [value for key, op, value in query.filter.to_list() + if key == 'device_name'] + if not device_names: + log.error('No device name in packet=', packet) + return + device_name = device_names[0] + node_name = query.params['node'] + node = ResourceManager().by_name(node_name) + if node is None: + print("no node") + return None + for interface in node.interfaces: + if not hasattr(interface, 'vppinterface') or not interface.vppinterface: + continue + if interface.vppinterface.device_name == device_name: + if interface.channel: + f = Filter.from_list([['id', '==', + interface.channel._state.uuid._uuid]]) + q = Query(ACTION_UPDATE, 'channel', filter = f, + params = query.params) + q.reply = True + self._ws.execute(q) + return None + print("no channel") + return None + print("no vpp interface found") + return None + return None def _on_netmon_channel_record(self, packet): @@ -546,14 +587,8 @@ class ResourceManager(metaclass=Singleton): await self.wait_attr_init(resource, attribute) return resource.get(attribute) - async def attribute_set(self, resource, attribute_name, value, - blocking=True): + async def _attribute_set(self, resource, attribute_name, value): with await resource._state.write_lock: - # Add the current operation to the pending list - # NOTE: collections are unordered and can be updated concurrently - #self._attribute_set_pending_value(resource, attribute_name) - resource._state.dirty[attribute_name].trigger(Operations.SET, - value) attr_state = resource._state.attr_state[attribute_name] if attr_state == AttributeState.CLEAN: @@ -597,8 +632,26 @@ class ResourceManager(metaclass=Singleton): raise RuntimeError("Resource cannot be in state".format( resource_state)) - if blocking: - await self.wait_attr_clean(resource, attribute_name) +# if blocking: +# await self.wait_attr_clean(resource, attribute_name) + + def attribute_set(self, resource, attribute_name, value): + # Add the current operation to the pending list + # NOTE: collections are unordered and can be updated concurrently + #self._attribute_set_pending_value(resource, attribute_name) + resource._state.dirty[attribute_name].trigger(Operations.SET, + value) + asyncio.ensure_future(self._attribute_set(resource, attribute_name, value)) + + async def attribute_set_async(self, resource, attribute_name, value): + # Add the current operation to the pending list + # NOTE: collections are unordered and can be updated concurrently + #self._attribute_set_pending_value(resource, attribute_name) + resource._state.dirty[attribute_name].trigger(Operations.SET, + value) + await self._attribute_set(resource, attribute_name, value) + + #--------------------------------------------------------------------------- # Resource dependency management @@ -623,7 +676,7 @@ class ResourceManager(metaclass=Singleton): deps = [deps] for dep in deps: - if attr.key: + if resource.has_key_attribute(attr): if not dep.managed: continue dep_pfx = '{}:{}'.format(dep.get_type(), dep.get_uuid()) @@ -687,13 +740,7 @@ class ResourceManager(metaclass=Singleton): async def _resource_wait_subresources(self, resource): self.log(resource, S_WAIT_SRS) - # We should accumulate subresources through the hierarchy - sr = EmptyResource() - for base in reversed(resource.__class__.mro()): - if '__subresources__' not in vars(base): - continue - sr = sr > base.__subresources__(resource) - + sr = resource.__subresources__() if sr is not None and not isinstance(sr, EmptyResource): resource.set_subresources(sr) pfx_sr = '{}:{}'.format(sr.get_type(), sr.get_uuid()) @@ -713,19 +760,8 @@ class ResourceManager(metaclass=Singleton): """Perform action: __get__, __create__, __delete__ on the full class hierarchy. """ - task = EmptyTask() - for base in reversed(resource.__class__.mro()): - # To avoid adding several times the same task - if action not in vars(base): - continue - func = getattr(base, action, None) - if func is None: - continue - t = func(resource) - - task = task > t - - return task + method = getattr(resource, action, None) + return method() if method else EmptyTask() #-------------------------------------------------------------------------- # Resource model @@ -788,7 +824,11 @@ class ResourceManager(metaclass=Singleton): ret = fut.result() resource._state.attr_change_success[attribute.name] = True resource._state.attr_change_value[attribute.name] = ret + except ResourceNotFound as e: + resource._state.attr_change_success[attribute.name] = False + resource._state.attr_change_value[attribute.name] = e except Exception as e: + import traceback; traceback.print_exc() resource._state.attr_change_success[attribute.name] = False resource._state.attr_change_value[attribute.name] = e resource._state.attr_change_event[attribute.name].set() @@ -899,7 +939,10 @@ class ResourceManager(metaclass=Singleton): attrs = resource._state.attr_change_value[attribute.name] self.attr_log(resource, attribute, 'INIT success. Value = {}'.format(attrs)) - found = self._process_attr_dict(resource, attribute, attrs) + if not isinstance(attrs, ReturnValue): + found = self._process_attr_dict(resource, attribute, attrs) + else: + found = self._process_attr_dict(resource, attribute, attrs.stdout) if not found: log.error('Attribute missing return attrs: {}'.format( attrs)) @@ -923,7 +966,7 @@ class ResourceManager(metaclass=Singleton): if resource._state.attr_change_success[attribute.name] == True: self.attr_log(resource, attribute, 'UPDATE success. Value = {}. Attribute is CLEAN'.format(attrs)) - if attrs != NEVER_SET: + if not isinstance(attrs, ReturnValue) and attrs != NEVER_SET: # None could be interpreted as the return value. Also, # we need not to overwrite the value from get self._process_attr_dict(resource, attribute, attrs) @@ -931,7 +974,7 @@ class ResourceManager(metaclass=Singleton): # We might do this for all returned attributes cur_value = vars(resource)[attribute.name] if attribute.is_collection: - tmp = InstrumentedList(pending_value.value) + tmp = Collection(pending_value.value) tmp._attribute = cur_value._attribute tmp._instance = cur_value._instance else: @@ -974,8 +1017,25 @@ class ResourceManager(metaclass=Singleton): return Query.from_dict(dic) + def _monitor_qtplayer(self, resource): + try: + ip = resource.node.hostname + except: + ip = str(resource.node.management_interface.ip4_address) + + hook = functools.partial(self._on_qtplayer_packet, resource.node.name) + ws = self._router.add_interface('websocketclient', address=ip, + port = DEFAULT_QTPLAYER_PORT, + hook = hook) + q_str = 'SUBSCRIBE * FROM stats' + q = self.parse_query(q_str) + packet = Packet.from_query(q) + self._router._flow_table.add(packet, None, set([ws])) + ws.send(packet) + def _monitor_netmon(self, resource): - ip = resource.node.management_interface.ip4_address + print("MONITOR NODE", resource.node) + ip = str(resource.node.management_interface.ip4_address) if not ip: log.error('IP of monitored Node {} is None'.format(resource.node)) import os; os._exit(1) @@ -986,49 +1046,146 @@ class ResourceManager(metaclass=Singleton): node = resource.node for interface in node.interfaces: if not interface.monitored: + print("non monitored interface", interface) continue + print("NETMON MONITOR INTERFACE", interface) + +# if interface.get_type() == 'dpdkdevice' and hasattr(node,'vpp'): +# +# # Check if vICN has already subscribed for one interface in +# # the channel +# if hasattr(interface.channel,'already_subscribed'): +# continue +# +# channel_id = interface.channel._state.uuid._uuid +# +# update_vpp = functools.partial(self._on_vpp_record, +# pylink_id = channel_id) +# ws_vpp = self._router.add_interface('websocketclient', +# address=ip, hook=update_vpp) +# +# aggregate_interfaces = list() +# for _interface in node.interfaces: +# if not _interface.get_type() == 'dpdkdevice' and \ +# _interface.monitored: +# aggregate_interfaces.append('"' + +# _interface.device_name + '"') +# +# q_str = Q_SUB_VPP.format(','.join(aggregate_interfaces)) +# q = self.parse_query(q_str) +# packet = Packet.from_query(q) +# self._router._flow_table.add(packet, None, ws_vpp) +# ws_vpp.send(packet) +# +# # Prevent vICN to subscribe to other interfaces of the same +# # channel +# interface.channel.already_subscribed = True +# +# else: + if hasattr(node, 'vpp') and node.vpp is not None: + q_str = Q_SUB_VPP_IF.format(interface.vppinterface.device_name) + else: + q_str = Q_SUB_IF.format(interface.device_name) + log.warning(" -- MONITOR {}".format(q_str)) + q = self.parse_query(q_str) + packet = Packet.from_query(q) + self._router._flow_table.add(packet, None, set([ws])) + ws.send(packet) - if interface.get_type() == 'dpdkdevice' and hasattr(node,'vpp'): + def _monitor_vpp_interface(self, vpp_interface): + print("MONITOR interface", vpp_interface) + interface = vpp_interface.parent + node = interface.node + # XXX only monitor in the topology group + if node.get_type() != 'lxccontainer': + print("MONITOR -> Ignored: not in container") + return - # Check if vICN has already subscribed for one interface in - # the channel - if hasattr(interface.channel,'already_subscribed'): - continue + # We only monitor interfaces to provide data for wired channels + channel = interface.channel + if channel is None: + print("MONITOR -> Ignored: no channel") + return + if channel.has_type('emulatedchannel'): + print("MONITOR -> Ignored: belong to wireless channel") + return - channel_id = interface.channel._state.uuid._uuid + # Don't monitor multiple interfaces per channel + if channel in self._monitored_channels: + print("MONITOR -> Ignored: channel already monitored") + return + self._monitored_channels.add(channel) - update_vpp = functools.partial(self._on_vpp_record, - pylink_id = channel_id) - ws_vpp = self._router.add_interface('websocketclient', - address=ip, hook=update_vpp) + ip = str(node.management_interface.ip4_address) + if not ip: + log.error('IP of monitored Node {} is None'.format(resource.node)) + import os; os._exit(1) - aggregate_interfaces = list() - for _interface in node.interfaces: - if not _interface.get_type() == 'dpdkdevice' and \ - _interface.monitored: - aggregate_interfaces.append('"' + - _interface.device_name + '"') + # Reuse existing websockets + ws = self._map_ip_interface.get(ip) + if not ws: + ws = self._router.add_interface('websocketclient', address=ip, + hook=self._on_netmon_record) + self._map_ip_interface[ip] = ws + + q_str = Q_SUB_VPP_IF.format(vpp_interface.device_name) + print("MONITOR -> query= {}".format(q_str)) + q = self.parse_query(q_str) + packet = Packet.from_query(q) + self._router._flow_table.add(packet, None, set([ws])) + ws.send(packet) + + def _monitor_interface(self, interface): + print("MONITOR interface", interface) + node = interface.node + # XXX only monitor in the topology group + if node.get_type() != 'lxccontainer': + print("MONITOR -> Ignored: not in container") + return - q_str = Q_SUB_VPP.format(','.join(aggregate_interfaces)) - q = self.parse_query(q_str) - packet = Packet.from_query(q) - self._router._flow_table.add(packet, None, ws_vpp) - ws_vpp.send(packet) + # Only monitor vpp interfaces on vpp node + if hasattr(node, 'vpp') and node.vpp is not None: + print("MONITOR -> Ignored: non-vpp interface on vpp node") + return - # Prevent vICN to subscribe to other interfaces of the same - # channel - interface.channel.already_subscribed = True + # We only monitor interfaces to provide data for wired channels + channel = interface.channel + if channel is None: + print("MONITOR -> Ignored: no channel") + return + if channel.has_type('emulatedchannel'): + print("MONITOR -> Ignored: belong to wireless channel") + return - else: - q_str = Q_SUB_IF.format(interface.device_name) - q = self.parse_query(q_str) - packet = Packet.from_query(q) - self._router._flow_table.add(packet, None, ws) - ws.send(packet) + # Don't monitor multiple interfaces per channel + if channel in self._monitored_channels: + print("MONITOR -> Ignored: channel already monitored") + return + self._monitored_channels.add(channel) + + ip = str(node.management_interface.ip4_address) + if not ip: + log.error('IP of monitored Node {} is None'.format(resource.node)) + import os; os._exit(1) + + # Reuse existing websockets + ws = self._map_ip_interface.get(ip) + if not ws: + ws = self._router.add_interface('websocketclient', address=ip, + hook=self._on_netmon_record) + self._map_ip_interface[ip] = ws + + q_str = Q_SUB_IF.format(interface.device_name) + print("MONITOR -> query= {}".format(q_str)) + q = self.parse_query(q_str) + packet = Packet.from_query(q) + self._router._flow_table.add(packet, None, set([ws])) + ws.send(packet) def _monitor_emulator(self, resource): ns = resource - ip = ns.node.bridge.ip4_address # management_interface.ip_address + # XXX UGLY, we have no management interface + ip = ns.node.hostname # str(ns.node.interfaces[0].ip4_address) ws_ns = self._router.add_interface('websocketclient', address = ip, port = ns.control_port, @@ -1050,7 +1207,7 @@ class ResourceManager(metaclass=Singleton): q_str = Q_SUB_EMULATOR_IF.format(identifier) q = self.parse_query(q_str) packet = Packet.from_query(q) - self._router._flow_table.add(packet, None, ws_ns) + self._router._flow_table.add(packet, None, set([ws_ns])) ws_ns.send(packet) # We also need to subscribe on the node for the tap interfaces @@ -1059,7 +1216,7 @@ class ResourceManager(metaclass=Singleton): q_str = Q_SUB_EMULATOR.format(tap.device_name) q = self.parse_query(q_str) packet = Packet.from_query(q) - self._router._flow_table.add(packet, None, ws) + self._router._flow_table.add(packet, None, set([ws])) ws.send(packet) def _monitor(self, resource): @@ -1070,29 +1227,37 @@ class ResourceManager(metaclass=Singleton): self._pending_monitoring.clear() return - central_ip = self.by_type_str('centralip') - if not central_ip: - raise NotImplementedError('Missing CentralIP in experiment') - central_ip = central_ip[0] - uuid = resource.get_uuid() - if central_ip._state.state != ResourceState.CLEAN: - self._pending_monitoring.add(uuid) - return + central_ip = self.by_type_str('centralip') + if central_ip: + central_ip = central_ip[0] + + if central_ip._state.state != ResourceState.CLEAN: + self._pending_monitoring.add(uuid) + return if uuid in self._monitored: return self._monitored.add(uuid) - if resource.get_type() == 'netmon': - if resource.node.get_type() != 'lxccontainer': - return - self._monitor_netmon(resource) +# if resource.get_type() == 'netmon': +# if resource.node.get_type() != 'lxccontainer': +# return +# self._monitor_netmon(resource) + + if resource.get_type() == 'qtplayer': + self._monitor_qtplayer(resource) elif resource.has_type('emulatedchannel'): self._monitor_emulator(resource) + elif resource.has_type('interface'): + self._monitor_interface(resource) + + elif resource.has_type('vppinterface'): + self._monitor_vpp_interface(resource) + async def __set_resource_state(self, resource, state): """Sets the resource state (no-lock version) @@ -1128,6 +1293,9 @@ class ResourceManager(metaclass=Singleton): ret = fut.result() resource._state.change_success = True resource._state.change_value = ret + except ResourceNotFound as e: + resource._state.change_success = False + resource._state.change_value = e except Exception as e: resource._state.change_success = False resource._state.change_value = e @@ -1148,7 +1316,7 @@ class ResourceManager(metaclass=Singleton): for attr in resource.iter_attributes(): if resource.is_local_attribute(attr.name): continue - if attr.key: + if resource.has_key_attribute(attr): # Those attributes are already done continue @@ -1182,12 +1350,14 @@ class ResourceManager(metaclass=Singleton): # Monitor all FSM one by one and inform about errors. futs = list() attrs = list() - for attr in resource.get_keys(): - if resource.is_local_attribute(attr.name): - continue - attrs.append(attr) - fut = self.attribute_process(resource, attr) - futs.append(fut) + + for key in resource.get_keys(): + for attr in key: + if resource.is_local_attribute(attr.name): + continue + attrs.append(attr) + fut = self.attribute_process(resource, attr) + futs.append(fut) if not futs: self.log(resource, 'No key attribute to update') @@ -1277,6 +1447,7 @@ class ResourceManager(metaclass=Singleton): print("------") import traceback; traceback.print_tb(e.__traceback__) log.error('Resource: {} - Exception: {}'.format(pfx, e)) + return import os; os._exit(1) elif state == ResourceState.UNINITIALIZED: pending_state = ResourceState.PENDING_DEPS @@ -1398,8 +1569,9 @@ class ResourceManager(metaclass=Singleton): elif pending_state == ResourceState.PENDING_GET: if resource._state.change_success == True: attrs = resource._state.change_value - self.log(resource, S_INIT_DONE.format(attrs)) - self._process_attr_dict(resource, None, attrs) + if not isinstance(attrs, ReturnValue): + self.log(resource, S_INIT_DONE.format(attrs)) + self._process_attr_dict(resource, None, attrs) new_state = ResourceState.CREATED else: e = resource._state.change_value @@ -1452,8 +1624,9 @@ class ResourceManager(metaclass=Singleton): elif pending_state == ResourceState.PENDING_CREATE: if resource._state.change_success == True: attrs = resource._state.change_value - self.log(resource, S_CREATE_OK.format(attrs)) - self._process_attr_dict(resource, None, attrs) + if not isinstance(attrs, ReturnValue): + self.log(resource, S_CREATE_OK.format(attrs)) + self._process_attr_dict(resource, None, attrs) new_state = ResourceState.CREATED else: e = resource._state.change_value diff --git a/vicn/core/sa_collections.py b/vicn/core/sa_collections.py deleted file mode 100644 index a4a24f85..00000000 --- a/vicn/core/sa_collections.py +++ /dev/null @@ -1,264 +0,0 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- -# -# This module is derived from code from SQLAlchemy -# -# orm/collections.py -# Copyright (C) 2005-2016 the SQLAlchemy authors and contributors -# -# This module is part of SQLAlchemy and is released under -# the MIT License: http://www.opensource.org/licenses/mit-license.php -# - -import logging - -from vicn.core.exception import VICNListException -from vicn.core.sa_compat import py2k -from vicn.core.state import UUID - -log = logging.getLogger(__name__) - -def _list_decorators(): - """Tailored instrumentation wrappers for any list-like class.""" - - def _tidy(fn): - fn._sa_instrumented = True - fn.__doc__ = getattr(list, fn.__name__).__doc__ - - def append(fn): - def append(self, item): - try: - item = self._attribute.do_list_add(self._instance, item) - fn(self, item) - except VICNListException as e: - pass - _tidy(append) - return append - - def remove(fn): - def remove(self, value): - # testlib.pragma exempt:__eq__ - try: - self._attribute.do_list_remove(self._instance, value) - fn(self, value) - except : pass - _tidy(remove) - return remove - - def insert(fn): - def insert(self, index, value): - try: - value = self._attribute.do_list_add(self._instance, item) - fn(self, index, value) - except : pass - _tidy(insert) - return insert - - def __getitem__(fn): - def __getitem__(self, index): - item = fn(self, index) - return self._attribute.handle_getitem(self._instance, item) - _tidy(__getitem__) - return __getitem__ - - def __setitem__(fn): - def __setitem__(self, index, value): - if not isinstance(index, slice): - existing = self[index] - if existing is not None: - try: - self._attribute.do_list_remove(self._instance, existing) - except: pass - try: - value = self._attribute.do_list_add(self._instance, value) - fn(self, index, value) - except: pass - else: - # slice assignment requires __delitem__, insert, __len__ - step = index.step or 1 - start = index.start or 0 - if start < 0: - start += len(self) - if index.stop is not None: - stop = index.stop - else: - stop = len(self) - if stop < 0: - stop += len(self) - - if step == 1: - for i in range(start, stop, step): - if len(self) > start: - del self[start] - - for i, item in enumerate(value): - self.insert(i + start, item) - else: - rng = list(range(start, stop, step)) - if len(value) != len(rng): - raise ValueError( - "attempt to assign sequence of size %s to " - "extended slice of size %s" % (len(value), - len(rng))) - for i, item in zip(rng, value): - self.__setitem__(i, item) - _tidy(__setitem__) - return __setitem__ - - def __delitem__(fn): - def __delitem__(self, index): - if not isinstance(index, slice): - item = self[index] - try: - self._attribute.do_list_remove(self._instance, item) - fn(self, index) - except : pass - else: - # slice deletion requires __getslice__ and a slice-groking - # __getitem__ for stepped deletion - # note: not breaking this into atomic dels - has_except = False - for item in self[index]: - try: - self._attribute.do_list_remove(self._instance, item) - except : has_except = True - if not has_except: - fn(self, index) - _tidy(__delitem__) - return __delitem__ - - if py2k: - def __setslice__(fn): - def __setslice__(self, start, end, values): - has_except = False - for value in self[start:end]: - try: - self._attribute.do_list_remove(self._instance, value) - except : has_except = True - #values = [self._attribute.do_list_add(self._instance, value) for value in values] - _values = list() - for value in values: - try: - _values.append(self._attribute.do_list_add(self._instance, value)) - except: has_except = True - if not has_except: - fn(self, start, end, _values) - _tidy(__setslice__) - return __setslice__ - - def __delslice__(fn): - def __delslice__(self, start, end): - has_except = False - for value in self[start:end]: - try: - self._attribute.do_list_remove(self._instance, value) - except : has_except = True - if not has_except: - fn(self, start, end) - _tidy(__delslice__) - return __delslice__ - - def extend(fn): - def extend(self, iterable): - for value in iterable: - self.append(value) - _tidy(extend) - return extend - - def __iadd__(fn): - def __iadd__(self, iterable): - # list.__iadd__ takes any iterable and seems to let TypeError - # raise as-is instead of returning NotImplemented - for value in iterable: - self.append(value) - return self - _tidy(__iadd__) - return __iadd__ - - def pop(fn): - def pop(self, index=-1): - try: - self._attribute.do_list_remove(self._instance, item) - item = fn(self, index) - return item - except : return None - _tidy(pop) - return pop - - def __iter__(fn): - def __iter__(self): - for item in fn(self): - yield self._attribute.handle_getitem(self._instance, item) - _tidy(__iter__) - return __iter__ - - def __repr__(fn): - def __repr__(self): - return '<Collection {} {}>'.format(id(self), list.__repr__(self)) - _tidy(__repr__) - return __repr__ - - __str__ = __repr__ - #def __str__(fn): - # def __str__(self): - # return str(list(self)) - # _tidy(__str__) - # return __str__ - - if not py2k: - def clear(fn): - def clear(self, index=-1): - has_except = False - for item in self: - try: - self._attribute.do_list_remove(self._instance, item) - except : has_except = True - if not has_except: - fn(self) - _tidy(clear) - return clear - - # __imul__ : not wrapping this. all members of the collection are already - # present, so no need to fire appends... wrapping it with an explicit - # decorator is still possible, so events on *= can be had if they're - # desired. hard to imagine a use case for __imul__, though. - - l = locals().copy() - l.pop('_tidy') - return l - -def _instrument_list(cls): - # inspired by sqlalchemy - for method, decorator in _list_decorators().items(): - fn = getattr(cls, method, None) - if fn: - #if (fn and method not in methods and - # not hasattr(fn, '_sa_instrumented')): - setattr(cls, method, decorator(fn)) - -class InstrumentedList(list): - - @classmethod - def from_list(cls, value, instance, attribute): - lst = list() - if value: - for x in value: - if isinstance(x, UUID): - x = instance.from_uuid(x) - lst.append(x) - # Having a class method is important for inheritance - value = cls(lst) - value._attribute = attribute - value._instance = instance - return value - - def __contains__(self, key): - from vicn.core.resource import Resource - if isinstance(key, Resource): - key = key.get_uuid() - return list.__contains__(self, key) - - def __lshift__(self, item): - self.append(item) - -_instrument_list(InstrumentedList) diff --git a/vicn/core/sa_compat.py b/vicn/core/sa_compat.py deleted file mode 100644 index 34211455..00000000 --- a/vicn/core/sa_compat.py +++ /dev/null @@ -1,270 +0,0 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- -# -# This module originates from SQLAlchemy -# -# util/compat.py -# Copyright (C) 2005-2016 the SQLAlchemy authors and contributors -# -# This module is part of SQLAlchemy and is released under -# the MIT License: http://www.opensource.org/licenses/mit-license.php -# - -"""Handle Python version/platform incompatibilities.""" - -import sys - -try: - import threading -except ImportError: - import dummy_threading as threading - -py36 = sys.version_info >= (3, 6) -py33 = sys.version_info >= (3, 3) -py32 = sys.version_info >= (3, 2) -py3k = sys.version_info >= (3, 0) -py2k = sys.version_info < (3, 0) -py265 = sys.version_info >= (2, 6, 5) -jython = sys.platform.startswith('java') -pypy = hasattr(sys, 'pypy_version_info') -win32 = sys.platform.startswith('win') -cpython = not pypy and not jython # TODO: something better for this ? - -import collections -next = next - -if py3k: - import pickle -else: - try: - import cPickle as pickle - except ImportError: - import pickle - -# work around http://bugs.python.org/issue2646 -if py265: - safe_kwarg = lambda arg: arg -else: - safe_kwarg = str - -ArgSpec = collections.namedtuple("ArgSpec", - ["args", "varargs", "keywords", "defaults"]) - -if py3k: - import builtins - - from inspect import getfullargspec as inspect_getfullargspec - from urllib.parse import (quote_plus, unquote_plus, - parse_qsl, quote, unquote) - import configparser - from io import StringIO - - from io import BytesIO as byte_buffer - - def inspect_getargspec(func): - return ArgSpec( - *inspect_getfullargspec(func)[0:4] - ) - - string_types = str, - binary_types = bytes, - binary_type = bytes - text_type = str - int_types = int, - iterbytes = iter - - def u(s): - return s - - def ue(s): - return s - - def b(s): - return s.encode("latin-1") - - if py32: - callable = callable - else: - def callable(fn): - return hasattr(fn, '__call__') - - def cmp(a, b): - return (a > b) - (a < b) - - from functools import reduce - - print_ = getattr(builtins, "print") - - import_ = getattr(builtins, '__import__') - - import itertools - itertools_filterfalse = itertools.filterfalse - itertools_filter = filter - itertools_imap = map - from itertools import zip_longest - - import base64 - - def b64encode(x): - return base64.b64encode(x).decode('ascii') - - def b64decode(x): - return base64.b64decode(x.encode('ascii')) - -else: - from inspect import getargspec as inspect_getfullargspec - inspect_getargspec = inspect_getfullargspec - from urllib import quote_plus, unquote_plus, quote, unquote - from urlparse import parse_qsl - import ConfigParser as configparser - from StringIO import StringIO - from cStringIO import StringIO as byte_buffer - - string_types = basestring, - binary_types = bytes, - binary_type = str - text_type = unicode - int_types = int, long - - def iterbytes(buf): - return (ord(byte) for byte in buf) - - def u(s): - # this differs from what six does, which doesn't support non-ASCII - # strings - we only use u() with - # literal source strings, and all our source files with non-ascii - # in them (all are tests) are utf-8 encoded. - return unicode(s, "utf-8") - - def ue(s): - return unicode(s, "unicode_escape") - - def b(s): - return s - - def import_(*args): - if len(args) == 4: - args = args[0:3] + ([str(arg) for arg in args[3]],) - return __import__(*args) - - callable = callable - cmp = cmp - reduce = reduce - - import base64 - b64encode = base64.b64encode - b64decode = base64.b64decode - - def print_(*args, **kwargs): - fp = kwargs.pop("file", sys.stdout) - if fp is None: - return - for arg in enumerate(args): - if not isinstance(arg, basestring): - arg = str(arg) - fp.write(arg) - - import itertools - itertools_filterfalse = itertools.ifilterfalse - itertools_filter = itertools.ifilter - itertools_imap = itertools.imap - from itertools import izip_longest as zip_longest - - -import time -if win32 or jython: - time_func = time.clock -else: - time_func = time.time - -from collections import namedtuple -from operator import attrgetter as dottedgetter - - -if py3k: - def reraise(tp, value, tb=None, cause=None): - if cause is not None: - assert cause is not value, "Same cause emitted" - value.__cause__ = cause - if value.__traceback__ is not tb: - raise value.with_traceback(tb) - raise value - -else: - # not as nice as that of Py3K, but at least preserves - # the code line where the issue occurred - exec("def reraise(tp, value, tb=None, cause=None):\n" - " if cause is not None:\n" - " assert cause is not value, 'Same cause emitted'\n" - " raise tp, value, tb\n") - - -def raise_from_cause(exception, exc_info=None): - if exc_info is None: - exc_info = sys.exc_info() - exc_type, exc_value, exc_tb = exc_info - cause = exc_value if exc_value is not exception else None - reraise(type(exception), exception, tb=exc_tb, cause=cause) - -if py3k: - exec_ = getattr(builtins, 'exec') -else: - def exec_(func_text, globals_, lcl=None): - if lcl is None: - exec('exec func_text in globals_') - else: - exec('exec func_text in globals_, lcl') - - -def with_metaclass(meta, *bases): - """Create a base class with a metaclass. - - Drops the middle class upon creation. - - Source: http://lucumr.pocoo.org/2013/5/21/porting-to-python-3-redux/ - - """ - - class metaclass(meta): - __call__ = type.__call__ - __init__ = type.__init__ - - def __new__(cls, name, this_bases, d): - if this_bases is None: - return type.__new__(cls, name, (), d) - return meta(name, bases, d) - return metaclass('temporary_class', None, {}) - - -from contextlib import contextmanager - -try: - from contextlib import nested -except ImportError: - # removed in py3k, credit to mitsuhiko for - # workaround - - @contextmanager - def nested(*managers): - exits = [] - vars = [] - exc = (None, None, None) - try: - for mgr in managers: - exit = mgr.__exit__ - enter = mgr.__enter__ - vars.append(enter()) - exits.append(exit) - yield vars - except: - exc = sys.exc_info() - finally: - while exits: - exit = exits.pop() - try: - if exit(*exc): - exc = (None, None, None) - except: - exc = sys.exc_info() - if exc != (None, None, None): - reraise(exc[0], exc[1], exc[2]) diff --git a/vicn/core/state.py b/vicn/core/state.py index a116ba82..aa68341e 100644 --- a/vicn/core/state.py +++ b/vicn/core/state.py @@ -17,17 +17,9 @@ # import asyncio -import random -import string -class NEVER_SET: - pass - -# Separator for components of the UUID -UUID_SEP = '-' - -# Length of the random component of the UUID -UUID_LEN = 5 +from netmodel.model.uuid import UUID +from vicn.core.attribute import NEVER_SET class ResourceState: UNINITIALIZED = 'UNINITIALIZED' @@ -64,31 +56,6 @@ class Operations: LIST_REMOVE = 'remove' LIST_CLEAR = 'clear' -class UUID: - def __init__(self, name, cls): - self._uuid = self._make_uuid(name, cls) - - def _make_uuid(self, name, cls): - """Generate a unique resource identifier - - The UUID consists in the type of the resource, to which is added a - random identifier of length UUID_LEN. Components of the UUID are - separated by UUID_SEP. - """ - uuid = ''.join(random.choice(string.ascii_uppercase + string.digits) - for _ in range(UUID_LEN)) - if name: - uuid = name # + UUID_SEP + uuid - return UUID_SEP.join([cls.__name__, uuid]) - - def __repr__(self): - return '<UUID {}>'.format(self._uuid) - - def __lt__(self, other): - return self._uuid < other._uuid - - __str__ = __repr__ - class PendingValue: def __init__(self, value = None): self.clear(value) diff --git a/vicn/core/task.py b/vicn/core/task.py index 49c34b1f..72c80716 100644 --- a/vicn/core/task.py +++ b/vicn/core/task.py @@ -24,11 +24,12 @@ import shlex import subprocess import os -from vicn.core.scheduling_algebra import SchedulingAlgebra +from netmodel.util.process import execute_local + from vicn.core.commands import ReturnValue -from vicn.core.exception import ResourceNotFound from vicn.core.commands import Command, SequentialCommands -from netmodel.util.process import execute_local +from vicn.core.exception import ResourceNotFound +from vicn.core.scheduling_algebra import SchedulingAlgebra log = logging.getLogger(__name__) @@ -150,6 +151,86 @@ async def wait_concurrent_tasks(tasks): wait_task_task = async_task(wait_task) +#------------------------------------------------------------------------------ +# Task inheritance +#------------------------------------------------------------------------------ + +# NOTES: +# - delete is a special case where we have to reverse operations +# - subresources is also a special case, since it deals with resources, but it +# works the same since the algebra is similar + +def inherit_parent(fn): + def f(self, *args, **kwargs): + # Break loops + if fn.__name__ not in vars(self.__class__): + return fn(self, *args, **kwargs) + + parent_tasks = EmptyTask() + for parent in self.__class__.__bases__: + if not fn.__name__ in vars(parent): + continue + for cls in self.__class__.__mro__: + if cls.__dict__.get(fn.__name__) is fn: + break + + parent_task = getattr(super(cls, self), fn.__name__, None) + if not parent_task: + continue + parent_tasks = parent_tasks | parent_task(self) + if fn.__name__ == 'delete': + return fn(*args, **kwargs) > parent_tasks + else: + return parent_tasks > fn(self, *args, **kwargs) + return f + +def override_parent(fn): + def f(self, *args, **kwargs): + # Break loops + if fn.__name__ not in vars(self.__class__): + return fn(self, *args, **kwargs) + + bases = set([ancestor for parent in self.__class__.__bases__ + for ancestor in parent.__bases__]) + + ancestor_tasks = EmptyTask() + for base in bases: + if not fn.__name__ in vars(base): + continue + ancestor_task = getattr(base, fn.__name__, None) + if not ancestor_task: + continue + ancestor_tasks = ancestor_tasks | ancestor_task(self) + if fn.__name__ == 'delete': + return fn(*args, **kwargs) > ancestor_tasks + else: + return ancestor_tasks > fn(self, *args, **kwargs) + return f + +def inherit(*classes): + def decorator(fn): + def f(self, *args, **kwargs): + # Break loops + if fn.__name__ not in vars(self.__class__): + return fn(self, *args, **kwargs) + + parent_tasks = EmptyTask() + for parent in classes: + if not fn.__name__ in vars(parent): + continue + parent_task = getattr(parent, fn.__name__, None) + if not parent_task: + continue + parent_tasks = parent_tasks | parent_task(self) + if fn.__name__ == 'delete': + return fn(*args, **kwargs) > parent_tasks + else: + return parent_tasks > fn(self, *args, **kwargs) + return f + return decorator + +#------------------------------------------------------------------------------ + def get_attribute_task(resource, attribute_name): @async_task async def func(): @@ -308,7 +389,10 @@ class BashTask(Task): def get_full_cmd(self): c = SequentialCommands() desc = None - for line in self._cmd.splitlines(): + cmd = self._cmd + if callable(cmd): + cmd = cmd() + for line in cmd.splitlines(): line = line.strip() if not line: continue @@ -341,10 +425,10 @@ class BashTask(Task): # executed, so that any object passed as parameters is deferenced right # on time. cmd = self.get_full_cmd() - partial = functools.partial(func, cmd, output = bool(self._parse)) + partial = functools.partial(func, cmd, output = self._output or bool(self._parse)) node_str = self._node.name if self._node else '(LOCAL)' - cmd_str = cmd[:77] + '...' if len(cmd) > 80 else cmd + cmd_str = cmd#[:77] + '...' if len(cmd) > 80 else cmd log.info('Execute: {} - {}'.format(node_str, cmd_str)) if self._lock: |