Context Navigation

source: fedd/fedd_experiment_control.py @ 11a08b0

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

Last change on this file since 11a08b0 was 11a08b0, checked in by Ted Faber <faber@…>, 16 years ago
decent logging
Property mode set to `100644`
File size: 48.4 KB

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

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

Download in other formats: