Context Navigation

source: fedd/federation/experiment_control.py @ 266e866

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

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

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

Download in other formats: