Ticket #29: Phasing out AccessConfiguration.
[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._deployment_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 trace(self, testbed_guid, guid, trace_id, attribute='value'):
332         return self._testbeds[testbed_guid].trace(guid, trace_id, attribute)
333
334     @staticmethod
335     def _parallel(callables):
336         threads = [ threading.Thread(target=callable) for callable in callables ]
337         for thread in threads:
338             thread.start()
339         for thread in threads:
340             thread.join()
341
342     def start(self):
343         self._init_testbed_controllers()
344         
345         # persist testbed connection data, for potential recovery
346         self._persist_testbed_proxies()
347         
348         # perform setup in parallel for all test beds,
349         # wait for all threads to finish
350         self._parallel([testbed.do_setup 
351                         for testbed in self._testbeds.itervalues()])
352         
353         # perform create-connect in parallel, wait
354         # (internal connections only)
355         self._parallel([testbed.do_create
356                         for testbed in self._testbeds.itervalues()])
357
358         self._parallel([testbed.do_connect_init
359                         for testbed in self._testbeds.itervalues()])
360
361         self._parallel([testbed.do_connect_compl
362                         for testbed in self._testbeds.itervalues()])
363
364         self._parallel([testbed.do_preconfigure
365                         for testbed in self._testbeds.itervalues()])
366
367         # resolve netrefs
368         self.do_netrefs(fail_if_undefined=True)
369         
370         # perform do_configure in parallel for al testbeds
371         # (it's internal configuration for each)
372         self._parallel([testbed.do_configure
373                         for testbed in self._testbeds.itervalues()])
374
375         # cross-connect (cannot be done in parallel)
376         for guid, testbed in self._testbeds.iteritems():
377             cross_data = self._get_cross_data(guid)
378             testbed.do_cross_connect_init(cross_data)
379         for guid, testbed in self._testbeds.iteritems():
380             cross_data = self._get_cross_data(guid)
381             testbed.do_cross_connect_compl(cross_data)
382        
383         # start experiment (parallel start on all testbeds)
384         self._parallel([testbed.start
385                         for testbed in self._testbeds.itervalues()])
386
387     def _persist_testbed_proxies(self):
388         TRANSIENT = ('Recover',)
389         
390         # persist access configuration for all testbeds, so that
391         # recovery mode can reconnect to them if it becomes necessary
392         conf = ConfigParser.RawConfigParser()
393         for testbed_guid, testbed_config in self._deployment_config.iteritems():
394             testbed_guid = str(testbed_guid)
395             conf.add_section(testbed_guid)
396             for attr in testbed_config.attributes_list:
397                 if attr not in TRANSIENT:
398                     conf.set(testbed_guid, attr, 
399                         testbed_config.get_attribute_value(attr))
400         
401         f = open(os.path.join(self._root_dir, 'deployment_config.ini'), 'w')
402         conf.write(f)
403         f.close()
404     
405     def _load_testbed_proxies(self):
406         TYPEMAP = {
407             STRING : 'get',
408             INTEGER : 'getint',
409             FLOAT : 'getfloat',
410             BOOLEAN : 'getboolean',
411         }
412         
413         conf = ConfigParser.RawConfigParser()
414         conf.read(os.path.join(self._root_dir, 'deployment_config.ini'))
415         for testbed_guid in conf.sections():
416             testbed_config = proxy.AccessConfiguration()
417             for attr in conf.options(testbed_guid):
418                 testbed_config.set_attribute_value(attr, 
419                     conf.get(testbed_guid, attr) )
420                 
421             testbed_guid = str(testbed_guid)
422             conf.add_section(testbed_guid)
423             for attr in testbed_config.attributes_list:
424                 if attr not in TRANSIENT:
425                     getter = getattr(conf, TYPEMAP.get(
426                         testbed_config.get_attribute_type(attr),
427                         'get') )
428                     testbed_config.set_attribute_value(
429                         testbed_guid, attr, getter(attr))
430     
431     def _unpersist_testbed_proxies(self):
432         try:
433             os.remove(os.path.join(self._root_dir, 'deployment_config.ini'))
434         except:
435             # Just print exceptions, this is just cleanup
436             import traceback
437             traceback.print_exc(file=sys.stderr)
438
439     def stop(self):
440        for testbed in self._testbeds.values():
441            testbed.stop()
442        self._unpersist_testbed_proxies()
443    
444     def recover(self):
445         # reload perviously persisted testbed access configurations
446         self._load_testbed_proxies()
447         
448         # recreate testbed proxies by reconnecting only
449         self._init_testbed_controllers(recover = True)
450
451     def is_finished(self, guid):
452         for testbed in self._testbeds.values():
453             for guid_ in testbed.guids:
454                 if guid_ == guid:
455                     return testbed.status(guid) == STATUS_FINISHED
456         raise RuntimeError("No element exists with guid %d" % guid)    
457
458     def set(self, testbed_guid, guid, name, value, time = TIME_NOW):
459         testbed = self._testbeds[testbed_guid]
460         testbed.set(guid, name, value, time)
461
462     def get(self, testbed_guid, guid, name, time = TIME_NOW):
463         testbed = self._testbeds[testbed_guid]
464         return testbed.get(guid, name, time)
465
466     def shutdown(self):
467        for testbed in self._testbeds.values():
468            testbed.shutdown()
469
470     @staticmethod
471     def _netref_component_split(component):
472         match = COMPONENT_PATTERN.match(component)
473         if match:
474             return match.group("kind"), match.group("index")
475         else:
476             return component, None
477
478     def do_netrefs(self, fail_if_undefined = False):
479         COMPONENT_GETTERS = {
480             'addr':
481                 lambda testbed, guid, index, name: 
482                     testbed.get_address(guid, index, name),
483             'route' :
484                 lambda testbed, guid, index, name: 
485                     testbed.get_route(guid, index, name),
486             'trace' :
487                 lambda testbed, guid, index, name: 
488                     testbed.trace(guid, index, name),
489             '' : 
490                 lambda testbed, guid, index, name: 
491                     testbed.get(guid, name),
492         }
493         
494         for (testbed_guid, guid), attrs in self._netrefs.iteritems():
495             testbed = self._testbeds[testbed_guid]
496             for name in attrs:
497                 value = testbed.get(guid, name)
498                 if isinstance(value, basestring):
499                     match = ATTRIBUTE_PATTERN_BASE.search(value)
500                     if match:
501                         label = match.group("label")
502                         if label.startswith('GUID-'):
503                             ref_guid = int(label[5:])
504                             if ref_guid:
505                                 expr = match.group("expr")
506                                 component = match.group("component")[1:] # skip the dot
507                                 attribute = match.group("attribute")
508                                 
509                                 # split compound components into component kind and index
510                                 # eg: 'addr[0]' -> ('addr', '0')
511                                 component, component_index = self._netref_component_split(component)
512                                 
513                                 # find object and resolve expression
514                                 for ref_testbed in self._testbeds.itervalues():
515                                     if component not in COMPONENT_GETTERS:
516                                         raise ValueError, "Malformed netref: %r - unknown component" % (expr,)
517                                     else:
518                                         ref_value = COMPONENT_GETTERS[component](
519                                             ref_testbed, ref_guid, component_index, attribute)
520                                         if ref_value:
521                                             testbed.set(guid, name, 
522                                                     value.replace(match.group(), ref_value))
523                                             break
524                                 else:
525                                     # couldn't find value
526                                     if fail_if_undefined:
527                                         raise ValueError, "Unresolvable GUID: %r, in netref: %r" % (ref_guid, expr)
528
529     def _init_testbed_controllers(self, recover = False):
530         parser = XmlExperimentParser()
531         data = parser.from_xml_to_data(self._experiment_xml)
532         element_guids = list()
533         label_guids = dict()
534         data_guids = data.guids
535
536         # create testbed controllers
537         for guid in data_guids:
538             if data.is_testbed_data(guid):
539                 self._create_testbed_controller(guid, data, element_guids, 
540                         recover)
541             else:
542                 element_guids.append(guid)
543                 label = data.get_attribute_data(guid, "label")
544                 if label is not None:
545                     if label in label_guids:
546                         raise RuntimeError, "Label %r is not unique" % (label,)
547                     label_guids[label] = guid
548
549         # replace references to elements labels for its guid
550         self._resolve_labels(data, data_guids, label_guids)
551     
552         # program testbed controllers
553         if not recover:
554             self._program_testbed_controllers(element_guids, data)
555
556     def _resolve_labels(self, data, data_guids, label_guids):
557         netrefs = self._netrefs
558         for guid in data_guids:
559             if not data.is_testbed_data(guid):
560                 for name, value in data.get_attribute_data(guid):
561                     if isinstance(value, basestring):
562                         match = ATTRIBUTE_PATTERN_BASE.search(value)
563                         if match:
564                             label = match.group("label")
565                             if not label.startswith('GUID-'):
566                                 ref_guid = label_guids.get(label)
567                                 if ref_guid is not None:
568                                     value = ATTRIBUTE_PATTERN_BASE.sub(
569                                         ATTRIBUTE_PATTERN_GUID_SUB % dict(
570                                             guid = 'GUID-%d' % (ref_guid,),
571                                             expr = match.group("expr"),
572                                             label = label), 
573                                         value)
574                                     data.set_attribute_data(guid, name, value)
575                                     
576                                     # memorize which guid-attribute pairs require
577                                     # postprocessing, to avoid excessive controller-testbed
578                                     # communication at configuration time
579                                     # (which could require high-latency network I/O)
580                                     (testbed_guid, factory_id) = data.get_box_data(guid)
581                                     netrefs.setdefault((testbed_guid, guid), set()).add(name)
582
583     def _create_testbed_controller(self, guid, data, element_guids, recover):
584         (testbed_id, testbed_version) = data.get_testbed_data(guid)
585         deployment_config = self._deployment_config.get(guid)
586         
587         if deployment_config is None:
588             # need to create one
589             deployment_config = self._deployment_config[guid] = proxy.AccessConfiguration()
590             
591             for (name, value) in data.get_attribute_data(guid):
592                 if value is not None and deployment_config.has_attribute(name):
593                     deployment_config.set_attribute_value(name, value)
594         
595         if deployment_config is not None:
596             # force recovery mode 
597             deployment_config.set_attribute_value("recover",recover)
598         
599         testbed = proxy.create_testbed_controller(testbed_id, 
600                 testbed_version, deployment_config)
601         for (name, value) in data.get_attribute_data(guid):
602             testbed.defer_configure(name, value)
603         self._testbeds[guid] = testbed
604
605     def _program_testbed_controllers(self, element_guids, data):
606         for guid in element_guids:
607             (testbed_guid, factory_id) = data.get_box_data(guid)
608             testbed = self._testbeds[testbed_guid]
609             testbed.defer_create(guid, factory_id)
610             for (name, value) in data.get_attribute_data(guid):
611                 testbed.defer_create_set(guid, name, value)
612
613         for guid in element_guids: 
614             (testbed_guid, factory_id) = data.get_box_data(guid)
615             testbed = self._testbeds[testbed_guid]
616             for (connector_type_name, cross_guid, cross_connector_type_name) \
617                     in data.get_connection_data(guid):
618                 (testbed_guid, factory_id) = data.get_box_data(guid)
619                 (cross_testbed_guid, cross_factory_id) = data.get_box_data(
620                         cross_guid)
621                 if testbed_guid == cross_testbed_guid:
622                     testbed.defer_connect(guid, connector_type_name, 
623                             cross_guid, cross_connector_type_name)
624                 else: 
625                     cross_testbed = self._testbeds[cross_testbed_guid]
626                     cross_testbed_id = cross_testbed.testbed_id
627                     testbed.defer_cross_connect(guid, connector_type_name, cross_guid, 
628                             cross_testbed_guid, cross_testbed_id, cross_factory_id, 
629                             cross_connector_type_name)
630                     # save cross data for later
631                     self._add_crossdata(testbed_guid, guid, cross_testbed_guid,
632                             cross_guid)
633             for trace_id in data.get_trace_data(guid):
634                 testbed.defer_add_trace(guid, trace_id)
635             for (autoconf, address, netprefix, broadcast) in \
636                     data.get_address_data(guid):
637                 if address != None:
638                     testbed.defer_add_address(guid, address, netprefix, 
639                             broadcast)
640             for (destination, netprefix, nexthop) in data.get_route_data(guid):
641                 testbed.defer_add_route(guid, destination, netprefix, nexthop)
642                 
643     def _add_crossdata(self, testbed_guid, guid, cross_testbed_guid, cross_guid):
644         if testbed_guid not in self._cross_data:
645             self._cross_data[testbed_guid] = dict()
646         if cross_testbed_guid not in self._cross_data[testbed_guid]:
647             self._cross_data[testbed_guid][cross_testbed_guid] = set()
648         self._cross_data[testbed_guid][cross_testbed_guid].add(cross_guid)
649
650     def _get_cross_data(self, testbed_guid):
651         cross_data = dict()
652         if not testbed_guid in self._cross_data:
653             return cross_data
654         for cross_testbed_guid, guid_list in \
655                 self._cross_data[testbed_guid].iteritems():
656             cross_data[cross_testbed_guid] = dict()
657             cross_testbed = self._testbeds[cross_testbed_guid]
658             for cross_guid in guid_list:
659                 elem_cross_data = dict()
660                 cross_data[cross_testbed_guid][cross_guid] = elem_cross_data
661                 attributes_list = cross_testbed.get_attribute_list(cross_guid)
662                 for attr_name in attributes_list:
663                     attr_value = cross_testbed.get(cross_guid, attr_name)
664                     elem_cross_data[attr_name] = attr_value
665         return cross_data
666