Changeset 593e901 for fedd/federation

Timestamp:

Mar 5, 2010 1:59:26 PM (15 years ago)

Author:

Ted Faber <faber@…>

Branches:

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

Children:

73e0a61

Parents:

c5b28bf

Message:

Checkpoint working federation w/PG (w/o routing yet...)

Location:

fedd/federation

Files:

• emulab_access.py (modified) (2 diffs)
• experiment_control.py (modified) (2 diffs)
• protogeni_access.py (modified) (4 diffs)
• proxy_protogeni_segment.py (modified) (10 diffs)

fedd/federation/emulab_access.py

@@ -986 +986 @@
             peer = info.get('peer', "")
             ldomain = self.domain;
+            ssh_port = info.get('ssh_port', 22)
 
             mexp = info.get('masterexperiment',"")
@@ -1001 +1002 @@
                 if active == 'True':
                     print >>f, "active: True"
+                    print >>f, "ssh_port: %s" % ssh_port
                     if type in ('control', 'both'):
                         for s in [s for s in services \

fedd/federation/experiment_control.py

@@ -1397 +1397 @@
                         },
                         {
+                            'name': 'ssh_port',
+                            'key': 'fedid:%s/%s-port' % (expid, myname),
+                            'store': self.store_url,
+                            'type': 'output',
+                        },
+                        {
                             'name': 'peer',
                             'key': 'fedid:%s/%s' % (expid, desthost),
+                            'store': self.store_url,
+                            'type': 'input',
+                        },
+                        {
+                            'name': 'ssh_port',
+                            'key': 'fedid:%s/%s-port' % (expid, desthost),
                             'store': self.store_url,
                             'type': 'input',
@@ -1409 +1421 @@
                 for h in (myname, desthost):
                     self.auth.set_attribute(aid, 'fedid:%s/%s' % (expid, h))
+                    self.auth.set_attribute(aid, 'fedid:%s/%s-port' % \
+                            (expid, h))
             else:
                 self.log.error("No aid for %s in new_portal_node" % st)

fedd/federation/protogeni_access.py

@@ -760 +760 @@
                 if active == 'True':
                     print >>f, "active: True"
+                    print >>f, "ssh_port: %s" % ssh_port
                     if type in ('control', 'both'):
                         for s in [s for s in services \
@@ -777 +778 @@
                 print >>f, "ssh_privkey: /usr/local/federation/etc/%s" % \
                         secretkey_base
-                print >>f, "ssh_port: %s" % ssh_port
                 f.close()
             except IOError, e:
@@ -1036 +1036 @@
             # generation off longer.
             self.import_store_info(certfile, connInfo)
-            self.generate_portal_configs(topo, pubkey_base, 
-                    secretkey_base, tmpdir, master, ename, connInfo, 
-                    services)
+            #self.generate_portal_configs(topo, pubkey_base, 
+                    #secretkey_base, tmpdir, master, ename, connInfo, 
+                    #services)
             rspec = self.generate_rspec(topo, "%s/%s/" \
                     % (self.staging_dir, ename), master, connInfo)
@@ -1047 +1047 @@
                     cm_url=self.cm_url)
             rv = starter(self, aid, user, rspec, pubkey_base, secretkey_base,
+                    master, ename,
                     "%s/%s" % (self.staging_dir, ename), tmpdir, cf, cpw,
-                    certfile, topo, connInfo)
+                    certfile, topo, connInfo, services)
         except service_error, e:
             err = e

fedd/federation/proxy_protogeni_segment.py

@@ -49 +49 @@
 
     def pg_call(self, url, method, params, context):
+        max_retries = 5
+        retries = 0
+
         s = service_caller(method, request_body_name="", strict=False)
         self.log.debug("Calling %s %s" % (url, method))
         if not self.debug:
-            r = s.call_xmlrpc_service(url, params, context=context)
-            if r.get('code', -1) != 0:
-                raise self.ProtoGENIError(op=method, 
-                        code=r.get('code', 'unknown'), 
-                        output=r.get('output', 'No output'))
-            else:
-                return r.get('value', None)
+            while retries < max_retries:
+                r = s.call_xmlrpc_service(url, params, context=context)
+                code = r.get('code', -1)
+                if code == 0:
+                    # Success leaves the loop here
+                    return r.get('value', None)
+                elif code == 14 and retries +1 < max_retries:
+                    # Busy resource
+                    retries+= 1
+                    self.log.info("Resource busy, retrying in 30 secs")
+                    time.sleep(30)
+                else:
+                    # NB: out of retries falls through to here
+                    raise self.ProtoGENIError(op=method, 
+                            code=r.get('code', 'unknown'), 
+                            output=r.get('output', 'No output'))
         else:
             if method in self.debug_fail:
@@ -80 +92 @@
     # interfaces in 'interfaces'.  I love having XML parser code lying around.
     def manifest_to_dict(self, manifest):
-        if self.debug: return { }
+        if self.debug: 
+            self.log.debug("Returning null manifest dict")
+            return { }
 
         # The class allows us to keep a little state - the dict under
@@ -118 +132 @@
         p.Parse(manifest)
         return mp.d
+
+
+    def generate_portal_configs(self, parent, topo, pubkey_base, 
+            secretkey_base, tmpdir, master, leid, connInfo, services, nodes):
+
+        def conninfo_to_dict(key, info):
+            """
+            Make a cpoy of the connection information about key, and flatten it
+            into a single dict by parsing out any feddAttrs.
+            """
+
+            rv = None
+            for i in info:
+                if key == i.get('portal', "") or \
+                        key in [e.get('element', "") \
+                        for e in i.get('member', [])]:
+                    rv = i.copy()
+                    break
+
+            else:
+                return rv
+
+            if 'fedAttr' in rv:
+                for a in rv['fedAttr']:
+                    attr = a.get('attribute', "")
+                    val = a.get('value', "")
+                    if attr and attr not in rv:
+                        rv[attr] = val
+                del rv['fedAttr']
+            return rv
+
+        # XXX: un hardcode this
+        def client_null(f, s):
+            print >>f, "Service: %s" % s['name']
+
+        def client_smb(f, s):
+            print >>f, "Service: %s" % s['name']
+            smbshare = None
+            smbuser = None
+            smbproj = None
+            for a in s.get('fedAttr', []):
+                if a.get('attribute', '') == 'SMBSHARE':
+                    smbshare = a.get('value', None)
+                elif a.get('attribute', '') == 'SMBUSER':
+                    smbuser = a.get('value', None)
+                elif a.get('attribute', '') == 'SMBPROJ':
+                    smbproj = a.get('value', None)
+
+            if all((smbshare, smbuser, smbproj)):
+                print >>f, "SMBshare: %s" % smbshare
+                print >>f, "ProjectUser: %s" % smbuser
+                print >>f, "ProjectName: %s" % smbproj
+
+        client_service_out = {
+                'SMB': client_smb,
+                'tmcd': client_null,
+                'seer': client_null,
+                'userconfig': client_null,
+            }
+        # XXX: end un hardcode this
+
+
+        seer_out = False
+        client_out = False
+        for e in [ e for e in topo.elements \
+                if isinstance(e, topdl.Computer) and e.get_attribute('portal')]:
+            myname = e.name[0]
+            type = e.get_attribute('portal_type')
+
+            info = conninfo_to_dict(myname, connInfo)
+
+            if not info:
+                raise service_error(service_error.req,
+                        "No connectivity info for %s" % myname)
+
+            # Translate to physical name (ProtoGENI doesn't have DNS)
+            physname = nodes.get(myname, {'hostname': myname})['hostname']
+            peer = info.get('peer', "")
+            ldomain = parent.domain;
+
+            mexp = info.get('masterexperiment',"")
+            mproj, meid = mexp.split("/", 1)
+            mdomain = info.get('masterdomain',"")
+            muser = info.get('masteruser','root')
+            smbshare = info.get('smbshare', 'USERS')
+            ssh_port = info.get('ssh_port', '22')
+
+            active = info.get('active', 'False')
+
+            cfn = "%s/%s.gw.conf" % (tmpdir, myname.lower())
+            tunnelconfig = parent.attrs.has_key('TunnelCfg')
+            try:
+                f = open(cfn, "w")
+                if active == 'True':
+                    print >>f, "active: True"
+                    print >>f, "ssh_port: %s" % ssh_port
+                    if type in ('control', 'both'):
+                        for s in [s for s in services \
+                                if s.get('name', "") in parent.imports]:
+                            p = urlparse(s.get('server', 'http://localhost'))
+                            print >>f, 'port: remote:%s:%s:%s' % \
+                                    (p.port, p.hostname, p.port)
+
+                if tunnelconfig:
+                    print >>f, "tunnelip: %s" % tunnelconfig
+                # XXX: send this an fedattr
+                #print >>f, "seercontrol: control.%s.%s%s" % \
+                        #(meid.lower(), mproj.lower(), mdomain)
+                print >>f, "peer: %s" % peer.lower()
+                print >>f, "ssh_pubkey: /usr/local/federation/etc/%s" % \
+                        pubkey_base
+                print >>f, "ssh_privkey: /usr/local/federation/etc/%s" % \
+                        secretkey_base
+                f.close()
+            except IOError, e:
+                raise service_error(service_error.internal,
+                        "Can't write protal config %s: %s" % (cfn, e))
+            
+            # XXX: This little seer config file needs to go away.
+            if not seer_out:
+                try:
+                    seerfn = "%s/seer.conf" % tmpdir
+                    f = open(seerfn, "w")
+                    if not master:
+                        print >>f, "ControlNode: control.%s.%s%s" % \
+                            (meid.lower(), mproj.lower(), mdomain)
+                    print >>f, "ExperimentID: %s" % mexp
+                    f.close()
+                except IOError, e:
+                    raise service_error(service_error.internal, 
+                            "Can't write seer.conf: %s" %e)
+                seer_out = True
+
+            if not client_out and type in ('control', 'both'):
+                try:
+                    f = open("%s/client.conf" % tmpdir, "w")
+                    print >>f, "ControlGateway: %s" % physname.lower()
+                    for s in services:
+                        if s.get('name',"") in parent.imports and \
+                                s.get('visibility','') == 'import':
+                            client_service_out[s['name']](f, s)
+                    # Does seer need this?
+                    # print >>f, "ExperimentID: %s/%s" % (mproj, meid)
+                    f.close()
+                except IOError, e:
+                    raise service_error(service_error.internal,
+                            "Cannot write client.conf: %s" %s)
+                client_out = True
 
 
@@ -153 +315 @@
                     surl = p.get('store', None)
                     if surl and k:
-                        req = { 'name': k, 'value': self.ssh_port }
+                        req = { 'name': k, 'value': ssh_port }
                         self.call_SetValue(surl, req, cf)
                     else:
@@ -211 +373 @@
                     print >>script, "%s %s@%s:%s %s/etc" % \
                             (scp, user, host, sshd_config, fed_dir)
+                # Look in tmpdir to get the names.  They've all been copied
+                # into the (remote) staging dir
                 if os.access("%s/%s.gw.conf" % (tmpdir, vname), os.R_OK):
                     print >>script, "%s %s@%s:%s/%s.gw.conf %s/etc" % \
@@ -240 +404 @@
         for n in nodes:
             t = Thread(target=self.ssh_cmd, args=(user, n,
-                    "sudo /bin/sh ./%s.startup" % n))
+                    "sudo /bin/sh ./%s.startup &" % n))
             t.start()
             threads.append(t)
@@ -253 +417 @@
 
 
-    def __call__(self, parent, aid, user, rspec, pubkey, secretkey, 
-            stagingdir, tmpdir, certfile, certpw, export_certfile, topo,
-            connInfo, timeout=0):
+    def __call__(self, parent, aid, user, rspec, pubkey, secretkey, master,
+            ename, stagingdir, tmpdir, certfile, certpw, export_certfile, topo,
+            connInfo, services, timeout=0):
         """
         Start a sub-experiment on a federant.
@@ -302 +466 @@
         if not self.ssh_cmd(user, host, "sh -x %s" % scriptbase):
             return False
-
-        # Copy software to the staging machine
-        for d in (tmpdir, lsoftdir):
-            if os.path.isdir(d):
-                for f in os.listdir(d):
-                    if not os.path.isdir("%s/%s" % (d, f)):
-                        if not self.scp_file("%s/%s" % (d, f), 
-                                user, host, "%s/%s" % (stagingdir, f)):
-                            self.log.error("Scp failed")
-                            return False
 
         try:
@@ -384 +538 @@
         except self.ProtoGENIError, e:
             raise service_error(service_error.federant,
-                    "ProtoGENI: %s" % e)
+                    "ProtoGENI: %s %s" % (e.code, e))
 
         # With manifest in hand, we can export the portal node names.
         nodes = self.manifest_to_dict(manifest)
+        print nodes
         self.export_store_info(export_certfile, nodes, parent.ssh_port,
                 connInfo)
+        self.generate_portal_configs(parent, topo, pubkey, secretkey, tmpdir, 
+                master, ename, connInfo, services, nodes)
+
+        # Copy software to the staging machine (done after generation to copy
+        # those, too)
+        for d in (tmpdir, lsoftdir):
+            if os.path.isdir(d):
+                for f in os.listdir(d):
+                    if not os.path.isdir("%s/%s" % (d, f)):
+                        if not self.scp_file("%s/%s" % (d, f), 
+                                user, host, "%s/%s" % (stagingdir, f)):
+                            self.log.error("Scp failed")
+                            return False
+
 
         # Now we wait for the nodes to start on PG
@@ -403 +572 @@
                 if status == 'notready':
                     time.sleep(30)
-        except Self.ProtoGENIError, e:
+        except self.ProtoGENIError, e:
             raise service_error(service_error.federant,
-                    "ProtoGENI: %s" % e)
+                    "ProtoGENI: %s %s" % (e.code, e))
 
         if status == 'failed':