Context Navigation

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
3	import os,sys
4
5	from ZSI import *
6	from M2Crypto import SSL
7	from M2Crypto.SSL.SSLServer import SSLServer
8	import M2Crypto.httpslib
9
10	import xml.parsers.expat
11
12	import re
13	import random
14	import string
15	import subprocess
16	import tempfile
17	import copy
18	import pickle
19
20	import traceback
21
22	from threading import *
23
24	from subprocess import *
25
26	from fedd_services import *
27	from fedd_internal_services import *
28	from fedd_util import *
29	import parse_detail
30	from service_error import *
31
32	class 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
1606	if __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.

Download in other formats: