Context Navigation

source: fedd/federation/experiment_control.py @ 4450b24

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

Last change on this file since 4450b24 was 7c3d547, checked in by Ted Faber <faber@…>, 16 years ago
add a line to the client config file telling the master experiment ID
Property mode set to `100644`
File size: 63.3 KB

Rev	Line
[6679c122]	1	#!/usr/local/bin/python
	2
	3	import os,sys
	4
	5	import re
	6	import random
	7	import string
	8	import subprocess
	9	import tempfile
	10	import copy
[eee2b2e]	11	import pickle
[c971895]	12	import logging
[6679c122]	13
[3441fe3]	14	import traceback
[c971895]	15	# For parsing visualization output and splitter output
	16	import xml.parsers.expat
[3441fe3]	17
[1af38d6]	18	from threading import *
[6679c122]	19	from subprocess import *
	20
[ec4fb42]	21	from util import *
[51cc9df]	22	from fedid import fedid, generate_fedid
[9460b1e]	23	from remote_service import xmlrpc_handler, soap_handler, service_caller
[c971895]	24	from service_error import service_error
[6679c122]	25
[11a08b0]	26
	27	class nullHandler(logging.Handler):
	28	def emit(self, record): pass
	29
	30	fl = logging.getLogger("fedd.experiment_control")
	31	fl.addHandler(nullHandler())
	32
[ec4fb42]	33	class experiment_control_local:
[0ea11af]	34	"""
	35	Control of experiments that this system can directly access.
	36
	37	Includes experiment creation, termination and information dissemination.
	38	Thred safe.
	39	"""
[6679c122]	40
[1af38d6]	41	class thread_pool:
[0ea11af]	42	"""
	43	A class to keep track of a set of threads all invoked for the same
	44	task. Manages the mutual exclusion of the states.
	45	"""
[8bc5754]	46	def __init__(self, nthreads):
[0ea11af]	47	"""
	48	Start a pool.
	49	"""
[1af38d6]	50	self.changed = Condition()
	51	self.started = 0
	52	self.terminated = 0
[8bc5754]	53	self.nthreads = nthreads
[1af38d6]	54
	55	def acquire(self):
[0ea11af]	56	"""
	57	Get the pool's lock.
	58	"""
[1af38d6]	59	self.changed.acquire()
	60
	61	def release(self):
[0ea11af]	62	"""
	63	Release the pool's lock.
	64	"""
[1af38d6]	65	self.changed.release()
	66
	67	def wait(self, timeout = None):
[0ea11af]	68	"""
	69	Wait for a pool thread to start or stop.
	70	"""
[1af38d6]	71	self.changed.wait(timeout)
	72
	73	def start(self):
[0ea11af]	74	"""
	75	Called by a pool thread to report starting.
	76	"""
[1af38d6]	77	self.changed.acquire()
	78	self.started += 1
	79	self.changed.notifyAll()
	80	self.changed.release()
	81
	82	def terminate(self):
[0ea11af]	83	"""
	84	Called by a pool thread to report finishing.
	85	"""
[1af38d6]	86	self.changed.acquire()
	87	self.terminated += 1
	88	self.changed.notifyAll()
	89	self.changed.release()
	90
	91	def clear(self):
[0ea11af]	92	"""
	93	Clear all pool data.
	94	"""
[1af38d6]	95	self.changed.acquire()
	96	self.started = 0
	97	self.terminated =0
	98	self.changed.notifyAll()
	99	self.changed.release()
	100
[8bc5754]	101	def wait_for_slot(self):
	102	"""
	103	Wait until we have a free slot to start another pooled thread
	104	"""
	105	self.acquire()
	106	while self.started - self.terminated >= self.nthreads:
	107	self.wait()
	108	self.release()
	109
	110	def wait_for_all_done(self):
	111	"""
	112	Wait until all active threads finish (and at least one has started)
	113	"""
	114	self.acquire()
	115	while self.started == 0 or self.started > self.terminated:
	116	self.wait()
	117	self.release()
	118
[1af38d6]	119	class pooled_thread(Thread):
[0ea11af]	120	"""
	121	One of a set of threads dedicated to a specific task. Uses the
	122	thread_pool class above for coordination.
	123	"""
[1af38d6]	124	def __init__(self, group=None, target=None, name=None, args=(),
[3441fe3]	125	kwargs={}, pdata=None, trace_file=None):
[1af38d6]	126	Thread.__init__(self, group, target, name, args, kwargs)
[0ea11af]	127	self.rv = None # Return value of the ops in this thread
	128	self.exception = None # Exception that terminated this thread
	129	self.target=target # Target function to run on start()
	130	self.args = args # Args to pass to target
	131	self.kwargs = kwargs # Additional kw args
	132	self.pdata = pdata # thread_pool for this class
	133	# Logger for this thread
	134	self.log = logging.getLogger("fedd.experiment_control")
[1af38d6]	135
	136	def run(self):
[0ea11af]	137	"""
	138	Emulate Thread.run, except add pool data manipulation and error
	139	logging.
	140	"""
[1af38d6]	141	if self.pdata:
	142	self.pdata.start()
	143
	144	if self.target:
	145	try:
	146	self.rv = self.target(self.args, *self.kwargs)
[3441fe3]	147	except service_error, s:
	148	self.exception = s
[0ea11af]	149	self.log.error("Thread exception: %s %s" % \
	150	(s.code_string(), s.desc))
[3441fe3]	151	except:
	152	self.exception = sys.exc_info()[1]
[0ea11af]	153	self.log.error(("Unexpected thread exception: %s" +\
	154	"Trace %s") % (self.exception,\
	155	traceback.format_exc()))
[1af38d6]	156	if self.pdata:
	157	self.pdata.terminate()
[6679c122]	158
[f069052]	159	call_RequestAccess = service_caller('RequestAccess')
	160	call_ReleaseAccess = service_caller('ReleaseAccess')
	161	call_Ns2Split = service_caller('Ns2Split')
[058f58e]	162
[3f6bc5f]	163	def __init__(self, config=None, auth=None):
[0ea11af]	164	"""
	165	Intialize the various attributes, most from the config object
	166	"""
[ec4fb42]	167	self.thread_with_rv = experiment_control_local.pooled_thread
	168	self.thread_pool = experiment_control_local.thread_pool
[19cc408]	169
[f005e44]	170	self.cert_file = config.get("experiment_control", "cert_file")
	171	if self.cert_file:
	172	self.cert_pwd = config.get("experiment_control", "cert_pwd")
	173	else:
[02786fc]	174	self.cert_file = config.get("globals", "cert_file")
	175	self.cert_pwd = config.get("globals", "cert_pwd")
[f005e44]	176
	177	self.trusted_certs = config.get("experiment_control", "trusted_certs") \
	178	or config.get("globals", "trusted_certs")
[4064742]	179
[19cc408]	180	self.exp_stem = "fed-stem"
[11a08b0]	181	self.log = logging.getLogger("fedd.experiment_control")
[34bc05c]	182	set_log_level(config, "experiment_control", self.log)
[19cc408]	183	self.muxmax = 2
	184	self.nthreads = 2
	185	self.randomize_experiments = False
[72ed6e4]	186
[19cc408]	187	self.scp_exec = "/usr/bin/scp"
	188	self.splitter = None
	189	self.ssh_exec="/usr/bin/ssh"
	190	self.ssh_keygen = "/usr/bin/ssh-keygen"
	191	self.ssh_identity_file = None
[72ed6e4]	192
[34bc05c]	193
[bf0a80e]	194	self.debug = config.getboolean("experiment_control", "create_debug")
	195	self.state_filename = config.get("experiment_control",
[61c71bf]	196	"experiment_state")
	197	self.splitter_url = config.get("experiment_control", "splitter_uri")
[bf0a80e]	198	self.fedkit = config.get("experiment_control", "fedkit")
	199	accessdb_file = config.get("experiment_control", "accessdb")
[72ed6e4]	200
[721705e9]	201	self.ssh_pubkey_file = config.get("experiment_control",
	202	"ssh_pubkey_file")
	203	self.ssh_privkey_file = config.get("experiment_control",
	204	"ssh_privkey_file")
	205	# NB for internal master/slave ops, not experiment setup
	206	self.ssh_type = config.get("experiment_control", "sshkeytype", "rsa")
[19cc408]	207	self.state = { }
[a97394b]	208	self.state_lock = Lock()
[19cc408]	209	self.tclsh = "/usr/local/bin/otclsh"
[77a7634]	210	self.tcl_splitter = config.get("splitter", "tcl_splitter") or \
	211	config.get("experiment_control", "tcl_splitter",
	212	"/usr/testbed/lib/ns2ir/parse.tcl")
[34bc05c]	213	mapdb_file = config.get("experiment_control", "mapdb")
[19cc408]	214	self.trace_file = sys.stderr
	215
	216	self.def_expstart = \
[2c6128f]	217	"sudo -H /bin/sh /usr/local/federation/bin/federate.sh >& " +\
	218	"/tmp/federate";
	219	self.def_mexpstart = "sudo -H /usr/local/federation/bin/make_hosts " +\
	220	"FEDDIR/hosts";
[19cc408]	221	self.def_gwstart = \
[2c6128f]	222	"sudo -H /usr/local/federation/bin/fed-tun.pl -f GWCONF>& " +\
	223	"/tmp/bridge.log";
[19cc408]	224	self.def_mgwstart = \
[2c6128f]	225	"sudo -H /usr/local/federation/bin/fed-tun.pl -f GWCONF >& " +\
	226	"/tmp/bridge.log";
[19cc408]	227	self.def_gwimage = "FBSD61-TUNNEL2";
	228	self.def_gwtype = "pc";
[5fffd82]	229	self.local_access = { }
[19cc408]	230
[34bc05c]	231	if auth:
	232	self.auth = auth
	233	else:
	234	self.log.error(\
	235	"[access]: No authorizer initialized, creating local one.")
	236	auth = authorizer()
	237
[19cc408]	238
	239	if self.ssh_pubkey_file:
	240	try:
	241	f = open(self.ssh_pubkey_file, 'r')
	242	self.ssh_pubkey = f.read()
	243	f.close()
	244	except IOError:
	245	raise service_error(service_error.internal,
	246	"Cannot read sshpubkey")
[721705e9]	247	else:
	248	raise service_error(service_error.internal,
	249	"No SSH public key file?")
	250
	251	if not self.ssh_privkey_file:
	252	raise service_error(service_error.internal,
	253	"No SSH public key file?")
[19cc408]	254
[11a08b0]	255
[34bc05c]	256	if mapdb_file:
	257	self.read_mapdb(mapdb_file)
[4064742]	258	else:
[34bc05c]	259	self.log.warn("[experiment_control] No testbed map, using defaults")
	260	self.tbmap = {
	261	'deter':'https://users.isi.deterlab.net:23235',
	262	'emulab':'https://users.isi.deterlab.net:23236',
	263	'ucb':'https://users.isi.deterlab.net:23237',
	264	}
[4064742]	265
	266	if accessdb_file:
[34bc05c]	267	self.read_accessdb(accessdb_file)
[4064742]	268	else:
	269	raise service_error(service_error.internal,
	270	"No accessdb specified in config")
	271
[0ea11af]	272	# Grab saved state. OK to do this w/o locking because it's read only
	273	# and only one thread should be in existence that can see self.state at
	274	# this point.
[eee2b2e]	275	if self.state_filename:
	276	self.read_state()
	277
[0ea11af]	278	# Dispatch tables
	279	self.soap_services = {\
[f069052]	280	'Create': soap_handler('Create', self.create_experiment),
	281	'Vtopo': soap_handler('Vtopo', self.get_vtopo),
	282	'Vis': soap_handler('Vis', self.get_vis),
	283	'Info': soap_handler('Info', self.get_info),
	284	'Terminate': soap_handler('Terminate',
	285	self.terminate_experiment),
[19cc408]	286	}
	287
[0ea11af]	288	self.xmlrpc_services = {\
[f069052]	289	'Create': xmlrpc_handler('Create', self.create_experiment),
	290	'Vtopo': xmlrpc_handler('Vtopo', self.get_vtopo),
	291	'Vis': xmlrpc_handler('Vis', self.get_vis),
	292	'Info': xmlrpc_handler('Info', self.get_info),
	293	'Terminate': xmlrpc_handler('Terminate',
	294	self.terminate_experiment),
[19cc408]	295	}
	296
[6679c122]	297	def copy_file(self, src, dest, size=1024):
	298	"""
	299	Exceedingly simple file copy.
	300	"""
	301	s = open(src,'r')
	302	d = open(dest, 'w')
	303
	304	buf = "x"
	305	while buf != "":
	306	buf = s.read(size)
	307	d.write(buf)
	308	s.close()
	309	d.close()
	310
[a97394b]	311	# Call while holding self.state_lock
[eee2b2e]	312	def write_state(self):
[0ea11af]	313	"""
	314	Write a new copy of experiment state after copying the existing state
	315	to a backup.
	316
	317	State format is a simple pickling of the state dictionary.
	318	"""
[eee2b2e]	319	if os.access(self.state_filename, os.W_OK):
	320	self.copy_file(self.state_filename, \
	321	"%s.bak" % self.state_filename)
	322	try:
	323	f = open(self.state_filename, 'w')
	324	pickle.dump(self.state, f)
	325	except IOError, e:
[11a08b0]	326	self.log.error("Can't write file %s: %s" % \
	327	(self.state_filename, e))
[eee2b2e]	328	except pickle.PicklingError, e:
[11a08b0]	329	self.log.error("Pickling problem: %s" % e)
[4ed10ae]	330	except TypeError, e:
	331	self.log.error("Pickling problem (TypeError): %s" % e)
[eee2b2e]	332
[a97394b]	333	# Call while holding self.state_lock
[eee2b2e]	334	def read_state(self):
[0ea11af]	335	"""
	336	Read a new copy of experiment state. Old state is overwritten.
	337
	338	State format is a simple pickling of the state dictionary.
	339	"""
[eee2b2e]	340	try:
	341	f = open(self.state_filename, "r")
	342	self.state = pickle.load(f)
[d199ced]	343	self.log.debug("[read_state]: Read state from %s" % \
	344	self.state_filename)
[eee2b2e]	345	except IOError, e:
[d199ced]	346	self.log.warning("[read_state]: No saved state: Can't open %s: %s"\
	347	% (self.state_filename, e))
[eee2b2e]	348	except pickle.UnpicklingError, e:
[d199ced]	349	self.log.warning(("[read_state]: No saved state: " + \
	350	"Unpickling failed: %s") % e)
[4064742]	351
	352	for k in self.state.keys():
	353	try:
	354	# This list should only have one element in it, but phrasing it
	355	# as a for loop doesn't cost much, really. We have to find the
	356	# fedid elements anyway.
	357	for eid in [ f['fedid'] \
	358	for f in self.state[k]['experimentID']\
	359	if f.has_key('fedid') ]:
	360	self.auth.set_attribute(self.state[k]['owner'], eid)
	361	except KeyError, e:
	362	self.log.warning("[read_state]: State ownership or identity " +\
	363	"misformatted in %s: %s" % (self.state_filename, e))
	364
	365
	366	def read_accessdb(self, accessdb_file):
[34bc05c]	367	"""
	368	Read the mapping from fedids that can create experiments to their name
	369	in the 3-level access namespace. All will be asserted from this
	370	testbed and can include the local username and porject that will be
	371	asserted on their behalf by this fedd. Each fedid is also added to the
	372	authorization system with the "create" attribute.
	373	"""
	374	self.accessdb = {}
	375	# These are the regexps for parsing the db
[4064742]	376	name_expr = "[" + string.ascii_letters + string.digits + "\.\-]+"
	377	project_line = re.compile("^\s*fedid:([" + string.hexdigits + "]+)"+ \
	378	"\s->\(\s("+name_expr+")\s,\s("+name_expr+")\s\)\s$")
	379	user_line = re.compile("^\s*fedid:([" + string.hexdigits + "]+)"+ \
	380	"\s->\s(" + name_expr + ")\s*$")
	381	lineno = 0
	382
[34bc05c]	383	# Parse the mappings and store in self.authdb, a dict of
	384	# fedid -> (proj, user)
	385	try:
	386	f = open(accessdb_file, "r")
	387	for line in f:
	388	lineno += 1
	389	line = line.strip()
	390	if len(line) == 0 or line.startswith('#'):
	391	continue
	392	m = project_line.match(line)
	393	if m:
	394	fid = fedid(hexstr=m.group(1))
	395	project, user = m.group(2,3)
	396	if not self.accessdb.has_key(fid):
	397	self.accessdb[fid] = []
	398	self.accessdb[fid].append((project, user))
	399	continue
	400
	401	m = user_line.match(line)
	402	if m:
	403	fid = fedid(hexstr=m.group(1))
	404	project = None
	405	user = m.group(2)
	406	if not self.accessdb.has_key(fid):
	407	self.accessdb[fid] = []
	408	self.accessdb[fid].append((project, user))
	409	continue
	410	self.log.warn("[experiment_control] Error parsing access " +\
	411	"db %s at line %d" % (accessdb_file, lineno))
	412	except IOError:
[4064742]	413	raise service_error(service_error.internal,
[34bc05c]	414	"Error opening/reading %s as experiment " +\
	415	"control accessdb" % accessdb_file)
[4064742]	416	f.close()
	417
[34bc05c]	418	# Initialize the authorization attributes
	419	for fid in self.accessdb.keys():
	420	self.auth.set_attribute(fid, 'create')
	421
	422	def read_mapdb(self, file):
	423	"""
	424	Read a simple colon separated list of mappings for the
	425	label-to-testbed-URL mappings. Clears or creates self.tbmap.
	426	"""
	427
	428	self.tbmap = { }
	429	lineno =0
	430	try:
	431	f = open(file, "r")
	432	for line in f:
	433	lineno += 1
	434	line = line.strip()
	435	if line.startswith('#') or len(line) == 0:
	436	continue
	437	try:
	438	label, url = line.split(':', 1)
	439	self.tbmap[label] = url
	440	except ValueError, e:
	441	self.log.warn("[read_mapdb] Ignored bad line (%d) in " +\
	442	"map db: %s %s" % (lineno, line, e))
	443	except IOError, e:
	444	self.log.warning("[read_mapdb]: No saved map database: Can't " +\
	445	"open %s: %s" % (file, e))
	446	f.close()
[eee2b2e]	447
[6679c122]	448	def scp_file(self, file, user, host, dest=""):
	449	"""
[0ea11af]	450	scp a file to the remote host. If debug is set the action is only
	451	logged.
[6679c122]	452	"""
[d0ae12d]	453
[f9665d1]	454	scp_cmd = [self.scp_exec, '-o', 'IdentitiesOnly yes',
	455	'-o', 'StrictHostKeyChecking yes', '-i',
[dceeef9]	456	self.ssh_privkey_file, file, "%s@%s:%s" % (user, host, dest)]
[11a08b0]	457	rv = 0
[3441fe3]	458
[d199ced]	459	try:
[03b9b14]	460	dnull = open("/dev/null", "w")
[d199ced]	461	except IOError:
	462	self.log.debug("[ssh_file]: failed to open /dev/null for redirect")
	463	dnull = Null
	464
[11a08b0]	465	self.log.debug("[scp_file]: %s" % " ".join(scp_cmd))
[d0ae12d]	466	if not self.debug:
[03b9b14]	467	rv = call(scp_cmd, stdout=dnull, stderr=dnull)
[d0ae12d]	468
	469	return rv == 0
[6679c122]	470
[3441fe3]	471	def ssh_cmd(self, user, host, cmd, wname=None):
[0ea11af]	472	"""
	473	Run a remote command on host as user. If debug is set, the action is
	474	only logged.
	475	"""
[f9665d1]	476	sh_str = ("%s -o 'IdentitiesOnly yes' -o " + \
	477	"'StrictHostKeyChecking yes' -i %s %s@%s %s") % \
[dceeef9]	478	(self.ssh_exec, self.ssh_privkey_file,
	479	user, host, cmd)
[d0ae12d]	480
[d199ced]	481	try:
	482	dnull = open("/dev/null", "r")
	483	except IOError:
	484	self.log.debug("[ssh_cmd]: failed to open /dev/null for redirect")
	485	dnull = Null
	486
[11a08b0]	487	self.log.debug("[ssh_cmd]: %s" % sh_str)
[d0ae12d]	488	if not self.debug:
[d199ced]	489	if dnull:
	490	sub = Popen(sh_str, shell=True, stdout=dnull, stderr=dnull)
	491	else:
	492	sub = Popen(sh_str, shell=True)
[d0ae12d]	493	return sub.wait() == 0
	494	else:
	495	return True
[6679c122]	496
	497	def ship_configs(self, host, user, src_dir, dest_dir):
[0ea11af]	498	"""
	499	Copy federant-specific configuration files to the federant.
	500	"""
[6679c122]	501	if not self.ssh_cmd(user, host, "mkdir -p %s" % dest_dir):
	502	return False
	503	if not self.ssh_cmd(user, host, "chmod 770 %s" % dest_dir):
	504	return False
	505
	506	for f in os.listdir(src_dir):
	507	if os.path.isdir(f):
	508	if not self.ship_configs(host, user, "%s/%s" % (src_dir, f),
	509	"%s/%s" % (dest_dir, f)):
	510	return False
	511	else:
	512	if not self.scp_file("%s/%s" % (src_dir, f),
	513	user, host, dest_dir):
	514	return False
	515	return True
	516
[6546868]	517	def start_segment(self, tb, eid, tbparams, tmpdir, timeout=0):
[0ea11af]	518	"""
	519	Start a sub-experiment on a federant.
	520
	521	Get the current state, modify or create as appropriate, ship data and
	522	configs and start the experiment. There are small ordering differences
	523	based on the initial state of the sub-experiment.
	524	"""
	525	# ops node in the federant
[6679c122]	526	host = "%s%s" % (tbparams[tb]['host'], tbparams[tb]['domain'])
[0ea11af]	527	user = tbparams[tb]['user'] # federant user
	528	pid = tbparams[tb]['project'] # federant project
[6679c122]	529	# XXX
[0d830de]	530	base_confs = ( "hosts",)
[0ea11af]	531	tclfile = "%s.%s.tcl" % (eid, tb) # sub-experiment description
	532	# command to test experiment state
	533	expinfo_exec = "/usr/testbed/bin/expinfo"
	534	# Configuration directories on the remote machine
[6679c122]	535	proj_dir = "/proj/%s/exp/%s/tmp" % (pid, eid)
	536	tarfiles_dir = "/proj/%s/tarfiles/%s" % (pid, eid)
	537	rpms_dir = "/proj/%s/rpms/%s" % (pid, eid)
[0ea11af]	538	# Regular expressions to parse the expinfo response
[6679c122]	539	state_re = re.compile("State:\s+(\w+)")
[3441fe3]	540	no_exp_re = re.compile("^No\s+such\s+experiment")
[0ea11af]	541	state = None # Experiment state parsed from expinfo
[dceeef9]	542	# The expinfo ssh command. Note the identity restriction to use only
	543	# the identity provided in the pubkey given.
[f9665d1]	544	cmd = [self.ssh_exec, '-o', 'IdentitiesOnly yes', '-o',
	545	'StrictHostKeyChecking yes', '-i',
[dceeef9]	546	self.ssh_privkey_file, "%s@%s" % (user, host),
	547	expinfo_exec, pid, eid]
[3441fe3]	548
[0ea11af]	549	# Get status
[11a08b0]	550	self.log.debug("[start_segment]: %s"% " ".join(cmd))
	551	dev_null = None
	552	try:
	553	dev_null = open("/dev/null", "a")
	554	except IOError, e:
	555	self.log.error("[start_segment]: can't open /dev/null: %s" %e)
[3441fe3]	556
[4ed10ae]	557	if self.debug:
	558	state = 'swapped'
	559	rv = 0
	560	else:
	561	status = Popen(cmd, stdout=PIPE, stderr=dev_null)
	562	for line in status.stdout:
	563	m = state_re.match(line)
	564	if m: state = m.group(1)
	565	else:
	566	m = no_exp_re.match(line)
	567	if m: state = "none"
	568	rv = status.wait()
[0ea11af]	569
[0d830de]	570	# If the experiment is not present the subcommand returns a non-zero
	571	# return value. If we successfully parsed a "none" outcome, ignore the
	572	# return code.
[3441fe3]	573	if rv != 0 and state != "none":
[6679c122]	574	raise service_error(service_error.internal,
	575	"Cannot get status of segment %s:%s/%s" % (tb, pid, eid))
[11a08b0]	576
	577	self.log.debug("[start_segment]: %s: %s" % (tb, state))
	578	self.log.info("[start_segment]:transferring experiment to %s" % tb)
[6679c122]	579
[6546868]	580	if not self.scp_file("%s/%s/%s" % (tmpdir, tb, tclfile), user, host):
[6679c122]	581	return False
[c8853fd]	582	# Clear the federation config dirs
[a8b42b5]	583	if not self.ssh_cmd(user, host,
	584	"/bin/sh -c \"'/bin/rm -rf %s'\"" % proj_dir):
	585	return False
[6679c122]	586	# Clear and create the tarfiles and rpm directories
	587	for d in (tarfiles_dir, rpms_dir):
	588	if not self.ssh_cmd(user, host,
	589	"/bin/sh -c \"'/bin/rm -rf %s/*'\"" % d):
	590	return False
	591	if not self.ssh_cmd(user, host, "mkdir -p %s" % d,
	592	"create tarfiles"):
	593	return False
	594
	595	if state == 'active':
[c8853fd]	596	# Create the federation config dirs (do not move outside the
	597	# conditional. Happens later in new expriment creation)
	598	if not self.ssh_cmd(user, host,
	599	"/bin/sh -c \"'mkdir -p %s'\"" % proj_dir):
	600	return False
[6679c122]	601	# Remote experiment is active. Modify it.
	602	for f in base_confs:
[6546868]	603	if not self.scp_file("%s/%s" % (tmpdir, f), user, host,
[6679c122]	604	"%s/%s" % (proj_dir, f)):
	605	return False
[6546868]	606	if not self.ship_configs(host, user, "%s/%s" % (tmpdir, tb),
[6679c122]	607	proj_dir):
	608	return False
[6546868]	609	if os.path.isdir("%s/tarfiles" % tmpdir):
[6679c122]	610	if not self.ship_configs(host, user,
[6546868]	611	"%s/tarfiles" % tmpdir, tarfiles_dir):
[6679c122]	612	return False
[6546868]	613	if os.path.isdir("%s/rpms" % tmpdir):
[6679c122]	614	if not self.ship_configs(host, user,
[6546868]	615	"%s/rpms" % tmpdir, tarfiles_dir):
[6679c122]	616	return False
[11a08b0]	617	self.log.info("[start_segment]: Modifying %s on %s" % (eid, tb))
[6679c122]	618	if not self.ssh_cmd(user, host,
	619	"/usr/testbed/bin/modexp -r -s -w %s %s %s" % \
	620	(pid, eid, tclfile), "modexp"):
	621	return False
	622	return True
	623	elif state == "swapped":
[c8853fd]	624	# Create the federation config dirs (do not move outside the
	625	# conditional. Happens later in new expriment creation)
	626	if not self.ssh_cmd(user, host,
	627	"/bin/sh -c \"'mkdir -p %s'\"" % proj_dir):
	628	return False
[6679c122]	629	# Remote experiment swapped out. Modify it and swap it in.
	630	for f in base_confs:
[6546868]	631	if not self.scp_file("%s/%s" % (tmpdir, f), user, host,
[6679c122]	632	"%s/%s" % (proj_dir, f)):
	633	return False
[6546868]	634	if not self.ship_configs(host, user, "%s/%s" % (tmpdir, tb),
[6679c122]	635	proj_dir):
	636	return False
[6546868]	637	if os.path.isdir("%s/tarfiles" % tmpdir):
[6679c122]	638	if not self.ship_configs(host, user,
[6546868]	639	"%s/tarfiles" % tmpdir, tarfiles_dir):
[6679c122]	640	return False
[6546868]	641	if os.path.isdir("%s/rpms" % tmpdir):
[6679c122]	642	if not self.ship_configs(host, user,
[6546868]	643	"%s/rpms" % tmpdir, tarfiles_dir):
[6679c122]	644	return False
[11a08b0]	645	self.log.info("[start_segment]: Modifying %s on %s" % (eid, tb))
[6679c122]	646	if not self.ssh_cmd(user, host,
	647	"/usr/testbed/bin/modexp -w %s %s %s" % (pid, eid, tclfile),
	648	"modexp"):
	649	return False
[11a08b0]	650	self.log.info("[start_segment]: Swapping %s in on %s" % (eid, tb))
[6679c122]	651	if not self.ssh_cmd(user, host,
	652	"/usr/testbed/bin/swapexp -w %s %s in" % (pid, eid),
[3441fe3]	653	"swapexp"):
[6679c122]	654	return False
	655	return True
	656	elif state == "none":
	657	# No remote experiment. Create one. We do this in 2 steps so we
	658	# can put the configuration files and scripts into the new
	659	# experiment directories.
	660
	661	# Tarfiles must be present for creation to work
[6546868]	662	if os.path.isdir("%s/tarfiles" % tmpdir):
[6679c122]	663	if not self.ship_configs(host, user,
[6546868]	664	"%s/tarfiles" % tmpdir, tarfiles_dir):
[6679c122]	665	return False
[6546868]	666	if os.path.isdir("%s/rpms" % tmpdir):
[6679c122]	667	if not self.ship_configs(host, user,
[6546868]	668	"%s/rpms" % tmpdir, tarfiles_dir):
[6679c122]	669	return False
[11a08b0]	670	self.log.info("[start_segment]: Creating %s on %s" % (eid, tb))
[6679c122]	671	if not self.ssh_cmd(user, host,
	672	"/usr/testbed/bin/startexp -i -f -w -p %s -e %s %s" % \
[3441fe3]	673	(pid, eid, tclfile), "startexp"):
[6679c122]	674	return False
[c8853fd]	675	# Create the federation config dirs (do not move outside the
	676	# conditional.)
	677	if not self.ssh_cmd(user, host,
	678	"/bin/sh -c \"'mkdir -p %s'\"" % proj_dir):
	679	return False
[6679c122]	680	# After startexp the per-experiment directories exist
	681	for f in base_confs:
[6546868]	682	if not self.scp_file("%s/%s" % (tmpdir, f), user, host,
[6679c122]	683	"%s/%s" % (proj_dir, f)):
	684	return False
[6546868]	685	if not self.ship_configs(host, user, "%s/%s" % (tmpdir, tb),
[6679c122]	686	proj_dir):
	687	return False
[11a08b0]	688	self.log.info("[start_segment]: Swapping %s in on %s" % (eid, tb))
[6679c122]	689	if not self.ssh_cmd(user, host,
	690	"/usr/testbed/bin/swapexp -w %s %s in" % (pid, eid),
[3441fe3]	691	"swapexp"):
[6679c122]	692	return False
	693	return True
	694	else:
[11a08b0]	695	self.log.debug("[start_segment]:unknown state %s" % state)
[6679c122]	696	return False
	697
	698	def stop_segment(self, tb, eid, tbparams):
[0ea11af]	699	"""
	700	Stop a sub experiment by calling swapexp on the federant
	701	"""
[6679c122]	702	user = tbparams[tb]['user']
[03b9b14]	703	host = "%s%s" % (tbparams[tb]['host'], tbparams[tb]['domain'])
[6679c122]	704	pid = tbparams[tb]['project']
	705
[11a08b0]	706	self.log.info("[stop_segment]: Stopping %s on %s" % (eid, tb))
[6679c122]	707	return self.ssh_cmd(user, host,
[3441fe3]	708	"/usr/testbed/bin/swapexp -w %s %s out" % (pid, eid))
[6679c122]	709
	710
	711	def generate_ssh_keys(self, dest, type="rsa" ):
	712	"""
	713	Generate a set of keys for the gateways to use to talk.
	714
	715	Keys are of type type and are stored in the required dest file.
	716	"""
	717	valid_types = ("rsa", "dsa")
	718	t = type.lower();
	719	if t not in valid_types: raise ValueError
[11a08b0]	720	cmd = [self.ssh_keygen, '-t', t, '-N', '', '-f', dest]
[3441fe3]	721
[11a08b0]	722	try:
	723	trace = open("/dev/null", "w")
	724	except IOError:
	725	raise service_error(service_error.internal,
	726	"Cannot open /dev/null??");
[3441fe3]	727
[6679c122]	728	# May raise CalledProcessError
[11a08b0]	729	self.log.debug("[generate_ssh_keys]: %s" % " ".join(cmd))
	730	rv = call(cmd, stdout=trace, stderr=trace)
[6679c122]	731	if rv != 0:
	732	raise service_error(service_error.internal,
	733	"Cannot generate nonce ssh keys. %s return code %d" \
	734	% (self.ssh_keygen, rv))
	735
[0d830de]	736	def gentopo(self, str):
[0ea11af]	737	"""
	738	Generate the topology dtat structure from the splitter's XML
	739	representation of it.
	740
	741	The topology XML looks like:
	742	<experiment>
	743	<nodes>
	744	<node><vname></vname><ips>ip1:ip2</ips></node>
	745	</nodes>
	746	<lans>
	747	<lan>
	748	<vname></vname><vnode></vnode><ip></ip>
	749	<bandwidth></bandwidth><member>node:port</member>
	750	</lan>
	751	</lans>
	752	"""
[d0ae12d]	753	class topo_parse:
[0ea11af]	754	"""
	755	Parse the topology XML and create the dats structure.
	756	"""
[d0ae12d]	757	def __init__(self):
[0ea11af]	758	# Typing of the subelements for data conversion
[0d830de]	759	self.str_subelements = ('vname', 'vnode', 'ips', 'ip', 'member')
	760	self.int_subelements = ( 'bandwidth',)
	761	self.float_subelements = ( 'delay',)
[0ea11af]	762	# The final data structure
[0d830de]	763	self.nodes = [ ]
	764	self.lans = [ ]
	765	self.topo = { \
[bcbf543]	766	'node': self.nodes,\
	767	'lan' : self.lans,\
[0d830de]	768	}
[0ea11af]	769	self.element = { } # Current element being created
	770	self.chars = "" # Last text seen
[0d830de]	771
[d0ae12d]	772	def end_element(self, name):
[0ea11af]	773	# After each sub element the contents is added to the current
	774	# element or to the appropriate list.
[0d830de]	775	if name == 'node':
	776	self.nodes.append(self.element)
	777	self.element = { }
	778	elif name == 'lan':
	779	self.lans.append(self.element)
	780	self.element = { }
	781	elif name in self.str_subelements:
	782	self.element[name] = self.chars
	783	self.chars = ""
	784	elif name in self.int_subelements:
	785	self.element[name] = int(self.chars)
	786	self.chars = ""
	787	elif name in self.float_subelements:
	788	self.element[name] = float(self.chars)
	789	self.chars = ""
[d0ae12d]	790
	791	def found_chars(self, data):
[0d830de]	792	self.chars += data.rstrip()
	793
	794
	795	tp = topo_parse();
	796	parser = xml.parsers.expat.ParserCreate()
	797	parser.EndElementHandler = tp.end_element
	798	parser.CharacterDataHandler = tp.found_chars
	799
	800	parser.Parse(str)
	801
	802	return tp.topo
	803
	804
	805	def genviz(self, topo):
	806	"""
	807	Generate the visualization the virtual topology
	808	"""
[d0ae12d]	809
	810	neato = "/usr/local/bin/neato"
	811	# These are used to parse neato output and to create the visualization
	812	# file.
	813	vis_re = re.compile('^\s"?([\w\-]+)"?\s+\[.pos="(\d+),(\d+)"')
	814	vis_fmt = "<node><name>%s</name><x>%s</x><y>%s</y><type>" + \
[0d830de]	815	"%s</type></node>"
	816
	817	try:
	818	# Node names
[bcbf543]	819	nodes = [ n['vname'] for n in topo['node'] ]
	820	topo_lans = topo['lan']
[0d830de]	821	except KeyError:
	822	raise service_error(service_error.internal, "Bad topology")
	823
	824	lans = { }
	825	links = { }
	826
	827	# Walk through the virtual topology, organizing the connections into
	828	# 2-node connections (links) and more-than-2-node connections (lans).
	829	# When a lan is created, it's added to the list of nodes (there's a
	830	# node in the visualization for the lan).
	831	for l in topo_lans:
	832	if links.has_key(l['vname']):
	833	if len(links[l['vname']]) < 2:
	834	links[l['vname']].append(l['vnode'])
	835	else:
	836	nodes.append(l['vname'])
	837	lans[l['vname']] = links[l['vname']]
	838	del links[l['vname']]
	839	lans[l['vname']].append(l['vnode'])
	840	elif lans.has_key(l['vname']):
	841	lans[l['vname']].append(l['vnode'])
	842	else:
	843	links[l['vname']] = [ l['vnode'] ]
	844
	845
	846	# Open up a temporary file for dot to turn into a visualization
[d0ae12d]	847	try:
	848	df, dotname = tempfile.mkstemp()
	849	dotfile = os.fdopen(df, 'w')
	850	except IOError:
	851	raise service_error(service_error.internal,
	852	"Failed to open file in genviz")
	853
	854	# Generate a dot/neato input file from the links, nodes and lans
	855	try:
	856	print >>dotfile, "graph G {"
[0d830de]	857	for n in nodes:
[d0ae12d]	858	print >>dotfile, '\t"%s"' % n
[0d830de]	859	for l in links.keys():
	860	print >>dotfile, '\t"%s" -- "%s"' % tuple(links[l])
	861	for l in lans.keys():
	862	for n in lans[l]:
[d0ae12d]	863	print >>dotfile, '\t "%s" -- "%s"' % (n,l)
	864	print >>dotfile, "}"
	865	dotfile.close()
[0d830de]	866	except TypeError:
	867	raise service_error(service_error.internal,
	868	"Single endpoint link in vtopo")
[d0ae12d]	869	except IOError:
	870	raise service_error(service_error.internal, "Cannot write dot file")
	871
	872	# Use dot to create a visualization
	873	dot = Popen([neato, '-Gstart=rand', '-Gepsilon=0.005', '-Gmaxiter=2000',
	874	'-Gpack=true', dotname], stdout=PIPE)
	875
[0d830de]	876	# Translate dot to vis format
	877	vis_nodes = [ ]
	878	vis = { 'node': vis_nodes }
	879	for line in dot.stdout:
	880	m = vis_re.match(line)
	881	if m:
	882	vn = m.group(1)
	883	vis_node = {'name': vn, \
[bcbf543]	884	'x': float(m.group(2)),\
	885	'y' : float(m.group(3)),\
[0d830de]	886	}
	887	if vn in links.keys() or vn in lans.keys():
	888	vis_node['type'] = 'lan'
	889	else:
	890	vis_node['type'] = 'node'
	891	vis_nodes.append(vis_node)
[d0ae12d]	892	rv = dot.wait()
[0d830de]	893
[d0ae12d]	894	os.remove(dotname)
[0d830de]	895	if rv == 0 : return vis
	896	else: return None
[d0ae12d]	897
[4064742]	898	def get_access(self, tb, nodes, user, tbparam, master, export_project,
	899	access_user):
[6679c122]	900	"""
	901	Get access to testbed through fedd and set the parameters for that tb
	902	"""
	903
[03b9b14]	904	# XXX This is needless complexity. It must vanish.
[6679c122]	905	translate_attr = {
	906	'slavenodestartcmd': 'expstart',
	907	'slaveconnectorstartcmd': 'gwstart',
	908	'masternodestartcmd': 'mexpstart',
	909	'masterconnectorstartcmd': 'mgwstart',
	910	'connectorimage': 'gwimage',
	911	'connectortype': 'gwtype',
	912	'tunnelcfg': 'tun',
[03b9b14]	913	'tunnelinterface': 'tunnelinterface',
[3441fe3]	914	'smbshare': 'smbshare',
[14a2f95]	915	'slaveconnectorcmd': 'gwcmd',
	916	'slaveconnectorcmdparams': 'gwcmdparams',
	917	'masterconnectorcmd': 'mgwcmd',
	918	'masterconnectorcmdparams': 'mgwcmdparams',
[6679c122]	919	}
	920
	921	uri = self.tbmap.get(tb, None)
	922	if not uri:
	923	raise service_error(serice_error.server_config,
	924	"Unknown testbed: %s" % tb)
	925
[721705e9]	926	# currently this lumps all users into one service access group
	927	service_keys = [ a for u in user \
	928	for a in u.get('access', []) \
	929	if a.has_key('sshPubkey')]
	930
	931	if len(service_keys) == 0:
	932	raise service_error(service_error.req,
	933	"Must have at least one SSH pubkey for services")
	934
	935
[4064742]	936	for p, u in access_user:
[d90f0fa]	937	self.log.debug(("[get_access] Attempting access from (%s, %s) " + \
	938	"to %s") % ((p or "None"), u, uri))
[4064742]	939
	940	if p:
	941	# Request with user and project specified
	942	req = {\
	943	'destinationTestbed' : { 'uri' : uri },
	944	'project': {
	945	'name': {'localname': p},
	946	'user': [ {'userID': { 'localname': u } } ],
	947	},
	948	'user': user,
	949	'allocID' : { 'localname': 'test' },
	950	'createAccess' : [ { 'sshPubkey' : self.ssh_pubkey } ],
	951	'serviceAccess' : service_keys
	952	}
	953	else:
	954	# Request with only user specified
	955	req = {\
	956	'destinationTestbed' : { 'uri' : uri },
	957	'user': [ {'userID': { 'localname': u } } ],
	958	'allocID' : { 'localname': 'test' },
	959	'createAccess' : [ { 'sshPubkey' : self.ssh_pubkey } ],
	960	'serviceAccess' : service_keys
	961	}
[721705e9]	962
[4064742]	963	if tb == master:
	964	# NB, the export_project parameter is a dict that includes
	965	# the type
	966	req['exportProject'] = export_project
	967
	968	# node resources if any
	969	if nodes != None and len(nodes) > 0:
	970	rnodes = [ ]
	971	for n in nodes:
	972	rn = { }
	973	image, hw, count = n.split(":")
	974	if image: rn['image'] = [ image ]
	975	if hw: rn['hardware'] = [ hw ]
[dceeef9]	976	if count and int(count) >0 : rn['count'] = int(count)
[4064742]	977	rnodes.append(rn)
	978	req['resources']= { }
	979	req['resources']['node'] = rnodes
[5576a47]	980
[4064742]	981	try:
[5fffd82]	982	if self.local_access.has_key(uri):
	983	# Local access call
	984	req = { 'RequestAccessRequestBody' : req }
	985	r = self.local_access[uri].RequestAccess(req,
	986	fedid(file=self.cert_file))
	987	r = { 'RequestAccessResponseBody' : r }
	988	else:
	989	r = self.call_RequestAccess(uri, req,
	990	self.cert_file, self.cert_pwd, self.trusted_certs)
[4064742]	991	except service_error, e:
	992	if e.code == service_error.access:
[d90f0fa]	993	self.log.debug("[get_access] Access denied")
[4064742]	994	r = None
	995	continue
	996	else:
	997	raise e
[6679c122]	998
[4064742]	999	if r.has_key('RequestAccessResponseBody'):
[d90f0fa]	1000	# Through to here we have a valid response, not a fault.
	1001	# Access denied is a fault, so something better or worse than
	1002	# access denied has happened.
[4064742]	1003	r = r['RequestAccessResponseBody']
[d90f0fa]	1004	self.log.debug("[get_access] Access granted")
	1005	break
[4064742]	1006	else:
	1007	raise service_error(service_error.protocol,
	1008	"Bad proxy response")
	1009
	1010	if not r:
	1011	raise service_error(service_error.access,
	1012	"Access denied by %s (%s)" % (tb, uri))
[6679c122]	1013
	1014	e = r['emulab']
	1015	p = e['project']
	1016	tbparam[tb] = {
	1017	"boss": e['boss'],
	1018	"host": e['ops'],
	1019	"domain": e['domain'],
	1020	"fs": e['fileServer'],
	1021	"eventserver": e['eventServer'],
[3441fe3]	1022	"project": unpack_id(p['name']),
[d81971a]	1023	"emulab" : e,
	1024	"allocID" : r['allocID'],
[6679c122]	1025	}
[7a8d667]	1026	# Make the testbed name be the label the user applied
	1027	p['testbed'] = {'localname': tb }
[6679c122]	1028
	1029	for u in p['user']:
[3bce099]	1030	role = u.get('role', None)
	1031	if role == 'experimentCreation':
	1032	tbparam[tb]['user'] = unpack_id(u['userID'])
	1033	break
	1034	else:
	1035	raise service_error(service_error.internal,
	1036	"No createExperimentUser from %s" %tb)
[6679c122]	1037
	1038	for a in e['fedAttr']:
	1039	if a['attribute']:
	1040	key = translate_attr.get(a['attribute'].lower(), None)
	1041	if key:
	1042	tbparam[tb][key]= a['value']
[4ed10ae]	1043
[d81971a]	1044	def release_access(self, tb, aid):
	1045	"""
	1046	Release access to testbed through fedd
	1047	"""
	1048
	1049	uri = self.tbmap.get(tb, None)
	1050	if not uri:
	1051	raise service_error(serice_error.server_config,
	1052	"Unknown testbed: %s" % tb)
	1053
[5fffd82]	1054	if self.local_access.has_key(uri):
	1055	resp = self.local_access[uri].ReleaseAccess(\
	1056	{ 'ReleaseAccessRequestBody' : {'allocID': aid},},
	1057	fedid(file=self.cert_file))
	1058	resp = { 'ReleaseAccessResponseBody': resp }
	1059	else:
	1060	resp = self.call_ReleaseAccess(uri, {'allocID': aid},
	1061	self.cert_file, self.cert_pwd, self.trusted_certs)
[d81971a]	1062
	1063	# better error coding
	1064
[f4f4117]	1065	def remote_splitter(self, uri, desc, master):
	1066
	1067	req = {
	1068	'description' : { 'ns2description': desc },
	1069	'master': master,
[2c6128f]	1070	'include_fedkit': bool(self.fedkit)
[f4f4117]	1071	}
	1072
[058f58e]	1073	r = self.call_Ns2Split(uri, req, self.cert_file, self.cert_pwd,
	1074	self.trusted_certs)
[f4f4117]	1075
	1076	if r.has_key('Ns2SplitResponseBody'):
	1077	r = r['Ns2SplitResponseBody']
	1078	if r.has_key('output'):
	1079	return r['output'].splitlines()
	1080	else:
[058f58e]	1081	raise service_error(service_error.protocol,
[f4f4117]	1082	"Bad splitter response (no output)")
	1083	else:
[058f58e]	1084	raise service_error(service_error.protocol, "Bad splitter response")
[6679c122]	1085
	1086	class current_testbed:
[0ea11af]	1087	"""
	1088	Object for collecting the current testbed description. The testbed
	1089	description is saved to a file with the local testbed variables
	1090	subsittuted line by line.
	1091	"""
[2c6128f]	1092	def __init__(self, eid, tmpdir, fedkit):
[6679c122]	1093	self.begin_testbed = re.compile("^#\s+Begin\s+Testbed\s+\((\w+)\)")
	1094	self.end_testbed = re.compile("^#\s+End\s+Testbed\s+\((\w+)\)")
	1095	self.current_testbed = None
	1096	self.testbed_file = None
	1097
	1098	self.def_expstart = \
	1099	"sudo -H /bin/sh FEDDIR/fed_bootstrap >& /tmp/federate";
	1100	self.def_mexpstart = "sudo -H FEDDIR/make_hosts FEDDIR/hosts";
	1101	self.def_gwstart = \
	1102	"sudo -H FEDDIR/fed-tun.pl -f GWCONF>& /tmp/bridge.log";
	1103	self.def_mgwstart = \
	1104	"sudo -H FEDDIR/fed-tun.pl -f GWCONF >& /tmp/bridge.log";
	1105	self.def_gwimage = "FBSD61-TUNNEL2";
	1106	self.def_gwtype = "pc";
[14a2f95]	1107	self.def_mgwcmd = '# '
	1108	self.def_mgwcmdparams = ''
	1109	self.def_gwcmd = '# '
	1110	self.def_gwcmdparams = ''
[6679c122]	1111
	1112	self.eid = eid
	1113	self.tmpdir = tmpdir
[2c6128f]	1114	self.fedkit = fedkit
[6679c122]	1115
	1116	def __call__(self, line, master, allocated, tbparams):
	1117	# Capture testbed topology descriptions
	1118	if self.current_testbed == None:
	1119	m = self.begin_testbed.match(line)
	1120	if m != None:
	1121	self.current_testbed = m.group(1)
	1122	if self.current_testbed == None:
	1123	raise service_error(service_error.req,
	1124	"Bad request format (unnamed testbed)")
	1125	allocated[self.current_testbed] = \
	1126	allocated.get(self.current_testbed,0) + 1
	1127	tb_dir = "%s/%s" % (self.tmpdir, self.current_testbed)
	1128	if not os.path.exists(tb_dir):
	1129	try:
	1130	os.mkdir(tb_dir)
	1131	except IOError:
	1132	raise service_error(service_error.internal,
	1133	"Cannot create %s" % tb_dir)
	1134	try:
	1135	self.testbed_file = open("%s/%s.%s.tcl" %
	1136	(tb_dir, self.eid, self.current_testbed), 'w')
	1137	except IOError:
	1138	self.testbed_file = None
	1139	return True
	1140	else: return False
	1141	else:
	1142	m = self.end_testbed.match(line)
	1143	if m != None:
	1144	if m.group(1) != self.current_testbed:
	1145	raise service_error(service_error.internal,
	1146	"Mismatched testbed markers!?")
	1147	if self.testbed_file != None:
	1148	self.testbed_file.close()
	1149	self.testbed_file = None
	1150	self.current_testbed = None
	1151	elif self.testbed_file:
	1152	# Substitute variables and put the line into the local
	1153	# testbed file.
	1154	gwtype = tbparams[self.current_testbed].get('gwtype',
	1155	self.def_gwtype)
	1156	gwimage = tbparams[self.current_testbed].get('gwimage',
	1157	self.def_gwimage)
	1158	mgwstart = tbparams[self.current_testbed].get('mgwstart',
	1159	self.def_mgwstart)
	1160	mexpstart = tbparams[self.current_testbed].get('mexpstart',
	1161	self.def_mexpstart)
	1162	gwstart = tbparams[self.current_testbed].get('gwstart',
	1163	self.def_gwstart)
	1164	expstart = tbparams[self.current_testbed].get('expstart',
	1165	self.def_expstart)
	1166	project = tbparams[self.current_testbed].get('project')
[14a2f95]	1167	gwcmd = tbparams[self.current_testbed].get('gwcmd',
	1168	self.def_gwcmd)
	1169	gwcmdparams = tbparams[self.current_testbed].get(\
	1170	'gwcmdparams', self.def_gwcmdparams)
	1171	mgwcmd = tbparams[self.current_testbed].get('mgwcmd',
	1172	self.def_gwcmd)
	1173	mgwcmdparams = tbparams[self.current_testbed].get(\
	1174	'mgwcmdparams', self.def_gwcmdparams)
[6679c122]	1175	line = re.sub("GWTYPE", gwtype, line)
	1176	line = re.sub("GWIMAGE", gwimage, line)
	1177	if self.current_testbed == master:
	1178	line = re.sub("GWSTART", mgwstart, line)
	1179	line = re.sub("EXPSTART", mexpstart, line)
[14a2f95]	1180	# NB GWCMDPARAMS is a prefix of GWCMD, so expand first
	1181	line = re.sub("GWCMDPARAMS", mgwcmdparams, line)
	1182	line = re.sub("(#\s*)?GWCMD", mgwcmd, line)
[6679c122]	1183	else:
	1184	line = re.sub("GWSTART", gwstart, line)
	1185	line = re.sub("EXPSTART", expstart, line)
[14a2f95]	1186	# NB GWCMDPARAMS is a prefix of GWCMD, so expand first
	1187	line = re.sub("GWCMDPARAMS", gwcmdparams, line)
	1188	line = re.sub("(#\s*)?GWCMD", gwcmd, line)
[6679c122]	1189	line = re.sub("GWCONF", "FEDDIR`hostname`.gw.conf", line)
	1190	line = re.sub("PROJDIR", "/proj/%s/" % project, line)
	1191	line = re.sub("EID", self.eid, line)
	1192	line = re.sub("FEDDIR", "/proj/%s/exp/%s/tmp/" % \
	1193	(project, self.eid), line)
[2c6128f]	1194	if self.fedkit:
	1195	line = re.sub("FEDKIT", os.path.basename(self.fedkit),
	1196	line)
[6679c122]	1197	print >>self.testbed_file, line
	1198	return True
	1199
	1200	class allbeds:
[0ea11af]	1201	"""
	1202	Process the Allbeds section. Get access to each federant and save the
	1203	parameters in tbparams
	1204	"""
[6679c122]	1205	def __init__(self, get_access):
	1206	self.begin_allbeds = re.compile("^#\s+Begin\s+Allbeds")
	1207	self.end_allbeds = re.compile("^#\s+End\s+Allbeds")
	1208	self.in_allbeds = False
	1209	self.get_access = get_access
	1210
[4064742]	1211	def __call__(self, line, user, tbparams, master, export_project,
	1212	access_user):
[6679c122]	1213	# Testbed access parameters
	1214	if not self.in_allbeds:
	1215	if self.begin_allbeds.match(line):
	1216	self.in_allbeds = True
	1217	return True
	1218	else:
	1219	return False
	1220	else:
	1221	if self.end_allbeds.match(line):
	1222	self.in_allbeds = False
	1223	else:
	1224	nodes = line.split('\|')
	1225	tb = nodes.pop(0)
[5576a47]	1226	self.get_access(tb, nodes, user, tbparams, master,
[4064742]	1227	export_project, access_user)
[6679c122]	1228	return True
	1229
	1230	class gateways:
[3441fe3]	1231	def __init__(self, eid, master, tmpdir, gw_pubkey,
[2c6128f]	1232	gw_secretkey, copy_file, fedkit):
[6679c122]	1233	self.begin_gateways = \
	1234	re.compile("^#\s+Begin\s+gateways\s+\((\w+)\)")
	1235	self.end_gateways = re.compile("^#\s+End\s+gateways\s+\((\w+)\)")
	1236	self.current_gateways = None
	1237	self.control_gateway = None
	1238	self.active_end = { }
	1239
	1240	self.eid = eid
	1241	self.master = master
	1242	self.tmpdir = tmpdir
	1243	self.gw_pubkey_base = gw_pubkey
	1244	self.gw_secretkey_base = gw_secretkey
	1245
	1246	self.copy_file = copy_file
[2c6128f]	1247	self.fedkit = fedkit
[6679c122]	1248
	1249
	1250	def gateway_conf_file(self, gw, master, eid, pubkey, privkey,
	1251	active_end, tbparams, dtb, myname, desthost, type):
	1252	"""
[1af38d6]	1253	Produce a gateway configuration file from a gateways line.
[6679c122]	1254	"""
	1255
	1256	sproject = tbparams[gw].get('project', 'project')
	1257	dproject = tbparams[dtb].get('project', 'project')
	1258	sdomain = ".%s.%s%s" % (eid, sproject,
	1259	tbparams[gw].get('domain', ".example.com"))
	1260	ddomain = ".%s.%s%s" % (eid, dproject,
	1261	tbparams[dtb].get('domain', ".example.com"))
	1262	boss = tbparams[master].get('boss', "boss")
	1263	fs = tbparams[master].get('fs', "fs")
	1264	event_server = "%s%s" % \
[27b6aea]	1265	(tbparams[gw].get('eventserver', "event_server"),
	1266	tbparams[gw].get('domain', "example.com"))
[6679c122]	1267	remote_event_server = "%s%s" % \
	1268	(tbparams[dtb].get('eventserver', "event_server"),
	1269	tbparams[dtb].get('domain', "example.com"))
[27b6aea]	1270	seer_control = "%s%s" % \
	1271	(tbparams[gw].get('control', "control"), sdomain)
[03b9b14]	1272	tunnel_iface = tbparams[gw].get("tunnelinterface", None)
[2c6128f]	1273
	1274	if self.fedkit:
	1275	remote_script_dir = "/usr/local/federation/bin"
	1276	local_script_dir = "/usr/local/federation/bin"
	1277	else:
	1278	remote_script_dir = "/proj/%s/exp/%s/tmp" % ( dproject, eid)
	1279	local_script_dir = "/proj/%s/exp/%s/tmp" % ( sproject, eid)
	1280
	1281	local_key_dir = "/proj/%s/exp/%s/tmp" % ( sproject, eid)
	1282	remote_conf_dir = "/proj/%s/exp/%s/tmp" % ( dproject, eid)
[6679c122]	1283	tunnel_cfg = tbparams[gw].get("tun", "false")
	1284
	1285	conf_file = "%s%s.gw.conf" % (myname, sdomain)
	1286	remote_conf_file = "%s%s.gw.conf" % (desthost, ddomain)
	1287
	1288	# translate to lower case so the `hostname` hack for specifying
	1289	# configuration files works.
	1290	conf_file = conf_file.lower();
	1291	remote_conf_file = remote_conf_file.lower();
	1292
	1293	if dtb == master:
	1294	active = "false"
	1295	elif gw == master:
	1296	active = "true"
[8dfc06c]	1297	elif active_end.has_key('%s-%s' % (dtb, gw)):
[6679c122]	1298	active = "false"
	1299	else:
	1300	active_end['%s-%s' % (gw, dtb)] = 1
	1301	active = "true"
	1302
	1303	gwconfig = open("%s/%s/%s" % (self.tmpdir, gw, conf_file), "w")
	1304	print >>gwconfig, "Active: %s" % active
	1305	print >>gwconfig, "TunnelCfg: %s" % tunnel_cfg
[03b9b14]	1306	if tunnel_iface:
	1307	print >>gwconfig, "Interface: %s" % tunnel_iface
[6679c122]	1308	print >>gwconfig, "BossName: %s" % boss
	1309	print >>gwconfig, "FsName: %s" % fs
	1310	print >>gwconfig, "EventServerName: %s" % event_server
	1311	print >>gwconfig, "RemoteEventServerName: %s" % remote_event_server
[27b6aea]	1312	print >>gwconfig, "SeerControl: %s" % seer_control
[6679c122]	1313	print >>gwconfig, "Type: %s" % type
	1314	print >>gwconfig, "RemoteScriptDir: %s" % remote_script_dir
	1315	print >>gwconfig, "EventRepeater: %s/fed_evrepeater" % \
	1316	local_script_dir
	1317	print >>gwconfig, "RemoteExperiment: %s/%s" % (dproject, eid)
	1318	print >>gwconfig, "LocalExperiment: %s/%s" % (sproject, eid)
	1319	print >>gwconfig, "RemoteConfigFile: %s/%s" % \
[2c6128f]	1320	(remote_conf_dir, remote_conf_file)
[6679c122]	1321	print >>gwconfig, "Peer: %s%s" % (desthost, ddomain)
[2c6128f]	1322	print >>gwconfig, "Pubkeys: %s/%s" % (local_key_dir, pubkey)
	1323	print >>gwconfig, "Privkeys: %s/%s" % (local_key_dir, privkey)
[6679c122]	1324	gwconfig.close()
	1325
	1326	return active == "true"
	1327
	1328	def __call__(self, line, allocated, tbparams):
	1329	# Process gateways
	1330	if not self.current_gateways:
	1331	m = self.begin_gateways.match(line)
	1332	if m:
	1333	self.current_gateways = m.group(1)
	1334	if allocated.has_key(self.current_gateways):
	1335	# This test should always succeed
	1336	tb_dir = "%s/%s" % (self.tmpdir, self.current_gateways)
	1337	if not os.path.exists(tb_dir):
	1338	try:
	1339	os.mkdir(tb_dir)
	1340	except IOError:
	1341	raise service_error(service_error.internal,
	1342	"Cannot create %s" % tb_dir)
	1343	else:
	1344	# XXX
[11a08b0]	1345	self.log.error("[gateways]: Ignoring gateways for " + \
	1346	"unknown testbed %s" % self.current_gateways)
[6679c122]	1347	self.current_gateways = None
	1348	return True
	1349	else:
	1350	return False
	1351	else:
	1352	m = self.end_gateways.match(line)
	1353	if m :
	1354	if m.group(1) != self.current_gateways:
	1355	raise service_error(service_error.internal,
	1356	"Mismatched gateway markers!?")
	1357	if self.control_gateway:
	1358	try:
	1359	cc = open("%s/%s/client.conf" %
	1360	(self.tmpdir, self.current_gateways), 'w')
	1361	print >>cc, "ControlGateway: %s" % \
	1362	self.control_gateway
[3441fe3]	1363	if tbparams[self.master].has_key('smbshare'):
	1364	print >>cc, "SMBSHare: %s" % \
	1365	tbparams[self.master]['smbshare']
[6679c122]	1366	print >>cc, "ProjectUser: %s" % \
[3441fe3]	1367	tbparams[self.master]['user']
[6679c122]	1368	print >>cc, "ProjectName: %s" % \
	1369	tbparams[self.master]['project']
[7c3d547]	1370	print >>cc, "ExperimentID: %s/%s" % \
	1371	( tbparams[self.master]['project'], \
	1372	self.eid )
[6679c122]	1373	cc.close()
	1374	except IOError:
	1375	raise service_error(service_error.internal,
	1376	"Error creating client config")
[7c3d547]	1377	# XXX: This seer specific file should disappear
[291423b]	1378	try:
	1379	cc = open("%s/%s/seer.conf" %
[a8b42b5]	1380	(self.tmpdir, self.current_gateways),
	1381	'w')
	1382	if self.current_gateways != self.master:
	1383	print >>cc, "ControlNode: %s" % \
	1384	self.control_gateway
[291423b]	1385	print >>cc, "ExperimentID: %s/%s" % \
	1386	( tbparams[self.master]['project'], \
	1387	self.eid )
	1388	cc.close()
	1389	except IOError:
	1390	raise service_error(service_error.internal,
	1391	"Error creating seer config")
[6679c122]	1392	else:
[11a08b0]	1393	debug.error("[gateways]: No control gateway for %s" %\
	1394	self.current_gateways)
[6679c122]	1395	self.current_gateways = None
	1396	else:
	1397	dtb, myname, desthost, type = line.split(" ")
	1398
	1399	if type == "control" or type == "both":
	1400	self.control_gateway = "%s.%s.%s%s" % (myname,
	1401	self.eid,
	1402	tbparams[self.current_gateways]['project'],
	1403	tbparams[self.current_gateways]['domain'])
	1404	try:
	1405	active = self.gateway_conf_file(self.current_gateways,
	1406	self.master, self.eid, self.gw_pubkey_base,
	1407	self.gw_secretkey_base,
	1408	self.active_end, tbparams, dtb, myname,
	1409	desthost, type)
	1410	except IOError, e:
	1411	raise service_error(service_error.internal,
	1412	"Failed to write config file for %s" % \
	1413	self.current_gateway)
	1414
	1415	gw_pubkey = "%s/keys/%s" % \
	1416	(self.tmpdir, self.gw_pubkey_base)
	1417	gw_secretkey = "%s/keys/%s" % \
	1418	(self.tmpdir, self.gw_secretkey_base)
	1419
	1420	pkfile = "%s/%s/%s" % \
	1421	( self.tmpdir, self.current_gateways,
	1422	self.gw_pubkey_base)
	1423	skfile = "%s/%s/%s" % \
	1424	( self.tmpdir, self.current_gateways,
	1425	self.gw_secretkey_base)
	1426
	1427	if not os.path.exists(pkfile):
	1428	try:
	1429	self.copy_file(gw_pubkey, pkfile)
	1430	except IOError:
	1431	service_error(service_error.internal,
	1432	"Failed to copy pubkey file")
	1433
	1434	if active and not os.path.exists(skfile):
	1435	try:
	1436	self.copy_file(gw_secretkey, skfile)
	1437	except IOError:
	1438	service_error(service_error.internal,
	1439	"Failed to copy secretkey file")
	1440	return True
	1441
	1442	class shunt_to_file:
[0ea11af]	1443	"""
	1444	Simple class to write data between two regexps to a file.
	1445	"""
[6679c122]	1446	def __init__(self, begin, end, filename):
[0ea11af]	1447	"""
	1448	Begin shunting on a match of begin, stop on end, send data to
	1449	filename.
	1450	"""
[6679c122]	1451	self.begin = re.compile(begin)
	1452	self.end = re.compile(end)
	1453	self.in_shunt = False
	1454	self.file = None
	1455	self.filename = filename
	1456
	1457	def __call__(self, line):
[0ea11af]	1458	"""
	1459	Call this on each line in the input that may be shunted.
	1460	"""
[6679c122]	1461	if not self.in_shunt:
	1462	if self.begin.match(line):
	1463	self.in_shunt = True
	1464	try:
	1465	self.file = open(self.filename, "w")
	1466	except:
	1467	self.file = None
	1468	raise
	1469	return True
	1470	else:
	1471	return False
	1472	else:
	1473	if self.end.match(line):
	1474	if self.file:
	1475	self.file.close()
	1476	self.file = None
	1477	self.in_shunt = False
	1478	else:
	1479	if self.file:
	1480	print >>self.file, line
	1481	return True
	1482
	1483	class shunt_to_list:
[0ea11af]	1484	"""
	1485	Same interface as shunt_to_file. Data collected in self.list, one list
	1486	element per line.
	1487	"""
[6679c122]	1488	def __init__(self, begin, end):
	1489	self.begin = re.compile(begin)
	1490	self.end = re.compile(end)
	1491	self.in_shunt = False
	1492	self.list = [ ]
	1493
	1494	def __call__(self, line):
	1495	if not self.in_shunt:
	1496	if self.begin.match(line):
	1497	self.in_shunt = True
	1498	return True
	1499	else:
	1500	return False
	1501	else:
	1502	if self.end.match(line):
	1503	self.in_shunt = False
	1504	else:
	1505	self.list.append(line)
	1506	return True
	1507
[0d830de]	1508	class shunt_to_string:
[0ea11af]	1509	"""
	1510	Same interface as shunt_to_file. Data collected in self.str, all in
	1511	one string.
	1512	"""
[0d830de]	1513	def __init__(self, begin, end):
	1514	self.begin = re.compile(begin)
	1515	self.end = re.compile(end)
	1516	self.in_shunt = False
	1517	self.str = ""
	1518
	1519	def __call__(self, line):
	1520	if not self.in_shunt:
	1521	if self.begin.match(line):
	1522	self.in_shunt = True
	1523	return True
	1524	else:
	1525	return False
	1526	else:
	1527	if self.end.match(line):
	1528	self.in_shunt = False
	1529	else:
	1530	self.str += line
	1531	return True
	1532
[6679c122]	1533	def create_experiment(self, req, fid):
[0ea11af]	1534	"""
	1535	The external interface to experiment creation called from the
	1536	dispatcher.
	1537
	1538	Creates a working directory, splits the incoming description using the
	1539	splitter script and parses out the avrious subsections using the
	1540	lcasses above. Once each sub-experiment is created, use pooled threads
	1541	to instantiate them and start it all up.
	1542	"""
[4064742]	1543
	1544	if not self.auth.check_attribute(fid, 'create'):
	1545	raise service_error(service_error.access, "Create access denied")
	1546
[6546868]	1547	try:
	1548	tmpdir = tempfile.mkdtemp(prefix="split-")
	1549	except IOError:
	1550	raise service_error(service_error.internal, "Cannot create tmp dir")
	1551
[6679c122]	1552	gw_pubkey_base = "fed.%s.pub" % self.ssh_type
	1553	gw_secretkey_base = "fed.%s" % self.ssh_type
[6546868]	1554	gw_pubkey = tmpdir + "/keys/" + gw_pubkey_base
	1555	gw_secretkey = tmpdir + "/keys/" + gw_secretkey_base
	1556	tclfile = tmpdir + "/experiment.tcl"
[6679c122]	1557	tbparams = { }
[4064742]	1558	try:
	1559	access_user = self.accessdb[fid]
	1560	except KeyError:
	1561	raise service_error(service_error.internal,
	1562	"Access map and authorizer out of sync in " + \
	1563	"create_experiment for fedid %s" % fid)
[6679c122]	1564
	1565	pid = "dummy"
	1566	gid = "dummy"
	1567	# XXX
	1568	fail_soft = False
	1569
	1570	try:
[6546868]	1571	os.mkdir(tmpdir+"/keys")
[6679c122]	1572	except OSError:
	1573	raise service_error(service_error.internal,
	1574	"Can't make temporary dir")
	1575
[b234bb9]	1576	req = req.get('CreateRequestBody', None)
	1577	if not req:
[2d5c8b6]	1578	raise service_error(service_error.req,
[b234bb9]	1579	"Bad request format (no CreateRequestBody)")
[6679c122]	1580	# The tcl parser needs to read a file so put the content into that file
[3925b50]	1581	descr=req.get('experimentdescription', None)
	1582	if descr:
	1583	file_content=descr.get('ns2description', None)
	1584	if file_content:
	1585	try:
	1586	f = open(tclfile, 'w')
	1587	f.write(file_content)
	1588	f.close()
	1589	except IOError:
	1590	raise service_error(service_error.internal,
	1591	"Cannot write temp experiment description")
	1592	else:
	1593	raise service_error(service_error.req,
	1594	"Only ns2descriptions supported")
[6679c122]	1595	else:
	1596	raise service_error(service_error.req, "No experiment description")
	1597
[e40c7ee]	1598	if req.has_key('experimentID') and \
	1599	req['experimentID'].has_key('localname'):
	1600	eid = req['experimentID']['localname']
[a97394b]	1601	self.state_lock.acquire()
[e40c7ee]	1602	while (self.state.has_key(eid)):
	1603	eid += random.choice(string.ascii_letters)
[0ea11af]	1604	# To avoid another thread picking this localname
[a97394b]	1605	self.state[eid] = "placeholder"
	1606	self.state_lock.release()
[e40c7ee]	1607	else:
	1608	eid = self.exp_stem
	1609	for i in range(0,5):
	1610	eid += random.choice(string.ascii_letters)
[a97394b]	1611	self.state_lock.acquire()
[e40c7ee]	1612	while (self.state.has_key(eid)):
	1613	eid = self.exp_stem
	1614	for i in range(0,5):
	1615	eid += random.choice(string.ascii_letters)
[0ea11af]	1616	# To avoid another thread picking this localname
[a97394b]	1617	self.state[eid] = "placeholder"
	1618	self.state_lock.release()
[e40c7ee]	1619
[c7e40f5]	1620	try:
[05191a6]	1621	# This catches exceptions to clear the placeholder if necessary
	1622	try:
	1623	self.generate_ssh_keys(gw_secretkey, self.ssh_type)
	1624	except ValueError:
	1625	raise service_error(service_error.server_config,
	1626	"Bad key type (%s)" % self.ssh_type)
	1627
	1628	user = req.get('user', None)
	1629	if user == None:
	1630	raise service_error(service_error.req, "No user")
	1631
	1632	master = req.get('master', None)
	1633	if not master:
	1634	raise service_error(service_error.req,
	1635	"No master testbed label")
	1636	export_project = req.get('exportProject', None)
	1637	if not export_project:
	1638	raise service_error(service_error.req, "No export project")
	1639
	1640	if self.splitter_url:
	1641	self.log.debug("Calling remote splitter at %s" % \
	1642	self.splitter_url)
	1643	split_data = self.remote_splitter(self.splitter_url,
	1644	file_content, master)
	1645	else:
	1646	tclcmd = [self.tclsh, self.tcl_splitter, '-s', '-x',
	1647	str(self.muxmax), '-m', master]
	1648
	1649	if self.fedkit:
	1650	tclcmd.append('-k')
	1651
	1652	tclcmd.extend([pid, gid, eid, tclfile])
	1653
	1654	self.log.debug("running local splitter %s", " ".join(tclcmd))
	1655	tclparser = Popen(tclcmd, stdout=PIPE)
	1656	split_data = tclparser.stdout
	1657
	1658	allocated = { } # Testbeds we can access
	1659	started = { } # Testbeds where a sub-experiment started
	1660	# successfully
	1661
	1662	# Objects to parse the splitter output (defined above)
	1663	parse_current_testbed = self.current_testbed(eid, tmpdir,
	1664	self.fedkit)
	1665	parse_allbeds = self.allbeds(self.get_access)
	1666	parse_gateways = self.gateways(eid, master, tmpdir,
	1667	gw_pubkey_base, gw_secretkey_base, self.copy_file,
	1668	self.fedkit)
	1669	parse_vtopo = self.shunt_to_string("^#\s+Begin\s+Vtopo",
	1670	"^#\s+End\s+Vtopo")
	1671	parse_hostnames = self.shunt_to_file("^#\s+Begin\s+hostnames",
	1672	"^#\s+End\s+hostnames", tmpdir + "/hosts")
	1673	parse_tarfiles = self.shunt_to_list("^#\s+Begin\s+tarfiles",
	1674	"^#\s+End\s+tarfiles")
	1675	parse_rpms = self.shunt_to_list("^#\s+Begin\s+rpms",
	1676	"^#\s+End\s+rpms")
	1677
[c7e40f5]	1678	# Working on the split data
	1679	for line in split_data:
	1680	line = line.rstrip()
	1681	if parse_current_testbed(line, master, allocated, tbparams):
	1682	continue
	1683	elif parse_allbeds(line, user, tbparams, master, export_project,
	1684	access_user):
	1685	continue
	1686	elif parse_gateways(line, allocated, tbparams):
	1687	continue
	1688	elif parse_vtopo(line):
	1689	continue
	1690	elif parse_hostnames(line):
	1691	continue
	1692	elif parse_tarfiles(line):
	1693	continue
	1694	elif parse_rpms(line):
	1695	continue
	1696	else:
	1697	raise service_error(service_error.internal,
	1698	"Bad tcl parse? %s" % line)
[05191a6]	1699	# Virtual topology and visualization
	1700	vtopo = self.gentopo(parse_vtopo.str)
	1701	if not vtopo:
	1702	raise service_error(service_error.internal,
	1703	"Failed to generate virtual topology")
	1704
	1705	vis = self.genviz(vtopo)
	1706	if not vis:
	1707	raise service_error(service_error.internal,
	1708	"Failed to generate visualization")
	1709
	1710	# save federant information
	1711	for k in allocated.keys():
	1712	tbparams[k]['federant'] = {\
	1713	'name': [ { 'localname' : eid} ],\
	1714	'emulab': tbparams[k]['emulab'],\
	1715	'allocID' : tbparams[k]['allocID'],\
	1716	'master' : k == master,\
	1717	}
	1718
	1719
	1720	# Copy tarfiles and rpms needed at remote sites into a staging area
	1721	try:
	1722	if self.fedkit:
	1723	parse_tarfiles.list.append(self.fedkit)
	1724	for t in parse_tarfiles.list:
	1725	if not os.path.exists("%s/tarfiles" % tmpdir):
	1726	os.mkdir("%s/tarfiles" % tmpdir)
	1727	self.copy_file(t, "%s/tarfiles/%s" % \
	1728	(tmpdir, os.path.basename(t)))
	1729	for r in parse_rpms.list:
	1730	if not os.path.exists("%s/rpms" % tmpdir):
	1731	os.mkdir("%s/rpms" % tmpdir)
	1732	self.copy_file(r, "%s/rpms/%s" % \
	1733	(tmpdir, os.path.basename(r)))
	1734	except IOError, e:
	1735	raise service_error(service_error.internal,
	1736	"Cannot stage tarfile/rpm: %s" % e.strerror)
	1737
[c7e40f5]	1738	except service_error, e:
	1739	# If something goes wrong in the parse (usually an access error)
[05191a6]	1740	# clear the placeholder state. From here on out the code delays
	1741	# exceptions.
[c7e40f5]	1742	self.state_lock.acquire()
	1743	del self.state[eid]
	1744	self.state_lock.release()
	1745	raise e
[6679c122]	1746
[8bc5754]	1747	thread_pool = self.thread_pool(self.nthreads)
[1af38d6]	1748	threads = [ ]
	1749
	1750	for tb in [ k for k in allocated.keys() if k != master]:
	1751	# Create and start a thread to start the segment, and save it to
	1752	# get the return value later
[8bc5754]	1753	thread_pool.wait_for_slot()
[1af38d6]	1754	t = self.pooled_thread(target=self.start_segment,
	1755	args=(tb, eid, tbparams, tmpdir, 0), name=tb,
[8bc5754]	1756	pdata=thread_pool, trace_file=self.trace_file)
[1af38d6]	1757	threads.append(t)
	1758	t.start()
	1759
[8bc5754]	1760	# Wait until all finish
	1761	thread_pool.wait_for_all_done()
[1af38d6]	1762
	1763	# If none failed, start the master
	1764	failed = [ t.getName() for t in threads if not t.rv ]
	1765
	1766	if len(failed) == 0:
	1767	if not self.start_segment(master, eid, tbparams, tmpdir):
	1768	failed.append(master)
	1769
[a97394b]	1770	succeeded = [tb for tb in allocated.keys() if tb not in failed]
	1771	# If one failed clean up, unless fail_soft is set
	1772	if failed:
	1773	if not fail_soft:
[8bc5754]	1774	thread_pool.clear()
[3441fe3]	1775	for tb in succeeded:
[8bc5754]	1776	# Create and start a thread to stop the segment
	1777	thread_pool.wait_for_slot()
	1778	t = self.pooled_thread(target=self.stop_segment,
	1779	args=(tb, eid, tbparams), name=tb,
	1780	pdata=thread_pool, trace_file=self.trace_file)
	1781	t.start()
	1782	# Wait until all finish
	1783	thread_pool.wait_for_all_done()
	1784
[03b9b14]	1785	# release the allocations
	1786	for tb in tbparams.keys():
[7ca25b3]	1787	self.release_access(tb, tbparams[tb]['allocID'])
[a97394b]	1788	# Remove the placeholder
	1789	self.state_lock.acquire()
	1790	del self.state[eid]
	1791	self.state_lock.release()
	1792
[3441fe3]	1793	raise service_error(service_error.federant,
	1794	"Swap in failed on %s" % ",".join(failed))
[6679c122]	1795	else:
[11a08b0]	1796	self.log.info("[start_segment]: Experiment %s started" % eid)
[6679c122]	1797
[4fc2250]	1798	# Generate an ID for the experiment (slice) and a certificate that the
	1799	# allocator can use to prove they own it. We'll ship it back through
	1800	# the encrypted connection.
[11a08b0]	1801	(expid, expcert) = generate_fedid("test", dir=tmpdir, log=self.log)
[4fc2250]	1802
[11a08b0]	1803	self.log.debug("[start_experiment]: removing %s" % tmpdir)
[4fc2250]	1804
	1805	# Walk up tmpdir, deleting as we go
	1806	for path, dirs, files in os.walk(tmpdir, topdown=False):
	1807	for f in files:
	1808	os.remove(os.path.join(path, f))
	1809	for d in dirs:
	1810	os.rmdir(os.path.join(path, d))
	1811	os.rmdir(tmpdir)
[987aaa1]	1812
[f8582c9]	1813	# The deepcopy prevents the allocation ID and other binaries from being
	1814	# translated into other formats
	1815	resp = { 'federant' : [ copy.deepcopy(tbparams[tb]['federant']) \
[3441fe3]	1816	for tb in tbparams.keys() \
[e40c7ee]	1817	if tbparams[tb].has_key('federant') ],\
[c52c48d]	1818	'vtopo': vtopo,\
[4fc2250]	1819	'vis' : vis,
[e40c7ee]	1820	'experimentID' : [\
[c52c48d]	1821	{ 'fedid': copy.copy(expid) }, \
	1822	{ 'localname': eid },\
[e40c7ee]	1823	],\
[4fc2250]	1824	'experimentAccess': { 'X509' : expcert },\
	1825	}
[ca24951]	1826	# remove the allocationID info from each federant
	1827	for f in resp['federant']:
	1828	if f.has_key('allocID'): del f['allocID']
[a97394b]	1829
[0ea11af]	1830	# Insert the experiment into our state and update the disk copy
[a97394b]	1831	self.state_lock.acquire()
[c52c48d]	1832	self.state[expid] = { 'federant' : [ tbparams[tb]['federant'] \
	1833	for tb in tbparams.keys() \
	1834	if tbparams[tb].has_key('federant') ],\
	1835	'vtopo': vtopo,\
	1836	'vis' : vis,
[4064742]	1837	'owner': fid,
[c52c48d]	1838	'experimentID' : [\
	1839	{ 'fedid': expid }, { 'localname': eid },\
	1840	],\
	1841	}
[e40c7ee]	1842	self.state[eid] = self.state[expid]
[eee2b2e]	1843	if self.state_filename: self.write_state()
[a97394b]	1844	self.state_lock.release()
	1845
[4064742]	1846	self.auth.set_attribute(fid, expid)
	1847	self.auth.set_attribute(expid, expid)
	1848
[a97394b]	1849	if not failed:
	1850	return resp
	1851	else:
	1852	raise service_error(service_error.partial, \
	1853	"Partial swap in on %s" % ",".join(succeeded))
	1854
[4064742]	1855	def check_experiment_access(self, fid, key):
	1856	"""
	1857	Confirm that the fid has access to the experiment. Though a request
	1858	may be made in terms of a local name, the access attribute is always
	1859	the experiment's fedid.
	1860	"""
	1861	if not isinstance(key, fedid):
	1862	self.state_lock.acquire()
	1863	if self.state.has_key(key):
	1864	try:
	1865	kl = [ f['fedid'] for f in self.state[key]['experimentID']\
	1866	if f.has_key('fedid') ]
	1867	except KeyError:
	1868	self.state_lock.release()
	1869	raise service_error(service_error.internal,
	1870	"No fedid for experiment %s when checking " +\
	1871	"access(!?)" % key)
	1872	if len(kl) == 1:
	1873	key = kl[0]
	1874	else:
	1875	self.state_lock.release()
	1876	raise service_error(service_error.internal,
	1877	"multiple fedids for experiment %s when " +\
	1878	"checking access(!?)" % key)
	1879	else:
	1880	self.state_lock.release()
	1881	raise service_error(service_error.access, "Access Denied")
	1882	self.state_lock.release()
	1883
	1884	if self.auth.check_attribute(fid, key):
	1885	return True
	1886	else:
	1887	raise service_error(service_error.access, "Access Denied")
	1888
	1889
[987aaa1]	1890
	1891	def get_vtopo(self, req, fid):
[0ea11af]	1892	"""
	1893	Return the stored virtual topology for this experiment
	1894	"""
[a97394b]	1895	rv = None
[987aaa1]	1896
	1897	req = req.get('VtopoRequestBody', None)
	1898	if not req:
	1899	raise service_error(service_error.req,
	1900	"Bad request format (no VtopoRequestBody)")
[e40c7ee]	1901	exp = req.get('experiment', None)
	1902	if exp:
	1903	if exp.has_key('fedid'):
[f8582c9]	1904	key = exp['fedid']
[e40c7ee]	1905	keytype = "fedid"
	1906	elif exp.has_key('localname'):
	1907	key = exp['localname']
	1908	keytype = "localname"
	1909	else:
	1910	raise service_error(service_error.req, "Unknown lookup type")
[987aaa1]	1911	else:
[e40c7ee]	1912	raise service_error(service_error.req, "No request?")
[987aaa1]	1913
[4064742]	1914	self.check_experiment_access(fid, key)
	1915
[a97394b]	1916	self.state_lock.acquire()
[e40c7ee]	1917	if self.state.has_key(key):
[a97394b]	1918	rv = { 'experiment' : {keytype: key },\
[e40c7ee]	1919	'vtopo': self.state[key]['vtopo'],\
[a97394b]	1920	}
	1921	self.state_lock.release()
	1922
	1923	if rv: return rv
	1924	else: raise service_error(service_error.req, "No such experiment")
[987aaa1]	1925
	1926	def get_vis(self, req, fid):
[0ea11af]	1927	"""
	1928	Return the stored visualization for this experiment
	1929	"""
[a97394b]	1930	rv = None
[987aaa1]	1931
	1932	req = req.get('VisRequestBody', None)
	1933	if not req:
	1934	raise service_error(service_error.req,
	1935	"Bad request format (no VisRequestBody)")
[e40c7ee]	1936	exp = req.get('experiment', None)
	1937	if exp:
	1938	if exp.has_key('fedid'):
[f8582c9]	1939	key = exp['fedid']
[e40c7ee]	1940	keytype = "fedid"
	1941	elif exp.has_key('localname'):
	1942	key = exp['localname']
	1943	keytype = "localname"
	1944	else:
	1945	raise service_error(service_error.req, "Unknown lookup type")
[987aaa1]	1946	else:
[e40c7ee]	1947	raise service_error(service_error.req, "No request?")
[987aaa1]	1948
[4064742]	1949	self.check_experiment_access(fid, key)
	1950
[a97394b]	1951	self.state_lock.acquire()
[e40c7ee]	1952	if self.state.has_key(key):
[a97394b]	1953	rv = { 'experiment' : {keytype: key },\
[e40c7ee]	1954	'vis': self.state[key]['vis'],\
[987aaa1]	1955	}
[a97394b]	1956	self.state_lock.release()
	1957
	1958	if rv: return rv
	1959	else: raise service_error(service_error.req, "No such experiment")
[987aaa1]	1960
[c52c48d]	1961	def get_info(self, req, fid):
[0ea11af]	1962	"""
	1963	Return all the stored info about this experiment
	1964	"""
[a97394b]	1965	rv = None
[c52c48d]	1966
	1967	req = req.get('InfoRequestBody', None)
	1968	if not req:
	1969	raise service_error(service_error.req,
	1970	"Bad request format (no VisRequestBody)")
	1971	exp = req.get('experiment', None)
	1972	if exp:
	1973	if exp.has_key('fedid'):
[f8582c9]	1974	key = exp['fedid']
[c52c48d]	1975	keytype = "fedid"
	1976	elif exp.has_key('localname'):
	1977	key = exp['localname']
	1978	keytype = "localname"
	1979	else:
	1980	raise service_error(service_error.req, "Unknown lookup type")
	1981	else:
	1982	raise service_error(service_error.req, "No request?")
	1983
[4064742]	1984	self.check_experiment_access(fid, key)
	1985
[c52c48d]	1986	# The state may be massaged by the service function that called
	1987	# get_info (e.g., encoded for XMLRPC transport) so send a copy of the
	1988	# state.
[a97394b]	1989	self.state_lock.acquire()
[c52c48d]	1990	if self.state.has_key(key):
[a97394b]	1991	rv = copy.deepcopy(self.state[key])
	1992	self.state_lock.release()
[38d2f3c]	1993	# Remove the owner info
	1994	del rv['owner']
	1995	# remove the allocationID info from each federant
	1996	for f in rv['federant']:
	1997	if f.has_key('allocID'): del f['allocID']
[6679c122]	1998
[a97394b]	1999	if rv: return rv
	2000	else: raise service_error(service_error.req, "No such experiment")
[7a8d667]	2001
	2002
	2003	def terminate_experiment(self, req, fid):
[0ea11af]	2004	"""
	2005	Swap this experiment out on the federants and delete the shared
	2006	information
	2007	"""
[7a8d667]	2008	tbparams = { }
	2009	req = req.get('TerminateRequestBody', None)
	2010	if not req:
	2011	raise service_error(service_error.req,
	2012	"Bad request format (no TerminateRequestBody)")
	2013	exp = req.get('experiment', None)
	2014	if exp:
	2015	if exp.has_key('fedid'):
[f8582c9]	2016	key = exp['fedid']
[7a8d667]	2017	keytype = "fedid"
	2018	elif exp.has_key('localname'):
	2019	key = exp['localname']
	2020	keytype = "localname"
	2021	else:
	2022	raise service_error(service_error.req, "Unknown lookup type")
	2023	else:
	2024	raise service_error(service_error.req, "No request?")
	2025
[4064742]	2026	self.check_experiment_access(fid, key)
	2027
[a97394b]	2028	self.state_lock.acquire()
[7a8d667]	2029	fed_exp = self.state.get(key, None)
	2030
	2031	if fed_exp:
[a97394b]	2032	# This branch of the conditional holds the lock to generate a
	2033	# consistent temporary tbparams variable to deallocate experiments.
	2034	# It releases the lock to do the deallocations and reacquires it to
	2035	# remove the experiment state when the termination is complete.
[eee2b2e]	2036	ids = []
	2037	# experimentID is a list of dicts that are self-describing
	2038	# identifiers. This finds all the fedids and localnames - the
	2039	# keys of self.state - and puts them into ids.
	2040	for id in fed_exp.get('experimentID', []):
	2041	if id.has_key('fedid'): ids.append(id['fedid'])
	2042	if id.has_key('localname'): ids.append(id['localname'])
	2043
[7a8d667]	2044	# Construct enough of the tbparams to make the stop_segment calls
	2045	# work
	2046	for fed in fed_exp['federant']:
	2047	try:
	2048	for e in fed['name']:
	2049	eid = e.get('localname', None)
	2050	if eid: break
	2051	else:
	2052	continue
	2053
	2054	p = fed['emulab']['project']
	2055
	2056	project = p['name']['localname']
	2057	tb = p['testbed']['localname']
	2058	user = p['user'][0]['userID']['localname']
	2059
	2060	domain = fed['emulab']['domain']
[03b9b14]	2061	host = fed['emulab']['ops']
[d81971a]	2062	aid = fed['allocID']
[7a8d667]	2063	except KeyError, e:
	2064	continue
	2065	tbparams[tb] = {\
	2066	'user': user,\
	2067	'domain': domain,\
	2068	'project': project,\
	2069	'host': host,\
	2070	'eid': eid,\
[d81971a]	2071	'aid': aid,\
[7a8d667]	2072	}
[a97394b]	2073	self.state_lock.release()
	2074
[7a8d667]	2075	# Stop everyone.
[8bc5754]	2076	thread_pool = self.thread_pool(self.nthreads)
[7a8d667]	2077	for tb in tbparams.keys():
[8bc5754]	2078	# Create and start a thread to stop the segment
	2079	thread_pool.wait_for_slot()
	2080	t = self.pooled_thread(target=self.stop_segment,
	2081	args=(tb, tbparams[tb]['eid'], tbparams), name=tb,
	2082	pdata=thread_pool, trace_file=self.trace_file)
	2083	t.start()
	2084	# Wait for completions
	2085	thread_pool.wait_for_all_done()
[eee2b2e]	2086
[d81971a]	2087	# release the allocations
	2088	for tb in tbparams.keys():
	2089	self.release_access(tb, tbparams[tb]['aid'])
	2090
	2091	# Remove the terminated experiment
[a97394b]	2092	self.state_lock.acquire()
[eee2b2e]	2093	for id in ids:
	2094	if self.state.has_key(id): del self.state[id]
	2095
	2096	if self.state_filename: self.write_state()
[a97394b]	2097	self.state_lock.release()
	2098
[7a8d667]	2099	return { 'experiment': exp }
	2100	else:
[a97394b]	2101	# Don't forget to release the lock
	2102	self.state_lock.release()
[7a8d667]	2103	raise service_error(service_error.req, "No saved state")

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

Download in other formats: