mock cross_connect test added to test/core/integration.py
[nepi.git] / src / nepi / core / execute.py
1 #!/usr/bin/env python
2 # -*- coding: utf-8 -*-
3
4 from nepi.core.attributes import Attribute, AttributesMap
5 from nepi.core.connector import ConnectorTypeBase
6 from nepi.util import proxy, validation
7 from nepi.util.constants import STATUS_FINISHED, TIME_NOW
8 from nepi.util.parser._xml import XmlExperimentParser
9 import sys
10 import re
11 import threading
12 import ConfigParser
13 import os
14
15 ATTRIBUTE_PATTERN_BASE = re.compile(r"\{#\[(?P<label>[-a-zA-Z0-9._]*)\](?P<expr>(?P<component>\.addr\[[0-9]+\]|\.route\[[0-9]+\]|\.trace\[[0-9]+\]|).\[(?P<attribute>[-a-zA-Z0-9._]*)\])#}")
16 ATTRIBUTE_PATTERN_GUID_SUB = r"{#[%(guid)s]%(expr)s#}"
17 COMPONENT_PATTERN = re.compile(r"(?P<kind>[a-z]*)\[(?P<index>.*)\]")
18
19 class ConnectorType(ConnectorTypeBase):
20     def __init__(self, testbed_id, factory_id, name, max = -1, min = 0):
21         super(ConnectorType, self).__init__(testbed_id, factory_id, name, max, min)
22         # from_connections -- connections where the other connector is the "From"
23         # to_connections -- connections where the other connector is the "To"
24         # keys in the dictionary correspond to the 
25         # connector_type_id for possible connections. The value is a tuple:
26         # (can_cross, connect)
27         # can_cross: indicates if the connection is allowed accros different
28         #    testbed instances
29         # code: is the connection function to be invoked when the elements
30         #    are connected
31         self._from_connections = dict()
32         self._to_connections = dict()
33
34     def add_from_connection(self, testbed_id, factory_id, name, can_cross, 
35             init_code, compl_code):
36         type_id = self.make_connector_type_id(testbed_id, factory_id, name)
37         self._from_connections[type_id] = (can_cross, init_code, compl_code)
38
39     def add_to_connection(self, testbed_id, factory_id, name, can_cross, 
40             init_code, compl_code):
41         type_id = self.make_connector_type_id(testbed_id, factory_id, name)
42         self._to_connections[type_id] = (can_cross, init_code, compl_code)
43
44     def can_connect(self, testbed_id, factory_id, name, count, 
45             must_cross = False):
46         connector_type_id = self.make_connector_type_id(testbed_id, factory_id, name)
47         for lookup_type_id in self._type_resolution_order(connector_type_id):
48             if lookup_type_id in self._from_connections:
49                 (can_cross, init_code, compl_code) = self._from_connections[lookup_type_id]
50             elif lookup_type_id in self._to_connections:
51                 (can_cross, init_code, compl_code) = self._to_connections[lookup_type_id]
52             else:
53                 # keey trying
54                 continue
55             return not must_cross or can_cross
56         else:
57             return False
58
59     def _connect_to_code(self, testbed_id, factory_id, name):
60         connector_type_id = self.make_connector_type_id(testbed_id, factory_id, name)
61         for lookup_type_id in self._type_resolution_order(connector_type_id):
62             if lookup_type_id in self._to_connections:
63                 (can_cross, init_code, compl_code) = self._to_connections[lookup_type_id]
64                 return (init_code, compl_code)
65         else:
66             return (False, False)
67     
68     def connect_to_init_code(self, testbed_id, factory_id, name):
69         return self._connect_to_code(testbed_id, factory_id, name)[0]
70
71     def connect_to_compl_code(self, testbed_id, factory_id, name):
72         return self._connect_to_code(testbed_id, factory_id, name)[1]
73
74 class Factory(AttributesMap):
75     def __init__(self, factory_id, create_function, start_function, 
76             stop_function, status_function, 
77             configure_function, preconfigure_function,
78             allow_addresses = False, has_addresses = False,
79             allow_routes = False, has_routes = False):
80         super(Factory, self).__init__()
81         self._factory_id = factory_id
82         self._allow_addresses = bool(allow_addresses)
83         self._allow_routes = bool(allow_routes)
84         self._has_addresses = bool(has_addresses) or self._allow_addresses
85         self._has_routes = bool(has_routes) or self._allow_routes
86         self._create_function = create_function
87         self._start_function = start_function
88         self._stop_function = stop_function
89         self._status_function = status_function
90         self._configure_function = configure_function
91         self._preconfigure_function = preconfigure_function
92         self._connector_types = dict()
93         self._traces = list()
94         self._box_attributes = AttributesMap()
95
96     @property
97     def factory_id(self):
98         return self._factory_id
99
100     @property
101     def allow_addresses(self):
102         return self._allow_addresses
103
104     @property
105     def allow_routes(self):
106         return self._allow_routes
107
108     @property
109     def has_addresses(self):
110         return self._has_addresses
111
112     @property
113     def has_routes(self):
114         return self._has_routes
115
116     @property
117     def box_attributes(self):
118         return self._box_attributes
119
120     @property
121     def create_function(self):
122         return self._create_function
123
124     @property
125     def start_function(self):
126         return self._start_function
127
128     @property
129     def stop_function(self):
130         return self._stop_function
131
132     @property
133     def status_function(self):
134         return self._status_function
135
136     @property
137     def configure_function(self):
138         return self._configure_function
139
140     @property
141     def preconfigure_function(self):
142         return self._preconfigure_function
143
144     @property
145     def traces(self):
146         return self._traces
147
148     def connector_type(self, name):
149         return self._connector_types[name]
150
151     def add_connector_type(self, connector_type):
152         self._connector_types[connector_type.name] = connector_type
153
154     def add_trace(self, trace_id):
155         self._traces.append(trace_id)
156
157     def add_box_attribute(self, name, help, type, value = None, range = None,
158         allowed = None, flags = Attribute.NoFlags, validation_function = None):
159         self._box_attributes.add_attribute(name, help, type, value, range, 
160                 allowed, flags, validation_function)
161
162 class TestbedController(object):
163     def __init__(self, testbed_id, testbed_version):
164         self._testbed_id = testbed_id
165         self._testbed_version = testbed_version
166
167     @property
168     def testbed_id(self):
169         return self._testbed_id
170
171     @property
172     def testbed_version(self):
173         return self._testbed_version
174
175     @property
176     def guids(self):
177         raise NotImplementedError
178
179     def defer_configure(self, name, value):
180         """Instructs setting a configuartion attribute for the testbed instance"""
181         raise NotImplementedError
182
183     def defer_create(self, guid, factory_id):
184         """Instructs creation of element """
185         raise NotImplementedError
186
187     def defer_create_set(self, guid, name, value):
188         """Instructs setting an initial attribute on an element"""
189         raise NotImplementedError
190
191     def defer_factory_set(self, guid, name, value):
192         """Instructs setting an attribute on a factory"""
193         raise NotImplementedError
194
195     def defer_connect(self, guid1, connector_type_name1, guid2, 
196             connector_type_name2): 
197         """Instructs creation of a connection between the given connectors"""
198         raise NotImplementedError
199
200     def defer_cross_connect(self, guid, connector_type_name, cross_guid, 
201             cross_testbed_id, cross_factory_id, cross_connector_type_name):
202         """
203         Instructs creation of a connection between the given connectors 
204         of different testbed instances
205         """
206         raise NotImplementedError
207
208     def defer_add_trace(self, guid, trace_id):
209         """Instructs the addition of a trace"""
210         raise NotImplementedError
211
212     def defer_add_address(self, guid, address, netprefix, broadcast): 
213         """Instructs the addition of an address"""
214         raise NotImplementedError
215
216     def defer_add_route(self, guid, destination, netprefix, nexthop):
217         """Instructs the addition of a route"""
218         raise NotImplementedError
219
220     def do_setup(self):
221         """After do_setup the testbed initial configuration is done"""
222         raise NotImplementedError
223
224     def do_create(self):
225         """
226         After do_create all instructed elements are created and 
227         attributes setted
228         """
229         raise NotImplementedError
230
231     def do_connect_init(self):
232         """
233         After do_connect_init all internal connections between testbed elements
234         are initiated
235         """
236         raise NotImplementedError
237
238     def do_connect_compl(self):
239         """
240         After do_connect all internal connections between testbed elements
241         are completed
242         """
243         raise NotImplementedError
244
245     def do_configure(self):
246         """After do_configure elements are configured"""
247         raise NotImplementedError
248
249     def do_cross_connect_init(self, cross_data):
250         """
251         After do_cross_connect_init initiation of all external connections 
252         between different testbed elements is performed
253         """
254         raise NotImplementedError
255
256     def do_cross_connect_compl(self, cross_data):
257         """
258         After do_cross_connect_compl completion of all external connections 
259         between different testbed elements is performed
260         """
261         raise NotImplementedError
262
263     def start(self):
264         raise NotImplementedError
265
266     def stop(self):
267         raise NotImplementedError
268
269     def set(self, guid, name, value, time = TIME_NOW):
270         raise NotImplementedError
271
272     def get(self, guid, name, time = TIME_NOW):
273         raise NotImplementedError
274     
275     def get_route(self, guid, index, attribute):
276         """
277         Params:
278             
279             guid: guid of box to query
280             index: number of routing entry to fetch
281             attribute: one of Destination, NextHop, NetPrefix
282         """
283         raise NotImplementedError
284
285     def get_address(self, guid, index, attribute='Address'):
286         """
287         Params:
288             
289             guid: guid of box to query
290             index: number of inteface to select
291             attribute: one of Address, NetPrefix, Broadcast
292         """
293         raise NotImplementedError
294
295     def get_attribute_list(self, guid):
296         raise NotImplementedError
297
298     def action(self, time, guid, action):
299         raise NotImplementedError
300
301     def status(self, guid):
302         raise NotImplementedError
303
304     def trace(self, guid, trace_id, attribute='value'):
305         raise NotImplementedError
306
307     def shutdown(self):
308         raise NotImplementedError
309
310 class ExperimentController(object):
311     def __init__(self, experiment_xml, root_dir):
312         self._experiment_xml = experiment_xml
313         self._testbeds = dict()
314         self._access_config = dict()
315         self._netrefs = dict()
316         self._cross_data = dict()
317         self._root_dir = root_dir
318
319         self.persist_experiment_xml()
320
321     @property
322     def experiment_xml(self):
323         return self._experiment_xml
324
325     def persist_experiment_xml(self):
326         xml_path = os.path.join(self._root_dir, "experiment.xml")
327         f = open(xml_path, "w")
328         f.write(self._experiment_xml)
329         f.close()
330
331     def set_access_configuration(self, testbed_guid, access_config):
332         self._access_config[testbed_guid] = access_config
333
334     def trace(self, testbed_guid, guid, trace_id, attribute='value'):
335         return self._testbeds[testbed_guid].trace(guid, trace_id, attribute)
336
337     @staticmethod
338     def _parallel(callables):
339         threads = [ threading.Thread(target=callable) for callable in callables ]
340         for thread in threads:
341             thread.start()
342         for thread in threads:
343             thread.join()
344
345     def start(self):
346         self._init_testbed_controllers()
347         
348         # persist testbed connection data, for potential recovery
349         self._persist_testbed_proxies()
350         
351         # perform setup in parallel for all test beds,
352         # wait for all threads to finish
353         self._parallel([testbed.do_setup 
354                         for testbed in self._testbeds.itervalues()])
355         
356         # perform create-connect in parallel, wait
357         # (internal connections only)
358         self._parallel([testbed.do_create
359                         for testbed in self._testbeds.itervalues()])
360
361         self._parallel([testbed.do_connect_init
362                         for testbed in self._testbeds.itervalues()])
363
364         self._parallel([testbed.do_connect_compl
365                         for testbed in self._testbeds.itervalues()])
366
367         self._parallel([testbed.do_preconfigure
368                         for testbed in self._testbeds.itervalues()])
369
370         # resolve netrefs
371         self.do_netrefs(fail_if_undefined=True)
372         
373         # perform do_configure in parallel for al testbeds
374         # (it's internal configuration for each)
375         self._parallel([testbed.do_configure
376                         for testbed in self._testbeds.itervalues()])
377
378         # cross-connect (cannot be done in parallel)
379         for guid, testbed in self._testbeds.iteritems():
380             cross_data = self._get_cross_data(guid)
381             testbed.do_cross_connect_init(cross_data)
382         for guid, testbed in self._testbeds.iteritems():
383             cross_data = self._get_cross_data(guid)
384             testbed.do_cross_connect_compl(cross_data)
385        
386         # start experiment (parallel start on all testbeds)
387         self._parallel([testbed.start
388                         for testbed in self._testbeds.itervalues()])
389
390     def _persist_testbed_proxies(self):
391         TRANSIENT = ('Recover',)
392         
393         # persist access configuration for all testbeds, so that
394         # recovery mode can reconnect to them if it becomes necessary
395         conf = ConfigParser.RawConfigParser()
396         for testbed_guid, testbed_config in self._access_config.iteritems():
397             testbed_guid = str(testbed_guid)
398             conf.add_section(testbed_guid)
399             for attr in testbed_config.attributes_list:
400                 if attr not in TRANSIENT:
401                     conf.set(testbed_guid, attr, 
402                         testbed_config.get_attribute_value(attr))
403         
404         f = open(os.path.join(self._root_dir, 'access_config.ini'), 'w')
405         conf.write(f)
406         f.close()
407     
408     def _load_testbed_proxies(self):
409         TYPEMAP = {
410             STRING : 'get',
411             INTEGER : 'getint',
412             FLOAT : 'getfloat',
413             BOOLEAN : 'getboolean',
414         }
415         
416         conf = ConfigParser.RawConfigParser()
417         conf.read(os.path.join(self._root_dir, 'access_config.ini'))
418         for testbed_guid in conf.sections():
419             testbed_config = proxy.AccessConfiguration()
420             for attr in conf.options(testbed_guid):
421                 testbed_config.set_attribute_value(attr, 
422                     conf.get(testbed_guid, attr) )
423                 
424             testbed_guid = str(testbed_guid)
425             conf.add_section(testbed_guid)
426             for attr in testbed_config.attributes_list:
427                 if attr not in TRANSIENT:
428                     getter = getattr(conf, TYPEMAP.get(
429                         testbed_config.get_attribute_type(attr),
430                         'get') )
431                     testbed_config.set_attribute_value(
432                         testbed_guid, attr, getter(attr))
433     
434     def _unpersist_testbed_proxies(self):
435         try:
436             os.remove(os.path.join(self._root_dir, 'access_config.ini'))
437         except:
438             # Just print exceptions, this is just cleanup
439             import traceback
440             traceback.print_exc(file=sys.stderr)
441
442     def stop(self):
443        for testbed in self._testbeds.values():
444            testbed.stop()
445        self._unpersist_testbed_proxies()
446    
447     def recover(self):
448         # reload perviously persisted testbed access configurations
449         self._load_testbed_proxies()
450         
451         # recreate testbed proxies by reconnecting only
452         self._init_testbed_controllers(recover = True)
453
454     def is_finished(self, guid):
455         for testbed in self._testbeds.values():
456             for guid_ in testbed.guids:
457                 if guid_ == guid:
458                     return testbed.status(guid) == STATUS_FINISHED
459         raise RuntimeError("No element exists with guid %d" % guid)    
460
461     def set(self, testbed_guid, guid, name, value, time = TIME_NOW):
462         testbed = self._testbeds[testbed_guid]
463         testbed.set(guid, name, value, time)
464
465     def get(self, testbed_guid, guid, name, time = TIME_NOW):
466         testbed = self._testbeds[testbed_guid]
467         return testbed.get(guid, name, time)
468
469     def shutdown(self):
470        for testbed in self._testbeds.values():
471            testbed.shutdown()
472
473     @staticmethod
474     def _netref_component_split(component):
475         match = COMPONENT_PATTERN.match(component)
476         if match:
477             return match.group("kind"), match.group("index")
478         else:
479             return component, None
480
481     def do_netrefs(self, fail_if_undefined = False):
482         COMPONENT_GETTERS = {
483             'addr':
484                 lambda testbed, guid, index, name: 
485                     testbed.get_address(guid, index, name),
486             'route' :
487                 lambda testbed, guid, index, name: 
488                     testbed.get_route(guid, index, name),
489             'trace' :
490                 lambda testbed, guid, index, name: 
491                     testbed.trace(guid, index, name),
492             '' : 
493                 lambda testbed, guid, index, name: 
494                     testbed.get(guid, name),
495         }
496         
497         for (testbed_guid, guid), attrs in self._netrefs.iteritems():
498             testbed = self._testbeds[testbed_guid]
499             for name in attrs:
500                 value = testbed.get(guid, name)
501                 if isinstance(value, basestring):
502                     match = ATTRIBUTE_PATTERN_BASE.search(value)
503                     if match:
504                         label = match.group("label")
505                         if label.startswith('GUID-'):
506                             ref_guid = int(label[5:])
507                             if ref_guid:
508                                 expr = match.group("expr")
509                                 component = match.group("component")[1:] # skip the dot
510                                 attribute = match.group("attribute")
511                                 
512                                 # split compound components into component kind and index
513                                 # eg: 'addr[0]' -> ('addr', '0')
514                                 component, component_index = self._netref_component_split(component)
515                                 
516                                 # find object and resolve expression
517                                 for ref_testbed in self._testbeds.itervalues():
518                                     if component not in COMPONENT_GETTERS:
519                                         raise ValueError, "Malformed netref: %r - unknown component" % (expr,)
520                                     else:
521                                         ref_value = COMPONENT_GETTERS[component](
522                                             ref_testbed, ref_guid, component_index, attribute)
523                                         if ref_value:
524                                             testbed.set(guid, name, 
525                                                     value.replace(match.group(), ref_value))
526                                             break
527                                 else:
528                                     # couldn't find value
529                                     if fail_if_undefined:
530                                         raise ValueError, "Unresolvable GUID: %r, in netref: %r" % (ref_guid, expr)
531
532     def _init_testbed_controllers(self, recover = False):
533         parser = XmlExperimentParser()
534         data = parser.from_xml_to_data(self._experiment_xml)
535         element_guids = list()
536         label_guids = dict()
537         data_guids = data.guids
538
539         # create testbed controllers
540         for guid in data_guids:
541             if data.is_testbed_data(guid):
542                 self._create_testbed_controller(guid, data, element_guids, 
543                         recover)
544             else:
545                 element_guids.append(guid)
546                 label = data.get_attribute_data(guid, "label")
547                 if label is not None:
548                     if label in label_guids:
549                         raise RuntimeError, "Label %r is not unique" % (label,)
550                     label_guids[label] = guid
551
552         # replace references to elements labels for its guid
553         self._resolve_labels(data, data_guids, label_guids)
554     
555         # program testbed controllers
556         if not recover:
557             self._program_testbed_controllers(element_guids, data)
558
559     def _resolve_labels(self, data, data_guids, label_guids):
560         netrefs = self._netrefs
561         for guid in data_guids:
562             if not data.is_testbed_data(guid):
563                 for name, value in data.get_attribute_data(guid):
564                     if isinstance(value, basestring):
565                         match = ATTRIBUTE_PATTERN_BASE.search(value)
566                         if match:
567                             label = match.group("label")
568                             if not label.startswith('GUID-'):
569                                 ref_guid = label_guids.get(label)
570                                 if ref_guid is not None:
571                                     value = ATTRIBUTE_PATTERN_BASE.sub(
572                                         ATTRIBUTE_PATTERN_GUID_SUB % dict(
573                                             guid = 'GUID-%d' % (ref_guid,),
574                                             expr = match.group("expr"),
575                                             label = label), 
576                                         value)
577                                     data.set_attribute_data(guid, name, value)
578                                     
579                                     # memorize which guid-attribute pairs require
580                                     # postprocessing, to avoid excessive controller-testbed
581                                     # communication at configuration time
582                                     # (which could require high-latency network I/O)
583                                     (testbed_guid, factory_id) = data.get_box_data(guid)
584                                     netrefs.setdefault((testbed_guid, guid), set()).add(name)
585
586     def _create_testbed_controller(self, guid, data, element_guids, recover):
587         (testbed_id, testbed_version) = data.get_testbed_data(guid)
588         access_config = None if guid not in self._access_config else\
589                 self._access_config[guid]
590         
591         if recover and access_config is None:
592             # need to create one
593             access_config = self._access_config[guid] = proxy.AccessConfiguration()
594         if access_config is not None:
595             # force recovery mode 
596             access_config.set_attribute_value("recover",recover)
597         
598         testbed = proxy.create_testbed_controller(testbed_id, 
599                 testbed_version, access_config)
600         for (name, value) in data.get_attribute_data(guid):
601             testbed.defer_configure(name, value)
602         self._testbeds[guid] = testbed
603
604     def _program_testbed_controllers(self, element_guids, data):
605         for guid in element_guids:
606             (testbed_guid, factory_id) = data.get_box_data(guid)
607             testbed = self._testbeds[testbed_guid]
608             testbed.defer_create(guid, factory_id)
609             for (name, value) in data.get_attribute_data(guid):
610                 testbed.defer_create_set(guid, name, value)
611
612         for guid in element_guids: 
613             (testbed_guid, factory_id) = data.get_box_data(guid)
614             testbed = self._testbeds[testbed_guid]
615             for (connector_type_name, cross_guid, cross_connector_type_name) \
616                     in data.get_connection_data(guid):
617                 (testbed_guid, factory_id) = data.get_box_data(guid)
618                 (cross_testbed_guid, cross_factory_id) = data.get_box_data(
619                         cross_guid)
620                 if testbed_guid == cross_testbed_guid:
621                     testbed.defer_connect(guid, connector_type_name, 
622                             cross_guid, cross_connector_type_name)
623                 else: 
624                     cross_testbed = self._testbeds[cross_testbed_guid]
625                     cross_testbed_id = cross_testbed.testbed_id
626                     testbed.defer_cross_connect(guid, connector_type_name, cross_guid, 
627                             cross_testbed_guid, cross_testbed_id, cross_factory_id, 
628                             cross_connector_type_name)
629                     # save cross data for later
630                     self._add_crossdata(testbed_guid, guid, cross_testbed_guid,
631                             cross_guid)
632             for trace_id in data.get_trace_data(guid):
633                 testbed.defer_add_trace(guid, trace_id)
634             for (autoconf, address, netprefix, broadcast) in \
635                     data.get_address_data(guid):
636                 if address != None:
637                     testbed.defer_add_address(guid, address, netprefix, 
638                             broadcast)
639             for (destination, netprefix, nexthop) in data.get_route_data(guid):
640                 testbed.defer_add_route(guid, destination, netprefix, nexthop)
641                 
642     def _add_crossdata(self, testbed_guid, guid, cross_testbed_guid, cross_guid):
643         if testbed_guid not in self._cross_data:
644             self._cross_data[testbed_guid] = dict()
645         if cross_testbed_guid not in self._cross_data[testbed_guid]:
646             self._cross_data[testbed_guid][cross_testbed_guid] = set()
647         self._cross_data[testbed_guid][cross_testbed_guid].add(cross_guid)
648
649     def _get_cross_data(self, testbed_guid):
650         cross_data = dict()
651         if not testbed_guid in self._cross_data:
652             return cross_data
653         for cross_testbed_guid, guid_list in \
654                 self._cross_data[testbed_guid].iteritems():
655             cross_data[cross_testbed_guid] = dict()
656             cross_testbed = self._testbeds[cross_testbed_guid]
657             for cross_guid in guid_list:
658                 elem_cross_data = dict()
659                 cross_data[cross_testbed_guid][cross_guid] = elem_cross_data
660                 attributes_list = cross_testbed.get_attribute_list(cross_guid)
661                 for attr_name in attributes_list:
662                     attr_value = cross_testbed.get(cross_guid, attr_name)
663                     elem_cross_data[attr_name] = attr_value
664         return cross_data
665