Context Navigation

source: fedd/fedd_experiment_control.py @ 19cc408

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

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

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

Download in other formats: