Context Navigation

experiment_control.py @ 9252414

axis_examplecompt_changesinfo-ops

Last change on this file since 9252414 was dee164e, checked in by Ted Faber <faber@…>, 14 years ago

Looks like internal works now.

Had to add default entries to the access list to accomodate that, and discovered that ABAC requires strings - not unicode.

Moved lookup_access into the aceess class as most should be able to use it directly now.

Property mode set to 100644

File size: 90.0 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
[79b6596]	13	import signal
	14	import time
[6679c122]	15
[3441fe3]	16	import traceback
[c971895]	17	# For parsing visualization output and splitter output
	18	import xml.parsers.expat
[3441fe3]	19
[6c57fe9]	20	from threading import Lock, Thread, Condition
	21	from subprocess import call, Popen, PIPE
[c573278]	22	from string import join
[6679c122]	23
[db6b092]	24	from urlparse import urlparse
	25	from urllib2 import urlopen
	26
[ec4fb42]	27	from util import *
[51cc9df]	28	from fedid import fedid, generate_fedid
[9460b1e]	29	from remote_service import xmlrpc_handler, soap_handler, service_caller
[c971895]	30	from service_error import service_error
[2761484]	31	from synch_store import synch_store
[73e7f5c]	32	from experiment_partition import experiment_partition
[7206e5a]	33	from authorizer import abac_authorizer
[6679c122]	34
[db6b092]	35	import topdl
[f07fa49]	36	import list_log
[db6b092]	37	from ip_allocator import ip_allocator
	38	from ip_addr import ip_addr
	39
[11a08b0]	40
	41	class nullHandler(logging.Handler):
	42	def emit(self, record): pass
	43
	44	fl = logging.getLogger("fedd.experiment_control")
	45	fl.addHandler(nullHandler())
	46
[43197eb]	47
	48	# Right now, no support for composition.
	49	class federated_service:
[5334044]	50	def __init__(self, name, exporter=None, importers=None, params=None,
	51	reqs=None, portal=None):
[43197eb]	52	self.name=name
	53	self.exporter=exporter
[5334044]	54	if importers is None: self.importers = []
	55	else: self.importers=importers
	56	if params is None: self.params = { }
	57	else: self.params = params
	58	if reqs is None: self.reqs = []
	59	else: self.reqs = reqs
	60
	61	if portal is not None:
	62	self.portal = portal
	63	else:
	64	self.portal = (name in federated_service.needs_portal)
[43197eb]	65
[d20823f]	66	def __str__(self):
	67	return "name %s export %s import %s params %s reqs %s" % \
	68	(self.name, self.exporter, self.importers, self.params,
	69	[ (r['name'], r['visibility']) for r in self.reqs] )
	70
[5334044]	71	needs_portal = ('SMB', 'seer', 'tmcd', 'project_export', 'seer_master')
	72
[ec4fb42]	73	class experiment_control_local:
[0ea11af]	74	"""
	75	Control of experiments that this system can directly access.
	76
	77	Includes experiment creation, termination and information dissemination.
	78	Thred safe.
	79	"""
[79b6596]	80
	81	class ssh_cmd_timeout(RuntimeError): pass
[6679c122]	82
[1af38d6]	83	class thread_pool:
[866c983]	84	"""
	85	A class to keep track of a set of threads all invoked for the same
	86	task. Manages the mutual exclusion of the states.
	87	"""
	88	def __init__(self, nthreads):
	89	"""
	90	Start a pool.
	91	"""
	92	self.changed = Condition()
	93	self.started = 0
	94	self.terminated = 0
	95	self.nthreads = nthreads
	96
	97	def acquire(self):
	98	"""
	99	Get the pool's lock.
	100	"""
	101	self.changed.acquire()
	102
	103	def release(self):
	104	"""
	105	Release the pool's lock.
	106	"""
	107	self.changed.release()
	108
	109	def wait(self, timeout = None):
	110	"""
	111	Wait for a pool thread to start or stop.
	112	"""
	113	self.changed.wait(timeout)
	114
	115	def start(self):
	116	"""
	117	Called by a pool thread to report starting.
	118	"""
	119	self.changed.acquire()
	120	self.started += 1
	121	self.changed.notifyAll()
	122	self.changed.release()
	123
	124	def terminate(self):
	125	"""
	126	Called by a pool thread to report finishing.
	127	"""
	128	self.changed.acquire()
	129	self.terminated += 1
	130	self.changed.notifyAll()
	131	self.changed.release()
	132
	133	def clear(self):
	134	"""
	135	Clear all pool data.
	136	"""
	137	self.changed.acquire()
	138	self.started = 0
	139	self.terminated =0
	140	self.changed.notifyAll()
	141	self.changed.release()
	142
	143	def wait_for_slot(self):
	144	"""
	145	Wait until we have a free slot to start another pooled thread
	146	"""
	147	self.acquire()
	148	while self.started - self.terminated >= self.nthreads:
	149	self.wait()
	150	self.release()
	151
[32e7d93]	152	def wait_for_all_done(self, timeout=None):
[866c983]	153	"""
[32e7d93]	154	Wait until all active threads finish (and at least one has
	155	started). If a timeout is given, return after waiting that long
	156	for termination. If all threads are done (and one has started in
	157	the since the last clear()) return True, otherwise False.
[866c983]	158	"""
[32e7d93]	159	if timeout:
	160	deadline = time.time() + timeout
[866c983]	161	self.acquire()
	162	while self.started == 0 or self.started > self.terminated:
[32e7d93]	163	self.wait(timeout)
	164	if timeout:
	165	if time.time() > deadline:
	166	break
	167	timeout = deadline - time.time()
[866c983]	168	self.release()
[32e7d93]	169	return not (self.started == 0 or self.started > self.terminated)
[8bc5754]	170
[1af38d6]	171	class pooled_thread(Thread):
[866c983]	172	"""
	173	One of a set of threads dedicated to a specific task. Uses the
	174	thread_pool class above for coordination.
	175	"""
	176	def __init__(self, group=None, target=None, name=None, args=(),
	177	kwargs={}, pdata=None, trace_file=None):
	178	Thread.__init__(self, group, target, name, args, kwargs)
	179	self.rv = None # Return value of the ops in this thread
	180	self.exception = None # Exception that terminated this thread
	181	self.target=target # Target function to run on start()
	182	self.args = args # Args to pass to target
	183	self.kwargs = kwargs # Additional kw args
	184	self.pdata = pdata # thread_pool for this class
	185	# Logger for this thread
	186	self.log = logging.getLogger("fedd.experiment_control")
	187
	188	def run(self):
	189	"""
	190	Emulate Thread.run, except add pool data manipulation and error
	191	logging.
	192	"""
	193	if self.pdata:
	194	self.pdata.start()
	195
	196	if self.target:
	197	try:
	198	self.rv = self.target(self.args, *self.kwargs)
	199	except service_error, s:
	200	self.exception = s
	201	self.log.error("Thread exception: %s %s" % \
	202	(s.code_string(), s.desc))
	203	except:
	204	self.exception = sys.exc_info()[1]
	205	self.log.error(("Unexpected thread exception: %s" +\
	206	"Trace %s") % (self.exception,\
	207	traceback.format_exc()))
	208	if self.pdata:
	209	self.pdata.terminate()
[6679c122]	210
[f069052]	211	call_RequestAccess = service_caller('RequestAccess')
	212	call_ReleaseAccess = service_caller('ReleaseAccess')
[cc8d8e9]	213	call_StartSegment = service_caller('StartSegment')
[5ae3857]	214	call_TerminateSegment = service_caller('TerminateSegment')
[5f6929a]	215	call_Ns2Topdl = service_caller('Ns2Topdl')
[058f58e]	216
[3f6bc5f]	217	def __init__(self, config=None, auth=None):
[866c983]	218	"""
	219	Intialize the various attributes, most from the config object
	220	"""
	221
	222	def parse_tarfile_list(tf):
	223	"""
	224	Parse a tarfile list from the configuration. This is a set of
	225	paths and tarfiles separated by spaces.
	226	"""
	227	rv = [ ]
	228	if tf is not None:
	229	tl = tf.split()
	230	while len(tl) > 1:
	231	p, t = tl[0:2]
	232	del tl[0:2]
	233	rv.append((p, t))
	234	return rv
	235
	236	self.thread_with_rv = experiment_control_local.pooled_thread
	237	self.thread_pool = experiment_control_local.thread_pool
[f07fa49]	238	self.list_log = list_log.list_log
[866c983]	239
	240	self.cert_file = config.get("experiment_control", "cert_file")
	241	if self.cert_file:
	242	self.cert_pwd = config.get("experiment_control", "cert_pwd")
	243	else:
	244	self.cert_file = config.get("globals", "cert_file")
	245	self.cert_pwd = config.get("globals", "cert_pwd")
	246
	247	self.trusted_certs = config.get("experiment_control", "trusted_certs") \
	248	or config.get("globals", "trusted_certs")
	249
[6c57fe9]	250	self.repodir = config.get("experiment_control", "repodir")
[7183b48]	251	self.repo_url = config.get("experiment_control", "repo_url",
	252	"https://users.isi.deterlab.net:23235");
[cc8d8e9]	253
[866c983]	254	self.exp_stem = "fed-stem"
	255	self.log = logging.getLogger("fedd.experiment_control")
	256	set_log_level(config, "experiment_control", self.log)
	257	self.muxmax = 2
[35a4c01]	258	self.nthreads = 10
[866c983]	259	self.randomize_experiments = False
	260
	261	self.splitter = None
	262	self.ssh_keygen = "/usr/bin/ssh-keygen"
	263	self.ssh_identity_file = None
	264
	265
	266	self.debug = config.getboolean("experiment_control", "create_debug")
[69692a9]	267	self.cleanup = not config.getboolean("experiment_control",
	268	"leave_tmpfiles")
[866c983]	269	self.state_filename = config.get("experiment_control",
	270	"experiment_state")
[2761484]	271	self.store_filename = config.get("experiment_control",
	272	"synch_store")
	273	self.store_url = config.get("experiment_control", "store_url")
[5f6929a]	274	self.splitter_url = config.get("experiment_control", "ns2topdl_uri")
[866c983]	275	self.fedkit = parse_tarfile_list(\
	276	config.get("experiment_control", "fedkit"))
	277	self.gatewaykit = parse_tarfile_list(\
	278	config.get("experiment_control", "gatewaykit"))
	279	accessdb_file = config.get("experiment_control", "accessdb")
	280
	281	self.ssh_pubkey_file = config.get("experiment_control",
	282	"ssh_pubkey_file")
	283	self.ssh_privkey_file = config.get("experiment_control",
	284	"ssh_privkey_file")
[175b444]	285	dt = config.get("experiment_control", "direct_transit")
[7206e5a]	286	self.auth_type = config.get('experiment_control', 'auth_type') \
	287	or 'legacy'
	288	self.auth_dir = config.get('experiment_control', 'auth_dir')
[139e2e2]	289	if dt: self.direct_transit = [ tb.strip() for tb in dt.split(",")]
	290	else: self.direct_transit = [ ]
[866c983]	291	# NB for internal master/slave ops, not experiment setup
	292	self.ssh_type = config.get("experiment_control", "sshkeytype", "rsa")
[ca489e8]	293
[db6b092]	294	self.overrides = set([])
	295	ovr = config.get('experiment_control', 'overrides')
	296	if ovr:
	297	for o in ovr.split(","):
	298	o = o.strip()
	299	if o.startswith('fedid:'): o = o[len('fedid:'):]
	300	self.overrides.add(fedid(hexstr=o))
[ca489e8]	301
[866c983]	302	self.state = { }
	303	self.state_lock = Lock()
	304	self.tclsh = "/usr/local/bin/otclsh"
[5f6929a]	305	self.tcl_splitter = config.get("ns2topdl", "tcl_splitter") or \
[866c983]	306	config.get("experiment_control", "tcl_splitter",
	307	"/usr/testbed/lib/ns2ir/parse.tcl")
	308	mapdb_file = config.get("experiment_control", "mapdb")
	309	self.trace_file = sys.stderr
	310
	311	self.def_expstart = \
	312	"sudo -H /bin/sh /usr/local/federation/bin/federate.sh >& " +\
	313	"/tmp/federate";
	314	self.def_mexpstart = "sudo -H /usr/local/federation/bin/make_hosts " +\
	315	"FEDDIR/hosts";
	316	self.def_gwstart = \
	317	"sudo -H /usr/local/federation/bin/fed-tun.pl -f GWCONF>& " +\
	318	"/tmp/bridge.log";
	319	self.def_mgwstart = \
	320	"sudo -H /usr/local/federation/bin/fed-tun.pl -f GWCONF >& " +\
	321	"/tmp/bridge.log";
	322	self.def_gwimage = "FBSD61-TUNNEL2";
	323	self.def_gwtype = "pc";
	324	self.local_access = { }
	325
[7206e5a]	326	if self.auth_type == 'legacy':
	327	if auth:
	328	self.auth = auth
	329	else:
	330	self.log.error( "[access]: No authorizer initialized, " +\
	331	"creating local one.")
	332	auth = authorizer()
	333	elif self.auth_type == 'abac':
	334	self.auth = abac_authorizer(load=self.auth_dir)
	335	else:
	336	raise service_error(service_error.internal,
	337	"Unknown auth_type: %s" % self.auth_type)
[866c983]	338
	339
	340	if self.ssh_pubkey_file:
	341	try:
	342	f = open(self.ssh_pubkey_file, 'r')
	343	self.ssh_pubkey = f.read()
	344	f.close()
[d3c8759]	345	except EnvironmentError:
[866c983]	346	raise service_error(service_error.internal,
	347	"Cannot read sshpubkey")
	348	else:
	349	raise service_error(service_error.internal,
	350	"No SSH public key file?")
	351
	352	if not self.ssh_privkey_file:
	353	raise service_error(service_error.internal,
	354	"No SSH public key file?")
	355
	356
	357	if mapdb_file:
	358	self.read_mapdb(mapdb_file)
	359	else:
	360	self.log.warn("[experiment_control] No testbed map, using defaults")
	361	self.tbmap = {
	362	'deter':'https://users.isi.deterlab.net:23235',
	363	'emulab':'https://users.isi.deterlab.net:23236',
	364	'ucb':'https://users.isi.deterlab.net:23237',
	365	}
	366
	367	if accessdb_file:
	368	self.read_accessdb(accessdb_file)
	369	else:
	370	raise service_error(service_error.internal,
	371	"No accessdb specified in config")
	372
	373	# Grab saved state. OK to do this w/o locking because it's read only
	374	# and only one thread should be in existence that can see self.state at
	375	# this point.
	376	if self.state_filename:
	377	self.read_state()
	378
[2761484]	379	if self.store_filename:
	380	self.read_store()
	381	else:
	382	self.log.warning("No saved synch store")
	383	self.synch_store = synch_store
	384
[866c983]	385	# Dispatch tables
	386	self.soap_services = {\
[a3ad8bd]	387	'New': soap_handler('New', self.new_experiment),
[e19b75c]	388	'Create': soap_handler('Create', self.create_experiment),
[866c983]	389	'Vtopo': soap_handler('Vtopo', self.get_vtopo),
	390	'Vis': soap_handler('Vis', self.get_vis),
	391	'Info': soap_handler('Info', self.get_info),
[65f3f29]	392	'MultiInfo': soap_handler('MultiInfo', self.get_multi_info),
[866c983]	393	'Terminate': soap_handler('Terminate',
[e19b75c]	394	self.terminate_experiment),
[2761484]	395	'GetValue': soap_handler('GetValue', self.GetValue),
	396	'SetValue': soap_handler('SetValue', self.SetValue),
[866c983]	397	}
	398
	399	self.xmlrpc_services = {\
[a3ad8bd]	400	'New': xmlrpc_handler('New', self.new_experiment),
[e19b75c]	401	'Create': xmlrpc_handler('Create', self.create_experiment),
[866c983]	402	'Vtopo': xmlrpc_handler('Vtopo', self.get_vtopo),
	403	'Vis': xmlrpc_handler('Vis', self.get_vis),
	404	'Info': xmlrpc_handler('Info', self.get_info),
[65f3f29]	405	'MultiInfo': xmlrpc_handler('MultiInfo', self.get_multi_info),
[866c983]	406	'Terminate': xmlrpc_handler('Terminate',
[e19b75c]	407	self.terminate_experiment),
[2761484]	408	'GetValue': xmlrpc_handler('GetValue', self.GetValue),
	409	'SetValue': xmlrpc_handler('SetValue', self.SetValue),
[866c983]	410	}
[19cc408]	411
[a97394b]	412	# Call while holding self.state_lock
[eee2b2e]	413	def write_state(self):
[866c983]	414	"""
	415	Write a new copy of experiment state after copying the existing state
	416	to a backup.
	417
	418	State format is a simple pickling of the state dictionary.
	419	"""
	420	if os.access(self.state_filename, os.W_OK):
[40dd8c1]	421	copy_file(self.state_filename, \
	422	"%s.bak" % self.state_filename)
[866c983]	423	try:
	424	f = open(self.state_filename, 'w')
	425	pickle.dump(self.state, f)
[d3c8759]	426	except EnvironmentError, e:
[866c983]	427	self.log.error("Can't write file %s: %s" % \
	428	(self.state_filename, e))
	429	except pickle.PicklingError, e:
	430	self.log.error("Pickling problem: %s" % e)
	431	except TypeError, e:
	432	self.log.error("Pickling problem (TypeError): %s" % e)
[eee2b2e]	433
[2761484]	434	@staticmethod
	435	def get_alloc_ids(state):
	436	"""
	437	Pull the fedids of the identifiers of each allocation from the
	438	state. Again, a dict dive that's best isolated.
	439
	440	Used by read_store and read state
	441	"""
	442
	443	return [ f['allocID']['fedid']
	444	for f in state.get('federant',[]) \
	445	if f.has_key('allocID') and \
	446	f['allocID'].has_key('fedid')]
	447
[a97394b]	448	# Call while holding self.state_lock
[eee2b2e]	449	def read_state(self):
[866c983]	450	"""
	451	Read a new copy of experiment state. Old state is overwritten.
	452
	453	State format is a simple pickling of the state dictionary.
	454	"""
[cc8d8e9]	455
	456	def get_experiment_id(state):
	457	"""
	458	Pull the fedid experimentID out of the saved state. This is kind
	459	of a gross walk through the dict.
	460	"""
	461
	462	if state.has_key('experimentID'):
	463	for e in state['experimentID']:
	464	if e.has_key('fedid'):
	465	return e['fedid']
	466	else:
	467	return None
	468	else:
	469	return None
	470
[866c983]	471	try:
	472	f = open(self.state_filename, "r")
	473	self.state = pickle.load(f)
	474	self.log.debug("[read_state]: Read state from %s" % \
	475	self.state_filename)
[d3c8759]	476	except EnvironmentError, e:
[866c983]	477	self.log.warning("[read_state]: No saved state: Can't open %s: %s"\
	478	% (self.state_filename, e))
	479	except pickle.UnpicklingError, e:
	480	self.log.warning(("[read_state]: No saved state: " + \
	481	"Unpickling failed: %s") % e)
	482
[cc8d8e9]	483	for s in self.state.values():
[866c983]	484	try:
[cc8d8e9]	485
	486	eid = get_experiment_id(s)
	487	if eid :
[7206e5a]	488	if self.auth_type == 'legacy':
	489	# XXX: legacy
	490	# Give the owner rights to the experiment
	491	self.auth.set_attribute(s['owner'], eid)
	492	# And holders of the eid as well
	493	self.auth.set_attribute(eid, eid)
	494	# allow overrides to control experiments as well
	495	for o in self.overrides:
	496	self.auth.set_attribute(o, eid)
	497	# Set permissions to allow reading of the software
	498	# repo, if any, as well.
	499	for a in self.get_alloc_ids(s):
	500	self.auth.set_attribute(a, 'repo/%s' % eid)
[cc8d8e9]	501	else:
	502	raise KeyError("No experiment id")
[866c983]	503	except KeyError, e:
	504	self.log.warning("[read_state]: State ownership or identity " +\
	505	"misformatted in %s: %s" % (self.state_filename, e))
[4064742]	506
	507
	508	def read_accessdb(self, accessdb_file):
[866c983]	509	"""
	510	Read the mapping from fedids that can create experiments to their name
	511	in the 3-level access namespace. All will be asserted from this
	512	testbed and can include the local username and porject that will be
	513	asserted on their behalf by this fedd. Each fedid is also added to the
	514	authorization system with the "create" attribute.
	515	"""
	516	self.accessdb = {}
	517	# These are the regexps for parsing the db
	518	name_expr = "[" + string.ascii_letters + string.digits + "\.\-]+"
	519	project_line = re.compile("^\s*fedid:([" + string.hexdigits + "]+)"+ \
	520	"\s->$\s("+name_expr+")\s,\s("+name_expr+")\s$\s$")
	521	user_line = re.compile("^\s*fedid:([" + string.hexdigits + "]+)"+ \
	522	"\s->\s(" + name_expr + ")\s*$")
	523	lineno = 0
	524
	525	# Parse the mappings and store in self.authdb, a dict of
	526	# fedid -> (proj, user)
	527	try:
	528	f = open(accessdb_file, "r")
	529	for line in f:
	530	lineno += 1
	531	line = line.strip()
	532	if len(line) == 0 or line.startswith('#'):
	533	continue
	534	m = project_line.match(line)
	535	if m:
	536	fid = fedid(hexstr=m.group(1))
	537	project, user = m.group(2,3)
	538	if not self.accessdb.has_key(fid):
	539	self.accessdb[fid] = []
	540	self.accessdb[fid].append((project, user))
	541	continue
	542
	543	m = user_line.match(line)
	544	if m:
	545	fid = fedid(hexstr=m.group(1))
	546	project = None
	547	user = m.group(2)
	548	if not self.accessdb.has_key(fid):
	549	self.accessdb[fid] = []
	550	self.accessdb[fid].append((project, user))
	551	continue
	552	self.log.warn("[experiment_control] Error parsing access " +\
	553	"db %s at line %d" % (accessdb_file, lineno))
[d3c8759]	554	except EnvironmentError:
[866c983]	555	raise service_error(service_error.internal,
[05fceef]	556	("Error opening/reading %s as experiment " +\
	557	"control accessdb") % accessdb_file)
[866c983]	558	f.close()
	559
	560	# Initialize the authorization attributes
[7206e5a]	561	# XXX: legacy
	562	if self.auth_type == 'legacy':
	563	for fid in self.accessdb.keys():
	564	self.auth.set_attribute(fid, 'create')
	565	self.auth.set_attribute(fid, 'new')
[34bc05c]	566
	567	def read_mapdb(self, file):
[866c983]	568	"""
	569	Read a simple colon separated list of mappings for the
	570	label-to-testbed-URL mappings. Clears or creates self.tbmap.
	571	"""
	572
	573	self.tbmap = { }
	574	lineno =0
	575	try:
	576	f = open(file, "r")
	577	for line in f:
	578	lineno += 1
	579	line = line.strip()
	580	if line.startswith('#') or len(line) == 0:
	581	continue
	582	try:
	583	label, url = line.split(':', 1)
	584	self.tbmap[label] = url
	585	except ValueError, e:
	586	self.log.warn("[read_mapdb] Ignored bad line (%d) in " +\
	587	"map db: %s %s" % (lineno, line, e))
[d3c8759]	588	except EnvironmentError, e:
[866c983]	589	self.log.warning("[read_mapdb]: No saved map database: Can't " +\
	590	"open %s: %s" % (file, e))
	591	f.close()
[2761484]	592
	593	def read_store(self):
	594	try:
	595	self.synch_store = synch_store()
	596	self.synch_store.load(self.store_filename)
	597	self.log.debug("[read_store]: Read store from %s" % \
	598	self.store_filename)
[d3c8759]	599	except EnvironmentError, e:
[2761484]	600	self.log.warning("[read_store]: No saved store: Can't open %s: %s"\
	601	% (self.state_filename, e))
	602	self.synch_store = synch_store()
	603
	604	# Set the initial permissions on data in the store. XXX: This ad hoc
	605	# authorization attribute initialization is getting out of hand.
[7206e5a]	606	# XXX: legacy
	607	if self.auth_type == 'legacy':
	608	for k in self.synch_store.all_keys():
	609	try:
	610	if k.startswith('fedid:'):
	611	fid = fedid(hexstr=k[6:46])
	612	if self.state.has_key(fid):
	613	for a in self.get_alloc_ids(self.state[fid]):
	614	self.auth.set_attribute(a, k)
	615	except ValueError, e:
	616	self.log.warn("Cannot deduce permissions for %s" % k)
[2761484]	617
	618
	619	def write_store(self):
	620	"""
	621	Write a new copy of synch_store after writing current state
	622	to a backup. We use the internal synch_store pickle method to avoid
	623	incinsistent data.
	624
	625	State format is a simple pickling of the store.
	626	"""
	627	if os.access(self.store_filename, os.W_OK):
	628	copy_file(self.store_filename, \
	629	"%s.bak" % self.store_filename)
	630	try:
	631	self.synch_store.save(self.store_filename)
[d3c8759]	632	except EnvironmentError, e:
[2761484]	633	self.log.error("Can't write file %s: %s" % \
	634	(self.store_filename, e))
	635	except TypeError, e:
	636	self.log.error("Pickling problem (TypeError): %s" % e)
	637
[866c983]	638
[6679c122]	639	def generate_ssh_keys(self, dest, type="rsa" ):
[866c983]	640	"""
	641	Generate a set of keys for the gateways to use to talk.
	642
	643	Keys are of type type and are stored in the required dest file.
	644	"""
	645	valid_types = ("rsa", "dsa")
	646	t = type.lower();
	647	if t not in valid_types: raise ValueError
	648	cmd = [self.ssh_keygen, '-t', t, '-N', '', '-f', dest]
	649
	650	try:
	651	trace = open("/dev/null", "w")
[d3c8759]	652	except EnvironmentError:
[866c983]	653	raise service_error(service_error.internal,
	654	"Cannot open /dev/null??");
	655
	656	# May raise CalledProcessError
	657	self.log.debug("[generate_ssh_keys]: %s" % " ".join(cmd))
[4ea1e22]	658	rv = call(cmd, stdout=trace, stderr=trace, close_fds=True)
[866c983]	659	if rv != 0:
	660	raise service_error(service_error.internal,
	661	"Cannot generate nonce ssh keys. %s return code %d" \
	662	% (self.ssh_keygen, rv))
[6679c122]	663
[0d830de]	664	def gentopo(self, str):
[866c983]	665	"""
	666	Generate the topology dtat structure from the splitter's XML
	667	representation of it.
	668
	669	The topology XML looks like:
	670	<experiment>
	671	<nodes>
	672	<node><vname></vname><ips>ip1:ip2</ips></node>
	673	</nodes>
	674	<lans>
	675	<lan>
	676	<vname></vname><vnode></vnode><ip></ip>
	677	<bandwidth></bandwidth><member>node:port</member>
	678	</lan>
	679	</lans>
	680	"""
	681	class topo_parse:
	682	"""
	683	Parse the topology XML and create the dats structure.
	684	"""
	685	def __init__(self):
	686	# Typing of the subelements for data conversion
	687	self.str_subelements = ('vname', 'vnode', 'ips', 'ip', 'member')
	688	self.int_subelements = ( 'bandwidth',)
	689	self.float_subelements = ( 'delay',)
	690	# The final data structure
	691	self.nodes = [ ]
	692	self.lans = [ ]
	693	self.topo = { \
	694	'node': self.nodes,\
	695	'lan' : self.lans,\
	696	}
	697	self.element = { } # Current element being created
	698	self.chars = "" # Last text seen
	699
	700	def end_element(self, name):
	701	# After each sub element the contents is added to the current
	702	# element or to the appropriate list.
	703	if name == 'node':
	704	self.nodes.append(self.element)
	705	self.element = { }
	706	elif name == 'lan':
	707	self.lans.append(self.element)
	708	self.element = { }
	709	elif name in self.str_subelements:
	710	self.element[name] = self.chars
	711	self.chars = ""
	712	elif name in self.int_subelements:
	713	self.element[name] = int(self.chars)
	714	self.chars = ""
	715	elif name in self.float_subelements:
	716	self.element[name] = float(self.chars)
	717	self.chars = ""
	718
	719	def found_chars(self, data):
	720	self.chars += data.rstrip()
	721
	722
	723	tp = topo_parse();
	724	parser = xml.parsers.expat.ParserCreate()
	725	parser.EndElementHandler = tp.end_element
	726	parser.CharacterDataHandler = tp.found_chars
	727
	728	parser.Parse(str)
	729
	730	return tp.topo
	731
[0d830de]	732
	733	def genviz(self, topo):
[866c983]	734	"""
	735	Generate the visualization the virtual topology
	736	"""
	737
	738	neato = "/usr/local/bin/neato"
	739	# These are used to parse neato output and to create the visualization
	740	# file.
[0ac1934]	741	vis_re = re.compile('^\s"?([\w\-]+)"?\s+\[.pos="([\d\.]+),([\d\.]+)"')
[866c983]	742	vis_fmt = "<node><name>%s</name><x>%s</x><y>%s</y><type>" + \
	743	"%s</type></node>"
	744
	745	try:
	746	# Node names
	747	nodes = [ n['vname'] for n in topo['node'] ]
	748	topo_lans = topo['lan']
[cc8d8e9]	749	except KeyError, e:
	750	raise service_error(service_error.internal, "Bad topology: %s" %e)
[866c983]	751
	752	lans = { }
	753	links = { }
	754
	755	# Walk through the virtual topology, organizing the connections into
	756	# 2-node connections (links) and more-than-2-node connections (lans).
	757	# When a lan is created, it's added to the list of nodes (there's a
	758	# node in the visualization for the lan).
	759	for l in topo_lans:
	760	if links.has_key(l['vname']):
	761	if len(links[l['vname']]) < 2:
	762	links[l['vname']].append(l['vnode'])
	763	else:
	764	nodes.append(l['vname'])
	765	lans[l['vname']] = links[l['vname']]
	766	del links[l['vname']]
	767	lans[l['vname']].append(l['vnode'])
	768	elif lans.has_key(l['vname']):
	769	lans[l['vname']].append(l['vnode'])
	770	else:
	771	links[l['vname']] = [ l['vnode'] ]
	772
	773
	774	# Open up a temporary file for dot to turn into a visualization
	775	try:
	776	df, dotname = tempfile.mkstemp()
	777	dotfile = os.fdopen(df, 'w')
[d3c8759]	778	except EnvironmentError:
[866c983]	779	raise service_error(service_error.internal,
	780	"Failed to open file in genviz")
	781
[db6b092]	782	try:
	783	dnull = open('/dev/null', 'w')
[d3c8759]	784	except EnvironmentError:
[db6b092]	785	service_error(service_error.internal,
[886307f]	786	"Failed to open /dev/null in genviz")
	787
[866c983]	788	# Generate a dot/neato input file from the links, nodes and lans
	789	try:
	790	print >>dotfile, "graph G {"
	791	for n in nodes:
	792	print >>dotfile, '\t"%s"' % n
	793	for l in links.keys():
	794	print >>dotfile, '\t"%s" -- "%s"' % tuple(links[l])
	795	for l in lans.keys():
	796	for n in lans[l]:
	797	print >>dotfile, '\t "%s" -- "%s"' % (n,l)
	798	print >>dotfile, "}"
	799	dotfile.close()
	800	except TypeError:
	801	raise service_error(service_error.internal,
	802	"Single endpoint link in vtopo")
[d3c8759]	803	except EnvironmentError:
[866c983]	804	raise service_error(service_error.internal, "Cannot write dot file")
	805
	806	# Use dot to create a visualization
	807	dot = Popen([neato, '-Gstart=rand', '-Gepsilon=0.005', '-Gmaxiter=2000',
[886307f]	808	'-Gpack=true', dotname], stdout=PIPE, stderr=dnull,
[db6b092]	809	close_fds=True)
	810	dnull.close()
[866c983]	811
	812	# Translate dot to vis format
	813	vis_nodes = [ ]
	814	vis = { 'node': vis_nodes }
	815	for line in dot.stdout:
	816	m = vis_re.match(line)
	817	if m:
	818	vn = m.group(1)
	819	vis_node = {'name': vn, \
	820	'x': float(m.group(2)),\
	821	'y' : float(m.group(3)),\
	822	}
	823	if vn in links.keys() or vn in lans.keys():
	824	vis_node['type'] = 'lan'
	825	else:
	826	vis_node['type'] = 'node'
	827	vis_nodes.append(vis_node)
	828	rv = dot.wait()
	829
	830	os.remove(dotname)
	831	if rv == 0 : return vis
	832	else: return None
[d0ae12d]	833
[fd07c48]	834	def get_access(self, tb, nodes, tbparam, access_user, masters, tbmap):
[866c983]	835	"""
	836	Get access to testbed through fedd and set the parameters for that tb
	837	"""
[43197eb]	838	def get_export_project(svcs):
	839	"""
	840	Look through for the list of federated_service for this testbed
	841	objects for a project_export service, and extract the project
	842	parameter.
	843	"""
	844
	845	pe = [s for s in svcs if s.name=='project_export']
	846	if len(pe) == 1:
	847	return pe[0].params.get('project', None)
	848	elif len(pe) == 0:
	849	return None
	850	else:
	851	raise service_error(service_error.req,
	852	"More than one project export is not supported")
	853
[fd07c48]	854	uri = tbmap.get(testbed_base(tb), None)
[866c983]	855	if not uri:
[b78c9ea]	856	raise service_error(service_error.server_config,
[866c983]	857	"Unknown testbed: %s" % tb)
	858
[43197eb]	859	export_svcs = masters.get(tb,[])
	860	import_svcs = [ s for m in masters.values() \
	861	for s in m \
	862	if tb in s.importers ]
	863
	864	export_project = get_export_project(export_svcs)
	865
[8218a3b]	866	# Tweak search order so that if there are entries in access_user that
	867	# have a project matching the export project, we try them first
[5f6929a]	868	if export_project:
	869	access_sequence = [ (p, u) for p, u in access_user \
	870	if p == export_project]
	871	access_sequence.extend([(p, u) for p, u in access_user \
	872	if p != export_project])
[8218a3b]	873	else:
	874	access_sequence = access_user
	875
	876	for p, u in access_sequence:
[866c983]	877	self.log.debug(("[get_access] Attempting access from (%s, %s) " + \
	878	"to %s") % ((p or "None"), u, uri))
	879
	880	if p:
	881	# Request with user and project specified
	882	req = {\
[3bddd24]	883	'credential': [ "project: %s" % p, "user: %s" % u],
[866c983]	884	}
	885	else:
	886	# Request with only user specified
	887	req = {\
[3bddd24]	888	'credential': [ 'user: %s' % u ],
[866c983]	889	}
	890
[43197eb]	891	# Make the service request from the services we're importing and
	892	# exporting. Keep track of the export request ids so we can
	893	# collect the resulting info from the access response.
	894	e_keys = { }
	895	if import_svcs or export_svcs:
	896	req['service'] = [ ]
	897
	898	for i, s in enumerate(import_svcs):
	899	idx = 'import%d' % i
	900	sr = {'id': idx, 'name': s.name, 'visibility': 'import' }
	901	if s.params:
	902	sr['fedAttr'] = [ { 'attribute': k, 'value': v } \
	903	for k, v in s.params.items()]
	904	req['service'].append(sr)
	905
	906	for i, s in enumerate(export_svcs):
	907	idx = 'export%d' % i
	908	e_keys[idx] = s
	909	sr = {'id': idx, 'name': s.name, 'visibility': 'export' }
	910	if s.params:
	911	sr['fedAttr'] = [ { 'attribute': k, 'value': v }
	912	for k, v in s.params.items()]
	913	req['service'].append(sr)
[866c983]	914
	915	# node resources if any
	916	if nodes != None and len(nodes) > 0:
	917	rnodes = [ ]
	918	for n in nodes:
	919	rn = { }
	920	image, hw, count = n.split(":")
	921	if image: rn['image'] = [ image ]
	922	if hw: rn['hardware'] = [ hw ]
	923	if count and int(count) >0 : rn['count'] = int(count)
	924	rnodes.append(rn)
	925	req['resources']= { }
	926	req['resources']['node'] = rnodes
	927
	928	try:
	929	if self.local_access.has_key(uri):
	930	# Local access call
	931	req = { 'RequestAccessRequestBody' : req }
	932	r = self.local_access[uri].RequestAccess(req,
	933	fedid(file=self.cert_file))
	934	r = { 'RequestAccessResponseBody' : r }
	935	else:
	936	r = self.call_RequestAccess(uri, req,
	937	self.cert_file, self.cert_pwd, self.trusted_certs)
	938	except service_error, e:
	939	if e.code == service_error.access:
	940	self.log.debug("[get_access] Access denied")
	941	r = None
	942	continue
	943	else:
	944	raise e
	945
[e19b75c]	946	if r.has_key('RequestAccessResponseBody'):
	947	# Through to here we have a valid response, not a fault.
	948	# Access denied is a fault, so something better or worse than
	949	# access denied has happened.
	950	r = r['RequestAccessResponseBody']
	951	self.log.debug("[get_access] Access granted")
	952	break
	953	else:
	954	raise service_error(service_error.protocol,
	955	"Bad proxy response")
	956
	957	if not r:
	958	raise service_error(service_error.access,
	959	"Access denied by %s (%s)" % (tb, uri))
[db6b092]	960
[4afcfc4]	961	tbparam[tb] = {
[69692a9]	962	"allocID" : r['allocID'],
	963	"uri": uri,
[4afcfc4]	964	}
[43197eb]	965
	966	# Collect the responses corresponding to the services this testbed
	967	# exports. These will be the service requests that we will include in
	968	# the start segment requests (with appropriate visibility values) to
	969	# import and export the segments.
	970	for s in r.get('service', []):
	971	id = s.get('id', None)
	972	if id and id in e_keys:
	973	e_keys[id].reqs.append(s)
[4afcfc4]	974
	975	# Add attributes to parameter space. We don't allow attributes to
	976	# overlay any parameters already installed.
[617592b]	977	for a in r.get('fedAttr', []):
[4afcfc4]	978	try:
	979	if a['attribute'] and \
	980	isinstance(a['attribute'], basestring)\
	981	and not tbparam[tb].has_key(a['attribute'].lower()):
	982	tbparam[tb][a['attribute'].lower()] = a['value']
	983	except KeyError:
	984	self.log.error("Bad attribute in response: %s" % a)
[db6b092]	985
[725c55d]	986	def release_access(self, tb, aid, tbmap=None, uri=None, cert_file=None,
	987	cert_pwd=None):
[e19b75c]	988	"""
	989	Release access to testbed through fedd
	990	"""
[db6b092]	991
[fd07c48]	992	if not uri and tbmap:
	993	uri = tbmap.get(tb, None)
[e19b75c]	994	if not uri:
[69692a9]	995	raise service_error(service_error.server_config,
[e19b75c]	996	"Unknown testbed: %s" % tb)
[db6b092]	997
[e19b75c]	998	if self.local_access.has_key(uri):
	999	resp = self.local_access[uri].ReleaseAccess(\
	1000	{ 'ReleaseAccessRequestBody' : {'allocID': aid},},
[725c55d]	1001	fedid(file=cert_file))
[e19b75c]	1002	resp = { 'ReleaseAccessResponseBody': resp }
	1003	else:
	1004	resp = self.call_ReleaseAccess(uri, {'allocID': aid},
[725c55d]	1005	cert_file, cert_pwd, self.trusted_certs)
[db6b092]	1006
[e19b75c]	1007	# better error coding
[db6b092]	1008
[5f6929a]	1009	def remote_ns2topdl(self, uri, desc):
[db6b092]	1010
[e19b75c]	1011	req = {
	1012	'description' : { 'ns2description': desc },
[db6b092]	1013	}
	1014
[5f6929a]	1015	r = self.call_Ns2Topdl(uri, req, self.cert_file, self.cert_pwd,
[e19b75c]	1016	self.trusted_certs)
	1017
[5f6929a]	1018	if r.has_key('Ns2TopdlResponseBody'):
	1019	r = r['Ns2TopdlResponseBody']
[1dcaff4]	1020	ed = r.get('experimentdescription', None)
	1021	if ed.has_key('topdldescription'):
	1022	return topdl.Topology(**ed['topdldescription'])
[e19b75c]	1023	else:
	1024	raise service_error(service_error.protocol,
	1025	"Bad splitter response (no output)")
	1026	else:
	1027	raise service_error(service_error.protocol, "Bad splitter response")
[cc8d8e9]	1028
[e19b75c]	1029	class start_segment:
[fd556d1]	1030	def __init__(self, debug=False, log=None, testbed="", cert_file=None,
[f07fa49]	1031	cert_pwd=None, trusted_certs=None, caller=None,
	1032	log_collector=None):
[cc8d8e9]	1033	self.log = log
	1034	self.debug = debug
	1035	self.cert_file = cert_file
	1036	self.cert_pwd = cert_pwd
	1037	self.trusted_certs = None
	1038	self.caller = caller
[fd556d1]	1039	self.testbed = testbed
[f07fa49]	1040	self.log_collector = log_collector
[69692a9]	1041	self.response = None
[b4b19c7]	1042	self.node = { }
	1043
	1044	def make_map(self, resp):
[c5869ef]	1045	for e in resp.get('embedding', []):
	1046	if 'toponame' in e and 'physname' in e:
	1047	self.node[e['toponame']] = e['physname'][0]
[cc8d8e9]	1048
[43197eb]	1049	def __call__(self, uri, aid, topo, masters, attrs=None, connInfo=None):
[cc8d8e9]	1050	req = {
	1051	'allocID': { 'fedid' : aid },
	1052	'segmentdescription': {
	1053	'topdldescription': topo.to_dict(),
	1054	},
	1055	}
[e02cd14]	1056
	1057	if connInfo:
	1058	req['connection'] = connInfo
[43197eb]	1059
	1060	import_svcs = [ s for m in masters.values() \
	1061	for s in m if self.testbed in s.importers]
	1062
	1063	if import_svcs or self.testbed in masters:
	1064	req['service'] = []
	1065
	1066	for s in import_svcs:
	1067	for r in s.reqs:
	1068	sr = copy.deepcopy(r)
	1069	sr['visibility'] = 'import';
	1070	req['service'].append(sr)
	1071
	1072	for s in masters.get(self.testbed, []):
	1073	for r in s.reqs:
	1074	sr = copy.deepcopy(r)
	1075	sr['visibility'] = 'export';
	1076	req['service'].append(sr)
	1077
[6c57fe9]	1078	if attrs:
	1079	req['fedAttr'] = attrs
[cc8d8e9]	1080
[fd556d1]	1081	try:
[13e3dd2]	1082	self.log.debug("Calling StartSegment at %s " % uri)
[fd556d1]	1083	r = self.caller(uri, req, self.cert_file, self.cert_pwd,
	1084	self.trusted_certs)
[f07fa49]	1085	if r.has_key('StartSegmentResponseBody'):
	1086	lval = r['StartSegmentResponseBody'].get('allocationLog',
	1087	None)
	1088	if lval and self.log_collector:
	1089	for line in lval.splitlines(True):
	1090	self.log_collector.write(line)
[b4b19c7]	1091	self.make_map(r['StartSegmentResponseBody'])
[69692a9]	1092	self.response = r
[f07fa49]	1093	else:
	1094	raise service_error(service_error.internal,
	1095	"Bad response!?: %s" %r)
[fd556d1]	1096	return True
	1097	except service_error, e:
	1098	self.log.error("Start segment failed on %s: %s" % \
	1099	(self.testbed, e))
	1100	return False
[cc8d8e9]	1101
	1102
[5ae3857]	1103
[e19b75c]	1104	class terminate_segment:
[fd556d1]	1105	def __init__(self, debug=False, log=None, testbed="", cert_file=None,
[5ae3857]	1106	cert_pwd=None, trusted_certs=None, caller=None):
	1107	self.log = log
	1108	self.debug = debug
	1109	self.cert_file = cert_file
	1110	self.cert_pwd = cert_pwd
	1111	self.trusted_certs = None
	1112	self.caller = caller
[fd556d1]	1113	self.testbed = testbed
[5ae3857]	1114
	1115	def __call__(self, uri, aid ):
	1116	req = {
	1117	'allocID': aid ,
	1118	}
[fd556d1]	1119	try:
	1120	r = self.caller(uri, req, self.cert_file, self.cert_pwd,
	1121	self.trusted_certs)
	1122	return True
	1123	except service_error, e:
	1124	self.log.error("Terminate segment failed on %s: %s" % \
	1125	(self.testbed, e))
	1126	return False
[db6b092]	1127
	1128
[43197eb]	1129	def allocate_resources(self, allocated, masters, eid, expid,
[b4b19c7]	1130	tbparams, top, topo, tmpdir, alloc_log=None, log_collector=None,
[c573278]	1131	attrs=None, connInfo={}, tbmap=None, expcert=None):
[69692a9]	1132
[cc8d8e9]	1133	started = { } # Testbeds where a sub-experiment started
	1134	# successfully
	1135
	1136	# XXX
	1137	fail_soft = False
	1138
[fd07c48]	1139	if tbmap is None: tbmap = { }
	1140
[cc8d8e9]	1141	log = alloc_log or self.log
	1142
	1143	thread_pool = self.thread_pool(self.nthreads)
	1144	threads = [ ]
[b4b19c7]	1145	starters = [ ]
[cc8d8e9]	1146
[c573278]	1147	if expcert:
	1148	cert = expcert
	1149	pw = None
	1150	else:
	1151	cert = self.cert_file
[822d31b]	1152	pw = self.cert_pwd
[c573278]	1153
[109a32a]	1154	for tb in allocated.keys():
	1155	# Create and start a thread to start the segment, and save it
	1156	# to get the return value later
[ab847bc]	1157	tb_attrs = copy.copy(attrs)
[109a32a]	1158	thread_pool.wait_for_slot()
[fd07c48]	1159	uri = tbparams[tb].get('uri', tbmap.get(testbed_base(tb), None))
[ab847bc]	1160	base, suffix = split_testbed(tb)
	1161	if suffix:
	1162	tb_attrs.append({'attribute': 'experiment_name',
[175b444]	1163	'value': "%s-%s" % (eid, suffix)})
[ab847bc]	1164	else:
	1165	tb_attrs.append({'attribute': 'experiment_name', 'value': eid})
[109a32a]	1166	if not uri:
	1167	raise service_error(service_error.internal,
	1168	"Unknown testbed %s !?" % tb)
	1169
[cc8d8e9]	1170	if tbparams[tb].has_key('allocID') and \
	1171	tbparams[tb]['allocID'].has_key('fedid'):
	1172	aid = tbparams[tb]['allocID']['fedid']
	1173	else:
	1174	raise service_error(service_error.internal,
	1175	"No alloc id for testbed %s !?" % tb)
	1176
[b4b19c7]	1177	s = self.start_segment(log=log, debug=self.debug,
[c573278]	1178	testbed=tb, cert_file=cert,
	1179	cert_pwd=pw, trusted_certs=self.trusted_certs,
[b4b19c7]	1180	caller=self.call_StartSegment,
	1181	log_collector=log_collector)
	1182	starters.append(s)
[109a32a]	1183	t = self.pooled_thread(\
[b4b19c7]	1184	target=s, name=tb,
[ab847bc]	1185	args=(uri, aid, topo[tb], masters, tb_attrs, connInfo[tb]),
[109a32a]	1186	pdata=thread_pool, trace_file=self.trace_file)
[69692a9]	1187	threads.append(t)
	1188	t.start()
[cc8d8e9]	1189
[109a32a]	1190	# Wait until all finish (keep pinging the log, though)
	1191	mins = 0
[dadc4da]	1192	revoked = False
[109a32a]	1193	while not thread_pool.wait_for_all_done(60.0):
	1194	mins += 1
	1195	alloc_log.info("Waiting for sub threads (it has been %d mins)" \
	1196	% mins)
[dadc4da]	1197	if not revoked and \
[f52f5df]	1198	len([ t.getName() for t in threads if t.rv == False]) > 0:
[dadc4da]	1199	# a testbed has failed. Revoke this experiment's
	1200	# synchronizarion values so that sub experiments will not
	1201	# deadlock waiting for synchronization that will never happen
	1202	self.log.info("A subexperiment has failed to swap in, " + \
	1203	"revoking synch keys")
	1204	var_key = "fedid:%s" % expid
	1205	for k in self.synch_store.all_keys():
	1206	if len(k) > 45 and k[0:46] == var_key:
	1207	self.synch_store.revoke_key(k)
	1208	revoked = True
[69692a9]	1209
[cc8d8e9]	1210	failed = [ t.getName() for t in threads if not t.rv ]
	1211	succeeded = [tb for tb in allocated.keys() if tb not in failed]
[3132419]	1212
[cc8d8e9]	1213	# If one failed clean up, unless fail_soft is set
[32e7d93]	1214	if failed:
[cc8d8e9]	1215	if not fail_soft:
	1216	thread_pool.clear()
	1217	for tb in succeeded:
	1218	# Create and start a thread to stop the segment
	1219	thread_pool.wait_for_slot()
[0fa1729]	1220	uri = tbparams[tb]['uri']
[cc8d8e9]	1221	t = self.pooled_thread(\
[32e7d93]	1222	target=self.terminate_segment(log=log,
[fd556d1]	1223	testbed=tb,
[32e7d93]	1224	cert_file=self.cert_file,
	1225	cert_pwd=self.cert_pwd,
	1226	trusted_certs=self.trusted_certs,
	1227	caller=self.call_TerminateSegment),
	1228	args=(uri, tbparams[tb]['federant']['allocID']),
	1229	name=tb,
[cc8d8e9]	1230	pdata=thread_pool, trace_file=self.trace_file)
	1231	t.start()
[f52f5df]	1232	# Wait until all finish (if any are being stopped)
	1233	if succeeded:
	1234	thread_pool.wait_for_all_done()
[cc8d8e9]	1235
	1236	# release the allocations
	1237	for tb in tbparams.keys():
[725c55d]	1238	try:
	1239	self.release_access(tb, tbparams[tb]['allocID'],
	1240	tbmap=tbmap, uri=tbparams[tb].get('uri', None))
	1241	except service_error, e:
	1242	self.log.warn("Error releasing access: %s" % e.desc)
[cc8d8e9]	1243	# Remove the placeholder
	1244	self.state_lock.acquire()
	1245	self.state[eid]['experimentStatus'] = 'failed'
	1246	if self.state_filename: self.write_state()
	1247	self.state_lock.release()
[05e8da8]	1248	# Remove the repo dir
	1249	self.remove_dirs("%s/%s" %(self.repodir, expid))
	1250	# Walk up tmpdir, deleting as we go
	1251	if self.cleanup:
	1252	self.remove_dirs(tmpdir)
	1253	else:
	1254	log.debug("[start_experiment]: not removing %s" % tmpdir)
	1255
[cc8d8e9]	1256
	1257	log.error("Swap in failed on %s" % ",".join(failed))
	1258	return
	1259	else:
[b4b19c7]	1260	# Walk through the successes and gather the virtual to physical
	1261	# mapping.
[2fd8f8c]	1262	embedding = [ ]
[b4b19c7]	1263	for s in starters:
[2fd8f8c]	1264	for k, v in s.node.items():
	1265	embedding.append({
	1266	'toponame': k,
	1267	'physname': [ v],
	1268	'testbed': s.testbed
	1269	})
[cc8d8e9]	1270	log.info("[start_segment]: Experiment %s active" % eid)
	1271
	1272
	1273	# Walk up tmpdir, deleting as we go
[69692a9]	1274	if self.cleanup:
[05e8da8]	1275	self.remove_dirs(tmpdir)
[69692a9]	1276	else:
	1277	log.debug("[start_experiment]: not removing %s" % tmpdir)
[cc8d8e9]	1278
[b4b19c7]	1279	# Insert the experiment into our state and update the disk copy.
[cc8d8e9]	1280	self.state_lock.acquire()
	1281	self.state[expid]['experimentStatus'] = 'active'
	1282	self.state[eid] = self.state[expid]
[b4b19c7]	1283	self.state[eid]['experimentdescription']['topdldescription'] = \
	1284	top.to_dict()
[2fd8f8c]	1285	self.state[eid]['embedding'] = embedding
[cc8d8e9]	1286	if self.state_filename: self.write_state()
	1287	self.state_lock.release()
	1288	return
	1289
	1290
[895a133]	1291	def add_kit(self, e, kit):
	1292	"""
	1293	Add a Software object created from the list of (install, location)
	1294	tuples passed as kit to the software attribute of an object e. We
	1295	do this enough to break out the code, but it's kind of a hack to
	1296	avoid changing the old tuple rep.
	1297	"""
	1298
	1299	s = [ topdl.Software(install=i, location=l) for i, l in kit]
	1300
	1301	if isinstance(e.software, list): e.software.extend(s)
	1302	else: e.software = s
	1303
	1304
[b4b19c7]	1305	def create_experiment_state(self, fid, req, expid, expcert,
[a3ad8bd]	1306	state='starting'):
[895a133]	1307	"""
	1308	Create the initial entry in the experiment's state. The expid and
	1309	expcert are the experiment's fedid and certifacte that represents that
	1310	ID, which are installed in the experiment state. If the request
	1311	includes a suggested local name that is used if possible. If the local
	1312	name is already taken by an experiment owned by this user that has
[a3ad8bd]	1313	failed, it is overwritten. Otherwise new letters are added until a
[895a133]	1314	valid localname is found. The generated local name is returned.
	1315	"""
	1316
	1317	if req.has_key('experimentID') and \
	1318	req['experimentID'].has_key('localname'):
	1319	overwrite = False
	1320	eid = req['experimentID']['localname']
	1321	# If there's an old failed experiment here with the same local name
	1322	# and accessible by this user, we'll overwrite it, otherwise we'll
	1323	# fall through and do the collision avoidance.
	1324	old_expid = self.get_experiment_fedid(eid)
	1325	if old_expid and self.check_experiment_access(fid, old_expid):
	1326	self.state_lock.acquire()
	1327	status = self.state[eid].get('experimentStatus', None)
	1328	if status and status == 'failed':
	1329	# remove the old access attribute
[7206e5a]	1330	self.auth.unset_attribute(fid, old_expid)
	1331	self.auth.save()
[895a133]	1332	overwrite = True
	1333	del self.state[eid]
	1334	del self.state[old_expid]
	1335	self.state_lock.release()
	1336	self.state_lock.acquire()
	1337	while (self.state.has_key(eid) and not overwrite):
	1338	eid += random.choice(string.ascii_letters)
	1339	# Initial state
	1340	self.state[eid] = {
	1341	'experimentID' : \
	1342	[ { 'localname' : eid }, {'fedid': expid } ],
[a3ad8bd]	1343	'experimentStatus': state,
[895a133]	1344	'experimentAccess': { 'X509' : expcert },
	1345	'owner': fid,
	1346	'log' : [],
	1347	}
	1348	self.state[expid] = self.state[eid]
	1349	if self.state_filename: self.write_state()
	1350	self.state_lock.release()
	1351	else:
	1352	eid = self.exp_stem
	1353	for i in range(0,5):
	1354	eid += random.choice(string.ascii_letters)
	1355	self.state_lock.acquire()
	1356	while (self.state.has_key(eid)):
	1357	eid = self.exp_stem
	1358	for i in range(0,5):
	1359	eid += random.choice(string.ascii_letters)
	1360	# Initial state
	1361	self.state[eid] = {
	1362	'experimentID' : \
	1363	[ { 'localname' : eid }, {'fedid': expid } ],
[a3ad8bd]	1364	'experimentStatus': state,
[895a133]	1365	'experimentAccess': { 'X509' : expcert },
	1366	'owner': fid,
	1367	'log' : [],
	1368	}
	1369	self.state[expid] = self.state[eid]
	1370	if self.state_filename: self.write_state()
	1371	self.state_lock.release()
	1372
	1373	return eid
	1374
	1375
	1376	def allocate_ips_to_topo(self, top):
	1377	"""
[69692a9]	1378	Add an ip4_address attribute to all the hosts in the topology, based on
[895a133]	1379	the shared substrates on which they sit. An /etc/hosts file is also
[69692a9]	1380	created and returned as a list of hostfiles entries. We also return
	1381	the allocator, because we may need to allocate IPs to portals
	1382	(specifically DRAGON portals).
[895a133]	1383	"""
	1384	subs = sorted(top.substrates,
	1385	cmp=lambda x,y: cmp(len(x.interfaces), len(y.interfaces)),
	1386	reverse=True)
	1387	ips = ip_allocator(int(ip_addr("10.0.0.0")), 2 **24)
	1388	ifs = { }
	1389	hosts = [ ]
	1390
	1391	for idx, s in enumerate(subs):
[289ff7e]	1392	net_size = len(s.interfaces)+2
	1393
	1394	a = ips.allocate(net_size)
[895a133]	1395	if a :
	1396	base, num = a
[289ff7e]	1397	if num < net_size:
[895a133]	1398	raise service_error(service_error.internal,
	1399	"Allocator returned wrong number of IPs??")
	1400	else:
	1401	raise service_error(service_error.req,
	1402	"Cannot allocate IP addresses")
[062b991]	1403	mask = ips.min_alloc
	1404	while mask < net_size:
	1405	mask *= 2
[289ff7e]	1406
[062b991]	1407	netmask = ((2**32-1) ^ (mask-1))
[895a133]	1408
	1409	base += 1
	1410	for i in s.interfaces:
	1411	i.attribute.append(
	1412	topdl.Attribute('ip4_address',
	1413	"%s" % ip_addr(base)))
[289ff7e]	1414	i.attribute.append(
	1415	topdl.Attribute('ip4_netmask',
	1416	"%s" % ip_addr(int(netmask))))
	1417
[1e7f268]	1418	hname = i.element.name
[895a133]	1419	if ifs.has_key(hname):
	1420	hosts.append("%s\t%s-%s %s-%d" % \
	1421	(ip_addr(base), hname, s.name, hname,
	1422	ifs[hname]))
	1423	else:
	1424	ifs[hname] = 0
	1425	hosts.append("%s\t%s-%s %s-%d %s" % \
	1426	(ip_addr(base), hname, s.name, hname,
	1427	ifs[hname], hname))
	1428
	1429	ifs[hname] += 1
	1430	base += 1
[69692a9]	1431	return hosts, ips
[895a133]	1432
[43197eb]	1433	def get_access_to_testbeds(self, testbeds, access_user, allocated,
[fd07c48]	1434	tbparams, masters, tbmap):
[895a133]	1435	"""
	1436	Request access to the various testbeds required for this instantiation
	1437	(passed in as testbeds). User, access_user, expoert_project and master
	1438	are used to construct the correct requests. Per-testbed parameters are
	1439	returned in tbparams.
	1440	"""
	1441	for tb in testbeds:
[fd07c48]	1442	self.get_access(tb, None, tbparams, access_user, masters, tbmap)
[895a133]	1443	allocated[tb] = 1
	1444
[6e63513]	1445	def get_abac_access_to_testbeds(self, testbeds, fid, allocated,
[725c55d]	1446	tbparam, masters, tbmap, expid=None, expcert=None):
[6e63513]	1447	for tb in testbeds:
[725c55d]	1448	self.get_abac_access(tb, tbparam, fid, masters, tbmap, expid,
[c573278]	1449	expcert)
[6e63513]	1450	allocated[tb] = 1
	1451
[725c55d]	1452	def get_abac_access(self, tb, tbparam,fid, masters, tbmap, expid=None, expcert=None):
[6e63513]	1453	"""
	1454	Get access to testbed through fedd and set the parameters for that tb
	1455	"""
	1456	def get_export_project(svcs):
	1457	"""
	1458	Look through for the list of federated_service for this testbed
	1459	objects for a project_export service, and extract the project
	1460	parameter.
	1461	"""
	1462
	1463	pe = [s for s in svcs if s.name=='project_export']
	1464	if len(pe) == 1:
	1465	return pe[0].params.get('project', None)
	1466	elif len(pe) == 0:
	1467	return None
	1468	else:
	1469	raise service_error(service_error.req,
	1470	"More than one project export is not supported")
	1471
	1472	uri = tbmap.get(testbed_base(tb), None)
	1473	if not uri:
	1474	raise service_error(service_error.server_config,
	1475	"Unknown testbed: %s" % tb)
	1476
	1477	export_svcs = masters.get(tb,[])
	1478	import_svcs = [ s for m in masters.values() \
	1479	for s in m \
	1480	if tb in s.importers ]
	1481
	1482	export_project = get_export_project(export_svcs)
	1483	# Compose the credential list so that IDs come before attributes
	1484	creds = set()
	1485	keys = set()
[c573278]	1486	certs = self.auth.get_creds_for_principal(fid)
	1487	if expid:
	1488	certs.update(self.auth.get_creds_for_principal(expid))
	1489	for c in certs:
[6e63513]	1490	keys.add(c.issuer_cert())
	1491	creds.add(c.attribute_cert())
	1492	creds = list(keys) + list(creds)
	1493
[c573278]	1494	if expcert: cert, pw = expcert, None
	1495	else: cert, pw = self.cert_file, self.cert_pw
	1496
[6e63513]	1497	# Request credentials
	1498	req = {
	1499	'abac_credential': creds,
	1500	}
	1501	# Make the service request from the services we're importing and
	1502	# exporting. Keep track of the export request ids so we can
	1503	# collect the resulting info from the access response.
	1504	e_keys = { }
	1505	if import_svcs or export_svcs:
	1506	req['service'] = [ ]
	1507
	1508	for i, s in enumerate(import_svcs):
	1509	idx = 'import%d' % i
	1510	sr = {'id': idx, 'name': s.name, 'visibility': 'import' }
	1511	if s.params:
	1512	sr['fedAttr'] = [ { 'attribute': k, 'value': v } \
	1513	for k, v in s.params.items()]
	1514	req['service'].append(sr)
	1515
	1516	for i, s in enumerate(export_svcs):
	1517	idx = 'export%d' % i
	1518	e_keys[idx] = s
	1519	sr = {'id': idx, 'name': s.name, 'visibility': 'export' }
	1520	if s.params:
	1521	sr['fedAttr'] = [ { 'attribute': k, 'value': v }
	1522	for k, v in s.params.items()]
	1523	req['service'].append(sr)
	1524
	1525
	1526	if self.local_access.has_key(uri):
	1527	# Local access call
	1528	req = { 'RequestAccessRequestBody' : req }
	1529	r = self.local_access[uri].RequestAccess(req,
	1530	fedid(file=self.cert_file))
	1531	r = { 'RequestAccessResponseBody' : r }
	1532	else:
[c573278]	1533	r = self.call_RequestAccess(uri, req, cert, pw, self.trusted_certs)
[6e63513]	1534
[725c55d]	1535	if r.has_key('RequestAccessResponseBody'):
	1536	# Through to here we have a valid response, not a fault.
	1537	# Access denied is a fault, so something better or worse than
	1538	# access denied has happened.
	1539	r = r['RequestAccessResponseBody']
	1540	self.log.debug("[get_access] Access granted")
	1541	else:
	1542	raise service_error(service_error.protocol,
	1543	"Bad proxy response")
	1544
[6e63513]	1545	tbparam[tb] = {
	1546	"allocID" : r['allocID'],
	1547	"uri": uri,
	1548	}
	1549
	1550	# Collect the responses corresponding to the services this testbed
	1551	# exports. These will be the service requests that we will include in
	1552	# the start segment requests (with appropriate visibility values) to
	1553	# import and export the segments.
	1554	for s in r.get('service', []):
	1555	id = s.get('id', None)
	1556	if id and id in e_keys:
	1557	e_keys[id].reqs.append(s)
	1558
	1559	# Add attributes to parameter space. We don't allow attributes to
	1560	# overlay any parameters already installed.
	1561	for a in r.get('fedAttr', []):
	1562	try:
	1563	if a['attribute'] and \
	1564	isinstance(a['attribute'], basestring)\
	1565	and not tbparam[tb].has_key(a['attribute'].lower()):
	1566	tbparam[tb][a['attribute'].lower()] = a['value']
	1567	except KeyError:
	1568	self.log.error("Bad attribute in response: %s" % a)
	1569
	1570
[7fe81be]	1571	def split_topology(self, top, topo, testbeds):
[895a133]	1572	"""
[e02cd14]	1573	Create the sub-topologies that are needed for experiment instantiation.
[895a133]	1574	"""
	1575	for tb in testbeds:
	1576	topo[tb] = top.clone()
[7fe81be]	1577	# copy in for loop allows deletions from the original
	1578	for e in [ e for e in topo[tb].elements]:
[895a133]	1579	etb = e.get_attribute('testbed')
[7fe81be]	1580	# NB: elements without a testbed attribute won't appear in any
	1581	# sub topologies.
	1582	if not etb or etb != tb:
[895a133]	1583	for i in e.interface:
	1584	for s in i.subs:
	1585	try:
	1586	s.interfaces.remove(i)
	1587	except ValueError:
	1588	raise service_error(service_error.internal,
	1589	"Can't remove interface??")
[7fe81be]	1590	topo[tb].elements.remove(e)
[895a133]	1591	topo[tb].make_indices()
	1592
	1593	def wrangle_software(self, expid, top, topo, tbparams):
	1594	"""
	1595	Copy software out to the repository directory, allocate permissions and
	1596	rewrite the segment topologies to look for the software in local
	1597	places.
	1598	"""
	1599
	1600	# Copy the rpms and tarfiles to a distribution directory from
	1601	# which the federants can retrieve them
	1602	linkpath = "%s/software" % expid
	1603	softdir ="%s/%s" % ( self.repodir, linkpath)
	1604	softmap = { }
	1605	# These are in a list of tuples format (each kit). This comprehension
	1606	# unwraps them into a single list of tuples that initilaizes the set of
	1607	# tuples.
	1608	pkgs = set([ t for l in [self.fedkit, self.gatewaykit] \
	1609	for p, t in l ])
	1610	pkgs.update([x.location for e in top.elements \
	1611	for x in e.software])
	1612	try:
	1613	os.makedirs(softdir)
[d3c8759]	1614	except EnvironmentError, e:
[895a133]	1615	raise service_error(
	1616	"Cannot create software directory: %s" % e)
	1617	# The actual copying. Everything's converted into a url for copying.
	1618	for pkg in pkgs:
	1619	loc = pkg
	1620
	1621	scheme, host, path = urlparse(loc)[0:3]
	1622	dest = os.path.basename(path)
	1623	if not scheme:
	1624	if not loc.startswith('/'):
	1625	loc = "/%s" % loc
	1626	loc = "file://%s" %loc
	1627	try:
	1628	u = urlopen(loc)
	1629	except Exception, e:
	1630	raise service_error(service_error.req,
	1631	"Cannot open %s: %s" % (loc, e))
	1632	try:
	1633	f = open("%s/%s" % (softdir, dest) , "w")
	1634	self.log.debug("Writing %s/%s" % (softdir,dest) )
	1635	data = u.read(4096)
	1636	while data:
	1637	f.write(data)
	1638	data = u.read(4096)
	1639	f.close()
	1640	u.close()
	1641	except Exception, e:
	1642	raise service_error(service_error.internal,
	1643	"Could not copy %s: %s" % (loc, e))
	1644	path = re.sub("/tmp", "", linkpath)
	1645	# XXX
	1646	softmap[pkg] = \
[7183b48]	1647	"%s/%s/%s" %\
	1648	( self.repo_url, path, dest)
[895a133]	1649
	1650	# Allow the individual segments to access the software.
	1651	for tb in tbparams.keys():
	1652	self.auth.set_attribute(tbparams[tb]['allocID']['fedid'],
	1653	"/%s/%s" % ( path, dest))
[7206e5a]	1654	self.auth.save()
[895a133]	1655
	1656	# Convert the software locations in the segments into the local
	1657	# copies on this host
	1658	for soft in [ s for tb in topo.values() \
	1659	for e in tb.elements \
	1660	if getattr(e, 'software', False) \
	1661	for s in e.software ]:
	1662	if softmap.has_key(soft.location):
	1663	soft.location = softmap[soft.location]
	1664
	1665
[a3ad8bd]	1666	def new_experiment(self, req, fid):
	1667	"""
	1668	The external interface to empty initial experiment creation called from
	1669	the dispatcher.
	1670
	1671	Creates a working directory, splits the incoming description using the
	1672	splitter script and parses out the avrious subsections using the
	1673	lcasses above. Once each sub-experiment is created, use pooled threads
	1674	to instantiate them and start it all up.
	1675	"""
[7206e5a]	1676	req = req.get('NewRequestBody', None)
	1677	if not req:
	1678	raise service_error(service_error.req,
	1679	"Bad request format (no NewRequestBody)")
	1680
	1681	if self.auth.import_credentials(data_list=req.get('credential', [])):
	1682	self.auth.save()
[c573278]	1683
[a3ad8bd]	1684	if not self.auth.check_attribute(fid, 'new'):
	1685	raise service_error(service_error.access, "New access denied")
	1686
	1687	try:
	1688	tmpdir = tempfile.mkdtemp(prefix="split-")
[d3c8759]	1689	except EnvironmentError:
[a3ad8bd]	1690	raise service_error(service_error.internal, "Cannot create tmp dir")
	1691
	1692	try:
	1693	access_user = self.accessdb[fid]
	1694	except KeyError:
	1695	raise service_error(service_error.internal,
	1696	"Access map and authorizer out of sync in " + \
[7183b48]	1697	"new_experiment for fedid %s" % fid)
[a3ad8bd]	1698
	1699	pid = "dummy"
	1700	gid = "dummy"
	1701
	1702	# Generate an ID for the experiment (slice) and a certificate that the
	1703	# allocator can use to prove they own it. We'll ship it back through
[7206e5a]	1704	# the encrypted connection. If the requester supplied one, use it.
	1705	if 'experimentAccess' in req and 'X509' in req['experimentAccess']:
	1706	expcert = req['experimentAccess']['X509']
[962ea25]	1707	expid = fedid(certstr=expcert)
[7206e5a]	1708	self.state_lock.acquire()
	1709	if expid in self.state:
	1710	self.state_lock.release()
	1711	raise service_error(service_error.req,
	1712	'fedid %s identifies an existing experiment' % expid)
	1713	self.state_lock.release()
	1714	else:
	1715	(expid, expcert) = generate_fedid("test", dir=tmpdir, log=self.log)
[a3ad8bd]	1716
	1717	#now we're done with the tmpdir, and it should be empty
	1718	if self.cleanup:
	1719	self.log.debug("[new_experiment]: removing %s" % tmpdir)
	1720	os.rmdir(tmpdir)
	1721	else:
	1722	self.log.debug("[new_experiment]: not removing %s" % tmpdir)
	1723
	1724	eid = self.create_experiment_state(fid, req, expid, expcert,
	1725	state='empty')
	1726
	1727	# Let users touch the state
[7206e5a]	1728	self.auth.set_attribute(fid, expid)
	1729	self.auth.set_attribute(expid, expid)
	1730	# Override fedids can manipulate state as well
	1731	for o in self.overrides:
	1732	self.auth.set_attribute(o, expid)
	1733	self.auth.save()
[a3ad8bd]	1734
	1735	rv = {
	1736	'experimentID': [
	1737	{'localname' : eid }, { 'fedid': copy.copy(expid) }
	1738	],
	1739	'experimentStatus': 'empty',
	1740	'experimentAccess': { 'X509' : expcert }
	1741	}
	1742
	1743	return rv
	1744
[e19b75c]	1745	def create_experiment(self, req, fid):
[db6b092]	1746	"""
	1747	The external interface to experiment creation called from the
	1748	dispatcher.
	1749
	1750	Creates a working directory, splits the incoming description using the
[43197eb]	1751	splitter script and parses out the various subsections using the
[1a4ee0f]	1752	classes above. Once each sub-experiment is created, use pooled threads
	1753	to instantiate them and start it all up.
[db6b092]	1754	"""
[7183b48]	1755
	1756	req = req.get('CreateRequestBody', None)
	1757	if not req:
	1758	raise service_error(service_error.req,
	1759	"Bad request format (no CreateRequestBody)")
	1760
	1761	# Get the experiment access
	1762	exp = req.get('experimentID', None)
	1763	if exp:
	1764	if exp.has_key('fedid'):
	1765	key = exp['fedid']
	1766	expid = key
	1767	eid = None
	1768	elif exp.has_key('localname'):
	1769	key = exp['localname']
	1770	eid = key
	1771	expid = None
	1772	else:
	1773	raise service_error(service_error.req, "Unknown lookup type")
	1774	else:
	1775	raise service_error(service_error.req, "No request?")
	1776
[6e63513]	1777	# Import information from the requester
	1778	if self.auth.import_credentials(data_list=req.get('credential', [])):
	1779	self.auth.save()
	1780
[7183b48]	1781	self.check_experiment_access(fid, key)
[db6b092]	1782
[fd07c48]	1783	# Install the testbed map entries supplied with the request into a copy
	1784	# of the testbed map.
	1785	tbmap = dict(self.tbmap)
	1786	for m in req.get('testbedmap', []):
	1787	if 'testbed' in m and 'uri' in m:
	1788	tbmap[m['testbed']] = m['uri']
	1789
[db6b092]	1790	try:
	1791	tmpdir = tempfile.mkdtemp(prefix="split-")
[895a133]	1792	os.mkdir(tmpdir+"/keys")
[d3c8759]	1793	except EnvironmentError:
[db6b092]	1794	raise service_error(service_error.internal, "Cannot create tmp dir")
	1795
	1796	gw_pubkey_base = "fed.%s.pub" % self.ssh_type
	1797	gw_secretkey_base = "fed.%s" % self.ssh_type
	1798	gw_pubkey = tmpdir + "/keys/" + gw_pubkey_base
	1799	gw_secretkey = tmpdir + "/keys/" + gw_secretkey_base
	1800	tclfile = tmpdir + "/experiment.tcl"
	1801	tbparams = { }
	1802	try:
	1803	access_user = self.accessdb[fid]
	1804	except KeyError:
	1805	raise service_error(service_error.internal,
	1806	"Access map and authorizer out of sync in " + \
	1807	"create_experiment for fedid %s" % fid)
	1808
	1809	pid = "dummy"
	1810	gid = "dummy"
	1811
	1812	# The tcl parser needs to read a file so put the content into that file
	1813	descr=req.get('experimentdescription', None)
	1814	if descr:
	1815	file_content=descr.get('ns2description', None)
	1816	if file_content:
	1817	try:
	1818	f = open(tclfile, 'w')
	1819	f.write(file_content)
	1820	f.close()
[d3c8759]	1821	except EnvironmentError:
[db6b092]	1822	raise service_error(service_error.internal,
	1823	"Cannot write temp experiment description")
	1824	else:
	1825	raise service_error(service_error.req,
	1826	"Only ns2descriptions supported")
	1827	else:
	1828	raise service_error(service_error.req, "No experiment description")
	1829
[7183b48]	1830	self.state_lock.acquire()
	1831	if self.state.has_key(key):
[4afcfc4]	1832	self.state[key]['experimentStatus'] = "starting"
[7183b48]	1833	for e in self.state[key].get('experimentID',[]):
	1834	if not expid and e.has_key('fedid'):
	1835	expid = e['fedid']
	1836	elif not eid and e.has_key('localname'):
	1837	eid = e['localname']
[c573278]	1838	if 'experimentAccess' in self.state[key] and \
	1839	'X509' in self.state[key]['experimentAccess']:
	1840	expcert = self.state[key]['experimentAccess']['X509']
	1841	else:
	1842	expcert = None
[7183b48]	1843	self.state_lock.release()
	1844
	1845	if not (eid and expid):
	1846	raise service_error(service_error.internal,
	1847	"Cannot find local experiment info!?")
[db6b092]	1848
[c573278]	1849	# make a protected copy of the access certificate so the experiment
[725c55d]	1850	# controller can act as the experiment principal.
[822d31b]	1851	if expcert and self.auth_type != 'legacy':
[725c55d]	1852	expcert_file = self.make_temp_certfile(expcert, tmpdir)
	1853	if not expcert_file:
[c573278]	1854	raise service_error(service_error.internal,
	1855	"Cannot create temp cert file?")
[725c55d]	1856	else:
	1857	expcert_file = None
[c573278]	1858
[db6b092]	1859	try:
	1860	# This catches exceptions to clear the placeholder if necessary
	1861	try:
	1862	self.generate_ssh_keys(gw_secretkey, self.ssh_type)
	1863	except ValueError:
	1864	raise service_error(service_error.server_config,
	1865	"Bad key type (%s)" % self.ssh_type)
[5f6929a]	1866
[43197eb]	1867	# Copy the service request
	1868	tb_services = [ s for s in req.get('service',[]) ]
[895a133]	1869	# Translate to topdl
[db6b092]	1870	if self.splitter_url:
[9b8e269]	1871	self.log.debug("Calling remote topdl translator at %s" % \
[db6b092]	1872	self.splitter_url)
[5f6929a]	1873	top = self.remote_ns2topdl(self.splitter_url, file_content)
[db6b092]	1874	else:
	1875	tclcmd = [self.tclsh, self.tcl_splitter, '-t', '-x',
[43197eb]	1876	str(self.muxmax), '-m', 'dummy']
[db6b092]	1877
	1878	tclcmd.extend([pid, gid, eid, tclfile])
	1879
	1880	self.log.debug("running local splitter %s", " ".join(tclcmd))
	1881	# This is just fantastic. As a side effect the parser copies
	1882	# tb_compat.tcl into the current directory, so that directory
	1883	# must be writable by the fedd user. Doing this in the
	1884	# temporary subdir ensures this is the case.
[70caa72]	1885	tclparser = Popen(tclcmd, stdout=PIPE, close_fds=True,
[db6b092]	1886	cwd=tmpdir)
[866c983]	1887	split_data = tclparser.stdout
	1888
[1dcaff4]	1889	top = topdl.topology_from_xml(file=split_data, top="experiment")
[895a133]	1890
[69692a9]	1891	hosts, ip_allocator = self.allocate_ips_to_topo(top)
[1a4ee0f]	1892	# Find the testbeds to look up
[895a133]	1893	testbeds = set([ a.value for e in top.elements \
	1894	for a in e.attribute \
[5f96438]	1895	if a.attribute == 'testbed'])
[895a133]	1896
[5334044]	1897	tb_hosts = { }
	1898	for tb in testbeds:
	1899	tb_hosts[tb] = [ e.name for e in top.elements \
	1900	if isinstance(e, topdl.Computer) and \
	1901	e.get_attribute('testbed') and \
	1902	e.get_attribute('testbed') == tb]
	1903
[43197eb]	1904	masters = { } # testbeds exporting services
[5334044]	1905	pmasters = { } # Testbeds exporting services that
	1906	# need portals
[43197eb]	1907	for s in tb_services:
[5334044]	1908	# If this is a service request with the importall field
[7e67ab9]	1909	# set, fill it out.
	1910
	1911	if s.get('importall', False):
	1912	s['import'] = [ tb for tb in testbeds \
	1913	if tb not in s.get('export',[])]
	1914	del s['importall']
	1915
[43197eb]	1916	# Add the service to masters
	1917	for tb in s.get('export', []):
[b4b19c7]	1918	if s.get('name', None):
[43197eb]	1919	if tb not in masters:
	1920	masters[tb] = [ ]
	1921
	1922	params = { }
	1923	if 'fedAttr' in s:
	1924	for a in s['fedAttr']:
	1925	params[a.get('attribute', '')] = \
	1926	a.get('value','')
	1927
[5334044]	1928	fser = federated_service(name=s['name'],
[43197eb]	1929	exporter=tb, importers=s.get('import',[]),
[5334044]	1930	params=params)
	1931	if fser.name == 'hide_hosts' \
	1932	and 'hosts' not in fser.params:
	1933	fser.params['hosts'] = \
	1934	",".join(tb_hosts.get(fser.exporter, []))
	1935	masters[tb].append(fser)
	1936
	1937	if fser.portal:
	1938	if tb not in pmasters: pmasters[tb] = [ fser ]
	1939	else: pmasters[tb].append(fser)
[43197eb]	1940	else:
[b4b19c7]	1941	self.log.error('Testbed service does not have name " + \
[43197eb]	1942	"and importers')
	1943
	1944
[895a133]	1945	allocated = { } # Testbeds we can access
	1946	topo ={ } # Sub topologies
[e02cd14]	1947	connInfo = { } # Connection information
[5334044]	1948
[6e63513]	1949	if self.auth_type == 'legacy':
	1950	self.get_access_to_testbeds(testbeds, access_user, allocated,
	1951	tbparams, masters, tbmap)
	1952	elif self.auth_type == 'abac':
	1953	self.get_abac_access_to_testbeds(testbeds, fid, allocated,
[725c55d]	1954	tbparams, masters, tbmap, expid, expcert_file)
[6e63513]	1955	else:
	1956	raise service_error(service_error.internal,
	1957	"Unknown auth_type %s" % self.auth_type)
[5f96438]	1958
[7fe81be]	1959	self.split_topology(top, topo, testbeds)
[895a133]	1960
	1961	# Copy configuration files into the remote file store
[6c57fe9]	1962	# The config urlpath
	1963	configpath = "/%s/config" % expid
	1964	# The config file system location
	1965	configdir ="%s%s" % ( self.repodir, configpath)
	1966	try:
	1967	os.makedirs(configdir)
[ab847bc]	1968	except EnvironmentError, e:
	1969	raise service_error(service_error.internal,
[6c57fe9]	1970	"Cannot create config directory: %s" % e)
	1971	try:
	1972	f = open("%s/hosts" % configdir, "w")
	1973	f.write('\n'.join(hosts))
	1974	f.close()
