Changeset 43197eb for fedd/federation/experiment_control.py

Timestamp:

Apr 16, 2010 8:16:22 AM (14 years ago)

Author:

Ted Faber <faber@…>

Branches:

axis_example, compt_changes, info-ops, master, version-3.01, version-3.02

Children:

4f2f41f

Parents:

0c4b12c

Message:

better service handling including project_export psuedo service done more or less right- tested on dry runs

File:

• fedd/federation/experiment_control.py (modified) (18 diffs)

fedd/federation/experiment_control.py

@@ -42 +42 @@
 fl = logging.getLogger("fedd.experiment_control")
 fl.addHandler(nullHandler())
+
+
+# Right now, no support for composition.
+class federated_service:
+    def __init__(self, name, exporter=None, importers=[], params={ }, reqs=[]):
+        self.name=name
+        self.exporter=exporter
+        self.importers=importers
+        self.params = params
+        self.reqs = reqs
 
 class experiment_control_local:
@@ -786 +796 @@
         else: return None
 
-    def get_access(self, tb, nodes, tbparam, master, export_project,
-            access_user, services):
+    def get_access(self, tb, nodes, tbparam, access_user, masters):
         """
         Get access to testbed through fedd and set the parameters for that tb
         """
+        def get_export_project(svcs):
+            """
+            Look through for the list of federated_service for this testbed
+            objects for a project_export service, and extract the project
+            parameter.
+            """
+
+            pe = [s for s in svcs if s.name=='project_export']
+            if len(pe) == 1:
+                return pe[0].params.get('project', None)
+            elif len(pe) == 0:
+                return None
+            else:
+                raise service_error(service_error.req,
+                        "More than one project export is not supported")
+
         uri = self.tbmap.get(tb, None)
         if not uri:
             raise service_error(service_error.server_config, 
                     "Unknown testbed: %s" % tb)
+
+        export_svcs = masters.get(tb,[])
+        import_svcs = [ s for m in masters.values() \
+                for s in m \
+                    if tb in s.importers ]
+
+        export_project = get_export_project(export_svcs)
 
         # Tweak search order so that if there are entries in access_user that 
@@ -822 +854 @@
                         'destinationTestbed' : { 'uri' : uri },
                         'credential': [ 'user: %s' % u ],
-                        'user':  [ {'userID': { 'localname': u } } ],
                         'allocID' : { 'localname': 'test' },
                     }
 
-            # If there is a master, and this is it, ask it to export services
-            # XXX move this to export pseudo-service
-            if tb == master:
-                req['service'] = [
-                        { 'name': 'userconfig', 'visibility': 'export'},
-                        { 'name': 'SMB', 'visibility': 'export'},
-                        { 'name': 'seer', 'visibility': 'export'},
-                        { 'name': 'tmcd', 'visibility': 'export'},
-                    ]
+            # Make the service request from the services we're importing and
+            # exporting.  Keep track of the export request ids so we can
+            # collect the resulting info from the access response.
+            e_keys = { }
+            if import_svcs or export_svcs:
+                req['service'] = [ ]
+
+                for i, s in enumerate(import_svcs):
+                    idx = 'import%d' % i
+                    sr = {'id': idx, 'name': s.name, 'visibility': 'import' }
+                    if s.params:
+                        sr['fedAttr'] = [ { 'attribute': k, 'value': v } \
+                                for k, v in s.params.items()]
+                    req['service'].append(sr)
+
+                for i, s in enumerate(export_svcs):
+                    idx = 'export%d' % i
+                    e_keys[idx] = s
+                    sr = {'id': idx, 'name': s.name, 'visibility': 'export' }
+                    if s.params:
+                        sr['fedAttr'] = [ { 'attribute': k, 'value': v } 
+                                for k, v in s.params.items()]
+                    req['service'].append(sr)
 
             # node resources if any
@@ -886 +931 @@
                 "uri": uri,
                 }
-        if 'service' in r:
-            for s in r['service']:
-                # Tag each service with the origin testbed
-                if s.has_key('fedAttr'):
-                    # The else connects with the for
-                    for a in s['fedAttr']:
-                        if a.get('attribute', None) == 'testbed':
-                            break
-                    else:
-                        s['fedAttr'].append({'attribute': 'testbed',
-                            'value': tb})
-                else:
-                    s['fedAttr'] = [ {'attribute': 'testbed', 'value': tb} ]
-            services.extend(r['service'])
+
+        # Collect the responses corresponding to the services this testbed
+        # exports.  These will be the service requests that we will include in
+        # the start segment requests (with appropriate visibility values) to
+        # import and export the segments.
+        for s in r.get('service', []):
+            id = s.get('id', None)
+            if id and id in e_keys:
+                e_keys[id].reqs.append(s)
 
         # Add attributes to parameter space.  We don't allow attributes to
@@ -968 +1008 @@
             self.response = None
 
-        def __call__(self, uri, aid, topo, master, attrs=None, connInfo=None,
-                services=None):
+        def __call__(self, uri, aid, topo, masters, attrs=None, connInfo=None):
             req = {
                     'allocID': { 'fedid' : aid }, 
@@ -975 +1014 @@
                         'topdldescription': topo.to_dict(),
                     },
-                    'master': master,
+                    # XXX: deprecated
+                    'master': False,
                 }
 
             if connInfo:
                 req['connection'] = connInfo
-            # Add services to request.  The master exports, everyone else
-            # imports.
-            if services:
-                svcs = [ x.copy() for x in services]
-                for s in svcs:
-                    if master: s['visibility'] = 'export'
-                    else: s['visibility'] = 'import'
-                req['service'] = svcs
+
+            import_svcs = [ s for m in masters.values() \
+                    for s in m if self.testbed in s.importers]
+
+            if import_svcs or self.testbed in masters:
+                req['service'] = []
+
+            for s in import_svcs:
+                for r in s.reqs:
+                    sr = copy.deepcopy(r)
+                    sr['visibility'] = 'import';
+                    req['service'].append(sr)
+
+            for s in masters.get(self.testbed, []):
+                for r in s.reqs:
+                    sr = copy.deepcopy(r)
+                    sr['visibility'] = 'export';
+                    req['service'].append(sr)
+
             if attrs:
                 req['fedAttr'] = attrs
@@ -1038 +1089 @@
     
 
-    def allocate_resources(self, allocated, master, eid, expid, 
+    def allocate_resources(self, allocated, masters, eid, expid, 
             tbparams, topo, tmpdir, alloc_log=None, log_collector=None, 
-            attrs=None, connInfo={}, services=[]):
+            attrs=None, connInfo={}):
 
         started = { }           # Testbeds where a sub-experiment started 
@@ -1076 +1127 @@
                         caller=self.call_StartSegment,
                         log_collector=log_collector), 
-                    args=(uri, aid, topo[tb], tb == master, 
-                        attrs, connInfo[tb], services),
+                    args=(uri, aid, topo[tb], masters, attrs, connInfo[tb]),
                     name=tb,
                     pdata=thread_pool, trace_file=self.trace_file)
@@ -1307 +1357 @@
         return hosts, ips
 
-    def get_access_to_testbeds(self, testbeds, access_user, 
-            export_project, master, allocated, tbparams, services):
+    def get_access_to_testbeds(self, testbeds, access_user, allocated, 
+            tbparams, masters):
         """
         Request access to the various testbeds required for this instantiation
@@ -1316 +1366 @@
         """
         for tb in testbeds:
-            self.get_access(tb, None, tbparams, master,
-                    export_project, access_user, services)
+            self.get_access(tb, None, tbparams, access_user, masters)
             allocated[tb] = 1
 
@@ -1479 +1528 @@
         return rv
 
-    def get_master_project(self, req):
-        master= None
-        export_project = None
-        for e in [ s for s in req.get('service', []) \
-                if s.get('name') == 'project_export']:
-            for a in e.get('fedAttr', []):
-                attr = a.get('attribute', None)
-                if attr == 'testbed':
-                    master = a.get('value', None)
-                elif attr == 'project':
-                    export_project = a.get('value', None)
-        
-        return (master, export_project)
-
-
-
     def create_experiment(self, req, fid):
         """
@@ -1501 +1534 @@
 
         Creates a working directory, splits the incoming description using the
-        splitter script and parses out the avrious subsections using the
-        lcasses above.  Once each sub-experiment is created, use pooled threads
+        splitter script and parses out the various subsections using the
+        classes above.  Once each sub-experiment is created, use pooled threads
         to instantiate them and start it all up.
         """
