source: fedd/fedd_experiment_control.py @ 8ecfbad

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

beginnings of a real multithreaded server

  • Property mode set to 100644
File size: 48.4 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
18import pickle
19
20import traceback
21
22from threading import *
23
24from subprocess import *
25
26from fedd_services import *
27from fedd_internal_services import *
28from fedd_util import *
29import parse_detail
30from service_error import *
31
32class fedd_experiment_control_local:
33    scripts = ["fed_bootstrap", "federate.sh", "smbmount.FreeBSD.pl",
34        "smbmount.Linux.pl", "make_hosts", "fed-tun.pl", "fed-tun.ucb.pl",
35        "fed_evrepeater", "rc.accounts.patch"]
36   
37    class thread_pool:
38        def __init__(self):
39            self.changed = Condition()
40            self.started = 0
41            self.terminated = 0
42
43        def acquire(self):
44            self.changed.acquire()
45
46        def release(self):
47            self.changed.release()
48
49        def wait(self, timeout = None):
50            self.changed.wait(timeout)
51
52        def start(self):
53            self.changed.acquire()
54            self.started += 1
55            self.changed.notifyAll()
56            self.changed.release()
57
58        def terminate(self):
59            self.changed.acquire()
60            self.terminated += 1
61            self.changed.notifyAll()
62            self.changed.release()
63
64        def clear(self):
65            self.changed.acquire()
66            self.started = 0
67            self.terminated =0
68            self.changed.notifyAll()
69            self.changed.release()
70
71    class pooled_thread(Thread):
72        def __init__(self, group=None, target=None, name=None, args=(), 
73                kwargs={}, pdata=None, trace_file=None):
74            Thread.__init__(self, group, target, name, args, kwargs)
75            self.rv = None
76            self.exception = None
77            self.target=target
78            self.args = args
79            self.kwargs = kwargs
80            self.pdata = pdata
81            self.trace_file = trace_file
82       
83        def run(self):
84            if self.pdata:
85                self.pdata.start()
86
87            if self.target:
88                try:
89                    self.rv = self.target(*self.args, **self.kwargs)
90                except service_error, s:
91                    self.exception = s
92                    if self.trace_file:
93                        print >>self.trace_file, "Thread exception: %s %s" % \
94                                (s.code_string(), s.desc)
95                   
96                except:
97                    self.exception = sys.exc_info()[1]
98                    if self.trace_file:
99                        print >>self.trace_file, \
100                                "Unexpected thread exception: %s" % \
101                                self.exception
102                        print >>self.trace_file, "Trace: %s" % \
103                                traceback.format_exc()
104            if self.pdata:
105                self.pdata.terminate()
106
107    def __init__(self, config=None):
108        self.scripts = fedd_experiment_control_local.scripts
109        self.thread_with_rv = fedd_experiment_control_local.pooled_thread
110        self.thread_pool = fedd_experiment_control_local.thread_pool
111
112        self.cert_file = None
113        self.cert_pwd = None
114        self.trusted_certs = None
115
116        # Walk through the various relevant certificat specifying config
117        # attributes until the local certificate attributes can be resolved.
118        # The walk is from omst specific to most general specification.
119        for p in ("create_experiment_", "proxy_", ""):
120            filen = "%scert_file" % p
121            pwn = "%scert_pwd" % p
122            trustn = "%strusted_certs" % p
123
124            if getattr(config, filen, None):
125                if not self.cert_file:
126                    self.cert_file = getattr(config, filen, None)
127                    self.cert_pwd = getattr(config, pwn, None)
128
129            if getattr(config, trustn, None):
130                if not self.trusted_certs:
131                    self.trusted_certs = getattr(config, trustn, None)
132
133        self.exp_stem = "fed-stem"
134        self.debug = config.create_debug
135        self.muxmax = 2
136        self.nthreads = 2
137        self.randomize_experiments = False
138        self.scp_exec = "/usr/bin/scp"
139        self.scripts_dir = "/users/faber/testbed/federation"
140        self.splitter = None
141        self.ssh_exec="/usr/bin/ssh"
142        self.ssh_keygen = "/usr/bin/ssh-keygen"
143        self.ssh_identity_file = None
144        # XXX
145        self.ssh_pubkey_file = "/users/faber/.ssh/id_rsa.pub"
146        self.ssh_type = "rsa"
147        self.state = { }
148        self.state_filename = config.experiment_state_file
149        self.state_lock = Lock()
150        self.tclsh = "/usr/local/bin/otclsh"
151        self.tcl_splitter = "/usr/testbed/lib/ns2ir/parse.tcl"
152        self.tbmap = { 
153                'deter':'https://users.isi.deterlab.net:23235',
154                'emulab':'https://users.isi.deterlab.net:23236',
155                'ucb':'https://users.isi.deterlab.net:23237',
156                }
157        self.trace_file = sys.stderr
158
159        self.def_expstart = \
160                "sudo -H /bin/sh FEDDIR/fed_bootstrap >& /tmp/federate";
161        self.def_mexpstart = "sudo -H FEDDIR/make_hosts FEDDIR/hosts";
162        self.def_gwstart = \
163                "sudo -H FEDDIR/fed-tun.pl -f GWCONF>& /tmp/bridge.log";
164        self.def_mgwstart = \
165                "sudo -H FEDDIR/fed-tun.pl -f GWCONF >& /tmp/bridge.log";
166        self.def_gwimage = "FBSD61-TUNNEL2";
167        self.def_gwtype = "pc";
168
169
170        if self.ssh_pubkey_file:
171            try:
172                f = open(self.ssh_pubkey_file, 'r')
173                self.ssh_pubkey = f.read()
174                f.close()
175            except IOError:
176                raise service_error(service_error.internal,
177                        "Cannot read sshpubkey")
178
179        # Grab saved state
180        if self.state_filename:
181            self.read_state()
182
183        # Confirm federation scripts in the right place
184        for s in self.scripts:
185            if not os.path.exists(self.scripts_dir + "/" + s):
186                raise service_error(service_error.server_config,
187                        "%s/%s not in local script dir" % (self.scripts_dir, s))
188
189        self.soap_handlers = {\
190                'Create': make_soap_handler(\
191                        CreateRequestMessage.typecode,
192                        getattr(self, "create_experiment"), 
193                        CreateResponseMessage,
194                        "CreateResponseBody"),
195                'Vtopo': make_soap_handler(\
196                        VtopoRequestMessage.typecode,
197                        getattr(self, "get_vtopo"),
198                        VtopoResponseMessage,
199                        "VtopoResponseBody"),
200                'Vis': make_soap_handler(\
201                        VisRequestMessage.typecode,
202                        getattr(self, "get_vis"),
203                        VisResponseMessage,
204                        "VisResponseBody"),
205                'Info': make_soap_handler(\
206                        InfoRequestMessage.typecode,
207                        getattr(self, "get_info"),
208                        InfoResponseMessage,
209                        "InfoResponseBody"),
210                'Terminate': make_soap_handler(\
211                        TerminateRequestMessage.typecode,
212                        getattr(self, "terminate_experiment"),
213                        TerminateResponseMessage,
214                        "TerminateResponseBody"),
215        }
216
217        self.xmlrpc_handlers = {\
218                'Create': make_xmlrpc_handler(\
219                        getattr(self, "create_experiment"), 
220                        "CreateResponseBody"),
221                'Vtopo': make_xmlrpc_handler(\
222                        getattr(self, "get_vtopo"),
223                        "VtopoResponseBody"),
224                'Vis': make_xmlrpc_handler(\
225                        getattr(self, "get_vis"),
226                        "VisResponseBody"),
227                'Info': make_xmlrpc_handler(\
228                        getattr(self, "get_info"),
229                        "InfoResponseBody"),
230                'Terminate': make_xmlrpc_handler(\
231                        getattr(self, "terminate_experiment"),
232                        "TerminateResponseBody"),
233        }
234
235    def get_soap_services(self):
236        return self.soap_handlers
237
238    def get_xmlrpc_services(self):
239        return self.xmlrpc_handlers
240
241    def copy_file(self, src, dest, size=1024):
242        """
243        Exceedingly simple file copy.
244        """
245        s = open(src,'r')
246        d = open(dest, 'w')
247
248        buf = "x"
249        while buf != "":
250            buf = s.read(size)
251            d.write(buf)
252        s.close()
253        d.close()
254
255    # Call while holding self.state_lock
256    def write_state(self):
257        if os.access(self.state_filename, os.W_OK):
258            self.copy_file(self.state_filename, \
259                    "%s.bak" % self.state_filename)
260        try:
261            f = open(self.state_filename, 'w')
262            pickle.dump(self.state, f)
263        except IOError, e:
264            print >>sys.stderr, "Can't write file %s: %s" % \
265                    (self.state_filename, e)
266        except pickle.PicklingError, e:
267            print >>sys.stderr, "Pickling problem: %s" % e
268
269    # Call while holding self.state_lock
270    def read_state(self):
271        try:
272            f = open(self.state_filename, "r")
273            self.state = pickle.load(f)
274        except IOError, e:
275            print >>sys.stderr, "Can't open %s: %s" % \
276                    (self.state_filename, e)
277        except pickle.UnpicklingError, e:
278            print >>sys.stderr, "Unpickling failed: %s" % e
279
280    def scp_file(self, file, user, host, dest=""):
281        """
282        scp a file to the remote host.
283        """
284
285        scp_cmd = [self.scp_exec, file, "%s@%s:%s" % (user, host, dest)]
286
287        trace = self.trace_file
288        if not trace:
289            try:
290                trace = open("/dev/null", "w")
291            except IOError:
292                raise service_error(service_error.internal,
293                        "Cannot open /dev/null??");
294
295        if not self.debug:
296            rv = call(scp_cmd, stdout=trace, stderr=trace)
297        else:
298            if self.trace_file: 
299                print >>self.trace_file, "debug [scp_file]: %s" % \
300                        " ".join(scp_cmd)
301            rv = 0
302
303        return rv == 0
304
305    def ssh_cmd(self, user, host, cmd, wname=None):
306        sh_str = "%s %s@%s %s" % (self.ssh_exec, user, host, cmd)
307
308        trace = self.trace_file
309        if not trace:
310            try:
311                trace = open("/dev/null", "w")
312            except IOError:
313                raise service_error(service_error.internal,
314                        "Cannot open /dev/null??");
315
316        if not self.debug:
317            sub = Popen(sh_str, shell=True, stdout=trace, stderr=trace)
318            return sub.wait() == 0
319        else:
320            if self.trace_file:
321                print >>self.trace_file,"debug [ssh_cmd]: %s" % sh_str
322            return True
323
324    def ship_scripts(self, host, user, dest_dir):
325        if self.ssh_cmd(user, host, "mkdir -p %s" % dest_dir):
326            for s in self.scripts:
327                if not self.scp_file("%s/%s" % (self.scripts_dir, s),
328                        user, host, dest_dir):
329                    return False
330            return True
331        else:
332            return False
333
334    def ship_configs(self, host, user, src_dir, dest_dir):
335        if not self.ssh_cmd(user, host, "mkdir -p %s" % dest_dir):
336            return False
337        if not self.ssh_cmd(user, host, "chmod 770 %s" % dest_dir):
338            return False
339
340        for f in os.listdir(src_dir):
341            if os.path.isdir(f):
342                if not self.ship_configs(host, user, "%s/%s" % (src_dir, f), 
343                        "%s/%s" % (dest_dir, f)):
344                    return False
345            else:
346                if not self.scp_file("%s/%s" % (src_dir, f), 
347                        user, host, dest_dir):
348                    return False
349        return True
350
351    def start_segment(self, tb, eid, tbparams, tmpdir, timeout=0):
352        host = "%s%s" % (tbparams[tb]['host'], tbparams[tb]['domain'])
353        user = tbparams[tb]['user']
354        pid = tbparams[tb]['project']
355        # XXX
356        base_confs = ( "hosts",)
357        tclfile = "%s.%s.tcl" % (eid, tb)
358        expinfo_exec = "/usr/testbed/bin/expinfo"
359        proj_dir = "/proj/%s/exp/%s/tmp" % (pid, eid)
360        tarfiles_dir = "/proj/%s/tarfiles/%s" % (pid, eid)
361        rpms_dir = "/proj/%s/rpms/%s" % (pid, eid)
362        state_re = re.compile("State:\s+(\w+)")
363        no_exp_re = re.compile("^No\s+such\s+experiment")
364        state = None
365        cmd = [self.ssh_exec, "%s@%s" % (user, host), expinfo_exec, pid, eid]
366
367
368        if self.trace_file:
369            print >>self.trace_file, "status request: %s" % " ".join(cmd)
370       
371        if not self.trace_file:
372            try:
373                st_file = open("/dev/null", "w")
374            except IOError:
375                raise service_error(service_error.internal, 
376                        "Cannot open /dev/null!?")
377        else:
378            st_file = self.trace_file
379
380        status = Popen(cmd, stdout=PIPE, stderr=st_file)
381        for line in status.stdout:
382            m = state_re.match(line)
383            if m: state = m.group(1)
384            else:
385                m = no_exp_re.match(line)
386                if m: state = "none"
387        rv = status.wait()
388        # If the experiment is not present the subcommand returns a non-zero
389        # return value.  If we successfully parsed a "none" outcome, ignore the
390        # return code.
391        if rv != 0 and state != "none":
392            raise service_error(service_error.internal,
393                    "Cannot get status of segment %s:%s/%s" % (tb, pid, eid))
394        if self.trace_file:
395            print >>self.trace_file, "%s: %s" % (tb, state)
396            print >>self.trace_file, "transferring experiment to %s" % tb
397
398        if not self.scp_file("%s/%s/%s" % (tmpdir, tb, tclfile), user, host):
399            return False
400        # Clear the federation files
401        if not self.ssh_cmd(user, host, 
402                "/bin/sh -c \"'/bin/rm -rf %s'\"" % proj_dir):
403            return False
404        if not self.ssh_cmd(user, host, 
405                "/bin/sh -c \"'mkdir -p %s'\"" % proj_dir):
406            return False
407        # Clear and create the tarfiles and rpm directories
408        for d in (tarfiles_dir, rpms_dir):
409            if not self.ssh_cmd(user, host, 
410                    "/bin/sh -c \"'/bin/rm -rf %s/*'\"" % d):
411                return False
412            if not self.ssh_cmd(user, host, "mkdir -p %s" % d, 
413                    "create tarfiles"):
414                return False
415       
416        if state == 'active':
417            # Remote experiment is active.  Modify it.
418            for f in base_confs:
419                if not self.scp_file("%s/%s" % (tmpdir, f), user, host,
420                        "%s/%s" % (proj_dir, f)):
421                    return False
422            if not self.ship_scripts(host, user, proj_dir):
423                return False
424            if not self.ship_configs(host, user, "%s/%s" % (tmpdir, tb),
425                    proj_dir):
426                return False
427            if os.path.isdir("%s/tarfiles" % tmpdir):
428                if not self.ship_configs(host, user,
429                        "%s/tarfiles" % tmpdir, tarfiles_dir):
430                    return False
431            if os.path.isdir("%s/rpms" % tmpdir):
432                if not self.ship_configs(host, user,
433                        "%s/rpms" % tmpdir, tarfiles_dir):
434                    return False
435            if self.trace_file:
436                print >>self.trace_file, "Modifying %s on %s" % (eid, tb)
437            if not self.ssh_cmd(user, host,
438                    "/usr/testbed/bin/modexp -r -s -w %s %s %s" % \
439                            (pid, eid, tclfile), "modexp"):
440                return False
441            return True
442        elif state == "swapped":
443            # Remote experiment swapped out.  Modify it and swap it in.
444            for f in base_confs:
445                if not self.scp_file("%s/%s" % (tmpdir, f), user, host,
446                        "%s/%s" % (proj_dir, f)):
447                    return False
448            if not self.ship_scripts(host, user, proj_dir):
449                return False
450            if not self.ship_configs(host, user, "%s/%s" % (tmpdir, tb),
451                    proj_dir):
452                return False
453            if os.path.isdir("%s/tarfiles" % tmpdir):
454                if not self.ship_configs(host, user,
455                        "%s/tarfiles" % tmpdir, tarfiles_dir):
456                    return False
457            if os.path.isdir("%s/rpms" % tmpdir):
458                if not self.ship_configs(host, user,
459                        "%s/rpms" % tmpdir, tarfiles_dir):
460                    return False
461            if self.trace_file:
462                print >>self.trace_file, "Modifying %s on %s" % (eid, tb)
463            if not self.ssh_cmd(user, host,
464                    "/usr/testbed/bin/modexp -w %s %s %s" % (pid, eid, tclfile),
465                    "modexp"):
466                return False
467            if self.trace_file:
468                print >>self.trace_file, "Swapping %s in on %s" % (eid, tb)
469            if not self.ssh_cmd(user, host,
470                    "/usr/testbed/bin/swapexp -w %s %s in" % (pid, eid),
471                    "swapexp"):
472                return False
473            return True
474        elif state == "none":
475            # No remote experiment.  Create one.  We do this in 2 steps so we
476            # can put the configuration files and scripts into the new
477            # experiment directories.
478
479            # Tarfiles must be present for creation to work
480            if os.path.isdir("%s/tarfiles" % tmpdir):
481                if not self.ship_configs(host, user,
482                        "%s/tarfiles" % tmpdir, tarfiles_dir):
483                    return False
484            if os.path.isdir("%s/rpms" % tmpdir):
485                if not self.ship_configs(host, user,
486                        "%s/rpms" % tmpdir, tarfiles_dir):
487                    return False
488            if self.trace_file:
489                print >>self.trace_file, "Creating %s on %s" % (eid, tb)
490            if not self.ssh_cmd(user, host,
491                    "/usr/testbed/bin/startexp -i -f -w -p %s -e %s %s" % \
492                            (pid, eid, tclfile), "startexp"):
493                return False
494            # After startexp the per-experiment directories exist
495            for f in base_confs:
496                if not self.scp_file("%s/%s" % (tmpdir, f), user, host,
497                        "%s/%s" % (proj_dir, f)):
498                    return False
499            if not self.ship_scripts(host, user, proj_dir):
500                return False
501            if not self.ship_configs(host, user, "%s/%s" % (tmpdir, tb),
502                    proj_dir):
503                return False
504            if self.trace_file:
505                print >>self.trace_file, "Swapping %s in on %s" % (eid, tb)
506            if not self.ssh_cmd(user, host,
507                    "/usr/testbed/bin/swapexp -w %s %s in" % (pid, eid),
508                    "swapexp"):
509                return False
510            return True
511        else:
512            if self.trace_file:
513                print >>self.trace_file, "unknown state %s" % state
514            return False
515
516    def stop_segment(self, tb, eid, tbparams):
517        user = tbparams[tb]['user']
518        host = tbparams[tb]['host']
519        pid = tbparams[tb]['project']
520
521        if self.trace_file:
522            print >>self.trace_file, "Stopping %s on %s" % (eid, tb)
523        return self.ssh_cmd(user, host,
524                "/usr/testbed/bin/swapexp -w %s %s out" % (pid, eid))
525
526       
527    def generate_ssh_keys(self, dest, type="rsa" ):
528        """
529        Generate a set of keys for the gateways to use to talk.
530
531        Keys are of type type and are stored in the required dest file.
532        """
533        valid_types = ("rsa", "dsa")
534        t = type.lower();
535        if t not in valid_types: raise ValueError
536
537        trace = self.trace_file
538        if not trace:
539            try:
540                trace = open("/dev/null", "w")
541            except IOError:
542                raise service_error(service_error.internal,
543                        "Cannot open /dev/null??");
544
545        # May raise CalledProcessError
546        rv = call([self.ssh_keygen, '-t', t, '-N', '', '-f', dest],
547                stdout=trace, stderr=trace)
548        if rv != 0:
549            raise service_error(service_error.internal, 
550                    "Cannot generate nonce ssh keys.  %s return code %d" \
551                            % (self.ssh_keygen, rv))
552
553    def gentopo(self, str):
554        class topo_parse:
555            def __init__(self):
556                self.str_subelements = ('vname', 'vnode', 'ips', 'ip', 'member')
557                self.int_subelements = ( 'bandwidth',)
558                self.float_subelements = ( 'delay',)
559                self.nodes = [ ]
560                self.lans =  [ ]
561                self.element = { }
562                self.topo = { \
563                        'node': self.nodes,\
564                        'lan' : self.lans,\
565                    }
566                self.chars = ""
567
568            def end_element(self, name):
569                if name == 'node':
570                    self.nodes.append(self.element)
571                    self.element = { }
572                elif name == 'lan':
573                    self.lans.append(self.element)
574                    self.element = { }
575                elif name in self.str_subelements:
576                    self.element[name] = self.chars
577                    self.chars = ""
578                elif name in self.int_subelements:
579                    self.element[name] = int(self.chars)
580                    self.chars = ""
581                elif name in self.float_subelements:
582                    self.element[name] = float(self.chars)
583                    self.chars = ""
584
585            def found_chars(self, data):
586                self.chars += data.rstrip()
587
588
589        tp = topo_parse();
590        parser = xml.parsers.expat.ParserCreate()
591        parser.EndElementHandler = tp.end_element
592        parser.CharacterDataHandler = tp.found_chars
593
594        parser.Parse(str)
595
596        return tp.topo
597       
598
599    def genviz(self, topo):
600        """
601        Generate the visualization the virtual topology
602        """
603
604        neato = "/usr/local/bin/neato"
605        # These are used to parse neato output and to create the visualization
606        # file.
607        vis_re = re.compile('^\s*"?([\w\-]+)"?\s+\[.*pos="(\d+),(\d+)"')
608        vis_fmt = "<node><name>%s</name><x>%s</x><y>%s</y><type>" + \
609                "%s</type></node>"
610
611        try:
612            # Node names
613            nodes = [ n['vname'] for n in topo['node'] ]
614            topo_lans = topo['lan']
615        except KeyError:
616            raise service_error(service_error.internal, "Bad topology")
617
618        lans = { }
619        links = { }
620
621        # Walk through the virtual topology, organizing the connections into
622        # 2-node connections (links) and more-than-2-node connections (lans).
623        # When a lan is created, it's added to the list of nodes (there's a
624        # node in the visualization for the lan).
625        for l in topo_lans:
626            if links.has_key(l['vname']):
627                if len(links[l['vname']]) < 2:
628                    links[l['vname']].append(l['vnode'])
629                else:
630                    nodes.append(l['vname'])
631                    lans[l['vname']] = links[l['vname']]
632                    del links[l['vname']]
633                    lans[l['vname']].append(l['vnode'])
634            elif lans.has_key(l['vname']):
635                lans[l['vname']].append(l['vnode'])
636            else:
637                links[l['vname']] = [ l['vnode'] ]
638
639
640        # Open up a temporary file for dot to turn into a visualization
641        try:
642            df, dotname = tempfile.mkstemp()
643            dotfile = os.fdopen(df, 'w')
644        except IOError:
645            raise service_error(service_error.internal,
646                    "Failed to open file in genviz")
647
648        # Generate a dot/neato input file from the links, nodes and lans
649        try:
650            print >>dotfile, "graph G {"
651            for n in nodes:
652                print >>dotfile, '\t"%s"' % n
653            for l in links.keys():
654                print >>dotfile, '\t"%s" -- "%s"' %  tuple(links[l])
655            for l in lans.keys():
656                for n in lans[l]:
657                    print >>dotfile, '\t "%s" -- "%s"' % (n,l)
658            print >>dotfile, "}"
659            dotfile.close()
660        except TypeError:
661            raise service_error(service_error.internal,
662                    "Single endpoint link in vtopo")
663        except IOError:
664            raise service_error(service_error.internal, "Cannot write dot file")
665
666        # Use dot to create a visualization
667        dot = Popen([neato, '-Gstart=rand', '-Gepsilon=0.005', '-Gmaxiter=2000',
668                '-Gpack=true', dotname], stdout=PIPE)
669
670        # Translate dot to vis format
671        vis_nodes = [ ]
672        vis = { 'node': vis_nodes }
673        for line in dot.stdout:
674            m = vis_re.match(line)
675            if m:
676                vn = m.group(1)
677                vis_node = {'name': vn, \
678                        'x': float(m.group(2)),\
679                        'y' : float(m.group(3)),\
680                    }
681                if vn in links.keys() or vn in lans.keys():
682                    vis_node['type'] = 'lan'
683                else:
684                    vis_node['type'] = 'node'
685                vis_nodes.append(vis_node)
686        rv = dot.wait()
687
688        os.remove(dotname)
689        if rv == 0 : return vis
690        else: return None
691
692
693    def get_access(self, tb, nodes, user, tbparam):
694        """
695        Get access to testbed through fedd and set the parameters for that tb
696        """
697
698        translate_attr = {
699            'slavenodestartcmd': 'expstart',
700            'slaveconnectorstartcmd': 'gwstart',
701            'masternodestartcmd': 'mexpstart',
702            'masterconnectorstartcmd': 'mgwstart',
703            'connectorimage': 'gwimage',
704            'connectortype': 'gwtype',
705            'tunnelcfg': 'tun',
706            'smbshare': 'smbshare',
707        }
708
709        # XXX multi-level access
710        uri = self.tbmap.get(tb, None)
711        if not uri:
712            raise service_error(serice_error.server_config, 
713                    "Unknown testbed: %s" % tb)
714
715        # The basic request
716        req = {\
717                'destinationTestbed' : { 'uri' : uri },
718                'user':  user,
719                'allocID' : { 'localname': 'test' },
720                'access' : [ { 'sshPubkey' : self.ssh_pubkey } ]
721            }
722       
723        # node resources if any
724        if nodes != None and len(nodes) > 0:
725            rnodes = [ ]
726            for n in nodes:
727                rn = { }
728                image, hw, count = n.split(":")
729                if image: rn['image'] = [ image ]
730                if hw: rn['hardware'] = [ hw ]
731                if count: rn['count'] = int(count)
732                rnodes.append(rn)
733            req['resources']= { }
734            req['resources']['node'] = rnodes
735
736        # No retry loop here.  Proxy servers must correctly authenticate
737        # themselves without help
738
739        try:
740            ctx = fedd_ssl_context(self.cert_file, 
741                    self.trusted_certs, password=self.cert_pwd)
742        except SSL.SSLError:
743            raise service_error(service_error.server_config, 
744                    "Server certificates misconfigured")
745
746        loc = feddServiceLocator();
747        port = loc.getfeddPortType(uri,
748                transport=M2Crypto.httpslib.HTTPSConnection, 
749                transdict={ 'ssl_context' : ctx })
750
751        # Reconstruct the full request message
752        msg = RequestAccessRequestMessage()
753        msg.set_element_RequestAccessRequestBody(
754                pack_soap(msg, "RequestAccessRequestBody", req))
755
756        try:
757            resp = port.RequestAccess(msg)
758        except ZSI.ParseException, e:
759            raise service_error(service_error.req,
760                    "Bad format message (XMLRPC??): %s" %
761                    str(e))
762        r = unpack_soap(resp)
763
764        if r.has_key('RequestAccessResponseBody'):
765            r = r['RequestAccessResponseBody']
766        else:
767            raise service_error(service_error.proxy,
768                    "Bad proxy response")
769
770
771        e = r['emulab']
772        p = e['project']
773        tbparam[tb] = { 
774                "boss": e['boss'],
775                "host": e['ops'],
776                "domain": e['domain'],
777                "fs": e['fileServer'],
778                "eventserver": e['eventServer'],
779                "project": unpack_id(p['name']),
780                "emulab" : e
781                }
782        # Make the testbed name be the label the user applied
783        p['testbed'] = {'localname': tb }
784
785        for u in p['user']:
786            tbparam[tb]['user'] = unpack_id(u['userID'])
787
788        for a in e['fedAttr']:
789            if a['attribute']:
790                key = translate_attr.get(a['attribute'].lower(), None)
791                if key:
792                    tbparam[tb][key]= a['value']
793       
794    class current_testbed:
795        def __init__(self, eid, tmpdir):
796            self.begin_testbed = re.compile("^#\s+Begin\s+Testbed\s+\((\w+)\)")
797            self.end_testbed = re.compile("^#\s+End\s+Testbed\s+\((\w+)\)")
798            self.current_testbed = None
799            self.testbed_file = None
800
801            self.def_expstart = \
802                    "sudo -H /bin/sh FEDDIR/fed_bootstrap >& /tmp/federate";
803            self.def_mexpstart = "sudo -H FEDDIR/make_hosts FEDDIR/hosts";
804            self.def_gwstart = \
805                    "sudo -H FEDDIR/fed-tun.pl -f GWCONF>& /tmp/bridge.log";
806            self.def_mgwstart = \
807                    "sudo -H FEDDIR/fed-tun.pl -f GWCONF >& /tmp/bridge.log";
808            self.def_gwimage = "FBSD61-TUNNEL2";
809            self.def_gwtype = "pc";
810
811            self.eid = eid
812            self.tmpdir = tmpdir
813
814        def __call__(self, line, master, allocated, tbparams):
815            # Capture testbed topology descriptions
816            if self.current_testbed == None:
817                m = self.begin_testbed.match(line)
818                if m != None:
819                    self.current_testbed = m.group(1)
820                    if self.current_testbed == None:
821                        raise service_error(service_error.req,
822                                "Bad request format (unnamed testbed)")
823                    allocated[self.current_testbed] = \
824                            allocated.get(self.current_testbed,0) + 1
825                    tb_dir = "%s/%s" % (self.tmpdir, self.current_testbed)
826                    if not os.path.exists(tb_dir):
827                        try:
828                            os.mkdir(tb_dir)
829                        except IOError:
830                            raise service_error(service_error.internal,
831                                    "Cannot create %s" % tb_dir)
832                    try:
833                        self.testbed_file = open("%s/%s.%s.tcl" %
834                                (tb_dir, self.eid, self.current_testbed), 'w')
835                    except IOError:
836                        self.testbed_file = None
837                    return True
838                else: return False
839            else:
840                m = self.end_testbed.match(line)
841                if m != None:
842                    if m.group(1) != self.current_testbed:
843                        raise service_error(service_error.internal, 
844                                "Mismatched testbed markers!?")
845                    if self.testbed_file != None: 
846                        self.testbed_file.close()
847                        self.testbed_file = None
848                    self.current_testbed = None
849                elif self.testbed_file:
850                    # Substitute variables and put the line into the local
851                    # testbed file.
852                    gwtype = tbparams[self.current_testbed].get('gwtype', 
853                            self.def_gwtype)
854                    gwimage = tbparams[self.current_testbed].get('gwimage', 
855                            self.def_gwimage)
856                    mgwstart = tbparams[self.current_testbed].get('mgwstart', 
857                            self.def_mgwstart)
858                    mexpstart = tbparams[self.current_testbed].get('mexpstart', 
859                            self.def_mexpstart)
860                    gwstart = tbparams[self.current_testbed].get('gwstart', 
861                            self.def_gwstart)
862                    expstart = tbparams[self.current_testbed].get('expstart', 
863                            self.def_expstart)
864                    project = tbparams[self.current_testbed].get('project')
865                    line = re.sub("GWTYPE", gwtype, line)
866                    line = re.sub("GWIMAGE", gwimage, line)
867                    if self.current_testbed == master:
868                        line = re.sub("GWSTART", mgwstart, line)
869                        line = re.sub("EXPSTART", mexpstart, line)
870                    else:
871                        line = re.sub("GWSTART", gwstart, line)
872                        line = re.sub("EXPSTART", expstart, line)
873                    # XXX: does `` embed without doing enything else?
874                    line = re.sub("GWCONF", "FEDDIR`hostname`.gw.conf", line)
875                    line = re.sub("PROJDIR", "/proj/%s/" % project, line)
876                    line = re.sub("EID", self.eid, line)
877                    line = re.sub("FEDDIR", "/proj/%s/exp/%s/tmp/" % \
878                            (project, self.eid), line)
879                    print >>self.testbed_file, line
880                return True
881
882    class allbeds:
883        def __init__(self, get_access):
884            self.begin_allbeds = re.compile("^#\s+Begin\s+Allbeds")
885            self.end_allbeds = re.compile("^#\s+End\s+Allbeds")
886            self.in_allbeds = False
887            self.get_access = get_access
888
889        def __call__(self, line, user, tbparams):
890            # Testbed access parameters
891            if not self.in_allbeds:
892                if self.begin_allbeds.match(line):
893                    self.in_allbeds = True
894                    return True
895                else:
896                    return False
897            else:
898                if self.end_allbeds.match(line):
899                    self.in_allbeds = False
900                else:
901                    nodes = line.split('|')
902                    tb = nodes.pop(0)
903                    self.get_access(tb, nodes, user, tbparams)
904                return True
905
906    class gateways:
907        def __init__(self, eid, master, tmpdir, gw_pubkey,
908                gw_secretkey, copy_file):
909            self.begin_gateways = \
910                    re.compile("^#\s+Begin\s+gateways\s+\((\w+)\)")
911            self.end_gateways = re.compile("^#\s+End\s+gateways\s+\((\w+)\)")
912            self.current_gateways = None
913            self.control_gateway = None
914            self.active_end = { }
915
916            self.eid = eid
917            self.master = master
918            self.tmpdir = tmpdir
919            self.gw_pubkey_base = gw_pubkey
920            self.gw_secretkey_base = gw_secretkey
921
922            self.copy_file = copy_file
923
924
925        def gateway_conf_file(self, gw, master, eid, pubkey, privkey,
926                active_end, tbparams, dtb, myname, desthost, type):
927            """
928            Produce a gateway configuration file from a gateways line.
929            """
930
931            sproject = tbparams[gw].get('project', 'project')
932            dproject = tbparams[dtb].get('project', 'project')
933            sdomain = ".%s.%s%s" % (eid, sproject,
934                    tbparams[gw].get('domain', ".example.com"))
935            ddomain = ".%s.%s%s" % (eid, dproject,
936                    tbparams[dtb].get('domain', ".example.com"))
937            boss = tbparams[master].get('boss', "boss")
938            fs = tbparams[master].get('fs', "fs")
939            event_server = "%s%s" % \
940                    (tbparams[gw].get('eventserver', "event_server"),
941                            tbparams[gw].get('domain', "example.com"))
942            remote_event_server = "%s%s" % \
943                    (tbparams[dtb].get('eventserver', "event_server"),
944                            tbparams[dtb].get('domain', "example.com"))
945            seer_control = "%s%s" % \
946                    (tbparams[gw].get('control', "control"), sdomain)
947            remote_script_dir = "/proj/%s/exp/%s/tmp" % ( dproject, eid)
948            local_script_dir = "/proj/%s/exp/%s/tmp" % ( sproject, eid)
949            tunnel_cfg = tbparams[gw].get("tun", "false")
950
951            conf_file = "%s%s.gw.conf" % (myname, sdomain)
952            remote_conf_file = "%s%s.gw.conf" % (desthost, ddomain)
953
954            # translate to lower case so the `hostname` hack for specifying
955            # configuration files works.
956            conf_file = conf_file.lower();
957            remote_conf_file = remote_conf_file.lower();
958
959            if dtb == master:
960                active = "false"
961            elif gw == master:
962                active = "true"
963            elif active_end.has_key['%s-%s' % (dtb, gw)]:
964                active = "false"
965            else:
966                active_end['%s-%s' % (gw, dtb)] = 1
967                active = "true"
968
969            gwconfig = open("%s/%s/%s" % (self.tmpdir, gw, conf_file), "w")
970            print >>gwconfig, "Active: %s" % active
971            print >>gwconfig, "TunnelCfg: %s" % tunnel_cfg
972            print >>gwconfig, "BossName: %s" % boss
973            print >>gwconfig, "FsName: %s" % fs
974            print >>gwconfig, "EventServerName: %s" % event_server
975            print >>gwconfig, "RemoteEventServerName: %s" % remote_event_server
976            print >>gwconfig, "SeerControl: %s" % seer_control
977            print >>gwconfig, "Type: %s" % type
978            print >>gwconfig, "RemoteScriptDir: %s" % remote_script_dir
979            print >>gwconfig, "EventRepeater: %s/fed_evrepeater" % \
980                    local_script_dir
981            print >>gwconfig, "RemoteExperiment: %s/%s" % (dproject, eid)
982            print >>gwconfig, "LocalExperiment: %s/%s" % (sproject, eid)
983            print >>gwconfig, "RemoteConfigFile: %s/%s" % \
984                    (remote_script_dir, remote_conf_file)
985            print >>gwconfig, "Peer: %s%s" % (desthost, ddomain)
986            print >>gwconfig, "Pubkeys: %s/%s" % (local_script_dir, pubkey)
987            print >>gwconfig, "Privkeys: %s/%s" % (local_script_dir, privkey)
988            gwconfig.close()
989
990            return active == "true"
991
992        def __call__(self, line, allocated, tbparams):
993            # Process gateways
994            if not self.current_gateways:
995                m = self.begin_gateways.match(line)
996                if m:
997                    self.current_gateways = m.group(1)
998                    if allocated.has_key(self.current_gateways):
999                        # This test should always succeed
1000                        tb_dir = "%s/%s" % (self.tmpdir, self.current_gateways)
1001                        if not os.path.exists(tb_dir):
1002                            try:
1003                                os.mkdir(tb_dir)
1004                            except IOError:
1005                                raise service_error(service_error.internal,
1006                                        "Cannot create %s" % tb_dir)
1007                    else:
1008                        # XXX
1009                        print >>sys.stderr, \
1010                            "Ignoring gateways for unknown testbed %s" \
1011                                    % self.current_gateways
1012                        self.current_gateways = None
1013                    return True
1014                else:
1015                    return False
1016            else:
1017                m = self.end_gateways.match(line)
1018                if m :
1019                    if m.group(1) != self.current_gateways:
1020                        raise service_error(service_error.internal,
1021                                "Mismatched gateway markers!?")
1022                    if self.control_gateway:
1023                        try:
1024                            cc = open("%s/%s/client.conf" %
1025                                    (self.tmpdir, self.current_gateways), 'w')
1026                            print >>cc, "ControlGateway: %s" % \
1027                                    self.control_gateway
1028                            if tbparams[self.master].has_key('smbshare'):
1029                                print >>cc, "SMBSHare: %s" % \
1030                                        tbparams[self.master]['smbshare']
1031                            print >>cc, "ProjectUser: %s" % \
1032                                    tbparams[self.master]['user']
1033                            print >>cc, "ProjectName: %s" % \
1034                                    tbparams[self.master]['project']
1035                            cc.close()
1036                        except IOError:
1037                            raise service_error(service_error.internal,
1038                                    "Error creating client config")
1039                        try:
1040                            cc = open("%s/%s/seer.conf" %
1041                                    (self.tmpdir, self.current_gateways),
1042                                    'w')
1043                            if self.current_gateways != self.master:
1044                                print >>cc, "ControlNode: %s" % \
1045                                        self.control_gateway
1046                            print >>cc, "ExperimentID: %s/%s" % \
1047                                    ( tbparams[self.master]['project'], \
1048                                    self.eid )
1049                            cc.close()
1050                        except IOError:
1051                            raise service_error(service_error.internal,
1052                                    "Error creating seer config")
1053                    else:
1054                        if self.trace_file:
1055                            print >>sys.stderr, "No control gateway for %s" %\
1056                                    self.current_gateways
1057                    self.current_gateways = None
1058                else:
1059                    dtb, myname, desthost, type = line.split(" ")
1060
1061                    if type == "control" or type == "both":
1062                        self.control_gateway = "%s.%s.%s%s" % (myname, 
1063                                self.eid, 
1064                                tbparams[self.current_gateways]['project'],
1065                                tbparams[self.current_gateways]['domain'])
1066                    try:
1067                        active = self.gateway_conf_file(self.current_gateways,
1068                                self.master, self.eid, self.gw_pubkey_base,
1069                                self.gw_secretkey_base,
1070                                self.active_end, tbparams, dtb, myname,
1071                                desthost, type)
1072                    except IOError, e:
1073                        raise service_error(service_error.internal,
1074                                "Failed to write config file for %s" % \
1075                                        self.current_gateway)
1076           
1077                    gw_pubkey = "%s/keys/%s" % \
1078                            (self.tmpdir, self.gw_pubkey_base)
1079                    gw_secretkey = "%s/keys/%s" % \
1080                            (self.tmpdir, self.gw_secretkey_base)
1081
1082                    pkfile = "%s/%s/%s" % \
1083                            ( self.tmpdir, self.current_gateways, 
1084                                    self.gw_pubkey_base)
1085                    skfile = "%s/%s/%s" % \
1086                            ( self.tmpdir, self.current_gateways, 
1087                                    self.gw_secretkey_base)
1088
1089                    if not os.path.exists(pkfile):
1090                        try:
1091                            self.copy_file(gw_pubkey, pkfile)
1092                        except IOError:
1093                            service_error(service_error.internal,
1094                                    "Failed to copy pubkey file")
1095
1096                    if active and not os.path.exists(skfile):
1097                        try:
1098                            self.copy_file(gw_secretkey, skfile)
1099                        except IOError:
1100                            service_error(service_error.internal,
1101                                    "Failed to copy secretkey file")
1102                return True
1103
1104    class shunt_to_file:
1105        def __init__(self, begin, end, filename):
1106            self.begin = re.compile(begin)
1107            self.end = re.compile(end)
1108            self.in_shunt = False
1109            self.file = None
1110            self.filename = filename
1111
1112        def __call__(self, line):
1113            if not self.in_shunt:
1114                if self.begin.match(line):
1115                    self.in_shunt = True
1116                    try:
1117                        self.file = open(self.filename, "w")
1118                    except:
1119                        self.file = None
1120                        raise
1121                    return True
1122                else:
1123                    return False
1124            else:
1125                if self.end.match(line):
1126                    if self.file: 
1127                        self.file.close()
1128                        self.file = None
1129                    self.in_shunt = False
1130                else:
1131                    if self.file:
1132                        print >>self.file, line
1133                return True
1134
1135    class shunt_to_list:
1136        def __init__(self, begin, end):
1137            self.begin = re.compile(begin)
1138            self.end = re.compile(end)
1139            self.in_shunt = False
1140            self.list = [ ]
1141       
1142        def __call__(self, line):
1143            if not self.in_shunt:
1144                if self.begin.match(line):
1145                    self.in_shunt = True
1146                    return True
1147                else:
1148                    return False
1149            else:
1150                if self.end.match(line):
1151                    self.in_shunt = False
1152                else:
1153                    self.list.append(line)
1154                return True
1155
1156    class shunt_to_string:
1157        def __init__(self, begin, end):
1158            self.begin = re.compile(begin)
1159            self.end = re.compile(end)
1160            self.in_shunt = False
1161            self.str = ""
1162       
1163        def __call__(self, line):
1164            if not self.in_shunt:
1165                if self.begin.match(line):
1166                    self.in_shunt = True
1167                    return True
1168                else:
1169                    return False
1170            else:
1171                if self.end.match(line):
1172                    self.in_shunt = False
1173                else:
1174                    self.str += line
1175                return True
1176
1177    def create_experiment(self, req, fid):
1178        try:
1179            tmpdir = tempfile.mkdtemp(prefix="split-")
1180        except IOError:
1181            raise service_error(service_error.internal, "Cannot create tmp dir")
1182
1183        gw_pubkey_base = "fed.%s.pub" % self.ssh_type
1184        gw_secretkey_base = "fed.%s" % self.ssh_type
1185        gw_pubkey = tmpdir + "/keys/" + gw_pubkey_base
1186        gw_secretkey = tmpdir + "/keys/" + gw_secretkey_base
1187        tclfile = tmpdir + "/experiment.tcl"
1188        tbparams = { }
1189
1190        pid = "dummy"
1191        gid = "dummy"
1192        # XXX
1193        fail_soft = False
1194
1195        try:
1196            os.mkdir(tmpdir+"/keys")
1197        except OSError:
1198            raise service_error(service_error.internal,
1199                    "Can't make temporary dir")
1200
1201        req = req.get('CreateRequestBody', None)
1202        if not req:
1203            raise service_error(service_error.req,
1204                    "Bad request format (no CreateRequestBody)")
1205        # The tcl parser needs to read a file so put the content into that file
1206        file_content=req.get('experimentdescription', None)
1207        if file_content:
1208            try:
1209                f = open(tclfile, 'w')
1210                f.write(file_content)
1211                f.close()
1212            except IOError:
1213                raise service_error(service_error.internal,
1214                        "Cannot write temp experiment description")
1215        else:
1216            raise service_error(service_error.req, "No experiment description")
1217
1218        if req.has_key('experimentID') and \
1219                req['experimentID'].has_key('localname'):
1220            eid = req['experimentID']['localname']
1221            self.state_lock.acquire()
1222            while (self.state.has_key(eid)):
1223                eid += random.choice(string.ascii_letters)
1224            self.state[eid] = "placeholder"
1225            self.state_lock.release()
1226        else:
1227            eid = self.exp_stem
1228            for i in range(0,5):
1229                eid += random.choice(string.ascii_letters)
1230            self.state_lock.acquire()
1231            while (self.state.has_key(eid)):
1232                eid = self.exp_stem
1233                for i in range(0,5):
1234                    eid += random.choice(string.ascii_letters)
1235            self.state[eid] = "placeholder"
1236            self.state_lock.release()
1237
1238        try:
1239            self.generate_ssh_keys(gw_secretkey, self.ssh_type)
1240        except ValueError:
1241            raise service_error(service_error.server_config, 
1242                    "Bad key type (%s)" % self.ssh_type)
1243
1244        user = req.get('user', None)
1245        if user == None:
1246            raise service_error(service_error.req, "No user")
1247
1248        master = req.get('master', None)
1249        if master == None:
1250            raise service_error(service_error.req, "No master testbed label")
1251       
1252       
1253        tclcmd = [self.tclsh, self.tcl_splitter, '-s', '-x', 
1254            str(self.muxmax), '-m', master, pid, gid, eid, tclfile]
1255        tclparser = Popen(tclcmd, stdout=PIPE)
1256
1257        allocated = { }
1258        started = { }
1259
1260        parse_current_testbed = self.current_testbed(eid, tmpdir)
1261        parse_allbeds = self.allbeds(self.get_access)
1262        parse_gateways = self.gateways(eid, master, tmpdir,
1263                gw_pubkey_base, gw_secretkey_base, self.copy_file)
1264        parse_vtopo = self.shunt_to_string("^#\s+Begin\s+Vtopo",
1265                    "^#\s+End\s+Vtopo")
1266        parse_hostnames = self.shunt_to_file("^#\s+Begin\s+hostnames",
1267                    "^#\s+End\s+hostnames", tmpdir + "/hosts")
1268        parse_tarfiles = self.shunt_to_list("^#\s+Begin\s+tarfiles",
1269                "^#\s+End\s+tarfiles")
1270        parse_rpms = self.shunt_to_list("^#\s+Begin\s+rpms",
1271                "^#\s+End\s+rpms")
1272
1273        for line in tclparser.stdout:
1274            line = line.rstrip()
1275            if parse_current_testbed(line, master, allocated, tbparams):
1276                continue
1277            elif parse_allbeds(line, user, tbparams):
1278                continue
1279            elif parse_gateways(line, allocated, tbparams):
1280                continue
1281            elif parse_vtopo(line):
1282                continue
1283            elif parse_hostnames(line):
1284                continue
1285            elif parse_tarfiles(line):
1286                continue
1287            elif parse_rpms(line):
1288                continue
1289            else:
1290                raise service_error(service_error.internal, 
1291                        "Bad tcl parse? %s" % line)
1292
1293        vtopo = self.gentopo(parse_vtopo.str)
1294        if not vtopo:
1295            raise service_error(service_error.internal, 
1296                    "Failed to generate virtual topology")
1297
1298        vis = self.genviz(vtopo)
1299        if not vis:
1300            raise service_error(service_error.internal, 
1301                    "Failed to generate visualization")
1302
1303        # save federant information
1304        for k in allocated.keys():
1305            tbparams[k]['federant'] = {\
1306                    'name': [ { 'localname' : eid} ],\
1307                    'emulab': tbparams[k]['emulab'],\
1308                    'master' : k == master,\
1309                }
1310
1311
1312        # Copy tarfiles and rpms needed at remote sites into a staging area
1313        try:
1314            for t in parse_tarfiles.list:
1315                if not os.path.exists("%s/tarfiles" % tmpdir):
1316                    os.mkdir("%s/tarfiles" % tmpdir)
1317                self.copy_file(t, "%s/tarfiles/%s" % \
1318                        (tmpdir, os.path.basename(t)))
1319            for r in parse_rpms.list:
1320                if not os.path.exists("%s/rpms" % tmpdir):
1321                    os.mkdir("%s/rpms" % tmpdir)
1322                self.copy_file(r, "%s/rpms/%s" % \
1323                        (tmpdir, os.path.basename(r)))
1324        except IOError, e:
1325            raise service_error(service_error.internal, 
1326                    "Cannot stage tarfile/rpm: %s" % e.strerror)
1327
1328        thread_pool_info = self.thread_pool()
1329        threads = [ ]
1330
1331        for tb in [ k for k in allocated.keys() if k != master]:
1332            # Wait until we have a free slot to start the next testbed load
1333            thread_pool_info.acquire()
1334            while thread_pool_info.started - \
1335                    thread_pool_info.terminated >= self.nthreads:
1336                thread_pool_info.wait()
1337            thread_pool_info.release()
1338
1339            # Create and start a thread to start the segment, and save it to
1340            # get the return value later
1341            t  = self.pooled_thread(target=self.start_segment, 
1342                    args=(tb, eid, tbparams, tmpdir, 0), name=tb,
1343                    pdata=thread_pool_info, trace_file=self.trace_file)
1344            threads.append(t)
1345            t.start()
1346
1347        # Wait until all finish (the first clause of the while is to make sure
1348        # one starts)
1349        thread_pool_info.acquire()
1350        while thread_pool_info.started == 0 or \
1351                thread_pool_info.started > thread_pool_info.terminated:
1352            thread_pool_info.wait()
1353        thread_pool_info.release()
1354
1355        # If none failed, start the master
1356        failed = [ t.getName() for t in threads if not t.rv ]
1357
1358        if len(failed) == 0:
1359            if not self.start_segment(master, eid, tbparams, tmpdir):
1360                failed.append(master)
1361
1362        succeeded = [tb for tb in allocated.keys() if tb not in failed]
1363        # If one failed clean up, unless fail_soft is set
1364        if failed:
1365            if not fail_soft:
1366                for tb in succeeded:
1367                    self.stop_segment(tb, eid, tbparams)
1368                # Remove the placeholder
1369                self.state_lock.acquire()
1370                del self.state[eid]
1371                self.state_lock.release()
1372
1373                raise service_error(service_error.federant,
1374                    "Swap in failed on %s" % ",".join(failed))
1375        else:
1376            if self.trace_file:
1377                print >>self.trace_file, "Experiment started"
1378
1379        # Generate an ID for the experiment (slice) and a certificate that the
1380        # allocator can use to prove they own it.  We'll ship it back through
1381        # the encrypted connection.
1382        (expid, expcert) = generate_fedid("test", dir=tmpdir, 
1383                trace=self.trace_file)
1384
1385        if self.trace_file:
1386            print >>self.trace_file, "removing %s" % tmpdir
1387
1388        # Walk up tmpdir, deleting as we go
1389        for path, dirs, files in os.walk(tmpdir, topdown=False):
1390            for f in files:
1391                os.remove(os.path.join(path, f))
1392            for d in dirs:
1393                os.rmdir(os.path.join(path, d))
1394        os.rmdir(tmpdir)
1395
1396        resp = { 'federant' : [ tbparams[tb]['federant'] \
1397                for tb in tbparams.keys() \
1398                    if tbparams[tb].has_key('federant') ],\
1399                    'vtopo': vtopo,\
1400                    'vis' : vis,
1401                    'experimentID' : [\
1402                            { 'fedid': copy.copy(expid) }, \
1403                            { 'localname': eid },\
1404                        ],\
1405                    'experimentAccess': { 'X509' : expcert },\
1406                }
1407
1408        self.state_lock.acquire()
1409        self.state[expid] = { 'federant' : [ tbparams[tb]['federant'] \
1410                for tb in tbparams.keys() \
1411                    if tbparams[tb].has_key('federant') ],\
1412                    'vtopo': vtopo,\
1413                    'vis' : vis,
1414                    'experimentID' : [\
1415                            { 'fedid': expid }, { 'localname': eid },\
1416                        ],\
1417                }
1418        self.state[eid] = self.state[expid]
1419        if self.state_filename: self.write_state()
1420        self.state_lock.release()
1421
1422        if not failed:
1423            return resp
1424        else:
1425            raise service_error(service_error.partial, \
1426                    "Partial swap in on %s" % ",".join(succeeded))
1427
1428
1429    def get_vtopo(self, req, fid):
1430        rv = None
1431
1432        req = req.get('VtopoRequestBody', None)
1433        if not req:
1434            raise service_error(service_error.req,
1435                    "Bad request format (no VtopoRequestBody)")
1436        exp = req.get('experiment', None)
1437        if exp:
1438            if exp.has_key('fedid'):
1439                key = fedid(bits=exp['fedid'])
1440                keytype = "fedid"
1441            elif exp.has_key('localname'):
1442                key = exp['localname']
1443                keytype = "localname"
1444            else:
1445                raise service_error(service_error.req, "Unknown lookup type")
1446        else:
1447            raise service_error(service_error.req, "No request?")
1448
1449        self.state_lock.acquire()
1450        if self.state.has_key(key):
1451            rv = { 'experiment' : {keytype: key },\
1452                    'vtopo': self.state[key]['vtopo'],\
1453                }
1454        self.state_lock.release()
1455
1456        if rv: return rv
1457        else: raise service_error(service_error.req, "No such experiment")
1458
1459    def get_vis(self, req, fid):
1460        rv = None
1461
1462        req = req.get('VisRequestBody', None)
1463        if not req:
1464            raise service_error(service_error.req,
1465                    "Bad request format (no VisRequestBody)")
1466        exp = req.get('experiment', None)
1467        if exp:
1468            if exp.has_key('fedid'):
1469                key = fedid(bits=exp['fedid'])
1470                keytype = "fedid"
1471            elif exp.has_key('localname'):
1472                key = exp['localname']
1473                keytype = "localname"
1474            else:
1475                raise service_error(service_error.req, "Unknown lookup type")
1476        else:
1477            raise service_error(service_error.req, "No request?")
1478
1479        self.state_lock.acquire()
1480        if self.state.has_key(key):
1481            rv =  { 'experiment' : {keytype: key },\
1482                    'vis': self.state[key]['vis'],\
1483                    }
1484        self.state_lock.release()
1485
1486        if rv: return rv
1487        else: raise service_error(service_error.req, "No such experiment")
1488
1489    def get_info(self, req, fid):
1490        rv = None
1491
1492        req = req.get('InfoRequestBody', None)
1493        if not req:
1494            raise service_error(service_error.req,
1495                    "Bad request format (no VisRequestBody)")
1496        exp = req.get('experiment', None)
1497        if exp:
1498            if exp.has_key('fedid'):
1499                key = fedid(bits=exp['fedid'])
1500                keytype = "fedid"
1501            elif exp.has_key('localname'):
1502                key = exp['localname']
1503                keytype = "localname"
1504            else:
1505                raise service_error(service_error.req, "Unknown lookup type")
1506        else:
1507            raise service_error(service_error.req, "No request?")
1508
1509        # The state may be massaged by the service function that called
1510        # get_info (e.g., encoded for XMLRPC transport) so send a copy of the
1511        # state.
1512        self.state_lock.acquire()
1513        if self.state.has_key(key):
1514            rv = copy.deepcopy(self.state[key])
1515        self.state_lock.release()
1516
1517        if rv: return rv
1518        else: raise service_error(service_error.req, "No such experiment")
1519
1520
1521    def terminate_experiment(self, req, fid):
1522        tbparams = { }
1523        req = req.get('TerminateRequestBody', None)
1524        if not req:
1525            raise service_error(service_error.req,
1526                    "Bad request format (no TerminateRequestBody)")
1527        exp = req.get('experiment', None)
1528        if exp:
1529            if exp.has_key('fedid'):
1530                key = fedid(bits=exp['fedid'])
1531                keytype = "fedid"
1532            elif exp.has_key('localname'):
1533                key = exp['localname']
1534                keytype = "localname"
1535            else:
1536                raise service_error(service_error.req, "Unknown lookup type")
1537        else:
1538            raise service_error(service_error.req, "No request?")
1539
1540        self.state_lock.acquire()
1541        fed_exp = self.state.get(key, None)
1542
1543        if fed_exp:
1544            # This branch of the conditional holds the lock to generate a
1545            # consistent temporary tbparams variable to deallocate experiments.
1546            # It releases the lock to do the deallocations and reacquires it to
1547            # remove the experiment state when the termination is complete.
1548            ids = []
1549            #  experimentID is a list of dicts that are self-describing
1550            #  identifiers.  This finds all the fedids and localnames - the
1551            #  keys of self.state - and puts them into ids.
1552            for id in fed_exp.get('experimentID', []):
1553                if id.has_key('fedid'): ids.append(id['fedid'])
1554                if id.has_key('localname'): ids.append(id['localname'])
1555
1556            # Construct enough of the tbparams to make the stop_segment calls
1557            # work
1558            for fed in fed_exp['federant']:
1559                try:
1560                    for e in fed['name']:
1561                        eid = e.get('localname', None)
1562                        if eid: break
1563                    else:
1564                        continue
1565
1566                    p = fed['emulab']['project']
1567
1568                    project = p['name']['localname']
1569                    tb = p['testbed']['localname']
1570                    user = p['user'][0]['userID']['localname']
1571
1572                    domain = fed['emulab']['domain']
1573                    host  = "%s%s" % (fed['emulab']['ops'], domain)
1574                except KeyError, e:
1575                    continue
1576                tbparams[tb] = {\
1577                        'user': user,\
1578                        'domain': domain,\
1579                        'project': project,\
1580                        'host': host,\
1581                        'eid': eid,\
1582                    }
1583            self.state_lock.release()
1584
1585            # Stop everyone.
1586            for tb in tbparams.keys():
1587                self.stop_segment(tb, tbparams[tb]['eid'], tbparams)
1588
1589            # Remove teh terminated experiment
1590            self.state_lock.acquire()
1591            for id in ids:
1592                if self.state.has_key(id): del self.state[id]
1593
1594            if self.state_filename: self.write_state()
1595            self.state_lock.release()
1596
1597            return { 'experiment': exp }
1598        else:
1599            # Don't forget to release the lock
1600            self.state_lock.release()
1601            raise service_error(service_error.req, "No saved state")
1602
1603
1604
1605
1606if __name__ == '__main__':
1607    from optparse import OptionParser
1608   
1609    parser = OptionParser()
1610    parser.add_option('-d', '--debug', dest='debug', default=False,
1611            action='store_true', help='print actions rather than take them')
1612    parser.add_option('-f', '--file', dest='tcl', help='tcl file to parse')
1613    parser.add_option('-m', '--master', dest='master', 
1614            help='testbed label for matster testbd')
1615    parser.add_option('-t', '--trace', dest='trace', default=None, 
1616            help='file to print intermediate messages to')
1617    parser.add_option('-T', '--trace-stderr', dest='trace', 
1618            action='store_const',const=sys.stderr,
1619            help='file to print intermediate messages to')
1620    opts, args  = parser.parse_args()
1621
1622    trace_file = None
1623    if opts.trace:
1624        try:
1625            trace_file = open(opts.trace, 'w')
1626        except IOError:
1627            print >>sys.stderr, "Can't open trace file"
1628
1629    if opts.debug:
1630        if not trace_file:
1631            trace_file = sys.stderr
1632
1633    if opts.tcl != None:
1634        try:
1635            f = open(opts.tcl, 'r')
1636            content = ''.join(f)
1637            f.close()
1638        except IOError, e:
1639            sys.exit("Can't read %s: %s" % (opts.tcl, e))
1640    else:
1641        sys.exit("Must specify a file name")
1642
1643    if not opts.master:
1644        sys.exit("Must supply master tb label (--master)");
1645
1646    obj = fedd_create_experiment_local(
1647            debug=opts.debug,
1648            scripts_dir="/users/faber/testbed/federation",
1649            cert_file="./fedd_client.pem", cert_pwd="faber", 
1650            ssh_pubkey_file='/users/faber/.ssh/id_rsa.pub',
1651            trusted_certs="./cacert.pem",
1652            tbmap = { 
1653                'deter':'https://users.isi.deterlab.net:23235',
1654                'emulab':'https://users.isi.deterlab.net:23236',
1655                'ucb':'https://users.isi.deterlab.net:23237',
1656                },
1657            trace_file=trace_file
1658        ) 
1659    rv = obj.create_experiment( {\
1660            'experimentdescription' : content, 
1661            'master' : opts.master, 
1662            'user': [ {'userID' : { 'localname' : 'faber' } } ],
1663            },
1664            None)
1665
1666    print rv
Note: See TracBrowser for help on using the repository browser.