-#!/usr/bin/env python
# -*- coding: utf-8 -*-
from nepi.core import execute
-from nepi.core.metadata import Metadata
+from nepi.core.metadata import Metadata, Parallel
from nepi.util import validation
from nepi.util.constants import TIME_NOW, \
ApplicationStatus as AS, \
TestbedStatus as TS, \
CONNECTION_DELAY
+from nepi.util.parallel import ParallelRun
import collections
import copy
+import logging
class TestbedController(execute.TestbedController):
def __init__(self, testbed_id, testbed_version):
self._factories[factory.factory_id] = factory
self._attributes = self._metadata.testbed_attributes()
self._root_directory = None
+
+ # Logging
+ self._logger = logging.getLogger("nepi.core.testbed_impl")
@property
def root_directory(self):
self._add_address[guid] = list()
self._add_address[guid].append((address, netprefix, broadcast))
- def defer_add_route(self, guid, destination, netprefix, nexthop, metric = 0):
+ def defer_add_route(self, guid, destination, netprefix, nexthop,
+ metric = 0, device = None):
self._validate_guid(guid)
self._validate_allow_routes(guid)
if not guid in self._add_route:
self._add_route[guid] = list()
- self._add_route[guid].append((destination, netprefix, nexthop, metric))
+ self._add_route[guid].append((destination, netprefix, nexthop,
+ metric, device))
def do_setup(self):
self._root_directory = self._attributes.\
self._status = TS.STATUS_CONNECTED
def _do_in_factory_order(self, action, order, postaction = None, poststep = None):
+ logger = self._logger
+
guids = collections.defaultdict(list)
# order guids (elements) according to factory_id
for guid, factory_id in self._create.iteritems():
guids[factory_id].append(guid)
+
# configure elements following the factory_id order
for factory_id in order:
+ # Create a parallel runner if we're given a Parallel() wrapper
+ runner = None
+ if isinstance(factory_id, Parallel):
+ runner = ParallelRun(factory_id.maxthreads)
+ factory_id = factory_id.factory
+
# omit the factories that have no element to create
if factory_id not in guids:
continue
+
+ # configure action
factory = self._factories[factory_id]
- if not getattr(factory, action):
+ if isinstance(action, basestring) and not getattr(factory, action):
continue
- for guid in guids[factory_id]:
- getattr(factory, action)(self, guid)
+ def perform_action(guid):
+ if isinstance(action, basestring):
+ getattr(factory, action)(self, guid)
+ else:
+ action(self, guid)
if postaction:
postaction(self, guid)
+
+ # perform the action on all elements, in parallel if so requested
+ if runner:
+ logger.debug("TestbedController: Starting parallel %s", action)
+ runner.start()
+
+ for guid in guids[factory_id]:
+ if runner:
+ logger.debug("TestbedController: Scheduling %s on %s", action, guid)
+ runner.put(perform_action, guid)
+ else:
+ logger.debug("TestbedController: Performing %s on %s", action, guid)
+ perform_action(guid)
+
+ # sync
+ if runner:
+ runner.sync()
+
+ # post hook
if poststep:
for guid in guids[factory_id]:
- poststep(self, guid)
+ if runner:
+ logger.debug("TestbedController: Scheduling post-%s on %s", action, guid)
+ runner.put(poststep, self, guid)
+ else:
+ logger.debug("TestbedController: Performing post-%s on %s", action, guid)
+ poststep(self, guid)
+
+ # sync
+ if runner:
+ runner.join()
+ logger.debug("TestbedController: Finished parallel %s", action)
@staticmethod
def do_poststep_preconfigure(self, guid):
cross_connector_type_name,
True)
if connect_code:
+ if hasattr(connect_code, "func"):
+ func_name = connect_code.func.__name__
+ elif hasattr(connect_code, "__name__"):
+ func_name = connect_code.__name__
+ else:
+ func_name = repr(connect_code)
+ self._logger.debug("Cross-connect - guid: %d, connect_code: %s " % (
+ guid, func_name))
elem_cross_data = cross_data[cross_testbed_guid][cross_guid]
connect_code(self, guid, elem_cross_data)
if status_function:
return status_function(self, guid)
return AS.STATUS_UNDETERMINED
+
+ def testbed_status(self):
+ return self._status
def trace(self, guid, trace_id, attribute='value'):
if attribute == 'value':
fd.close()
elif attribute == 'path':
content = self.trace_filepath(guid, trace_id)
+ elif attribute == 'filename':
+ content = self.trace_filename(guid, trace_id)
else:
content = None
return content
"""
raise NotImplementedError
+ def trace_filename(self, guid, trace_id):
+ """
+ Return a trace's file name, for TestbedController's default
+ implementation of trace()
+ """
+ raise NotImplementedError
+
#shutdown: NotImplementedError
def get_connected(self, guid, connector_type_name,
def _validate_testbed_value(self, name, value):
if not self._attributes.is_attribute_value_valid(name, value):
- raise AttributeError("Invalid value %s for testbed attribute %s" % \
+ raise AttributeError("Invalid value %r for testbed attribute %s" % \
(value, name))
def _validate_box_attribute(self, guid, name):
def _validate_box_value(self, guid, name, value):
factory = self._get_factory(guid)
if not factory.box_attributes.is_attribute_value_valid(name, value):
- raise AttributeError("Invalid value %s for attribute %s" % \
+ raise AttributeError("Invalid value %r for attribute %s" % \
(value, name))
def _validate_factory_attribute(self, guid, name):
def _validate_factory_value(self, guid, name, value):
factory = self._get_factory(guid)
if not factory.is_attribute_value_valid(name, value):
- raise AttributeError("Invalid value %s for attribute %s" % \
+ raise AttributeError("Invalid value %r for attribute %s" % \
(value, name))
def _validate_trace(self, guid, trace_name):
git://git.onelab.eu / nepi.git / blobdiff