@@ -1590 +1623 @@
                 raise service_error(service_error.server_config, 
                         "Bad key type (%s)" % self.ssh_type)
-            master, export_project = self.get_master_project(req)
-            # XXX get these out when master and project are optional
-            if not master:
-                raise service_error(service_error.req,
-                        "No master testbed label")
-            if not export_project:
-                raise service_error(service_error.req, "No export project")
-            # XXX
-
+
+            # Copy the service request
+            tb_services = [ s for s in req.get('service',[]) ]
             # Translate to topdl
             if self.splitter_url:
@@ -1607 +1634 @@
             else:
                 tclcmd = [self.tclsh, self.tcl_splitter, '-t', '-x', 
-                    str(self.muxmax), '-m', master]
+                    str(self.muxmax), '-m', 'dummy']
 
                 tclcmd.extend([pid, gid, eid, tclfile])
@@ -1628 +1655 @@
                         if a.attribute == 'testbed'])
 
+            masters = { }           # testbeds exporting services
+            for s in tb_services:
+                # If this is a project_export request with the imports field
+                # blank, fill it in.
+                if s.get('name', '') == 'project_export':
+                    if 'import' not in s or len(s['import']) == 0:
+                        s['import'] = [ tb for tb in testbeds \
+                                if tb not in s.get('export',[])]
+                # Add the service to masters
+                for tb in s.get('export', []):
+                    if s.get('name', None) and s.get('import', None):
+                        if tb not in masters:
+                            masters[tb] = [ ]
+
+                        params = { }
+                        if 'fedAttr' in s:
+                            for a in s['fedAttr']:
+                                params[a.get('attribute', '')] = \
+                                        a.get('value','')
+
+                        masters[tb].append(federated_service(name=s['name'],
+                                exporter=tb, importers=s.get('import',[]),
+                                params=params))
+                    else:
+                        log.error('Testbed service does not have name " + \
+                                "and importers')
+
+
             allocated = { }         # Testbeds we can access
             topo ={ }               # Sub topologies
             connInfo = { }          # Connection information
-            services = [ ]
-            masters = { }           # testbeds exporting services
-            self.get_access_to_testbeds(testbeds, access_user, 
-                    export_project, master, allocated, tbparams, services)
-
-            # After this masters will hold a set of services exported by each
-            # testbed
-            for s in services:
-                i = s.get('visibility', 'import')
-                if i == 'export':
-                    for a in s.get('fedAttr', []):
-                        if a.get('attribute', '') == 'testbed':
-                            tb = a.get('value', None)
-                            if tb:
-                                if masters.has_key(tb): 
-                                    masters[tb].add(s.get('name',''))
-                                else:
-                                    masters[tb] = set([s.get('name','')])
-                            else:
-                                log.error('Testbed attribute with no value?')
-
+            self.get_access_to_testbeds(testbeds, access_user, allocated, 
+                    tbparams, masters)
 
             self.split_topology(top, topo, testbeds)
@@ -1697 +1733 @@
                 tb = t.get_attribute('testbed')
                 if tb: 
-                    self.get_access(tb, None, tbparams, master, 
-                            export_project, access_user, services)
+                    self.get_access(tb, None, tbparams, export_project, 
+                            access_user, masters)
                     tbparams[k] = tbparams[tb]
                     del tbparams[tb]
@@ -1722 +1758 @@
                         'name': [ { 'localname' : eid} ],
                         'allocID' : tbparams[k]['allocID'],
-                        'master' : k == master,
+                        # XXX:
+                        #'master' : k == master,
                         'uri': tbparams[k]['uri'],
                     }
@@ -1803 +1840 @@
         # Start a thread to do the resource allocation
         t  = Thread(target=self.allocate_resources,
-                args=(allocated, master, eid, expid, tbparams, 
-                    topo, tmpdir, alloc_log, alloc_collector, attrs, connInfo,
-                    services),
+                args=(allocated, masters, eid, expid, tbparams, 
+                    topo, tmpdir, alloc_log, alloc_collector, attrs, connInfo),
                 name=eid)
         t.start()