Context Navigation

source: fedd/federation/experiment_control.py @ 4909fcf

compt_changesinfo-ops

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

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

Download in other formats: