source: fedd/fedd_create_experiment.py @ fca8740

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

generic fault support (they're not all named RequestAccessFault? any more)

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