[d3c8759]	1975	except EnvironmentError, e:
[6c57fe9]	1976	raise service_error(service_error.internal,
	1977	"Cannot write hosts file: %s" % e)
	1978	try:
[40dd8c1]	1979	copy_file("%s" % gw_pubkey, "%s/%s" % \
[6c57fe9]	1980	(configdir, gw_pubkey_base))
[40dd8c1]	1981	copy_file("%s" % gw_secretkey, "%s/%s" % \
[6c57fe9]	1982	(configdir, gw_secretkey_base))
[d3c8759]	1983	except EnvironmentError, e:
[6c57fe9]	1984	raise service_error(service_error.internal,
	1985	"Cannot copy keyfiles: %s" % e)
[cc8d8e9]	1986
[6c57fe9]	1987	# Allow the individual testbeds to access the configuration files.
	1988	for tb in tbparams.keys():
	1989	asignee = tbparams[tb]['allocID']['fedid']
	1990	for f in ("hosts", gw_secretkey_base, gw_pubkey_base):
[7206e5a]	1991	self.auth.set_attribute(asignee, "%s/%s" % \
	1992	(configpath, f))
[cc8d8e9]	1993
[fd07c48]	1994	part = experiment_partition(self.auth, self.store_url, tbmap,
[175b444]	1995	self.muxmax, self.direct_transit)
[5334044]	1996	part.add_portals(top, topo, eid, pmasters, tbparams, ip_allocator,
[2761484]	1997	connInfo, expid)
[ab847bc]	1998	# Now get access to the dynamic testbeds (those added above)
	1999	for tb in [ t for t in topo if t not in allocated]:
[c573278]	2000	#XXX: ABAC
[dee164e]	2001	if self.auth_type =='legacy':
	2002	self.get_access(tb, None, tbparams, access_user,
	2003	masters, tbmap)
	2004	elif self.auth_type == 'abac':
	2005	self.get_abac_access(tb, tbparams, fid, masters, tbmap,
	2006	expid, expcert_file)
	2007	else:
	2008	raise service_error(service_error.internal,
	2009	"Unknown auth_type %s" % self.auth_type)
[ab847bc]	2010	allocated[tb] = 1
	2011	store_keys = topo[tb].get_attribute('store_keys')
	2012	# Give the testbed access to keys it exports or imports
	2013	if store_keys:
	2014	for sk in store_keys.split(" "):
	2015	self.auth.set_attribute(\
	2016	tbparams[tb]['allocID']['fedid'], sk)
[7206e5a]	2017	self.auth.save()
[69692a9]	2018
[895a133]	2019	self.wrangle_software(expid, top, topo, tbparams)
[cc8d8e9]	2020
	2021	vtopo = topdl.topology_to_vtopo(top)
	2022	vis = self.genviz(vtopo)
[db6b092]	2023
[866c983]	2024	# save federant information
	2025	for k in allocated.keys():
[ecf679e]	2026	tbparams[k]['federant'] = {
	2027	'name': [ { 'localname' : eid} ],
	2028	'allocID' : tbparams[k]['allocID'],
	2029	'uri': tbparams[k]['uri'],
[866c983]	2030	}
	2031
[db6b092]	2032	self.state_lock.acquire()
	2033	self.state[eid]['vtopo'] = vtopo
	2034	self.state[eid]['vis'] = vis
[b4b19c7]	2035	self.state[eid]['experimentdescription'] = \
[1a4ee0f]	2036	{ 'topdldescription': top.to_dict() }
	2037	self.state[eid]['federant'] = \
[db6b092]	2038	[ tbparams[tb]['federant'] for tb in tbparams.keys() \
	2039	if tbparams[tb].has_key('federant') ]
[cc8d8e9]	2040	if self.state_filename:
	2041	self.write_state()
[db6b092]	2042	self.state_lock.release()
[866c983]	2043	except service_error, e:
	2044	# If something goes wrong in the parse (usually an access error)
	2045	# clear the placeholder state. From here on out the code delays
[db6b092]	2046	# exceptions. Failing at this point returns a fault to the remote
	2047	# caller.
[cc8d8e9]	2048
[866c983]	2049	self.state_lock.acquire()
	2050	del self.state[eid]
[bd3e314]	2051	del self.state[expid]
	2052	if self.state_filename: self.write_state()
[866c983]	2053	self.state_lock.release()
[dee164e]	2054	if tmpdir and self.cleanup:
	2055	self.remove_dirs(tmpdir)
[866c983]	2056	raise e
	2057
	2058
[db6b092]	2059	# Start the background swapper and return the starting state. From
	2060	# here on out, the state will stick around a while.
[866c983]	2061
[7206e5a]	2062	# Let users touch the state
	2063	self.auth.set_attribute(fid, expid)
	2064	self.auth.set_attribute(expid, expid)
	2065	# Override fedids can manipulate state as well
	2066	for o in self.overrides:
	2067	self.auth.set_attribute(o, expid)
	2068	self.auth.save()
[db6b092]	2069
	2070	# Create a logger that logs to the experiment's state object as well as
	2071	# to the main log file.
	2072	alloc_log = logging.getLogger('fedd.experiment_control.%s' % eid)
[f07fa49]	2073	alloc_collector = self.list_log(self.state[eid]['log'])
	2074	h = logging.StreamHandler(alloc_collector)
[db6b092]	2075	# XXX: there should be a global one of these rather than repeating the
	2076	# code.
	2077	h.setFormatter(logging.Formatter("%(asctime)s %(name)s %(message)s",
	2078	'%d %b %y %H:%M:%S'))
	2079	alloc_log.addHandler(h)
	2080
[6c57fe9]	2081	attrs = [
	2082	{
	2083	'attribute': 'ssh_pubkey',
	2084	'value': '%s/%s/config/%s' % \
[7183b48]	2085	(self.repo_url, expid, gw_pubkey_base)
[6c57fe9]	2086	},
	2087	{
	2088	'attribute': 'ssh_secretkey',
	2089	'value': '%s/%s/config/%s' % \
[7183b48]	2090	(self.repo_url, expid, gw_secretkey_base)
[6c57fe9]	2091	},
	2092	{
	2093	'attribute': 'hosts',
	2094	'value': '%s/%s/config/hosts' % \
[7183b48]	2095	(self.repo_url, expid)
[6c57fe9]	2096	},
	2097	]
	2098
[617592b]	2099	# transit and disconnected testbeds may not have a connInfo entry.
	2100	# Fill in the blanks.
	2101	for t in allocated.keys():
	2102	if not connInfo.has_key(t):
	2103	connInfo[t] = { }
	2104
[db6b092]	2105	# Start a thread to do the resource allocation
[e19b75c]	2106	t = Thread(target=self.allocate_resources,
[43197eb]	2107	args=(allocated, masters, eid, expid, tbparams,
[b4b19c7]	2108	top, topo, tmpdir, alloc_log, alloc_collector, attrs,
[725c55d]	2109	connInfo, tbmap, expcert_file),
[db6b092]	2110	name=eid)
	2111	t.start()
	2112
	2113	rv = {
	2114	'experimentID': [
	2115	{'localname' : eid }, { 'fedid': copy.copy(expid) }
	2116	],
	2117	'experimentStatus': 'starting',
	2118	}
	2119
	2120	return rv
[9479343]	2121
	2122	def get_experiment_fedid(self, key):
	2123	"""
[db6b092]	2124	find the fedid associated with the localname key in the state database.
[9479343]	2125	"""
	2126
[db6b092]	2127	rv = None
	2128	self.state_lock.acquire()
	2129	if self.state.has_key(key):
	2130	if isinstance(self.state[key], dict):
	2131	try:
	2132	kl = [ f['fedid'] for f in \
	2133	self.state[key]['experimentID']\
	2134	if f.has_key('fedid') ]
	2135	except KeyError:
	2136	self.state_lock.release()
	2137	raise service_error(service_error.internal,
	2138	"No fedid for experiment %s when getting "+\
	2139	"fedid(!?)" % key)
	2140	if len(kl) == 1:
	2141	rv = kl[0]
	2142	else:
	2143	self.state_lock.release()
	2144	raise service_error(service_error.internal,
	2145	"multiple fedids for experiment %s when " +\
	2146	"getting fedid(!?)" % key)
	2147	else:
	2148	self.state_lock.release()
	2149	raise service_error(service_error.internal,
	2150	"Unexpected state for %s" % key)
	2151	self.state_lock.release()
	2152	return rv
[a97394b]	2153
[4064742]	2154	def check_experiment_access(self, fid, key):
[866c983]	2155	"""
	2156	Confirm that the fid has access to the experiment. Though a request
	2157	may be made in terms of a local name, the access attribute is always
	2158	the experiment's fedid.
	2159	"""
	2160	if not isinstance(key, fedid):
[db6b092]	2161	key = self.get_experiment_fedid(key)
[866c983]	2162
	2163	if self.auth.check_attribute(fid, key):
	2164	return True
	2165	else:
	2166	raise service_error(service_error.access, "Access Denied")
[4064742]	2167
	2168
[db6b092]	2169	def get_handler(self, path, fid):
[7183b48]	2170	self.log.info("Get handler %s %s" % (path, fid))
[6c57fe9]	2171	if self.auth.check_attribute(fid, path):
	2172	return ("%s/%s" % (self.repodir, path), "application/binary")
	2173	else:
	2174	return (None, None)
[987aaa1]	2175
	2176	def get_vtopo(self, req, fid):
[866c983]	2177	"""
	2178	Return the stored virtual topology for this experiment
	2179	"""
	2180	rv = None
[db6b092]	2181	state = None
[866c983]	2182
	2183	req = req.get('VtopoRequestBody', None)
	2184	if not req:
	2185	raise service_error(service_error.req,
	2186	"Bad request format (no VtopoRequestBody)")
	2187	exp = req.get('experiment', None)
	2188	if exp:
	2189	if exp.has_key('fedid'):
	2190	key = exp['fedid']
	2191	keytype = "fedid"
	2192	elif exp.has_key('localname'):
	2193	key = exp['localname']
	2194	keytype = "localname"
	2195	else:
	2196	raise service_error(service_error.req, "Unknown lookup type")
	2197	else:
	2198	raise service_error(service_error.req, "No request?")
	2199
	2200	self.check_experiment_access(fid, key)
	2201
	2202	self.state_lock.acquire()
	2203	if self.state.has_key(key):
[db6b092]	2204	if self.state[key].has_key('vtopo'):
	2205	rv = { 'experiment' : {keytype: key },\
	2206	'vtopo': self.state[key]['vtopo'],\
	2207	}
	2208	else:
	2209	state = self.state[key]['experimentStatus']
[866c983]	2210	self.state_lock.release()
	2211
	2212	if rv: return rv
[bd3e314]	2213	else:
[db6b092]	2214	if state:
	2215	raise service_error(service_error.partial,
	2216	"Not ready: %s" % state)
	2217	else:
	2218	raise service_error(service_error.req, "No such experiment")
[987aaa1]	2219
	2220	def get_vis(self, req, fid):
[866c983]	2221	"""
	2222	Return the stored visualization for this experiment
	2223	"""
	2224	rv = None
[db6b092]	2225	state = None
[866c983]	2226
	2227	req = req.get('VisRequestBody', None)
	2228	if not req:
	2229	raise service_error(service_error.req,
	2230	"Bad request format (no VisRequestBody)")
	2231	exp = req.get('experiment', None)
	2232	if exp:
	2233	if exp.has_key('fedid'):
	2234	key = exp['fedid']
	2235	keytype = "fedid"
	2236	elif exp.has_key('localname'):
	2237	key = exp['localname']
	2238	keytype = "localname"
	2239	else:
	2240	raise service_error(service_error.req, "Unknown lookup type")
	2241	else:
	2242	raise service_error(service_error.req, "No request?")
	2243
	2244	self.check_experiment_access(fid, key)
	2245
	2246	self.state_lock.acquire()
	2247	if self.state.has_key(key):
[db6b092]	2248	if self.state[key].has_key('vis'):
	2249	rv = { 'experiment' : {keytype: key },\
	2250	'vis': self.state[key]['vis'],\
	2251	}
	2252	else:
	2253	state = self.state[key]['experimentStatus']
[866c983]	2254	self.state_lock.release()
	2255
	2256	if rv: return rv
[bd3e314]	2257	else:
[db6b092]	2258	if state:
	2259	raise service_error(service_error.partial,
	2260	"Not ready: %s" % state)
	2261	else:
	2262	raise service_error(service_error.req, "No such experiment")
[987aaa1]	2263
[65f3f29]	2264	def clean_info_response(self, rv):
[db6b092]	2265	"""
	2266	Remove the information in the experiment's state object that is not in
	2267	the info response.
	2268	"""
	2269	# Remove the owner info (should always be there, but...)
	2270	if rv.has_key('owner'): del rv['owner']
	2271
	2272	# Convert the log into the allocationLog parameter and remove the
	2273	# log entry (with defensive programming)
	2274	if rv.has_key('log'):
	2275	rv['allocationLog'] = "".join(rv['log'])
	2276	del rv['log']
	2277	else:
	2278	rv['allocationLog'] = ""
	2279
	2280	if rv['experimentStatus'] != 'active':
	2281	if rv.has_key('federant'): del rv['federant']
	2282	else:
[69692a9]	2283	# remove the allocationID and uri info from each federant
[db6b092]	2284	for f in rv.get('federant', []):
	2285	if f.has_key('allocID'): del f['allocID']
[69692a9]	2286	if f.has_key('uri'): del f['uri']
[b4b19c7]	2287
[db6b092]	2288	return rv
[65f3f29]	2289
[c52c48d]	2290	def get_info(self, req, fid):
[866c983]	2291	"""
	2292	Return all the stored info about this experiment
	2293	"""
	2294	rv = None
	2295
	2296	req = req.get('InfoRequestBody', None)
	2297	if not req:
	2298	raise service_error(service_error.req,
[65f3f29]	2299	"Bad request format (no InfoRequestBody)")
[866c983]	2300	exp = req.get('experiment', None)
	2301	if exp:
	2302	if exp.has_key('fedid'):
	2303	key = exp['fedid']
	2304	keytype = "fedid"
	2305	elif exp.has_key('localname'):
	2306	key = exp['localname']
	2307	keytype = "localname"
	2308	else:
	2309	raise service_error(service_error.req, "Unknown lookup type")
	2310	else:
	2311	raise service_error(service_error.req, "No request?")
	2312
	2313	self.check_experiment_access(fid, key)
	2314
	2315	# The state may be massaged by the service function that called
	2316	# get_info (e.g., encoded for XMLRPC transport) so send a copy of the
	2317	# state.
	2318	self.state_lock.acquire()
	2319	if self.state.has_key(key):
	2320	rv = copy.deepcopy(self.state[key])
	2321	self.state_lock.release()
	2322
[db6b092]	2323	if rv:
	2324	return self.clean_info_response(rv)
[bd3e314]	2325	else:
[db6b092]	2326	raise service_error(service_error.req, "No such experiment")
[7a8d667]	2327
[65f3f29]	2328	def get_multi_info(self, req, fid):
	2329	"""
	2330	Return all the stored info that this fedid can access
	2331	"""
[db6b092]	2332	rv = { 'info': [ ] }
[65f3f29]	2333
[db6b092]	2334	self.state_lock.acquire()
	2335	for key in [ k for k in self.state.keys() if isinstance(k, fedid)]:
[829246e]	2336	try:
	2337	self.check_experiment_access(fid, key)
	2338	except service_error, e:
	2339	if e.code == service_error.access:
	2340	continue
	2341	else:
	2342	self.state_lock.release()
	2343	raise e
[65f3f29]	2344
[db6b092]	2345	if self.state.has_key(key):
	2346	e = copy.deepcopy(self.state[key])
	2347	e = self.clean_info_response(e)
	2348	rv['info'].append(e)
[65f3f29]	2349	self.state_lock.release()
[db6b092]	2350	return rv
[65f3f29]	2351
[e07c8f3]	2352	def remove_dirs(self, dir):
	2353	"""
	2354	Remove the directory tree and all files rooted at dir. Log any errors,
	2355	but continue.
	2356	"""
	2357	self.log.debug("[removedirs]: removing %s" % dir)
	2358	try:
	2359	for path, dirs, files in os.walk(dir, topdown=False):
	2360	for f in files:
	2361	os.remove(os.path.join(path, f))
	2362	for d in dirs:
	2363	os.rmdir(os.path.join(path, d))
	2364	os.rmdir(dir)
	2365	except EnvironmentError, e:
	2366	self.log.error("Error deleting directory tree in %s" % e);
	2367
[725c55d]	2368	@staticmethod
	2369	def make_temp_certfile(expcert, tmpdir):
	2370	"""
	2371	make a protected copy of the access certificate so the experiment
	2372	controller can act as the experiment principal. mkstemp is the most
	2373	secure way to do that. The directory should be created by
	2374	mkdtemp. Return the filename.
	2375	"""
	2376	if expcert and tmpdir:
	2377	try:
	2378	certf, certfn = tempfile.mkstemp(suffix=".pem", dir=tmpdir)
	2379	f = os.fdopen(certf, 'w')
	2380	print >> f, expcert
	2381	f.close()
	2382	except EnvironmentError, e:
	2383	raise service_error(service_error.internal,
	2384	"Cannot create temp cert file?")
	2385	return certfn
	2386	else:
	2387	return None
	2388
[7a8d667]	2389	def terminate_experiment(self, req, fid):
[866c983]	2390	"""
	2391	Swap this experiment out on the federants and delete the shared
	2392	information
	2393	"""
	2394	tbparams = { }
	2395	req = req.get('TerminateRequestBody', None)
	2396	if not req:
	2397	raise service_error(service_error.req,
	2398	"Bad request format (no TerminateRequestBody)")
[db6b092]	2399	force = req.get('force', False)
[866c983]	2400	exp = req.get('experiment', None)
	2401	if exp:
	2402	if exp.has_key('fedid'):
	2403	key = exp['fedid']
	2404	keytype = "fedid"
	2405	elif exp.has_key('localname'):
	2406	key = exp['localname']
	2407	keytype = "localname"
	2408	else:
	2409	raise service_error(service_error.req, "Unknown lookup type")
	2410	else:
	2411	raise service_error(service_error.req, "No request?")
	2412
	2413	self.check_experiment_access(fid, key)
	2414
[db6b092]	2415	dealloc_list = [ ]
[46e4682]	2416
	2417
[5ae3857]	2418	# Create a logger that logs to the dealloc_list as well as to the main
	2419	# log file.
	2420	dealloc_log = logging.getLogger('fedd.experiment_control.%s' % key)
	2421	h = logging.StreamHandler(self.list_log(dealloc_list))
	2422	# XXX: there should be a global one of these rather than repeating the
	2423	# code.
	2424	h.setFormatter(logging.Formatter("%(asctime)s %(name)s %(message)s",
	2425	'%d %b %y %H:%M:%S'))
	2426	dealloc_log.addHandler(h)
	2427
	2428	self.state_lock.acquire()
	2429	fed_exp = self.state.get(key, None)
[e07c8f3]	2430	repo = None
[5ae3857]	2431
	2432	if fed_exp:
	2433	# This branch of the conditional holds the lock to generate a
	2434	# consistent temporary tbparams variable to deallocate experiments.
	2435	# It releases the lock to do the deallocations and reacquires it to
	2436	# remove the experiment state when the termination is complete.
	2437
	2438	# First make sure that the experiment creation is complete.
	2439	status = fed_exp.get('experimentStatus', None)
	2440
	2441	if status:
	2442	if status in ('starting', 'terminating'):
	2443	if not force:
	2444	self.state_lock.release()
	2445	raise service_error(service_error.partial,
	2446	'Experiment still being created or destroyed')
	2447	else:
	2448	self.log.warning('Experiment in %s state ' % status + \
	2449	'being terminated by force.')
	2450	else:
	2451	# No status??? trouble
	2452	self.state_lock.release()
	2453	raise service_error(service_error.internal,
	2454	"Experiment has no status!?")
	2455
	2456	ids = []
	2457	# experimentID is a list of dicts that are self-describing
	2458	# identifiers. This finds all the fedids and localnames - the
	2459	# keys of self.state - and puts them into ids.
	2460	for id in fed_exp.get('experimentID', []):
[e07c8f3]	2461	if id.has_key('fedid'):
	2462	ids.append(id['fedid'])
	2463	repo = "%s" % id['fedid']
[5ae3857]	2464	if id.has_key('localname'): ids.append(id['localname'])
	2465
[725c55d]	2466	# Get the experimentAccess - the principal for this experiment. It
	2467	# is this principal to which credentials have been delegated, and
	2468	# as which the experiment controller must act.
	2469	if 'experimentAccess' in self.state[key] and \
	2470	'X509' in self.state[key]['experimentAccess']:
	2471	expcert = self.state[key]['experimentAccess']['X509']
	2472	else:
	2473	expcert = None
[63a35b7]	2474	# Collect the allocation/segment ids into a dict keyed by the fedid
	2475	# of the allocation (or a monotonically increasing integer) that
	2476	# contains a tuple of uri, aid (which is a dict...)
	2477	for i, fed in enumerate(fed_exp.get('federant', [])):
[5ae3857]	2478	try:
[63a35b7]	2479	uri = fed['uri']
	2480	aid = fed['allocID']
	2481	k = fed['allocID'].get('fedid', i)
[5ae3857]	2482	except KeyError, e:
	2483	continue
[63a35b7]	2484	tbparams[k] = (uri, aid)
[5ae3857]	2485	fed_exp['experimentStatus'] = 'terminating'
	2486	if self.state_filename: self.write_state()
	2487	self.state_lock.release()
	2488
[725c55d]	2489	try:
	2490	tmpdir = tempfile.mkdtemp(prefix="split-")
	2491	except EnvironmentError:
	2492	raise service_error(service_error.internal,
	2493	"Cannot create tmp dir")
	2494	# This try block makes sure the tempdir is cleared
	2495	try:
	2496	# If no expcert, try the deallocation as the experiment
	2497	# controller instance.
[822d31b]	2498	if expcert and self.auth_type != 'legacy':
[725c55d]	2499	cert_file = self.make_temp_certfile(expcert, tmpdir)
	2500	pw = None
	2501	else:
	2502	cert_file = self.cert_file
[822d31b]	2503	pw = self.cert_pwd
[725c55d]	2504
	2505	# Stop everyone. NB, wait_for_all waits until a thread starts
	2506	# and then completes, so we can't wait if nothing starts. So,
	2507	# no tbparams, no start.
	2508	if len(tbparams) > 0:
	2509	thread_pool = self.thread_pool(self.nthreads)
	2510	for k in tbparams.keys():
	2511	# Create and start a thread to stop the segment
	2512	thread_pool.wait_for_slot()
	2513	uri, aid = tbparams[k]
	2514	t = self.pooled_thread(\
	2515	target=self.terminate_segment(log=dealloc_log,
	2516	testbed=uri,
	2517	cert_file=cert_file,
	2518	cert_pwd=pw,
	2519	trusted_certs=self.trusted_certs,
	2520	caller=self.call_TerminateSegment),
	2521	args=(uri, aid), name=k,
	2522	pdata=thread_pool, trace_file=self.trace_file)
	2523	t.start()
	2524	# Wait for completions
	2525	thread_pool.wait_for_all_done()
	2526
	2527	# release the allocations (failed experiments have done this
	2528	# already, and starting experiments may be in odd states, so we
	2529	# ignore errors releasing those allocations
	2530	try:
	2531	for k in tbparams.keys():
	2532	# This releases access by uri
	2533	uri, aid = tbparams[k]
	2534	self.release_access(None, aid, uri=uri,
	2535	cert_file=cert_file, cert_pwd=pw)
	2536	except service_error, e:
	2537	if status != 'failed' and not force:
	2538	raise e
	2539
	2540	# Clean up the tmpdir no matter what
	2541	finally:
	2542	self.remove_dirs(tmpdir)
[5ae3857]	2543
	2544	# Remove the terminated experiment
	2545	self.state_lock.acquire()
	2546	for id in ids:
	2547	if self.state.has_key(id): del self.state[id]
	2548
	2549	if self.state_filename: self.write_state()
	2550	self.state_lock.release()
	2551
[2761484]	2552	# Delete any synch points associated with this experiment. All
	2553	# synch points begin with the fedid of the experiment.
	2554	fedid_keys = set(["fedid:%s" % f for f in ids \
	2555	if isinstance(f, fedid)])
	2556	for k in self.synch_store.all_keys():
	2557	try:
	2558	if len(k) > 45 and k[0:46] in fedid_keys:
	2559	self.synch_store.del_value(k)
[dadc4da]	2560	except synch_store.BadDeletionError:
[2761484]	2561	pass
	2562	self.write_store()
[e07c8f3]	2563
	2564	# Remove software and other cached stuff from the filesystem.
	2565	if repo:
	2566	self.remove_dirs("%s/%s" % (self.repodir, repo))
[2761484]	2567
[5ae3857]	2568	return {
	2569	'experiment': exp ,
	2570	'deallocationLog': "".join(dealloc_list),
	2571	}
	2572	else:
	2573	# Don't forget to release the lock
	2574	self.state_lock.release()
	2575	raise service_error(service_error.req, "No saved state")
[2761484]	2576
	2577
	2578	def GetValue(self, req, fid):
	2579	"""
	2580	Get a value from the synchronized store
	2581	"""
	2582	req = req.get('GetValueRequestBody', None)
	2583	if not req:
	2584	raise service_error(service_error.req,
	2585	"Bad request format (no GetValueRequestBody)")
	2586
	2587	name = req['name']
	2588	wait = req['wait']
	2589	rv = { 'name': name }
	2590
	2591	if self.auth.check_attribute(fid, name):
[d8442da]	2592	self.log.debug("[GetValue] asking for %s " % name)
[dadc4da]	2593	try:
	2594	v = self.synch_store.get_value(name, wait)
	2595	except synch_store.RevokedKeyError:
	2596	# No more synch on this key
	2597	raise service_error(service_error.federant,
	2598	"Synch key %s revoked" % name)
[2761484]	2599	if v is not None:
	2600	rv['value'] = v
[109a32a]	2601	self.log.debug("[GetValue] got %s from %s" % (v, name))
[2761484]	2602	return rv
	2603	else:
	2604	raise service_error(service_error.access, "Access Denied")
	2605
	2606
	2607	def SetValue(self, req, fid):
	2608	"""
	2609	Set a value in the synchronized store
	2610	"""
	2611	req = req.get('SetValueRequestBody', None)
	2612	if not req:
	2613	raise service_error(service_error.req,
	2614	"Bad request format (no SetValueRequestBody)")
	2615
	2616	name = req['name']
	2617	v = req['value']
	2618
	2619	if self.auth.check_attribute(fid, name):
	2620	try:
	2621	self.synch_store.set_value(name, v)
	2622	self.write_store()
[109a32a]	2623	self.log.debug("[SetValue] set %s to %s" % (name, v))
[2761484]	2624	except synch_store.CollisionError:
	2625	# Translate into a service_error
	2626	raise service_error(service_error.req,
	2627	"Value already set: %s" %name)
[dadc4da]	2628	except synch_store.RevokedKeyError:
	2629	# No more synch on this key
	2630	raise service_error(service_error.federant,
	2631	"Synch key %s revoked" % name)
[2761484]	2632	return { 'name': name, 'value': v }
	2633	else:
	2634	raise service_error(service_error.access, "Access Denied")

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

Context Navigation

source: fedd/federation/experiment_control.py @ 9252414

Download in other formats: