Context Navigation

source: fedd/fedd_experiment_control.py @ 0b123ff

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

Last change on this file since 0b123ff was 5576a47, checked in by Ted Faber <faber@…>, 16 years ago
project exporting in place
Property mode set to `100644`
File size: 53.2 KB

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

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

Download in other formats: