Context Navigation

source: fedd/fedd_experiment_control.py @ eee2b2e

axis_examplecompt_changesinfo-opsversion-1.30version-2.00version-3.01version-3.02

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

Note: See TracBrowser for help on using the repository browser.

Download in other formats: