source: fedd/fedd_create_experiment.py @ 1af38d6

axis_examplecompt_changesinfo-opsversion-1.30version-2.00version-3.01version-3.02
Last change on this file since 1af38d6 was 1af38d6, checked in by Ted Faber <faber@…>, 16 years ago

multithreading for startup. Needs more testing
  • Property mode set to 100644
File size: 33.9 KB
Line 
1#!/usr/local/bin/python
2
3import os,sys
4
5from ZSI import *
6from M2Crypto import SSL
7from M2Crypto.SSL.SSLServer import SSLServer
8import M2Crypto.httpslib
9
10import xml.parsers.expat
11
12import re
13import random
14import string
15import subprocess
16import tempfile
17import copy
18
19from threading import *
20
21from subprocess import *
22
23from fedd_services import *
24from fedd_internal_services import *
25from fedd_util import *
26import parse_detail
27from service_error import *
28
29class fedd_create_experiment_local:
30    scripts = ["fed_bootstrap", "federate.sh", "smbmount.FreeBSD.pl",
31        "smbmount.Linux.pl", "make_hosts", "fed-tun.pl", "fed-tun.ucb.pl",
32        "fed_evrepeater", "rc.accounts.patch"]
33   
34    def __init__(self, 
35            cert_file=None,
36            cert_pwd=None,
37            debug=False,
38            muxmax=2,
39            nthreads=2,
40            project_user = "faber",
41            scp_exec="/usr/bin/scp",
42            scripts_dir="./", 
43            splitter=None,
44            ssh_exec="/usr/bin/ssh",
45            ssh_identity_file=None,
46            ssh_keygen="/usr/bin/ssh-keygen",
47            ssh_pubkey_file=None,
48            ssh_type="rsa",
49            tbmap=None,
50            tclsh="/usr/local/bin/otclsh",
51            tcl_splitter="/usr/testbed/lib/ns2ir/parse.tcl",
52            trusted_certs=None,
53            ):
54        self.scripts = fedd_create_experiment_local.scripts
55        self.thread_with_rv = fedd_create_experiment_local.pooled_thread
56        self.thread_pool = fedd_create_experiment_local.thread_pool
57
58        self.cert_file = cert_file
59        self.cert_pwd = cert_pwd
60        self.debug = debug
61        self.muxmax = muxmax
62        self.nthreads = nthreads
63        self.project_user = project_user
64        self.scp_exec = scp_exec
65        self.scripts_dir = scripts_dir
66        self.splitter = splitter
67        self.ssh_exec=ssh_exec
68        self.ssh_keygen = ssh_keygen
69        self.ssh_identity_file = ssh_identity_file
70        self.ssh_type = ssh_type
71        self.tclsh = tclsh
72        self.tcl_splitter = tcl_splitter
73        self.tbmap = tbmap
74        self.trusted_certs=trusted_certs
75
76        self.def_expstart = \
77                "sudo -H /bin/sh FEDDIR/fed_bootstrap >& /tmp/federate";
78        self.def_mexpstart = "sudo -H FEDDIR/make_hosts FEDDIR/hosts";
79        self.def_gwstart = \
80                "sudo -H FEDDIR/fed-tun.pl -f GWCONF>& /tmp/bridge.log";
81        self.def_mgwstart = \
82                "sudo -H FEDDIR/fed-tun.pl -f GWCONF >& /tmp/bridge.log";
83        self.def_gwimage = "FBSD61-TUNNEL2";
84        self.def_gwtype = "pc";
85
86
87        if ssh_pubkey_file:
88            try:
89                f = open(ssh_pubkey_file, 'r')
90                self.ssh_pubkey = f.read()
91                f.close()
92            except IOError:
93                raise service_error(service_error.internal,
94                        "Cannot read sshpubkey")
95
96        # Confirm federation scripts in the right place
97        for s in self.scripts:
98            if not os.path.exists(self.scripts_dir + "/" + s):
99                raise service_error(service_error.server_config,
100                        "%s/%s not in local script dir" % (self.scripts_dir, s))
101    class thread_pool:
102        def __init__(self):
103            self.changed = Condition()
104            self.started = 0
105            self.terminated = 0
106
107        def acquire(self):
108            self.changed.acquire()
109
110        def release(self):
111            self.changed.release()
112
113        def wait(self, timeout = None):
114            self.changed.wait(timeout)
115
116        def start(self):
117            self.changed.acquire()
118            self.started += 1
119            self.changed.notifyAll()
120            self.changed.release()
121
122        def terminate(self):
123            self.changed.acquire()
124            self.terminated += 1
125            self.changed.notifyAll()
126            self.changed.release()
127
128        def clear(self):
129            self.changed.acquire()
130            self.started = 0
131            self.terminated =0
132            self.changed.notifyAll()
133            self.changed.release()
134
135
136
137    class pooled_thread(Thread):
138        def __init__(self, group=None, target=None, name=None, args=(), 
139                kwargs={}, pdata=None):
140            Thread.__init__(self, group, target, name, args, kwargs)
141            self.rv = None
142            self.exception = None
143            self.target=target
144            self.args = args
145            self.kwargs = kwargs
146            self.pdata = pdata
147       
148        def run(self):
149            if self.pdata:
150                self.pdata.start()
151
152            if self.target:
153                try:
154                    self.rv = self.target(*self.args, **self.kwargs)
155                except e:
156                    self.exception = e
157
158            print "%s done: %s" % (self.getName(), self.rv)
159            if self.pdata:
160                self.pdata.terminate()
161
162    def copy_file(self, src, dest, size=1024):
163        """
164        Exceedingly simple file copy.
165        """
166        s = open(src,'r')
167        d = open(dest, 'w')
168
169        buf = "x"
170        while buf != "":
171            buf = s.read(size)
172            d.write(buf)
173        s.close()
174        d.close()
175
176    def scp_file(self, file, user, host, dest=""):
177        """
178        scp a file to the remote host.
179        """
180
181        scp_cmd = [self.scp_exec, file, "%s@%s:%s" % (user, host, dest)]
182        if not self.debug:
183            rv = call(scp_cmd)
184        else:
185            print "debug: %s" % " ".join(scp_cmd)
186            rv = 0
187
188        return rv == 0
189
190    def ssh_cmd(self, user, host, cmd, wname=None, timeout=0):
191        sh_str = "%s %s@%s %s" % (self.ssh_exec, user, host, cmd)
192
193        if not self.debug:
194            # This should be done more carefully
195            sub = Popen(sh_str, shell=True)
196            return sub.wait() == 0
197        else:
198            print "debug: %s" % sh_str
199            return True
200
201    def ship_scripts(self, host, user, dest_dir):
202        if self.ssh_cmd(user, host, "mkdir -p %s" % dest_dir):
203            for s in self.scripts:
204                if not self.scp_file("%s/%s" % (self.scripts_dir, s),
205                        user, host, dest_dir):
206                    return False
207            return True
208        else:
209            return False
210
211    def ship_configs(self, host, user, src_dir, dest_dir):
212        if not self.ssh_cmd(user, host, "mkdir -p %s" % dest_dir):
213            return False
214        if not self.ssh_cmd(user, host, "chmod 770 %s" % dest_dir):
215            return False
216
217        for f in os.listdir(src_dir):
218            if os.path.isdir(f):
219                if not self.ship_configs(host, user, "%s/%s" % (src_dir, f), 
220                        "%s/%s" % (dest_dir, f)):
221                    return False
222            else:
223                if not self.scp_file("%s/%s" % (src_dir, f), 
224                        user, host, dest_dir):
225                    return False
226        return True
227
228    def start_segment(self, tb, eid, tbparams, tmpdir, timeout=0):
229        host = "%s%s" % (tbparams[tb]['host'], tbparams[tb]['domain'])
230        user = tbparams[tb]['user']
231        pid = tbparams[tb]['project']
232        # XXX
233        base_confs = ( "hosts", "vtopo.xml", "viz.xml")
234        tclfile = "%s.%s.tcl" % (eid, tb)
235        expinfo_exec = "/usr/testbed/bin/expinfo"
236        proj_dir = "/proj/%s/exp/%s/tmp" % (pid, eid)
237        tarfiles_dir = "/proj/%s/tarfiles/%s" % (pid, eid)
238        rpms_dir = "/proj/%s/rpms/%s" % (pid, eid)
239        state_re = re.compile("State:\s+(\w+)")
240        state = "none"
241
242        status = Popen([self.ssh_exec, "%s@%s" % (user, host), 
243                expinfo_exec, pid, eid], stdout=PIPE)
244        for line in status.stdout:
245            m = state_re.match(line)
246            if m: state = m.group(1)
247        rv = status.wait()
248        if rv != 0:
249            raise service_error(service_error.internal,
250                    "Cannot get status of segment %s:%s/%s" % (tb, pid, eid))
251        # XXX
252        print "%s: %s" % (tb, state)
253        print "transferring experiment to %s" % tb
254
255        if not self.scp_file("%s/%s/%s" % (tmpdir, tb, tclfile), user, host):
256            return False
257        # Clear and create the tarfiles and rpm directories
258        for d in (tarfiles_dir, rpms_dir):
259            if not self.ssh_cmd(user, host, 
260                    "/bin/sh -c \"'/bin/rm -rf %s/*'\"" % d):
261                return False
262            if not self.ssh_cmd(user, host, "mkdir -p %s" % d, 
263                    "create tarfiles"):
264                return False
265       
266        if state == 'active':
267            # Remote experiment is active.  Modify it.
268            for f in base_confs:
269                if not self.scp_file("%s/%s" % (tmpdir, f), user, host,
270                        "%s/%s" % (proj_dir, f)):
271                    return False
272            if not self.ship_scripts(host, user, proj_dir):
273                return False
274            if not self.ship_configs(host, user, "%s/%s" % (tmpdir, tb),
275                    proj_dir):
276                return False
277            if os.path.isdir("%s/tarfiles" % tmpdir):
278                if not self.ship_configs(host, user,
279                        "%s/tarfiles" % tmpdir, tarfiles_dir):
280                    return False
281            if os.path.isdir("%s/rpms" % tmpdir):
282                if not self.ship_configs(host, user,
283                        "%s/rpms" % tmpdir, tarfiles_dir):
284                    return False
285            print "Modifying %s on %s" % (eid, tb)
286            if not self.ssh_cmd(user, host,
287                    "/usr/testbed/bin/modexp -r -s -w %s %s %s" % \
288                            (pid, eid, tclfile), "modexp"):
289                return False
290            return True
291        elif state == "swapped":
292            # Remote experiment swapped out.  Modify it and swap it in.
293            for f in base_confs:
294                if not self.scp_file("%s/%s" % (tmpdir, f), user, host,
295                        "%s/%s" % (proj_dir, f)):
296                    return False
297            if not self.ship_scripts(host, user, proj_dir):
298                return False
299            if not self.ship_configs(host, user, "%s/%s" % (tmpdir, tb),
300                    proj_dir):
301                return False
302            if os.path.isdir("%s/tarfiles" % tmpdir):
303                if not self.ship_configs(host, user,
304                        "%s/tarfiles" % tmpdir, tarfiles_dir):
305                    return False
306            if os.path.isdir("%s/rpms" % tmpdir):
307                if not self.ship_configs(host, user,
308                        "%s/rpms" % tmpdir, tarfiles_dir):
309                    return False
310            print "Modifying %s on %s" % (eid, tb)
311            if not self.ssh_cmd(user, host,
312                    "/usr/testbed/bin/modexp -w %s %s %s" % (pid, eid, tclfile),
313                    "modexp"):
314                return False
315            print "Swapping %s in on %s" % (eid, tb)
316            if not self.ssh_cmd(user, host,
317                    "/usr/testbed/bin/swapexp -w %s %s in" % (pid, eid),
318                    "swapexp", timeout):
319                return False
320            return True
321        elif state == "none":
322            # No remote experiment.  Create one.  We do this in 2 steps so we
323            # can put the configuration files and scripts into the new
324            # experiment directories.
325
326            # Tarfiles must be present for creation to work
327            if os.path.isdir("%s/tarfiles" % tmpdir):
328                if not self.ship_configs(host, user,
329                        "%s/tarfiles" % tmpdir, tarfiles_dir):
330                    return False
331            if os.path.isdir("%s/rpms" % tmpdir):
332                if not self.ship_configs(host, user,
333                        "%s/rpms" % tmpdir, tarfiles_dir):
334                    return False
335            print "Creating %s on %s" % (eid, tb)
336            if not self.ssh_cmd(user, host,
337                    "/usr/testbed/bin/startexp -i -f -w -p %s -e %s %s" % \
338                            (pid, eid, tclfile), "startexp", timeout):
339                return False
340            # After startexp the per-experiment directories exist
341            for f in base_confs:
342                if not self.scp_file("%s/%s" % (tmpdir, f), user, host,
343                        "%s/%s" % (proj_dir, f)):
344                    return False
345            if not self.ship_scripts(host, user, proj_dir):
346                return False
347            if not self.ship_configs(host, user, "%s/%s" % (tmpdir, tb),
348                    proj_dir):
349                return False
350            print "Swapping %s in on %s" % (eid, tb)
351            if not self.ssh_cmd(user, host,
352                    "/usr/testbed/bin/swapexp -w %s %s in" % (pid, eid),
353                    "swapexp", timeout):
354                return False
355            return True
356        else:
357            # XXX
358            print "unknown state %s" % state
359            return False
360
361    def stop_segment(self, tb, eid, tbparams):
362        user = tbparams[tb]['user']
363        host = "%s%s" % (tbparams[tb]['host'], tbparams[tb]['domain'])
364        pid = tbparams[tb]['project']
365
366        # XXX:
367        print "Stopping %s on %s" % (eid, tb)
368        return self.ssh_cmd(user, host,
369                "/usr/testbed/bin/swapexp -w %d %d out" % (pid, eid))
370
371       
372    def generate_ssh_keys(self, dest, type="rsa" ):
373        """
374        Generate a set of keys for the gateways to use to talk.
375
376        Keys are of type type and are stored in the required dest file.
377        """
378        valid_types = ("rsa", "dsa")
379        t = type.lower();
380        if t not in valid_types: raise ValueError
381        # May raise CalledProcessError
382        rv = call([self.ssh_keygen, '-t', t, '-N', '', '-f', dest])
383        if rv != 0:
384            raise service_error(service_error.internal, 
385                    "Cannot generate nonce ssh keys.  %s return code %d" \
386                            % (self.ssh_keygen, rv))
387
388    def genviz(self, topo_file, viz_file):
389        """
390        Generate the visualization file from the topology file
391        """
392
393        class topo_parse:
394            """
395            Parse the vtopo file into a set of lans, links and nodes.
396            """
397            def __init__(self):
398                self.links = {}         # Each link is a list of at most 2 nodes
399                self.lans = {}          # Lans have more than 2 nodes
400                self.nodes = {}         # The nodes in the experiment
401                self.lan = {}           # The current link/len being collected
402                self.in_lan = False     # Is the conatining element lan?
403                self.in_node = False    # Is the conatining element node?
404                self.chars = None       # Last set of marked-up chars
405
406            def start_element(self, name, attrs):
407                """
408                New element started.  Set flags and clear lan
409                """
410                if name == "node":
411                    self.in_node = True
412                elif name == "lan":
413                    self.in_lan = True
414                    self.lan.clear()
415           
416            def end_element(self, name):
417                """
418                End of element.  Collect data if appropriate
419
420                If a node or lan is ending, create an entry in nodes or
421                lans/links.  If a vname/vnode in a node or lan is ending,
422                capture its name.  If a lan is ending, add the node to the
423                evolving link or lan.
424                """
425                if name == "node":
426                    self.in_node = False
427                if self.in_node and name == "vname":
428                    self.nodes[self.chars] = "node"
429                if self.in_lan:
430                    if name != "lan":
431                        if name == 'vname' or name == 'vnode':
432                            self.lan[name] = self.chars
433                    else:
434                        self.in_lan = False
435                        vname = self.lan['vname']
436                        links = self.links.get(vname, [])
437                        if len(links) == 2:
438                            # This link needs to be a lan instead
439                            self.nodes[vname] = "lan"
440                            self.lans[vname] = \
441                                    [ l for l in links, self.lan['vnode'] ]
442                            del self.links[vname]
443                            self.lan = {}
444                            return
445                        lans = self.lans.get(vname, [])
446                        if len(lans) > 0:
447                            lans.append(self.lan['vnode'])
448                            self.lan = {}
449                            return
450                        if not vname in self.links:
451                            self.links[vname] = []
452                        self.links[vname].append(self.lan['vnode'])
453                        self.lan = {}
454                        return
455
456            def found_chars(self, data):
457                """
458                Capture marked up chars for later
459                """
460                self.chars = data
461
462        neato = "/usr/local/bin/neato"
463        # These are used to parse neato output and to create the visualization
464        # file.
465        vis_re = re.compile('^\s*"?([\w\-]+)"?\s+\[.*pos="(\d+),(\d+)"')
466        vis_fmt = "<node><name>%s</name><x>%s</x><y>%s</y><type>" + \
467                "%s</type></node>" 
468        try:
469            df, dotname = tempfile.mkstemp()
470            dotfile = os.fdopen(df, 'w')
471            infile = open(topo_file, 'r')
472            out = open(viz_file, "w")
473        except IOError:
474            raise service_error(service_error.internal,
475                    "Failed to open file in genviz")
476
477        # Parse the topology file using XML tools
478        tp = topo_parse();
479        parser = xml.parsers.expat.ParserCreate()
480        parser.StartElementHandler = tp.start_element
481        parser.EndElementHandler = tp.end_element
482        parser.CharacterDataHandler = tp.found_chars
483
484        parser.ParseFile(infile)
485
486        # Generate a dot/neato input file from the links, nodes and lans
487        try:
488            print >>dotfile, "graph G {"
489            for n in tp.nodes.keys():
490                print >>dotfile, '\t"%s"' % n
491            for l in tp.links.keys():
492                print >>dotfile, " -- ".join(tp.links[l])
493            for l in tp.lans.keys():
494                for n in tp.lans[l]:
495                    print >>dotfile, '\t "%s" -- "%s"' % (n,l)
496            print >>dotfile, "}"
497            dotfile.close()
498        except IOError:
499            raise service_error(service_error.internal, "Cannot write dot file")
500
501        # Use dot to create a visualization
502        dot = Popen([neato, '-Gstart=rand', '-Gepsilon=0.005', '-Gmaxiter=2000',
503                '-Gpack=true', dotname], stdout=PIPE)
504
505        # Translate dot to emulab format
506        try:
507            print >>out, "<vis>"
508            for line in dot.stdout:
509                m = vis_re.match(line)
510                if m:
511                    n, x, y = (m.group(1), m.group(2), m.group(3))
512                    if tp.nodes.has_key(n):
513                        print >>out, vis_fmt % (n, x, y, tp.nodes[n])
514            print >>out, "</vis>"
515            out.close()
516        except IOError:
517            raise service_error(service_error.internal,
518                    "Failed to write visualization file")
519        rv = dot.wait()
520        os.remove(dotname)
521        return rv == 0
522
523
524    def get_access(self, tb, nodes, user, tbparam):
525        """
526        Get access to testbed through fedd and set the parameters for that tb
527        """
528
529        translate_attr = {
530            'slavenodestartcmd': 'expstart',
531            'slaveconnectorstartcmd': 'gwstart',
532            'masternodestartcmd': 'mexpstart',
533            'masterconnectorstartcmd': 'mgwstart',
534            'connectorimage': 'gwimage',
535            'connectortype': 'gwtype',
536            'tunnelcfg': 'tun',
537        }
538
539        # XXX multi-level access
540        uri = self.tbmap.get(tb, None)
541        if not uri:
542            raise service_error(serice_error.server_config, 
543                    "Unknown testbed: %s" % tb)
544
545        # The basic request
546        req = {\
547                'destinationTestbed' : { 'uri' : uri },
548                'user':  user,
549                'allocID' : { 'username': 'test' },
550                'access' : [ { 'sshPubkey' : self.ssh_pubkey } ]
551            }
552       
553        # node resources if any
554        if nodes != None and len(nodes) > 0:
555            rnodes = [ ]
556            for n in nodes:
557                rn = { }
558                image, hw, count = n.split(":")
559                if image: rn['image'] = [ image ]
560                if hw: rn['hardware'] = [ hw ]
561                if count: rn['count'] = int(count)
562                rnodes.append(rn)
563            req['resources']= { }
564            req['resources']['node'] = rnodes
565
566        # No retry loop here.  Proxy servers must correctly authenticate
567        # themselves without help
568        try:
569            ctx = fedd_ssl_context(self.cert_file, 
570                    self.trusted_certs, password=self.cert_pwd)
571        except SSL.SSLError:
572            raise service_error(service_error.server_config, 
573                    "Server certificates misconfigured")
574
575        loc = feddServiceLocator();
576        port = loc.getfeddPortType(uri,
577                transport=M2Crypto.httpslib.HTTPSConnection, 
578                transdict={ 'ssl_context' : ctx })
579
580        # Reconstruct the full request message
581        msg = RequestAccessRequestMessage()
582        msg.set_element_RequestAccessRequestBody(
583                pack_soap(msg, "RequestAccessRequestBody", req))
584        try:
585            resp = port.RequestAccess(msg)
586        except ZSI.ParseException, e:
587            raise service_error(service_error.req,
588                    "Bad format message (XMLRPC??): %s" %
589                    str(e))
590        r = unpack_soap(resp)
591
592        if r.has_key('RequestAccessResponseBody'):
593            r = r['RequestAccessResponseBody']
594        else:
595            raise service_error(service_error.proxy,
596                    "Bad proxy response")
597
598
599        e = r['emulab']
600        p = e['project']
601        tbparam[tb] = { 
602                "boss": e['boss'],
603                "host": e['ops'],
604                "domain": e['domain'],
605                "fs": e['fileServer'],
606                "eventserver": e['eventServer'],
607                "project": unpack_id(p['name'])
608                }
609
610        for u in p['user']:
611            tbparam[tb]['user'] = unpack_id(u['userID'])
612
613        for a in e['fedAttr']:
614            if a['attribute']:
615                key = translate_attr.get(a['attribute'].lower(), None)
616                if key:
617                    tbparam[tb][key]= a['value']
618       
619    class current_testbed:
620        def __init__(self, eid, tmpdir):
621            self.begin_testbed = re.compile("^#\s+Begin\s+Testbed\s+\((\w+)\)")
622            self.end_testbed = re.compile("^#\s+End\s+Testbed\s+\((\w+)\)")
623            self.current_testbed = None
624            self.testbed_file = None
625
626            self.def_expstart = \
627                    "sudo -H /bin/sh FEDDIR/fed_bootstrap >& /tmp/federate";
628            self.def_mexpstart = "sudo -H FEDDIR/make_hosts FEDDIR/hosts";
629            self.def_gwstart = \
630                    "sudo -H FEDDIR/fed-tun.pl -f GWCONF>& /tmp/bridge.log";
631            self.def_mgwstart = \
632                    "sudo -H FEDDIR/fed-tun.pl -f GWCONF >& /tmp/bridge.log";
633            self.def_gwimage = "FBSD61-TUNNEL2";
634            self.def_gwtype = "pc";
635
636            self.eid = eid
637            self.tmpdir = tmpdir
638
639        def __call__(self, line, master, allocated, tbparams):
640            # Capture testbed topology descriptions
641            if self.current_testbed == None:
642                m = self.begin_testbed.match(line)
643                if m != None:
644                    self.current_testbed = m.group(1)
645                    if self.current_testbed == None:
646                        raise service_error(service_error.req,
647                                "Bad request format (unnamed testbed)")
648                    allocated[self.current_testbed] = \
649                            allocated.get(self.current_testbed,0) + 1
650                    tb_dir = "%s/%s" % (self.tmpdir, self.current_testbed)
651                    if not os.path.exists(tb_dir):
652                        try:
653                            os.mkdir(tb_dir)
654                        except IOError:
655                            raise service_error(service_error.internal,
656                                    "Cannot create %s" % tb_dir)
657                    try:
658                        self.testbed_file = open("%s/%s.%s.tcl" %
659                                (tb_dir, self.eid, self.current_testbed), 'w')
660                    except IOError:
661                        self.testbed_file = None
662                    return True
663                else: return False
664            else:
665                m = self.end_testbed.match(line)
666                if m != None:
667                    if m.group(1) != self.current_testbed:
668                        raise service_error(service_error.internal, 
669                                "Mismatched testbed markers!?")
670                    if self.testbed_file != None: 
671                        self.testbed_file.close()
672                        self.testbed_file = None
673                    self.current_testbed = None
674                elif self.testbed_file:
675                    # Substitute variables and put the line into the local
676                    # testbed file.
677                    gwtype = tbparams[self.current_testbed].get('gwtype', 
678                            self.def_gwtype)
679                    gwimage = tbparams[self.current_testbed].get('gwimage', 
680                            self.def_gwimage)
681                    mgwstart = tbparams[self.current_testbed].get('mgwstart', 
682                            self.def_mgwstart)
683                    mexpstart = tbparams[self.current_testbed].get('mexpstart', 
684                            self.def_mexpstart)
685                    gwstart = tbparams[self.current_testbed].get('gwstart', 
686                            self.def_gwstart)
687                    expstart = tbparams[self.current_testbed].get('expstart', 
688                            self.def_expstart)
689                    project = tbparams[self.current_testbed].get('project')
690                    line = re.sub("GWTYPE", gwtype, line)
691                    line = re.sub("GWIMAGE", gwimage, line)
692                    if self.current_testbed == master:
693                        line = re.sub("GWSTART", mgwstart, line)
694                        line = re.sub("EXPSTART", mexpstart, line)
695                    else:
696                        line = re.sub("GWSTART", gwstart, line)
697                        line = re.sub("EXPSTART", expstart, line)
698                    # XXX: does `` embed without doing enything else?
699                    line = re.sub("GWCONF", "FEDDIR`hostname`.gw.conf", line)
700                    line = re.sub("PROJDIR", "/proj/%s/" % project, line)
701                    line = re.sub("EID", self.eid, line)
702                    line = re.sub("FEDDIR", "/proj/%s/exp/%s/tmp/" % \
703                            (project, self.eid), line)
704                    print >>self.testbed_file, line
705                return True
706
707    class allbeds:
708        def __init__(self, get_access):
709            self.begin_allbeds = re.compile("^#\s+Begin\s+Allbeds")
710            self.end_allbeds = re.compile("^#\s+End\s+Allbeds")
711            self.in_allbeds = False
712            self.get_access = get_access
713
714        def __call__(self, line, user, tbparams):
715            # Testbed access parameters
716            if not self.in_allbeds:
717                if self.begin_allbeds.match(line):
718                    self.in_allbeds = True
719                    return True
720                else:
721                    return False
722            else:
723                if self.end_allbeds.match(line):
724                    self.in_allbeds = False
725                else:
726                    nodes = line.split('|')
727                    tb = nodes.pop(0)
728                    self.get_access(tb, nodes, user, tbparams)
729                return True
730
731    class gateways:
732        def __init__(self, eid, smbshare, master, tmpdir, gw_pubkey,
733                gw_secretkey, copy_file):
734            self.begin_gateways = \
735                    re.compile("^#\s+Begin\s+gateways\s+\((\w+)\)")
736            self.end_gateways = re.compile("^#\s+End\s+gateways\s+\((\w+)\)")
737            self.current_gateways = None
738            self.control_gateway = None
739            self.active_end = { }
740
741            self.eid = eid
742            self.smbshare = smbshare
743            self.master = master
744            self.tmpdir = tmpdir
745            self.gw_pubkey_base = gw_pubkey
746            self.gw_secretkey_base = gw_secretkey
747
748            self.copy_file = copy_file
749
750
751        def gateway_conf_file(self, gw, master, eid, pubkey, privkey,
752                active_end, tbparams, dtb, myname, desthost, type):
753            """
754            Produce a gateway configuration file from a gateways line.
755            """
756
757            sproject = tbparams[gw].get('project', 'project')
758            dproject = tbparams[dtb].get('project', 'project')
759            sdomain = ".%s.%s%s" % (eid, sproject,
760                    tbparams[gw].get('domain', ".example.com"))
761            ddomain = ".%s.%s%s" % (eid, dproject,
762                    tbparams[dtb].get('domain', ".example.com"))
763            boss = tbparams[master].get('boss', "boss")
764            fs = tbparams[master].get('fs', "fs")
765            event_server = "%s%s" % \
766                    (tbparams[master].get('eventserver', "event_server"),
767                            tbparams[master].get('domain', "example.com"))
768            remote_event_server = "%s%s" % \
769                    (tbparams[dtb].get('eventserver', "event_server"),
770                            tbparams[dtb].get('domain', "example.com"))
771
772            remote_script_dir = "/proj/%s/exp/%s/tmp" % ( dproject, eid)
773            local_script_dir = "/proj/%s/exp/%s/tmp" % ( sproject, eid)
774            tunnel_cfg = tbparams[gw].get("tun", "false")
775
776            conf_file = "%s%s.gw.conf" % (myname, sdomain)
777            remote_conf_file = "%s%s.gw.conf" % (desthost, ddomain)
778
779            # translate to lower case so the `hostname` hack for specifying
780            # configuration files works.
781            conf_file = conf_file.lower();
782            remote_conf_file = remote_conf_file.lower();
783
784            if dtb == master:
785                active = "false"
786            elif gw == master:
787                active = "true"
788            elif active_end.has_key['%s-%s' % (dtb, gw)]:
789                active = "false"
790            else:
791                active_end['%s-%s' % (gw, dtb)] = 1
792                active = "true"
793
794            gwconfig = open("%s/%s/%s" % (self.tmpdir, gw, conf_file), "w")
795            print >>gwconfig, "Active: %s" % active
796            print >>gwconfig, "TunnelCfg: %s" % tunnel_cfg
797            print >>gwconfig, "BossName: %s" % boss
798            print >>gwconfig, "FsName: %s" % fs
799            print >>gwconfig, "EventServerName: %s" % event_server
800            print >>gwconfig, "RemoteEventServerName: %s" % remote_event_server
801            print >>gwconfig, "Type: %s" % type
802            print >>gwconfig, "RemoteScriptDir: %s" % remote_script_dir
803            print >>gwconfig, "EventRepeater: %s/fed_evrepeater" % \
804                    local_script_dir
805            print >>gwconfig, "RemoteExperiment: %s/%s" % (dproject, eid)
806            print >>gwconfig, "LocalExperiment: %s/%s" % (sproject, eid)
807            print >>gwconfig, "RemoteConfigFile: %s/%s" % \
808                    (remote_script_dir, remote_conf_file)
809            print >>gwconfig, "Peer: %s%s" % (desthost, ddomain)
810            print >>gwconfig, "Pubkeys: %s/%s" % (local_script_dir, pubkey)
811            print >>gwconfig, "Privkeys: %s/%s" % (local_script_dir, privkey)
812            gwconfig.close()
813
814            return active == "true"
815
816        def __call__(self, line, allocated, tbparams):
817            # Process gateways
818            if not self.current_gateways:
819                m = self.begin_gateways.match(line)
820                if m:
821                    self.current_gateways = m.group(1)
822                    if allocated.has_key(self.current_gateways):
823                        # This test should always succeed
824                        tb_dir = "%s/%s" % (self.tmpdir, self.current_gateways)
825                        if not os.path.exists(tb_dir):
826                            try:
827                                os.mkdir(tb_dir)
828                            except IOError:
829                                raise service_error(service_error.internal,
830                                        "Cannot create %s" % tb_dir)
831                    else:
832                        # XXX
833                        print >>sys.stderr, \
834                            "Ignoring gateways for unknown testbed %s" \
835                                    % self.current_gateways
836                        self.current_gateways = None
837                    return True
838                else:
839                    return False
840            else:
841                m = self.end_gateways.match(line)
842                if m :
843                    if m.group(1) != self.current_gateways:
844                        raise service_error(service_error.internal,
845                                "Mismatched gateway markers!?")
846                    if self.control_gateway:
847                        try:
848                            cc = open("%s/%s/client.conf" %
849                                    (self.tmpdir, self.current_gateways), 'w')
850                            print >>cc, "ControlGateway: %s" % \
851                                    self.control_gateway
852                            print >>cc, "SMBSHare: %s" % self.smbshare
853                            print >>cc, "ProjectUser: %s" % \
854                                    tbparams[self.current_gateways]['user']
855                            print >>cc, "ProjectName: %s" % \
856                                    tbparams[self.master]['project']
857                            cc.close()
858                        except IOError:
859                            raise service_error(service_error.internal,
860                                    "Error creating client config")
861                    else:
862                        # XXX
863                        print >>sys.stderr, "No control gateway for %s" %\
864                                self.current_gateways
865                    self.current_gateways = None
866                else:
867                    dtb, myname, desthost, type = line.split(" ")
868
869                    if type == "control" or type == "both":
870                        self.control_gateway = "%s.%s.%s%s" % (myname, 
871                                self.eid, 
872                                tbparams[self.current_gateways]['project'],
873                                tbparams[self.current_gateways]['domain'])
874                    try:
875                        active = self.gateway_conf_file(self.current_gateways,
876                                self.master, self.eid, self.gw_pubkey_base,
877                                self.gw_secretkey_base,
878                                self.active_end, tbparams, dtb, myname,
879                                desthost, type)
880                    except IOError, e:
881                        raise service_error(service_error.internal,
882                                "Failed to write config file for %s" % \
883                                        self.current_gateway)
884           
885                    gw_pubkey = "%s/keys/%s" % \
886                            (self.tmpdir, self.gw_pubkey_base)
887                    gw_secretkey = "%s/keys/%s" % \
888                            (self.tmpdir, self.gw_secretkey_base)
889
890                    pkfile = "%s/%s/%s" % \
891                            ( self.tmpdir, self.current_gateways, 
892                                    self.gw_pubkey_base)
893                    skfile = "%s/%s/%s" % \
894                            ( self.tmpdir, self.current_gateways, 
895                                    self.gw_secretkey_base)
896
897                    if not os.path.exists(pkfile):
898                        try:
899                            self.copy_file(gw_pubkey, pkfile)
900                        except IOError:
901                            service_error(service_error.internal,
902                                    "Failed to copy pubkey file")
903
904                    if active and not os.path.exists(skfile):
905                        try:
906                            self.copy_file(gw_secretkey, skfile)
907                        except IOError:
908                            service_error(service_error.internal,
909                                    "Failed to copy secretkey file")
910                return True
911
912    class shunt_to_file:
913        def __init__(self, begin, end, filename):
914            self.begin = re.compile(begin)
915            self.end = re.compile(end)
916            self.in_shunt = False
917            self.file = None
918            self.filename = filename
919
920        def __call__(self, line):
921            if not self.in_shunt:
922                if self.begin.match(line):
923                    self.in_shunt = True
924                    try:
925                        self.file = open(self.filename, "w")
926                    except:
927                        self.file = None
928                        raise
929                    return True
930                else:
931                    return False
932            else:
933                if self.end.match(line):
934                    if self.file: 
935                        self.file.close()
936                        self.file = None
937                    self.in_shunt = False
938                else:
939                    if self.file:
940                        print >>self.file, line
941                return True
942
943    class shunt_to_list:
944        def __init__(self, begin, end):
945            self.begin = re.compile(begin)
946            self.end = re.compile(end)
947            self.in_shunt = False
948            self.list = [ ]
949       
950        def __call__(self, line):
951            if not self.in_shunt:
952                if self.begin.match(line):
953                    self.in_shunt = True
954                    return True
955                else:
956                    return False
957            else:
958                if self.end.match(line):
959                    self.in_shunt = False
960                else:
961                    self.list.append(line)
962                return True
963
964    def create_experiment(self, req, fid):
965        try:
966            tmpdir = tempfile.mkdtemp(prefix="split-")
967        except IOError:
968            raise service_error(service_error.internal, "Cannot create tmp dir")
969
970        gw_pubkey_base = "fed.%s.pub" % self.ssh_type
971        gw_secretkey_base = "fed.%s" % self.ssh_type
972        gw_pubkey = tmpdir + "/keys/" + gw_pubkey_base
973        gw_secretkey = tmpdir + "/keys/" + gw_secretkey_base
974        tclfile = tmpdir + "/experiment.tcl"
975        tbparams = { }
976
977        pid = "dummy"
978        gid = "dummy"
979        # XXX
980        eid = "faber-splitter"
981        # XXX
982        master = "deter"
983        # XXX
984        smbshare="USERS"
985        # XXX
986        fail_soft = False
987        # XXX
988        startem = True
989
990
991        try:
992            os.mkdir(tmpdir+"/keys")
993        except OSError:
994            raise service_error(service_error.internal,
995                    "Can't make temporary dir")
996
997        # The tcl parser needs to read a file so put the content into that file
998        file_content=req.get('experimentdescription', None)
999        if file_content != None:
1000            try:
1001                f = open(tclfile, 'w')
1002                f.write(file_content)
1003                f.close()
1004            except IOError:
1005                raise service_error(service_error.internal,
1006                        "Cannot write temp experiment description")
1007        else:
1008            raise service_error(service_error.req, "No experiment description")
1009
1010        try:
1011            self.generate_ssh_keys(gw_secretkey, self.ssh_type)
1012        except ValueError:
1013            raise service_error(service_error.server_config, 
1014                    "Bad key type (%s)" % self.ssh_type)
1015
1016        user = req.get('user', None)
1017        if user == None:
1018            raise service_error(service_error.req, "No user")
1019       
1020       
1021        tclcmd = [self.tclsh, self.tcl_splitter, '-s', '-x', 
1022            str(self.muxmax), '-m', master, pid, gid, eid, tclfile]
1023        tclparser = Popen(tclcmd, stdout=PIPE)
1024
1025        allocated = { }
1026        started = { }
1027
1028        parse_current_testbed = self.current_testbed(eid, tmpdir)
1029        parse_allbeds = self.allbeds(self.get_access)
1030        parse_gateways = self.gateways(eid, smbshare, master, tmpdir,
1031                gw_pubkey_base, gw_secretkey_base, self.copy_file)
1032        parse_vtopo = self.shunt_to_file("^#\s+Begin\s+Vtopo",
1033                    "^#\s+End\s+Vtopo", tmpdir + "/vtopo.xml")
1034        parse_hostnames = self.shunt_to_file("^#\s+Begin\s+hostnames",
1035                    "^#\s+End\s+hostnames", tmpdir + "/hosts")
1036        parse_tarfiles = self.shunt_to_list("^#\s+Begin\s+tarfiles",
1037                "^#\s+End\s+tarfiles")
1038        parse_rpms = self.shunt_to_list("^#\s+Begin\s+rpms",
1039                "^#\s+End\s+rpms")
1040
1041        for line in tclparser.stdout:
1042            line = line.rstrip()
1043            if parse_current_testbed(line, master, allocated, tbparams):
1044                continue
1045            elif parse_allbeds(line, user, tbparams):
1046                continue
1047            elif parse_gateways(line, allocated, tbparams):
1048                continue
1049            elif parse_vtopo(line):
1050                continue
1051            elif parse_hostnames(line):
1052                continue
1053            elif parse_tarfiles(line):
1054                continue
1055            elif parse_rpms(line):
1056                continue
1057            else:
1058                raise service_error(service_error.internal, 
1059                        "Bad tcl parse? %s" % line)
1060
1061        self.genviz(tmpdir + "/vtopo.xml", tmpdir + "/viz.xml")
1062        if not startem: return True
1063
1064        # Copy tarfiles and rpms needed at remote sites into a staging area
1065        try:
1066            for t in parse_tarfiles.list:
1067                if not os.path.exists("%s/tarfiles" % tmpdir):
1068                    os.mkdir("%s/tarfiles" % tmpdir)
1069                self.copy_file(t, "%s/tarfiles/%s" % \
1070                        (tmpdir, os.path.basename(t)))
1071            for r in parse_rpms.list:
1072                if not os.path.exists("%s/rpms" % tmpdir):
1073                    os.mkdir("%s/rpms" % tmpdir)
1074                self.copy_file(r, "%s/rpms/%s" % \
1075                        (tmpdir, os.path.basename(r)))
1076        except IOError, e:
1077            raise service_error(service_error.internal, 
1078                    "Cannot stage tarfile/rpm: %s" % e.strerror)
1079
1080        thread_pool_info = self.thread_pool()
1081        threads = [ ]
1082
1083        for tb in [ k for k in allocated.keys() if k != master]:
1084            # Wait until we have a free slot to start the next testbed load
1085            thread_pool_info.acquire()
1086            while thread_pool_info.started - \
1087                    thread_pool_info.terminated >= self.nthreads:
1088                thread_pool_info.wait()
1089            thread_pool_info.release()
1090
1091            # Create and start a thread to start the segment, and save it to
1092            # get the return value later
1093            t  = self.pooled_thread(target=self.start_segment, 
1094                    args=(tb, eid, tbparams, tmpdir, 0), name=tb,
1095                    pdata=thread_pool_info)
1096            threads.append(t)
1097            t.start()
1098
1099        # Wait until all finish (the first clause of the while is to make sure
1100        # one starts)
1101        thread_pool_info.acquire()
1102        while thread_pool_info.started == 0 or \
1103                thread_pool_info.started > thread_pool_info.terminated:
1104            thread_pool_info.wait()
1105        thread_pool_info.release()
1106
1107        # If none failed, start the master
1108        failed = [ t.getName() for t in threads if not t.rv ]
1109
1110        if len(failed) == 0:
1111            if not self.start_segment(master, eid, tbparams, tmpdir):
1112                failed.append(master)
1113
1114        # If one failed clean up
1115        if not fail_soft and len(failed) > 0:
1116            for tb in failed:
1117                self.stop_segment(tb, eid, tbparams)
1118        else:
1119            print "Experiment started"
1120
1121        # XXX: return value
1122
1123if __name__ == '__main__':
1124    from optparse import OptionParser
1125   
1126    parser = OptionParser()
1127    parser.add_option('-d', '--debug', dest='debug', default=False,
1128            action='store_true', help='print actions rather than take them')
1129    parser.add_option('-f', '--file', dest='tcl', help='tcl file to parse')
1130    opts, args  = parser.parse_args()
1131
1132    if opts.tcl != None:
1133        try:
1134            f = open(opts.tcl, 'r')
1135            content = ''.join(f)
1136            f.close()
1137        except IOError, e:
1138            sys.exit("Can't read %s: %s" % (opts.tcl, e))
1139    else:
1140        sys.exit("Must specify a file name")
1141
1142    obj = fedd_create_experiment_local(
1143            debug=opts.debug,
1144            scripts_dir="/users/faber/testbed/federation",
1145            cert_file="./fedd_client.pem", cert_pwd="faber", 
1146            ssh_pubkey_file='/users/faber/.ssh/id_rsa.pub',
1147            trusted_certs="./cacert.pem",
1148            tbmap = { 
1149                'deter':'https://users.isi.deterlab.net:23235',
1150                'emulab':'https://users.isi.deterlab.net:23236',
1151                'ucb':'https://users.isi.deterlab.net:23237',
1152                },
1153        ) 
1154    obj.create_experiment( {\
1155            'experimentdescription' : content, 
1156            'user': [ {'userID' : { 'username' : 'faber' } } ],
1157            },
1158            None)
Note: See TracBrowser for help on using the repository browser.