Context Navigation

source: fedd/fedd_create_experiment.py @ 291423b

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

Last change on this file since 291423b was 291423b, checked in by Ted Faber <faber@…>, 16 years ago
SEER config support
Property mode set to `100644`
File size: 42.3 KB

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

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

Download in other formats: