src/nepi/core/testbed_impl.py

   1 #!/usr/bin/env python
   2 # -*- coding: utf-8 -*-
   3
   4 from nepi.core import execute
   5 from nepi.core.metadata import Metadata
   6 from nepi.util import validation
   7 from nepi.util.constants import TIME_NOW, \
   8         ApplicationStatus as AS, \
   9         TestbedStatus as TS, \
  10         CONNECTION_DELAY
  11
  12 import collections
  13 import copy
  14
  15 class TestbedController(execute.TestbedController):
  16     def __init__(self, testbed_id, testbed_version):
  17         super(TestbedController, self).__init__(testbed_id, testbed_version)
  18         self._status = TS.STATUS_ZERO
  19         # testbed attributes for validation
  20         self._attributes = None
  21         # element factories for validation
  22         self._factories = dict()
  23
  24         # experiment construction instructions
  25         self._create = dict()
  26         self._create_set = dict()
  27         self._factory_set = dict()
  28         self._connect = dict()
  29         self._cross_connect = dict()
  30         self._add_trace = dict()
  31         self._add_address = dict()
  32         self._add_route = dict()
  33         self._configure = dict()
  34
  35         # log of set operations
  36         self._setlog = dict()
  37         # last set operations
  38         self._set = dict()
  39
  40         # testbed element instances
  41         self._elements = dict()
  42
  43         self._metadata = Metadata(self._testbed_id, self._testbed_version)
  44         for factory in self._metadata.build_factories():
  45             self._factories[factory.factory_id] = factory
  46         self._attributes = self._metadata.testbed_attributes()
  47         self._root_directory = None
  48
  49     @property
  50     def root_directory(self):
  51         return self._root_directory
  52
  53     @property
  54     def guids(self):
  55         return self._create.keys()
  56
  57     @property
  58     def elements(self):
  59         return self._elements
  60
  61     def defer_configure(self, name, value):
  62         self._validate_testbed_attribute(name)
  63         self._validate_testbed_value(name, value)
  64         self._attributes.set_attribute_value(name, value)
  65         self._configure[name] = value
  66
  67     def defer_create(self, guid, factory_id):
  68         self._validate_factory_id(factory_id)
  69         self._validate_not_guid(guid)
  70         self._create[guid] = factory_id
  71
  72     def defer_create_set(self, guid, name, value):
  73         self._validate_guid(guid)
  74         self._validate_box_attribute(guid, name)
  75         self._validate_box_value(guid, name, value)
  76         if guid not in self._create_set:
  77             self._create_set[guid] = dict()
  78         self._create_set[guid][name] = value
  79
  80     def defer_factory_set(self, guid, name, value):
  81         self._validate_guid(guid)
  82         self._validate_factory_attribute(guid, name)
  83         self._validate_factory_value(guid, name, value)
  84         if guid not in self._factory_set:
  85             self._factory_set[guid] = dict()
  86         self._factory_set[guid][name] = value
  87
  88     def defer_connect(self, guid1, connector_type_name1, guid2,
  89             connector_type_name2):
  90         self._validate_guid(guid1)
  91         self._validate_guid(guid2)
  92         factory1 = self._get_factory(guid1)
  93         factory_id2 = self._create[guid2]
  94         connector_type = factory1.connector_type(connector_type_name1)
  95         connector_type.can_connect(self._testbed_id, factory_id2,
  96                 connector_type_name2, False)
  97         self._validate_connection(guid1, connector_type_name1, guid2,
  98             connector_type_name2)
  99
 100         if not guid1 in self._connect:
 101             self._connect[guid1] = dict()
 102         if not connector_type_name1 in self._connect[guid1]:
 103              self._connect[guid1][connector_type_name1] = dict()
 104         self._connect[guid1][connector_type_name1][guid2] = \
 105                connector_type_name2
 106         if not guid2 in self._connect:
 107             self._connect[guid2] = dict()
 108         if not connector_type_name2 in self._connect[guid2]:
 109              self._connect[guid2][connector_type_name2] = dict()
 110         self._connect[guid2][connector_type_name2][guid1] = \
 111                connector_type_name1
 112
 113     def defer_cross_connect(self, guid, connector_type_name, cross_guid,
 114             cross_testbed_guid, cross_testbed_id, cross_factory_id,
 115             cross_connector_type_name):
 116         self._validate_guid(guid)
 117         factory = self._get_factory(guid)
 118         connector_type = factory.connector_type(connector_type_name)
 119         connector_type.can_connect(cross_testbed_id, cross_factory_id,
 120                 cross_connector_type_name, True)
 121         self._validate_connection(guid, connector_type_name, cross_guid,
 122             cross_connector_type_name)
 123
 124         if not guid in self._cross_connect:
 125             self._cross_connect[guid] = dict()
 126         if not connector_type_name in self._cross_connect[guid]:
 127              self._cross_connect[guid][connector_type_name] = dict()
 128         self._cross_connect[guid][connector_type_name] = \
 129                 (cross_guid, cross_testbed_guid, cross_testbed_id,
 130                 cross_factory_id, cross_connector_type_name)
 131
 132     def defer_add_trace(self, guid, trace_name):
 133         self._validate_guid(guid)
 134         self._validate_trace(guid, trace_name)
 135         if not guid in self._add_trace:
 136             self._add_trace[guid] = list()
 137         self._add_trace[guid].append(trace_name)
 138
 139     def defer_add_address(self, guid, address, netprefix, broadcast):
 140         self._validate_guid(guid)
 141         self._validate_allow_addresses(guid)
 142         if guid not in self._add_address:
 143             self._add_address[guid] = list()
 144         self._add_address[guid].append((address, netprefix, broadcast))
 145
 146     def defer_add_route(self, guid, destination, netprefix, nexthop, metric = 0):
 147         self._validate_guid(guid)
 148         self._validate_allow_routes(guid)
 149         if not guid in self._add_route:
 150             self._add_route[guid] = list()
 151         self._add_route[guid].append((destination, netprefix, nexthop, metric))
 152
 153     def do_setup(self):
 154         self._root_directory = self._attributes.\
 155             get_attribute_value("rootDirectory")
 156         self._status = TS.STATUS_SETUP
 157
 158     def do_create(self):
 159         def set_params(self, guid):
 160             parameters = self._get_parameters(guid)
 161             for name, value in parameters.iteritems():
 162                 self.set(guid, name, value)
 163
 164         self._do_in_factory_order(
 165             'create_function',
 166             self._metadata.create_order,
 167             postaction = set_params )
 168         self._status = TS.STATUS_CREATED
 169
 170     def _do_connect(self, init = True):
 171         unconnected = copy.deepcopy(self._connect)
 172
 173         while unconnected:
 174             for guid1, connections in unconnected.items():
 175                 factory1 = self._get_factory(guid1)
 176                 for connector_type_name1, connections2 in connections.items():
 177                     connector_type1 = factory1.connector_type(connector_type_name1)
 178                     for guid2, connector_type_name2 in connections2.items():
 179                         factory_id2 = self._create[guid2]
 180                         # Connections are executed in a "From -> To" direction only
 181                         # This explicitly ignores the "To -> From" (mirror)
 182                         # connections of every connection pair.
 183                         if init:
 184                             connect_code = connector_type1.connect_to_init_code(
 185                                     self._testbed_id, factory_id2,
 186                                     connector_type_name2,
 187                                     False)
 188                         else:
 189                             connect_code = connector_type1.connect_to_compl_code(
 190                                     self._testbed_id, factory_id2,
 191                                     connector_type_name2,
 192                                     False)
 193                         delay = None
 194                         if connect_code:
 195                             delay = connect_code(self, guid1, guid2)
 196
 197                         if delay is not CONNECTION_DELAY:
 198                             del unconnected[guid1][connector_type_name1][guid2]
 199                     if not unconnected[guid1][connector_type_name1]:
 200                         del unconnected[guid1][connector_type_name1]
 201                 if not unconnected[guid1]:
 202                     del unconnected[guid1]
 203
 204     def do_connect_init(self):
 205         self._do_connect()
 206
 207     def do_connect_compl(self):
 208         self._do_connect(init = False)
 209         self._status = TS.STATUS_CONNECTED
 210
 211     def _do_in_factory_order(self, action, order, postaction = None, poststep = None):
 212         guids = collections.defaultdict(list)
 213         # order guids (elements) according to factory_id
 214         for guid, factory_id in self._create.iteritems():
 215             guids[factory_id].append(guid)
 216         # configure elements following the factory_id order
 217         for factory_id in order:
 218             # omit the factories that have no element to create
 219             if factory_id not in guids:
 220                 continue
 221             factory = self._factories[factory_id]
 222             if not getattr(factory, action):
 223                 continue
 224             for guid in guids[factory_id]:
 225                 getattr(factory, action)(self, guid)
 226                 if postaction:
 227                     postaction(self, guid)
 228             if poststep:
 229                 for guid in guids[factory_id]:
 230                     poststep(self, guid)
 231
 232     @staticmethod
 233     def do_poststep_preconfigure(self, guid):
 234         # dummy hook for implementations interested in
 235         # two-phase configuration
 236         pass
 237
 238     def do_preconfigure(self):
 239         self._do_in_factory_order(
 240             'preconfigure_function',
 241             self._metadata.preconfigure_order,
 242             poststep = self.do_poststep_preconfigure )
 243
 244     @staticmethod
 245     def do_poststep_configure(self, guid):
 246         # dummy hook for implementations interested in
 247         # two-phase configuration
 248         pass
 249
 250     def do_configure(self):
 251         self._do_in_factory_order(
 252             'configure_function',
 253             self._metadata.configure_order,
 254             poststep = self.do_poststep_configure )
 255         self._status = TS.STATUS_CONFIGURED
 256
 257     def do_prestart(self):
 258         self._do_in_factory_order(
 259             'prestart_function',
 260             self._metadata.prestart_order )
 261
 262     def _do_cross_connect(self, cross_data, init = True):
 263         for guid, cross_connections in self._cross_connect.iteritems():
 264             factory = self._get_factory(guid)
 265             for connector_type_name, cross_connection in \
 266                     cross_connections.iteritems():
 267                 connector_type = factory.connector_type(connector_type_name)
 268                 (cross_guid, cross_testbed_guid, cross_testbed_id,
 269                     cross_factory_id, cross_connector_type_name) = cross_connection
 270                 if init:
 271                     connect_code = connector_type.connect_to_init_code(
 272                         cross_testbed_id, cross_factory_id,
 273                         cross_connector_type_name,
 274                         True)
 275                 else:
 276                     connect_code = connector_type.connect_to_compl_code(
 277                         cross_testbed_id, cross_factory_id,
 278                         cross_connector_type_name,
 279                         True)
 280                 if connect_code:
 281                     elem_cross_data = cross_data[cross_testbed_guid][cross_guid]
 282                     connect_code(self, guid, elem_cross_data)
 283
 284     def do_cross_connect_init(self, cross_data):
 285         self._do_cross_connect(cross_data)
 286
 287     def do_cross_connect_compl(self, cross_data):
 288         self._do_cross_connect(cross_data, init = False)
 289         self._status = TS.STATUS_CROSS_CONNECTED
 290
 291     def set(self, guid, name, value, time = TIME_NOW):
 292         self._validate_guid(guid)
 293         self._validate_box_attribute(guid, name)
 294         self._validate_box_value(guid, name, value)
 295         self._validate_modify_box_value(guid, name)
 296         if guid not in self._set:
 297             self._set[guid] = dict()
 298             self._setlog[guid] = dict()
 299         if time not in self._setlog[guid]:
 300             self._setlog[guid][time] = dict()
 301         self._setlog[guid][time][name] = value
 302         self._set[guid][name] = value
 303
 304     def get(self, guid, name, time = TIME_NOW):
 305         """
 306         gets an attribute from box definitions if available.
 307         Throws KeyError if the GUID wasn't created
 308         through the defer_create interface, and AttributeError if the
 309         attribute isn't available (doesn't exist or is design-only)
 310         """
 311         self._validate_guid(guid)
 312         self._validate_box_attribute(guid, name)
 313         if guid in self._set and name in self._set[guid]:
 314             return self._set[guid][name]
 315         if guid in self._create_set and name in self._create_set[guid]:
 316             return self._create_set[guid][name]
 317         # if nothing else found, returns the factory default value
 318         factory = self._get_factory(guid)
 319         return factory.box_attributes.get_attribute_value(name)
 320
 321     def get_route(self, guid, index, attribute):
 322         """
 323         returns information given to defer_add_route.
 324
 325         Raises AttributeError if an invalid attribute is requested
 326             or if the indexed routing rule does not exist.
 327
 328         Raises KeyError if the GUID has not been seen by
 329             defer_add_route
 330         """
 331         ATTRIBUTES = ['Destination', 'NetPrefix', 'NextHop']
 332
 333         if attribute not in ATTRIBUTES:
 334             raise AttributeError, "Attribute %r invalid for addresses of %r" % (attribute, guid)
 335
 336         attribute_index = ATTRIBUTES.index(attribute)
 337
 338         routes = self._add_route.get(guid)
 339         if not routes:
 340             raise KeyError, "GUID %r not found in %s" % (guid, self._testbed_id)
 341
 342         index = int(index)
 343         if not (0 <= index < len(addresses)):
 344             raise AttributeError, "GUID %r at %s does not have a routing entry #%s" % (
 345                 guid, self._testbed_id, index)
 346
 347         return routes[index][attribute_index]
 348
 349     def get_address(self, guid, index, attribute='Address'):
 350         """
 351         returns information given to defer_add_address
 352
 353         Raises AttributeError if an invalid attribute is requested
 354             or if the indexed routing rule does not exist.
 355
 356         Raises KeyError if the GUID has not been seen by
 357             defer_add_address
 358         """
 359         ATTRIBUTES = ['Address', 'NetPrefix', 'Broadcast']
 360
 361         if attribute not in ATTRIBUTES:
 362             raise AttributeError, "Attribute %r invalid for addresses of %r" % (attribute, guid)
 363
 364         attribute_index = ATTRIBUTES.index(attribute)
 365
 366         addresses = self._add_address.get(guid)
 367         if not addresses:
 368             raise KeyError, "GUID %r not found in %s" % (guid, self._testbed_id)
 369
 370         index = int(index)
 371         if not (0 <= index < len(addresses)):
 372             raise AttributeError, "GUID %r at %s does not have an address #%s" % (
 373                 guid, self._testbed_id, index)
 374
 375         return addresses[index][attribute_index]
 376
 377     def get_attribute_list(self, guid, filter_flags = None, exclude = False):
 378         factory = self._get_factory(guid)
 379         attribute_list = list()
 380         return factory.box_attributes.get_attribute_list(filter_flags, exclude)
 381
 382     def get_factory_id(self, guid):
 383         factory = self._get_factory(guid)
 384         return factory.factory_id
 385
 386     def start(self, time = TIME_NOW):
 387         self._do_in_factory_order(
 388             'start_function',
 389             self._metadata.start_order )
 390         self._status = TS.STATUS_STARTED
 391
 392     #action: NotImplementedError
 393
 394     def stop(self, time = TIME_NOW):
 395         self._do_in_factory_order(
 396             'stop_function',
 397             reversed(self._metadata.start_order) )
 398         self._status = TS.STATUS_STOPPED
 399
 400     def status(self, guid = None):
 401         if not guid:
 402             return self._status
 403         self._validate_guid(guid)
 404         factory = self._get_factory(guid)
 405         status_function = factory.status_function
 406         if status_function:
 407             return status_function(self, guid)
 408         return AS.STATUS_UNDETERMINED
 409
 410     def trace(self, guid, trace_id, attribute='value'):
 411         if attribute == 'value':
 412             fd = open("%s" % self.trace_filepath(guid, trace_id), "r")
 413             content = fd.read()
 414             fd.close()
 415         elif attribute == 'path':
 416             content = self.trace_filepath(guid, trace_id)
 417         else:
 418             content = None
 419         return content
 420
 421     def traces_info(self):
 422         traces_info = dict()
 423         host = self._attributes.get_attribute_value("deployment_host")
 424         user = self._attributes.get_attribute_value("deployment_user")
 425         for guid, trace_list in self._add_trace.iteritems():
 426             traces_info[guid] = dict()
 427             for trace_id in trace_list:
 428                 traces_info[guid][trace_id] = dict()
 429                 filepath = self.trace(guid, trace_id, attribute = "path")
 430                 traces_info[guid][trace_id]["host"] = host
 431                 traces_info[guid][trace_id]["user"] = user
 432                 traces_info[guid][trace_id]["filepath"] = filepath
 433         return traces_info
 434
 435     def trace_filepath(self, guid, trace_id):
 436         """
 437         Return a trace's file path, for TestbedController's default
 438         implementation of trace()
 439         """
 440         raise NotImplementedError
 441
 442     #shutdown: NotImplementedError
 443
 444     def get_connected(self, guid, connector_type_name,
 445             other_connector_type_name):
 446         """searchs the connected elements for the specific connector_type_name
 447         pair"""
 448         if guid not in self._connect:
 449             return []
 450         # all connections for all connectors for guid
 451         all_connections = self._connect[guid]
 452         if connector_type_name not in all_connections:
 453             return []
 454         # all connections for the specific connector
 455         connections = all_connections[connector_type_name]
 456         specific_connections = [otr_guid for otr_guid, otr_connector_type_name \
 457                 in connections.iteritems() if \
 458                 otr_connector_type_name == other_connector_type_name]
 459         return specific_connections
 460
 461     def _get_connection_count(self, guid, connection_type_name):
 462         count = 0
 463         cross_count = 0
 464         if guid in self._connect and connection_type_name in \
 465                 self._connect[guid]:
 466             count = len(self._connect[guid][connection_type_name])
 467         if guid in self._cross_connect and connection_type_name in \
 468                 self._cross_connect[guid]:
 469             cross_count = len(self._cross_connect[guid][connection_type_name])
 470         return count + cross_count
 471
 472     def _get_traces(self, guid):
 473         return [] if guid not in self._add_trace else self._add_trace[guid]
 474
 475     def _get_parameters(self, guid):
 476         return dict() if guid not in self._create_set else \
 477                 self._create_set[guid]
 478
 479     def _get_factory(self, guid):
 480         factory_id = self._create[guid]
 481         return self._factories[factory_id]
 482
 483     def _get_factory_id(self, guid):
 484         """ Returns the factory ID of the (perhaps not yet) created object """
 485         return self._create.get(guid, None)
 486
 487     def _validate_guid(self, guid):
 488         if not guid in self._create:
 489             raise RuntimeError("Element guid %d doesn't exist" % guid)
 490
 491     def _validate_not_guid(self, guid):
 492         if guid in self._create:
 493             raise AttributeError("Cannot add elements with the same guid: %d" %
 494                     guid)
 495
 496     def _validate_factory_id(self, factory_id):
 497         if factory_id not in self._factories:
 498             raise AttributeError("Invalid element type %s for testbed version %s" %
 499                     (factory_id, self._testbed_version))
 500
 501     def _validate_testbed_attribute(self, name):
 502         if not self._attributes.has_attribute(name):
 503             raise AttributeError("Invalid testbed attribute %s for testbed" % \
 504                     name)
 505
 506     def _validate_testbed_value(self, name, value):
 507         if not self._attributes.is_attribute_value_valid(name, value):
 508             raise AttributeError("Invalid value %s for testbed attribute %s" % \
 509                 (value, name))
 510
 511     def _validate_box_attribute(self, guid, name):
 512         factory = self._get_factory(guid)
 513         if not factory.box_attributes.has_attribute(name):
 514             raise AttributeError("Invalid attribute %s for element type %s" %
 515                     (name, factory.factory_id))
 516
 517     def _validate_box_value(self, guid, name, value):
 518         factory = self._get_factory(guid)
 519         if not factory.box_attributes.is_attribute_value_valid(name, value):
 520             raise AttributeError("Invalid value %s for attribute %s" % \
 521                 (value, name))
 522
 523     def _validate_factory_attribute(self, guid, name):
 524         factory = self._get_factory(guid)
 525         if not factory.has_attribute(name):
 526             raise AttributeError("Invalid attribute %s for element type %s" %
 527                     (name, factory.factory_id))
 528
 529     def _validate_factory_value(self, guid, name, value):
 530         factory = self._get_factory(guid)
 531         if not factory.is_attribute_value_valid(name, value):
 532             raise AttributeError("Invalid value %s for attribute %s" % \
 533                 (value, name))
 534
 535     def _validate_trace(self, guid, trace_name):
 536         factory = self._get_factory(guid)
 537         if not trace_name in factory.traces_list:
 538             raise RuntimeError("Element type '%s' has no trace '%s'" %
 539                     (factory.factory_id, trace_name))
 540
 541     def _validate_allow_addresses(self, guid):
 542         factory = self._get_factory(guid)
 543         if not factory.allow_addresses:
 544             raise RuntimeError("Element type '%s' doesn't support addresses" %
 545                     factory.factory_id)
 546         attr_name = "maxAddresses"
 547         if guid in self._create_set and attr_name in self._create_set[guid]:
 548             max_addresses = self._create_set[guid][attr_name]
 549         else:
 550             factory = self._get_factory(guid)
 551             max_addresses = factory.box_attributes.get_attribute_value(attr_name)
 552         if guid in self._add_address:
 553             count_addresses = len(self._add_address[guid])
 554             if max_addresses == count_addresses:
 555                 raise RuntimeError("Element guid %d of type '%s' can't accept \
 556                         more addresses" % (guid, factory.factory_id))
 557
 558     def _validate_allow_routes(self, guid):
 559         factory = self._get_factory(guid)
 560         if not factory.allow_routes:
 561             raise RuntimeError("Element type '%s' doesn't support routes" %
 562                     factory.factory_id)
 563
 564     def _validate_connection(self, guid1, connector_type_name1, guid2,
 565             connector_type_name2, cross = False):
 566         # can't connect with self
 567         if guid1 == guid2:
 568             raise AttributeError("Can't connect guid %d to self" % \
 569                 (guid1))
 570         # the connection is already done, so ignore
 571         connected = self.get_connected(guid1, connector_type_name1,
 572                 connector_type_name2)
 573         if guid2 in connected:
 574             return
 575         count1 = self._get_connection_count(guid1, connector_type_name1)
 576         factory1 = self._get_factory(guid1)
 577         connector_type1 = factory1.connector_type(connector_type_name1)
 578         if count1 == connector_type1.max:
 579             raise AttributeError("Connector %s is full for guid %d" % \
 580                 (connector_type_name1, guid1))
 581
 582     def _validate_modify_box_value(self, guid, name):
 583         factory = self._get_factory(guid)
 584         if self._status > TS.STATUS_STARTED and \
 585                 (factory.box_attributes.is_attribute_exec_read_only(name) or \
 586                 factory.box_attributes.is_attribute_exec_immutable(name)):
 587             raise AttributeError("Attribute %s can only be modified during experiment design" % name)
 588