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