Context Navigation

source: fedd/federation/experiment_control.py @ 8f1db21

compt_changesinfo-ops

Last change on this file since 8f1db21 was d0912be, checked in by Ted Faber <faber@…>, 14 years ago
Need to keep track of service records to update.
Property mode set to `100644`
File size: 83.9 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
	697	dot = Popen([neato, '-Gstart=rand', '-Gepsilon=0.005', '-Gmaxiter=2000',
[886307f]	698	'-Gpack=true', dotname], stdout=PIPE, stderr=dnull,
[db6b092]	699	close_fds=True)
	700	dnull.close()
[866c983]	701
	702	# Translate dot to vis format
	703	vis_nodes = [ ]
	704	vis = { 'node': vis_nodes }
	705	for line in dot.stdout:
	706	m = vis_re.match(line)
	707	if m:
	708	vn = m.group(1)
	709	vis_node = {'name': vn, \
	710	'x': float(m.group(2)),\
	711	'y' : float(m.group(3)),\
	712	}
	713	if vn in links.keys() or vn in lans.keys():
	714	vis_node['type'] = 'lan'
	715	else:
	716	vis_node['type'] = 'node'
	717	vis_nodes.append(vis_node)
	718	rv = dot.wait()
	719
	720	os.remove(dotname)
	721	if rv == 0 : return vis
	722	else: return None
[d0ae12d]	723
[db6b092]	724
[725c55d]	725	def release_access(self, tb, aid, tbmap=None, uri=None, cert_file=None,
	726	cert_pwd=None):
[e19b75c]	727	"""
	728	Release access to testbed through fedd
	729	"""
[db6b092]	730
[fd07c48]	731	if not uri and tbmap:
	732	uri = tbmap.get(tb, None)
[e19b75c]	733	if not uri:
[69692a9]	734	raise service_error(service_error.server_config,
[e19b75c]	735	"Unknown testbed: %s" % tb)
[db6b092]	736
[e19b75c]	737	if self.local_access.has_key(uri):
	738	resp = self.local_access[uri].ReleaseAccess(\
	739	{ 'ReleaseAccessRequestBody' : {'allocID': aid},},
[725c55d]	740	fedid(file=cert_file))
[e19b75c]	741	resp = { 'ReleaseAccessResponseBody': resp }
	742	else:
	743	resp = self.call_ReleaseAccess(uri, {'allocID': aid},
[725c55d]	744	cert_file, cert_pwd, self.trusted_certs)
[db6b092]	745
[e19b75c]	746	# better error coding
[db6b092]	747
[5f6929a]	748	def remote_ns2topdl(self, uri, desc):
[db6b092]	749
[e19b75c]	750	req = {
	751	'description' : { 'ns2description': desc },
[db6b092]	752	}
	753
[5f6929a]	754	r = self.call_Ns2Topdl(uri, req, self.cert_file, self.cert_pwd,
[e19b75c]	755	self.trusted_certs)
	756
[5f6929a]	757	if r.has_key('Ns2TopdlResponseBody'):
	758	r = r['Ns2TopdlResponseBody']
[1dcaff4]	759	ed = r.get('experimentdescription', None)
	760	if ed.has_key('topdldescription'):
	761	return topdl.Topology(**ed['topdldescription'])
[e19b75c]	762	else:
	763	raise service_error(service_error.protocol,
	764	"Bad splitter response (no output)")
	765	else:
	766	raise service_error(service_error.protocol, "Bad splitter response")
[cc8d8e9]	767
[e19b75c]	768	class start_segment:
[fd556d1]	769	def __init__(self, debug=False, log=None, testbed="", cert_file=None,
[f07fa49]	770	cert_pwd=None, trusted_certs=None, caller=None,
	771	log_collector=None):
[cc8d8e9]	772	self.log = log
	773	self.debug = debug
	774	self.cert_file = cert_file
	775	self.cert_pwd = cert_pwd
	776	self.trusted_certs = None
	777	self.caller = caller
[fd556d1]	778	self.testbed = testbed
[f07fa49]	779	self.log_collector = log_collector
[69692a9]	780	self.response = None
[b4b19c7]	781	self.node = { }
[e83f2f2]	782	self.proof = None
[b4b19c7]	783
	784	def make_map(self, resp):
[c5869ef]	785	for e in resp.get('embedding', []):
	786	if 'toponame' in e and 'physname' in e:
	787	self.node[e['toponame']] = e['physname'][0]
[cc8d8e9]	788
[43197eb]	789	def __call__(self, uri, aid, topo, masters, attrs=None, connInfo=None):
[cc8d8e9]	790	req = {
	791	'allocID': { 'fedid' : aid },
	792	'segmentdescription': {
	793	'topdldescription': topo.to_dict(),
	794	},
	795	}
[e02cd14]	796
	797	if connInfo:
	798	req['connection'] = connInfo
[43197eb]	799
	800	import_svcs = [ s for m in masters.values() \
	801	for s in m if self.testbed in s.importers]
	802
	803	if import_svcs or self.testbed in masters:
	804	req['service'] = []
	805
	806	for s in import_svcs:
	807	for r in s.reqs:
	808	sr = copy.deepcopy(r)
	809	sr['visibility'] = 'import';
	810	req['service'].append(sr)
	811
	812	for s in masters.get(self.testbed, []):
	813	for r in s.reqs:
	814	sr = copy.deepcopy(r)
	815	sr['visibility'] = 'export';
	816	req['service'].append(sr)
	817
[6c57fe9]	818	if attrs:
	819	req['fedAttr'] = attrs
[cc8d8e9]	820
[fd556d1]	821	try:
[13e3dd2]	822	self.log.debug("Calling StartSegment at %s " % uri)
[fd556d1]	823	r = self.caller(uri, req, self.cert_file, self.cert_pwd,
	824	self.trusted_certs)
[f07fa49]	825	if r.has_key('StartSegmentResponseBody'):
	826	lval = r['StartSegmentResponseBody'].get('allocationLog',
	827	None)
	828	if lval and self.log_collector:
	829	for line in lval.splitlines(True):
	830	self.log_collector.write(line)
[b4b19c7]	831	self.make_map(r['StartSegmentResponseBody'])
[e83f2f2]	832	if 'proof' in r: self.proof = r['proof']
[69692a9]	833	self.response = r
[f07fa49]	834	else:
	835	raise service_error(service_error.internal,
	836	"Bad response!?: %s" %r)
[fd556d1]	837	return True
	838	except service_error, e:
	839	self.log.error("Start segment failed on %s: %s" % \
	840	(self.testbed, e))
	841	return False
[cc8d8e9]	842
	843
[5ae3857]	844
[e19b75c]	845	class terminate_segment:
[fd556d1]	846	def __init__(self, debug=False, log=None, testbed="", cert_file=None,
[5ae3857]	847	cert_pwd=None, trusted_certs=None, caller=None):
	848	self.log = log
	849	self.debug = debug
	850	self.cert_file = cert_file
	851	self.cert_pwd = cert_pwd
	852	self.trusted_certs = None
	853	self.caller = caller
[fd556d1]	854	self.testbed = testbed
[5ae3857]	855
	856	def __call__(self, uri, aid ):
	857	req = {
	858	'allocID': aid ,
	859	}
[fd556d1]	860	try:
	861	r = self.caller(uri, req, self.cert_file, self.cert_pwd,
	862	self.trusted_certs)
	863	return True
	864	except service_error, e:
	865	self.log.error("Terminate segment failed on %s: %s" % \
	866	(self.testbed, e))
	867	return False
[db6b092]	868
[43197eb]	869	def allocate_resources(self, allocated, masters, eid, expid,
[b4b19c7]	870	tbparams, top, topo, tmpdir, alloc_log=None, log_collector=None,
[c573278]	871	attrs=None, connInfo={}, tbmap=None, expcert=None):
[69692a9]	872
[cc8d8e9]	873	started = { } # Testbeds where a sub-experiment started
	874	# successfully
	875
	876	# XXX
	877	fail_soft = False
	878
[fd07c48]	879	if tbmap is None: tbmap = { }
	880
[cc8d8e9]	881	log = alloc_log or self.log
	882
[faea607]	883	tp = thread_pool(self.nthreads)
[cc8d8e9]	884	threads = [ ]
[b4b19c7]	885	starters = [ ]
[cc8d8e9]	886
[c573278]	887	if expcert:
	888	cert = expcert
	889	pw = None
	890	else:
	891	cert = self.cert_file
[822d31b]	892	pw = self.cert_pwd
[c573278]	893
[109a32a]	894	for tb in allocated.keys():
	895	# Create and start a thread to start the segment, and save it
	896	# to get the return value later
[ab847bc]	897	tb_attrs = copy.copy(attrs)
[faea607]	898	tp.wait_for_slot()
[fd07c48]	899	uri = tbparams[tb].get('uri', tbmap.get(testbed_base(tb), None))
[ab847bc]	900	base, suffix = split_testbed(tb)
	901	if suffix:
	902	tb_attrs.append({'attribute': 'experiment_name',
[175b444]	903	'value': "%s-%s" % (eid, suffix)})
[ab847bc]	904	else:
	905	tb_attrs.append({'attribute': 'experiment_name', 'value': eid})
[109a32a]	906	if not uri:
	907	raise service_error(service_error.internal,
	908	"Unknown testbed %s !?" % tb)
	909
[cc8d8e9]	910	if tbparams[tb].has_key('allocID') and \
	911	tbparams[tb]['allocID'].has_key('fedid'):
	912	aid = tbparams[tb]['allocID']['fedid']
	913	else:
	914	raise service_error(service_error.internal,
	915	"No alloc id for testbed %s !?" % tb)
	916
[b4b19c7]	917	s = self.start_segment(log=log, debug=self.debug,
[c573278]	918	testbed=tb, cert_file=cert,
	919	cert_pwd=pw, trusted_certs=self.trusted_certs,
[b4b19c7]	920	caller=self.call_StartSegment,
	921	log_collector=log_collector)
	922	starters.append(s)
[faea607]	923	t = pooled_thread(\
[b4b19c7]	924	target=s, name=tb,
[ab847bc]	925	args=(uri, aid, topo[tb], masters, tb_attrs, connInfo[tb]),
[faea607]	926	pdata=tp, trace_file=self.trace_file)
[69692a9]	927	threads.append(t)
	928	t.start()
[cc8d8e9]	929
[109a32a]	930	# Wait until all finish (keep pinging the log, though)
	931	mins = 0
[dadc4da]	932	revoked = False
[faea607]	933	while not tp.wait_for_all_done(60.0):
[109a32a]	934	mins += 1
	935	alloc_log.info("Waiting for sub threads (it has been %d mins)" \
	936	% mins)
[dadc4da]	937	if not revoked and \
[f52f5df]	938	len([ t.getName() for t in threads if t.rv == False]) > 0:
[dadc4da]	939	# a testbed has failed. Revoke this experiment's
	940	# synchronizarion values so that sub experiments will not
	941	# deadlock waiting for synchronization that will never happen
	942	self.log.info("A subexperiment has failed to swap in, " + \
	943	"revoking synch keys")
	944	var_key = "fedid:%s" % expid
	945	for k in self.synch_store.all_keys():
	946	if len(k) > 45 and k[0:46] == var_key:
	947	self.synch_store.revoke_key(k)
	948	revoked = True
[69692a9]	949
[cc8d8e9]	950	failed = [ t.getName() for t in threads if not t.rv ]
	951	succeeded = [tb for tb in allocated.keys() if tb not in failed]
[3132419]	952
[cc8d8e9]	953	# If one failed clean up, unless fail_soft is set
[32e7d93]	954	if failed:
[cc8d8e9]	955	if not fail_soft:
[faea607]	956	tp.clear()
[cc8d8e9]	957	for tb in succeeded:
	958	# Create and start a thread to stop the segment
[faea607]	959	tp.wait_for_slot()
[0fa1729]	960	uri = tbparams[tb]['uri']
[faea607]	961	t = pooled_thread(\
[32e7d93]	962	target=self.terminate_segment(log=log,
[fd556d1]	963	testbed=tb,
[696a689]	964	cert_file=cert,
	965	cert_pwd=pw,
[32e7d93]	966	trusted_certs=self.trusted_certs,
	967	caller=self.call_TerminateSegment),
	968	args=(uri, tbparams[tb]['federant']['allocID']),
	969	name=tb,
[faea607]	970	pdata=tp, trace_file=self.trace_file)
[cc8d8e9]	971	t.start()
[f52f5df]	972	# Wait until all finish (if any are being stopped)
	973	if succeeded:
[faea607]	974	tp.wait_for_all_done()
[cc8d8e9]	975
	976	# release the allocations
	977	for tb in tbparams.keys():
[725c55d]	978	try:
	979	self.release_access(tb, tbparams[tb]['allocID'],
[696a689]	980	tbmap=tbmap, uri=tbparams[tb].get('uri', None),
	981	cert_file=cert, cert_pwd=pw)
[725c55d]	982	except service_error, e:
	983	self.log.warn("Error releasing access: %s" % e.desc)
[cc8d8e9]	984	# Remove the placeholder
	985	self.state_lock.acquire()
	986	self.state[eid]['experimentStatus'] = 'failed'
	987	if self.state_filename: self.write_state()
	988	self.state_lock.release()
[05e8da8]	989	# Remove the repo dir
	990	self.remove_dirs("%s/%s" %(self.repodir, expid))
	991	# Walk up tmpdir, deleting as we go
	992	if self.cleanup:
	993	self.remove_dirs(tmpdir)
	994	else:
	995	log.debug("[start_experiment]: not removing %s" % tmpdir)
	996
[cc8d8e9]	997
	998	log.error("Swap in failed on %s" % ",".join(failed))
	999	return
	1000	else:
[b4b19c7]	1001	# Walk through the successes and gather the virtual to physical
	1002	# mapping.
[2fd8f8c]	1003	embedding = [ ]
[e83f2f2]	1004	proofs = { }
[b4b19c7]	1005	for s in starters:
[2fd8f8c]	1006	for k, v in s.node.items():
	1007	embedding.append({
	1008	'toponame': k,
	1009	'physname': [ v],
	1010	'testbed': s.testbed
	1011	})
[e83f2f2]	1012	if s.proof:
	1013	proofs[s.testbed] = s.proof
[cc8d8e9]	1014	log.info("[start_segment]: Experiment %s active" % eid)
	1015
	1016
	1017	# Walk up tmpdir, deleting as we go
[69692a9]	1018	if self.cleanup:
[05e8da8]	1019	self.remove_dirs(tmpdir)
[69692a9]	1020	else:
	1021	log.debug("[start_experiment]: not removing %s" % tmpdir)
[cc8d8e9]	1022
[b4b19c7]	1023	# Insert the experiment into our state and update the disk copy.
[cc8d8e9]	1024	self.state_lock.acquire()
	1025	self.state[expid]['experimentStatus'] = 'active'
	1026	self.state[eid] = self.state[expid]
[b4b19c7]	1027	self.state[eid]['experimentdescription']['topdldescription'] = \
	1028	top.to_dict()
[2fd8f8c]	1029	self.state[eid]['embedding'] = embedding
[e83f2f2]	1030	# Append startup proofs
	1031	for f in self.state[eid]['federant']:
	1032	if 'name' in f and 'localname' in f['name']:
	1033	if f['name']['localname'] in proofs:
	1034	f['proof'].append(proofs[f['name']['localname']])
	1035
[cc8d8e9]	1036	if self.state_filename: self.write_state()
	1037	self.state_lock.release()
	1038	return
	1039
	1040
[895a133]	1041	def add_kit(self, e, kit):
	1042	"""
	1043	Add a Software object created from the list of (install, location)
	1044	tuples passed as kit to the software attribute of an object e. We
	1045	do this enough to break out the code, but it's kind of a hack to
	1046	avoid changing the old tuple rep.
	1047	"""
	1048
	1049	s = [ topdl.Software(install=i, location=l) for i, l in kit]
	1050
	1051	if isinstance(e.software, list): e.software.extend(s)
	1052	else: e.software = s
	1053
[913dc7a]	1054	def append_experiment_authorization(self, expid, attrs,
	1055	need_state_lock=True):
	1056	"""
	1057	Append the authorization information to system state
	1058	"""
	1059
	1060	for p, a in attrs:
	1061	self.auth.set_attribute(p, a)
	1062	self.auth.save()
	1063
	1064	if need_state_lock: self.state_lock.acquire()
	1065	self.state[expid]['auth'].update(attrs)
	1066	if self.state_filename: self.write_state()
	1067	if need_state_lock: self.state_lock.release()
	1068
[a96d946]	1069	def clear_experiment_authorization(self, expid, need_state_lock=True):
[913dc7a]	1070	"""
	1071	Attrs is a set of attribute principal pairs that need to be removed
	1072	from the authenticator. Remove them and save the authenticator.
	1073	"""
	1074
	1075	if need_state_lock: self.state_lock.acquire()
[b67fd22]	1076	if expid in self.state and 'auth' in self.state[expid]:
	1077	for p, a in self.state[expid]['auth']:
	1078	self.auth.unset_attribute(p, a)
	1079	self.state[expid]['auth'] = set()
[913dc7a]	1080	if self.state_filename: self.write_state()
	1081	if need_state_lock: self.state_lock.release()
[b67fd22]	1082	self.auth.save()
[913dc7a]	1083
[895a133]	1084
[b4b19c7]	1085	def create_experiment_state(self, fid, req, expid, expcert,
[a3ad8bd]	1086	state='starting'):
[895a133]	1087	"""
	1088	Create the initial entry in the experiment's state. The expid and
	1089	expcert are the experiment's fedid and certifacte that represents that
	1090	ID, which are installed in the experiment state. If the request
	1091	includes a suggested local name that is used if possible. If the local
	1092	name is already taken by an experiment owned by this user that has
[a3ad8bd]	1093	failed, it is overwritten. Otherwise new letters are added until a
[895a133]	1094	valid localname is found. The generated local name is returned.
	1095	"""
	1096
	1097	if req.has_key('experimentID') and \
	1098	req['experimentID'].has_key('localname'):
	1099	overwrite = False
	1100	eid = req['experimentID']['localname']
	1101	# If there's an old failed experiment here with the same local name
	1102	# and accessible by this user, we'll overwrite it, otherwise we'll
	1103	# fall through and do the collision avoidance.
	1104	old_expid = self.get_experiment_fedid(eid)
	1105	if old_expid and self.check_experiment_access(fid, old_expid):
	1106	self.state_lock.acquire()
	1107	status = self.state[eid].get('experimentStatus', None)
	1108	if status and status == 'failed':
[913dc7a]	1109	# remove the old access attributes
[b67fd22]	1110	self.clear_experiment_authorization(eid,
[913dc7a]	1111	need_state_lock=False)
[895a133]	1112	overwrite = True
	1113	del self.state[eid]
	1114	del self.state[old_expid]
	1115	self.state_lock.release()
	1116	self.state_lock.acquire()
	1117	while (self.state.has_key(eid) and not overwrite):
	1118	eid += random.choice(string.ascii_letters)
	1119	# Initial state
	1120	self.state[eid] = {
	1121	'experimentID' : \
	1122	[ { 'localname' : eid }, {'fedid': expid } ],
[a3ad8bd]	1123	'experimentStatus': state,
[895a133]	1124	'experimentAccess': { 'X509' : expcert },
	1125	'owner': fid,
	1126	'log' : [],
[913dc7a]	1127	'auth': set(),
[895a133]	1128	}
	1129	self.state[expid] = self.state[eid]
[913dc7a]	1130	if self.state_filename: self.write_state()
	1131	self.state_lock.release()
[895a133]	1132	else:
	1133	eid = self.exp_stem
	1134	for i in range(0,5):
	1135	eid += random.choice(string.ascii_letters)
	1136	self.state_lock.acquire()
	1137	while (self.state.has_key(eid)):
	1138	eid = self.exp_stem
	1139	for i in range(0,5):
	1140	eid += random.choice(string.ascii_letters)
	1141	# Initial state
	1142	self.state[eid] = {
	1143	'experimentID' : \
	1144	[ { 'localname' : eid }, {'fedid': expid } ],
[a3ad8bd]	1145	'experimentStatus': state,
[895a133]	1146	'experimentAccess': { 'X509' : expcert },
	1147	'owner': fid,
	1148	'log' : [],
[913dc7a]	1149	'auth': set(),
[895a133]	1150	}
	1151	self.state[expid] = self.state[eid]
[913dc7a]	1152	if self.state_filename: self.write_state()
	1153	self.state_lock.release()
	1154
	1155	# Let users touch the state. Authorize this fid and the expid itself
	1156	# to touch the experiment, as well as allowing th eoverrides.
	1157	self.append_experiment_authorization(eid,
	1158	set([(fid, expid), (expid,expid)] + \
	1159	[ (o, expid) for o in self.overrides]))
[895a133]	1160
	1161	return eid
	1162
	1163
	1164	def allocate_ips_to_topo(self, top):
	1165	"""
[69692a9]	1166	Add an ip4_address attribute to all the hosts in the topology, based on
[895a133]	1167	the shared substrates on which they sit. An /etc/hosts file is also
[69692a9]	1168	created and returned as a list of hostfiles entries. We also return
	1169	the allocator, because we may need to allocate IPs to portals
	1170	(specifically DRAGON portals).
[895a133]	1171	"""
	1172	subs = sorted(top.substrates,
	1173	cmp=lambda x,y: cmp(len(x.interfaces), len(y.interfaces)),
	1174	reverse=True)
	1175	ips = ip_allocator(int(ip_addr("10.0.0.0")), 2 **24)
	1176	ifs = { }
	1177	hosts = [ ]
	1178
	1179	for idx, s in enumerate(subs):
[289ff7e]	1180	net_size = len(s.interfaces)+2
	1181
	1182	a = ips.allocate(net_size)
[895a133]	1183	if a :
	1184	base, num = a
[289ff7e]	1185	if num < net_size:
[895a133]	1186	raise service_error(service_error.internal,
	1187	"Allocator returned wrong number of IPs??")
	1188	else:
	1189	raise service_error(service_error.req,
	1190	"Cannot allocate IP addresses")
[062b991]	1191	mask = ips.min_alloc
	1192	while mask < net_size:
	1193	mask *= 2
[289ff7e]	1194
[062b991]	1195	netmask = ((2**32-1) ^ (mask-1))
[895a133]	1196
	1197	base += 1
	1198	for i in s.interfaces:
	1199	i.attribute.append(
	1200	topdl.Attribute('ip4_address',
	1201	"%s" % ip_addr(base)))
[289ff7e]	1202	i.attribute.append(
	1203	topdl.Attribute('ip4_netmask',
	1204	"%s" % ip_addr(int(netmask))))
	1205
[1e7f268]	1206	hname = i.element.name
[895a133]	1207	if ifs.has_key(hname):
	1208	hosts.append("%s\t%s-%s %s-%d" % \
	1209	(ip_addr(base), hname, s.name, hname,
	1210	ifs[hname]))
	1211	else:
	1212	ifs[hname] = 0
	1213	hosts.append("%s\t%s-%s %s-%d %s" % \
	1214	(ip_addr(base), hname, s.name, hname,
	1215	ifs[hname], hname))
	1216
	1217	ifs[hname] += 1
	1218	base += 1
[69692a9]	1219	return hosts, ips
[895a133]	1220
[1d73342]	1221	def get_access_to_testbeds(self, testbeds, fid, allocated,
[725c55d]	1222	tbparam, masters, tbmap, expid=None, expcert=None):
[6e63513]	1223	for tb in testbeds:
[1d73342]	1224	self.get_access(tb, tbparam, fid, masters, tbmap, expid,
[c573278]	1225	expcert)
[6e63513]	1226	allocated[tb] = 1
	1227
[1d73342]	1228	def get_access(self, tb, tbparam,fid, masters, tbmap, expid=None,
	1229	expcert=None):
[6e63513]	1230	"""
	1231	Get access to testbed through fedd and set the parameters for that tb
	1232	"""
	1233	def get_export_project(svcs):
	1234	"""
	1235	Look through for the list of federated_service for this testbed
	1236	objects for a project_export service, and extract the project
	1237	parameter.
	1238	"""
	1239
	1240	pe = [s for s in svcs if s.name=='project_export']
	1241	if len(pe) == 1:
	1242	return pe[0].params.get('project', None)
	1243	elif len(pe) == 0:
	1244	return None
	1245	else:
	1246	raise service_error(service_error.req,
	1247	"More than one project export is not supported")
	1248
[d0912be]	1249	def add_services(svcs, type, slist, keys):
[63c6664]	1250	"""
[d0912be]	1251	Add the given services to slist. type is import or export. Also
	1252	add a mapping entry from the assigned id to the original service
	1253	record.
[63c6664]	1254	"""
	1255	for i, s in enumerate(svcs):
	1256	idx = '%s%d' % (type, i)
[d0912be]	1257	keys[idx] = s
[63c6664]	1258	sr = {'id': idx, 'name': s.name, 'visibility': type }
	1259	if s.params:
	1260	sr['fedAttr'] = [ { 'attribute': k, 'value': v } \
	1261	for k, v in s.params.items()]
	1262	slist.append(sr)
	1263
[6e63513]	1264	uri = tbmap.get(testbed_base(tb), None)
	1265	if not uri:
	1266	raise service_error(service_error.server_config,
	1267	"Unknown testbed: %s" % tb)
	1268
	1269	export_svcs = masters.get(tb,[])
	1270	import_svcs = [ s for m in masters.values() \
	1271	for s in m \
	1272	if tb in s.importers ]
	1273
	1274	export_project = get_export_project(export_svcs)
	1275	# Compose the credential list so that IDs come before attributes
	1276	creds = set()
	1277	keys = set()
[c573278]	1278	certs = self.auth.get_creds_for_principal(fid)
	1279	if expid:
	1280	certs.update(self.auth.get_creds_for_principal(expid))
	1281	for c in certs:
[6e63513]	1282	keys.add(c.issuer_cert())
	1283	creds.add(c.attribute_cert())
	1284	creds = list(keys) + list(creds)
	1285
[c573278]	1286	if expcert: cert, pw = expcert, None
	1287	else: cert, pw = self.cert_file, self.cert_pw
	1288
[6e63513]	1289	# Request credentials
	1290	req = {
	1291	'abac_credential': creds,
	1292	}
	1293	# Make the service request from the services we're importing and
	1294	# exporting. Keep track of the export request ids so we can
	1295	# collect the resulting info from the access response.
	1296	e_keys = { }
	1297	if import_svcs or export_svcs:
[63c6664]	1298	slist = []
[d0912be]	1299	add_services(import_svcs, 'import', slist, e_keys)
	1300	add_services(export_svcs, 'export', slist, e_keys)
[63c6664]	1301	req['service'] = slist
[6e63513]	1302
	1303	if self.local_access.has_key(uri):
	1304	# Local access call
	1305	req = { 'RequestAccessRequestBody' : req }
	1306	r = self.local_access[uri].RequestAccess(req,
	1307	fedid(file=self.cert_file))
	1308	r = { 'RequestAccessResponseBody' : r }
	1309	else:
[c573278]	1310	r = self.call_RequestAccess(uri, req, cert, pw, self.trusted_certs)
[6e63513]	1311
[725c55d]	1312	if r.has_key('RequestAccessResponseBody'):
	1313	# Through to here we have a valid response, not a fault.
	1314	# Access denied is a fault, so something better or worse than
	1315	# access denied has happened.
	1316	r = r['RequestAccessResponseBody']
	1317	self.log.debug("[get_access] Access granted")
	1318	else:
	1319	raise service_error(service_error.protocol,
	1320	"Bad proxy response")
[e83f2f2]	1321	if 'proof' not in r:
	1322	raise service_error(service_error.protocol,
	1323	"Bad access response (no access proof)")
[725c55d]	1324
[6e63513]	1325	tbparam[tb] = {
	1326	"allocID" : r['allocID'],
	1327	"uri": uri,
[e83f2f2]	1328	"proof": [ r['proof'] ],
[6e63513]	1329	}
	1330
	1331	# Collect the responses corresponding to the services this testbed
	1332	# exports. These will be the service requests that we will include in
	1333	# the start segment requests (with appropriate visibility values) to
	1334	# import and export the segments.
	1335	for s in r.get('service', []):
	1336	id = s.get('id', None)
	1337	if id and id in e_keys:
	1338	e_keys[id].reqs.append(s)
	1339
	1340	# Add attributes to parameter space. We don't allow attributes to
	1341	# overlay any parameters already installed.
	1342	for a in r.get('fedAttr', []):
	1343	try:
	1344	if a['attribute'] and \
	1345	isinstance(a['attribute'], basestring)\
	1346	and not tbparam[tb].has_key(a['attribute'].lower()):
	1347	tbparam[tb][a['attribute'].lower()] = a['value']
	1348	except KeyError:
	1349	self.log.error("Bad attribute in response: %s" % a)
	1350
	1351
[7fe81be]	1352	def split_topology(self, top, topo, testbeds):
[895a133]	1353	"""
[e02cd14]	1354	Create the sub-topologies that are needed for experiment instantiation.
[895a133]	1355	"""
	1356	for tb in testbeds:
	1357	topo[tb] = top.clone()
[7fe81be]	1358	# copy in for loop allows deletions from the original
	1359	for e in [ e for e in topo[tb].elements]:
[895a133]	1360	etb = e.get_attribute('testbed')
[7fe81be]	1361	# NB: elements without a testbed attribute won't appear in any
	1362	# sub topologies.
	1363	if not etb or etb != tb:
[895a133]	1364	for i in e.interface:
	1365	for s in i.subs:
	1366	try:
	1367	s.interfaces.remove(i)
	1368	except ValueError:
	1369	raise service_error(service_error.internal,
	1370	"Can't remove interface??")
[7fe81be]	1371	topo[tb].elements.remove(e)
[895a133]	1372	topo[tb].make_indices()
	1373
[2627eb3]	1374	def confirm_software(self, top):
	1375	"""
	1376	Make sure that the software to be loaded in the topo is all available
	1377	before we begin making access requests, etc. This is a subset of
	1378	wrangle_software.
	1379	"""
	1380	pkgs = set([ d for i, d in self.fedkit + self.gatewaykit ])
	1381	pkgs.update([x.location for e in top.elements for x in e.software])
	1382
	1383	for pkg in pkgs:
	1384	loc = pkg
	1385
	1386	scheme, host, path = urlparse(loc)[0:3]
	1387	dest = os.path.basename(path)
	1388	if not scheme:
	1389	if not loc.startswith('/'):
	1390	loc = "/%s" % loc
	1391	loc = "file://%s" %loc
	1392	# NB: if scheme was found, loc == pkg
	1393	try:
	1394	u = urlopen(loc)
	1395	u.close()
	1396	except Exception, e:
	1397	raise service_error(service_error.req,
	1398	"Cannot open %s: %s" % (loc, e))
	1399	return True
	1400
[895a133]	1401	def wrangle_software(self, expid, top, topo, tbparams):
	1402	"""
	1403	Copy software out to the repository directory, allocate permissions and
	1404	rewrite the segment topologies to look for the software in local
	1405	places.
	1406	"""
	1407
	1408	# Copy the rpms and tarfiles to a distribution directory from
	1409	# which the federants can retrieve them
	1410	linkpath = "%s/software" % expid
	1411	softdir ="%s/%s" % ( self.repodir, linkpath)
	1412	softmap = { }
[2627eb3]	1413
	1414	# self.fedkit and self.gateway kit are lists of tuples of
	1415	# (install_location, download_location) this extracts the download
	1416	# locations.
	1417	pkgs = set([ d for i, d in self.fedkit + self.gatewaykit ])
	1418	pkgs.update([x.location for e in top.elements for x in e.software])
[895a133]	1419	try:
	1420	os.makedirs(softdir)
[d3c8759]	1421	except EnvironmentError, e:
[895a133]	1422	raise service_error(
	1423	"Cannot create software directory: %s" % e)
	1424	# The actual copying. Everything's converted into a url for copying.
[913dc7a]	1425	auth_attrs = set()
[895a133]	1426	for pkg in pkgs:
	1427	loc = pkg
	1428
	1429	scheme, host, path = urlparse(loc)[0:3]
	1430	dest = os.path.basename(path)
	1431	if not scheme:
	1432	if not loc.startswith('/'):
	1433	loc = "/%s" % loc
	1434	loc = "file://%s" %loc
[2627eb3]	1435	# NB: if scheme was found, loc == pkg
[895a133]	1436	try:
	1437	u = urlopen(loc)
	1438	except Exception, e:
	1439	raise service_error(service_error.req,
	1440	"Cannot open %s: %s" % (loc, e))
	1441	try:
	1442	f = open("%s/%s" % (softdir, dest) , "w")
	1443	self.log.debug("Writing %s/%s" % (softdir,dest) )
	1444	data = u.read(4096)
	1445	while data:
	1446	f.write(data)
	1447	data = u.read(4096)
	1448	f.close()
	1449	u.close()
	1450	except Exception, e:
	1451	raise service_error(service_error.internal,
	1452	"Could not copy %s: %s" % (loc, e))
	1453	path = re.sub("/tmp", "", linkpath)
	1454	# XXX
	1455	softmap[pkg] = \
[7183b48]	1456	"%s/%s/%s" %\
	1457	( self.repo_url, path, dest)
[895a133]	1458
[913dc7a]	1459	# Allow the individual segments to access the software by assigning
	1460	# an attribute to each testbed allocation that encodes the data to
	1461	# be released. This expression collects the data for each run of
	1462	# the loop.
	1463	auth_attrs.update([
	1464	(tbparams[tb]['allocID']['fedid'], "/%s/%s" % ( path, dest)) \
	1465	for tb in tbparams.keys()])
	1466
	1467	self.append_experiment_authorization(expid, auth_attrs)
[895a133]	1468
	1469	# Convert the software locations in the segments into the local
	1470	# copies on this host
	1471	for soft in [ s for tb in topo.values() \
	1472	for e in tb.elements \
	1473	if getattr(e, 'software', False) \
	1474	for s in e.software ]:
	1475	if softmap.has_key(soft.location):
	1476	soft.location = softmap[soft.location]
	1477
	1478
[a3ad8bd]	1479	def new_experiment(self, req, fid):
	1480	"""
	1481	The external interface to empty initial experiment creation called from
	1482	the dispatcher.
	1483
	1484	Creates a working directory, splits the incoming description using the
	1485	splitter script and parses out the avrious subsections using the
	1486	lcasses above. Once each sub-experiment is created, use pooled threads
	1487	to instantiate them and start it all up.
	1488	"""
[7206e5a]	1489	req = req.get('NewRequestBody', None)
	1490	if not req:
	1491	raise service_error(service_error.req,
	1492	"Bad request format (no NewRequestBody)")
	1493
	1494	if self.auth.import_credentials(data_list=req.get('credential', [])):
	1495	self.auth.save()
[c573278]	1496
[e83f2f2]	1497	access_ok, proof = self.auth.check_attribute(fid, 'new',
	1498	with_proof=True)
	1499
	1500	if not access_ok:
	1501	raise service_error(service_error.access, "New access denied",
	1502	proof=[proof])
[a3ad8bd]	1503
	1504	try:
	1505	tmpdir = tempfile.mkdtemp(prefix="split-")
[d3c8759]	1506	except EnvironmentError:
[a3ad8bd]	1507	raise service_error(service_error.internal, "Cannot create tmp dir")
	1508
	1509	try:
	1510	access_user = self.accessdb[fid]
	1511	except KeyError:
	1512	raise service_error(service_error.internal,
	1513	"Access map and authorizer out of sync in " + \
[7183b48]	1514	"new_experiment for fedid %s" % fid)
[a3ad8bd]	1515
	1516	# Generate an ID for the experiment (slice) and a certificate that the
	1517	# allocator can use to prove they own it. We'll ship it back through
[7206e5a]	1518	# the encrypted connection. If the requester supplied one, use it.
	1519	if 'experimentAccess' in req and 'X509' in req['experimentAccess']:
	1520	expcert = req['experimentAccess']['X509']
[962ea25]	1521	expid = fedid(certstr=expcert)
[7206e5a]	1522	self.state_lock.acquire()
	1523	if expid in self.state:
	1524	self.state_lock.release()
	1525	raise service_error(service_error.req,
	1526	'fedid %s identifies an existing experiment' % expid)
	1527	self.state_lock.release()
	1528	else:
	1529	(expid, expcert) = generate_fedid("test", dir=tmpdir, log=self.log)
[a3ad8bd]	1530
	1531	#now we're done with the tmpdir, and it should be empty
	1532	if self.cleanup:
	1533	self.log.debug("[new_experiment]: removing %s" % tmpdir)
	1534	os.rmdir(tmpdir)
	1535	else:
	1536	self.log.debug("[new_experiment]: not removing %s" % tmpdir)
	1537
	1538	eid = self.create_experiment_state(fid, req, expid, expcert,
	1539	state='empty')
	1540
	1541	rv = {
	1542	'experimentID': [
	1543	{'localname' : eid }, { 'fedid': copy.copy(expid) }
	1544	],
	1545	'experimentStatus': 'empty',
[e83f2f2]	1546	'experimentAccess': { 'X509' : expcert },
	1547	'proof': proof.to_dict(),
[a3ad8bd]	1548	}
	1549
	1550	return rv
	1551
[cf0ff4f]	1552	# create_experiment sub-functions
	1553
[5ecb9a3]	1554	@staticmethod
[cf0ff4f]	1555	def get_experiment_key(req, field='experimentID'):
[5ecb9a3]	1556	"""
	1557	Parse the experiment identifiers out of the request (the request body
	1558	tag has been removed). Specifically this pulls either the fedid or the
	1559	localname out of the experimentID field. A fedid is preferred. If
	1560	neither is present or the request does not contain the fields,
	1561	service_errors are raised.
	1562	"""
	1563	# Get the experiment access
[cf0ff4f]	1564	exp = req.get(field, None)
[5ecb9a3]	1565	if exp:
	1566	if exp.has_key('fedid'):
	1567	key = exp['fedid']
	1568	elif exp.has_key('localname'):
	1569	key = exp['localname']
	1570	else:
	1571	raise service_error(service_error.req, "Unknown lookup type")
	1572	else:
	1573	raise service_error(service_error.req, "No request?")
	1574
	1575	return key
	1576
	1577	def get_experiment_ids_and_start(self, key, tmpdir):
	1578	"""
	1579	Get the experiment name, id and access certificate from the state, and
	1580	set the experiment state to 'starting'. returns a triple (fedid,
	1581	localname, access_cert_file). The access_cert_file is a copy of the
	1582	contents of the access certificate, created in the tempdir with
	1583	restricted permissions. If things are confused, raise an exception.
	1584	"""
	1585
	1586	expid = eid = None
	1587	self.state_lock.acquire()
	1588	if self.state.has_key(key):
	1589	self.state[key]['experimentStatus'] = "starting"
	1590	for e in self.state[key].get('experimentID',[]):
	1591	if not expid and e.has_key('fedid'):
	1592	expid = e['fedid']
	1593	elif not eid and e.has_key('localname'):
	1594	eid = e['localname']
	1595	if 'experimentAccess' in self.state[key] and \
	1596	'X509' in self.state[key]['experimentAccess']:
	1597	expcert = self.state[key]['experimentAccess']['X509']
	1598	else:
	1599	expcert = None
	1600	self.state_lock.release()
	1601
	1602	# make a protected copy of the access certificate so the experiment
	1603	# controller can act as the experiment principal.
	1604	if expcert:
	1605	expcert_file = self.make_temp_certfile(expcert, tmpdir)
	1606	if not expcert_file:
	1607	raise service_error(service_error.internal,
	1608	"Cannot create temp cert file?")
	1609	else:
	1610	expcert_file = None
	1611
	1612	return (eid, expid, expcert_file)
	1613
	1614	def get_topology(self, req, tmpdir):
	1615	"""
	1616	Get the ns2 content and put it into a file for parsing. Call the local
	1617	or remote parser and return the topdl.Topology. Errors result in
	1618	exceptions. req is the request and tmpdir is a work directory.
	1619	"""
	1620
	1621	# The tcl parser needs to read a file so put the content into that file
	1622	descr=req.get('experimentdescription', None)
	1623	if descr:
[a7c0bcb]	1624	if 'ns2description' in descr:
	1625	file_content=descr['ns2description']
	1626	elif 'topdldescription' in descr:
	1627	return topdl.Topology(**descr['topdldescription'])
	1628	else:
	1629	raise service_error(service_error.req,
	1630	'Unknown experiment description type')
[5ecb9a3]	1631	else:
	1632	raise service_error(service_error.req, "No experiment description")
	1633
	1634
	1635	if self.splitter_url:
	1636	self.log.debug("Calling remote topdl translator at %s" % \
	1637	self.splitter_url)
	1638	top = self.remote_ns2topdl(self.splitter_url, file_content)
	1639	else:
	1640	tclfile = os.path.join(tmpdir, "experiment.tcl")
	1641	if file_content:
	1642	try:
	1643	f = open(tclfile, 'w')
	1644	f.write(file_content)
	1645	f.close()
	1646	except EnvironmentError:
	1647	raise service_error(service_error.internal,
	1648	"Cannot write temp experiment description")
	1649	else:
	1650	raise service_error(service_error.req,
	1651	"Only ns2descriptions supported")
	1652	pid = "dummy"
	1653	gid = "dummy"
	1654	eid = "dummy"
	1655
	1656	tclcmd = [self.tclsh, self.tcl_splitter, '-t', '-x',
	1657	str(self.muxmax), '-m', 'dummy']
	1658
	1659	tclcmd.extend([pid, gid, eid, tclfile])
	1660
	1661	self.log.debug("running local splitter %s", " ".join(tclcmd))
	1662	# This is just fantastic. As a side effect the parser copies
	1663	# tb_compat.tcl into the current directory, so that directory
	1664	# must be writable by the fedd user. Doing this in the
	1665	# temporary subdir ensures this is the case.
	1666	tclparser = Popen(tclcmd, stdout=PIPE, close_fds=True,
	1667	cwd=tmpdir)
	1668	split_data = tclparser.stdout
	1669
	1670	top = topdl.topology_from_xml(file=split_data, top="experiment")
	1671	os.remove(tclfile)
	1672
	1673	return top
	1674
[1660f7c]	1675	def get_testbed_services(self, req, testbeds):
[5ecb9a3]	1676	"""
	1677	Parse the services section of the request into into two dicts mapping
	1678	testbed to lists of federated_service objects. The first lists all
	1679	exporters of services, and the second all exporters of services that
	1680	need control portals int the experiment.
	1681	"""
	1682	masters = { }
	1683	pmasters = { }
	1684	for s in req.get('service', []):
	1685	# If this is a service request with the importall field
	1686	# set, fill it out.
	1687
	1688	if s.get('importall', False):
	1689	s['import'] = [ tb for tb in testbeds \
	1690	if tb not in s.get('export',[])]
	1691	del s['importall']
	1692
	1693	# Add the service to masters
	1694	for tb in s.get('export', []):
	1695	if s.get('name', None):
	1696
	1697	params = { }
	1698	for a in s.get('fedAttr', []):
	1699	params[a.get('attribute', '')] = a.get('value','')
	1700
	1701	fser = federated_service(name=s['name'],
	1702	exporter=tb, importers=s.get('import',[]),
	1703	params=params)
	1704	if fser.name == 'hide_hosts' \
	1705	and 'hosts' not in fser.params:
	1706	fser.params['hosts'] = \
	1707	",".join(tb_hosts.get(fser.exporter, []))
	1708	if tb in masters: masters[tb].append(fser)
	1709	else: masters[tb] = [fser]
	1710
	1711	if fser.portal:
	1712	if tb not in pmasters: pmasters[tb] = [ fser ]
	1713	else: pmasters[tb].append(fser)
	1714	else:
	1715	self.log.error('Testbed service does not have name " + \
	1716	"and importers')
	1717	return masters, pmasters
	1718
[cf0ff4f]	1719	def generate_keys_and_hosts(self, tmpdir, expid, hosts, tbparams):
	1720	"""
	1721	Create the ssh keys necessary for interconnecting the potral nodes and
	1722	the global hosts file for letting each segment know about the IP
	1723	addresses in play. Save these into the repo. Add attributes to the
	1724	autorizer allowing access controllers to download them and return a set
	1725	of attributes that inform the segments where to find this stuff. Mau
	1726	raise service_errors in if there are problems.
	1727	"""
	1728	gw_pubkey_base = "fed.%s.pub" % self.ssh_type
	1729	gw_secretkey_base = "fed.%s" % self.ssh_type
	1730	gw_pubkey = tmpdir + "/keys/" + gw_pubkey_base
	1731	gw_secretkey = tmpdir + "/keys/" + gw_secretkey_base
	1732
	1733	try:
	1734	self.generate_ssh_keys(gw_secretkey, self.ssh_type)
	1735	except ValueError:
	1736	raise service_error(service_error.server_config,
	1737	"Bad key type (%s)" % self.ssh_type)
	1738
	1739
	1740	# Copy configuration files into the remote file store
	1741	# The config urlpath
	1742	configpath = "/%s/config" % expid
	1743	# The config file system location
	1744	configdir ="%s%s" % ( self.repodir, configpath)
	1745	try:
	1746	os.makedirs(configdir)
	1747	except EnvironmentError, e:
	1748	raise service_error(service_error.internal,
	1749	"Cannot create config directory: %s" % e)
	1750	try:
	1751	f = open("%s/hosts" % configdir, "w")
	1752	print >> f, string.join(hosts, '\n')
	1753	f.close()
	1754	except EnvironmentError, e:
	1755	raise service_error(service_error.internal,
	1756	"Cannot write hosts file: %s" % e)
	1757	try:
	1758	copy_file("%s" % gw_pubkey, "%s/%s" % \
	1759	(configdir, gw_pubkey_base))
	1760	copy_file("%s" % gw_secretkey, "%s/%s" % \
	1761	(configdir, gw_secretkey_base))
	1762	except EnvironmentError, e:
	1763	raise service_error(service_error.internal,
	1764	"Cannot copy keyfiles: %s" % e)
	1765
[913dc7a]	1766	# Allow the individual testbeds to access the configuration files,
	1767	# again by setting an attribute for the relevant pathnames on each
	1768	# allocation principal. Yeah, that's a long list comprehension.
	1769	self.append_experiment_authorization(expid, set([
	1770	(tbparams[tb]['allocID']['fedid'], "%s/%s" % (configpath, f)) \
	1771	for tb in tbparams.keys() \
	1772	for f in ("hosts", gw_secretkey_base, gw_pubkey_base)]))
[cf0ff4f]	1773
	1774	attrs = [
	1775	{
	1776	'attribute': 'ssh_pubkey',
	1777	'value': '%s/%s/config/%s' % \
	1778	(self.repo_url, expid, gw_pubkey_base)
	1779	},
	1780	{
	1781	'attribute': 'ssh_secretkey',
	1782	'value': '%s/%s/config/%s' % \
	1783	(self.repo_url, expid, gw_secretkey_base)
	1784	},
	1785	{
	1786	'attribute': 'hosts',
	1787	'value': '%s/%s/config/hosts' % \
	1788	(self.repo_url, expid)
	1789	},
	1790	]
	1791	return attrs
	1792
	1793
	1794	def get_vtopo(self, req, fid):
	1795	"""
	1796	Return the stored virtual topology for this experiment
	1797	"""
	1798	rv = None
	1799	state = None
	1800
	1801	req = req.get('VtopoRequestBody', None)
	1802	if not req:
	1803	raise service_error(service_error.req,
	1804	"Bad request format (no VtopoRequestBody)")
	1805	exp = req.get('experiment', None)
	1806	if exp:
	1807	if exp.has_key('fedid'):
	1808	key = exp['fedid']
	1809	keytype = "fedid"
	1810	elif exp.has_key('localname'):
	1811	key = exp['localname']
	1812	keytype = "localname"
	1813	else:
	1814	raise service_error(service_error.req, "Unknown lookup type")
	1815	else:
	1816	raise service_error(service_error.req, "No request?")
	1817
[e83f2f2]	1818	proof = self.check_experiment_access(fid, key)
[cf0ff4f]	1819
	1820	self.state_lock.acquire()
	1821	if self.state.has_key(key):
	1822	if self.state[key].has_key('vtopo'):
[e83f2f2]	1823	rv = { 'experiment' : {keytype: key },
	1824	'vtopo': self.state[key]['vtopo'],
	1825	'proof': proof.to_dict(),
[cf0ff4f]	1826	}
	1827	else:
	1828	state = self.state[key]['experimentStatus']
	1829	self.state_lock.release()
	1830
	1831	if rv: return rv
	1832	else:
	1833	if state:
	1834	raise service_error(service_error.partial,
	1835	"Not ready: %s" % state)
	1836	else:
	1837	raise service_error(service_error.req, "No such experiment")
	1838
	1839	def get_vis(self, req, fid):
	1840	"""
	1841	Return the stored visualization for this experiment
	1842	"""
	1843	rv = None
	1844	state = None
	1845
	1846	req = req.get('VisRequestBody', None)
	1847	if not req:
	1848	raise service_error(service_error.req,
	1849	"Bad request format (no VisRequestBody)")
	1850	exp = req.get('experiment', None)
	1851	if exp:
	1852	if exp.has_key('fedid'):
	1853	key = exp['fedid']
	1854	keytype = "fedid"
	1855	elif exp.has_key('localname'):
	1856	key = exp['localname']
	1857	keytype = "localname"
	1858	else:
	1859	raise service_error(service_error.req, "Unknown lookup type")
	1860	else:
	1861	raise service_error(service_error.req, "No request?")
	1862
[e83f2f2]	1863	proof = self.check_experiment_access(fid, key)
[cf0ff4f]	1864
	1865	self.state_lock.acquire()
	1866	if self.state.has_key(key):
	1867	if self.state[key].has_key('vis'):
[e83f2f2]	1868	rv = { 'experiment' : {keytype: key },
	1869	'vis': self.state[key]['vis'],
	1870	'proof': proof.to_dict(),
[cf0ff4f]	1871	}
	1872	else:
	1873	state = self.state[key]['experimentStatus']
	1874	self.state_lock.release()
	1875
	1876	if rv: return rv
	1877	else:
	1878	if state:
	1879	raise service_error(service_error.partial,
	1880	"Not ready: %s" % state)
	1881	else:
	1882	raise service_error(service_error.req, "No such experiment")
	1883
	1884
	1885	def save_federant_information(self, allocated, tbparams, eid, vtopo, vis,
	1886	top):
	1887	"""
	1888	Store the various data that have changed in the experiment state
	1889	between when it was started and the beginning of resource allocation.
	1890	This is basically the information about each local allocation. This
[e83f2f2]	1891	fills in the values of the placeholder allocation in the state. It
	1892	also collects the access proofs and returns them as dicts for a
	1893	response message.
[cf0ff4f]	1894	"""
	1895	# save federant information
	1896	for k in allocated.keys():
	1897	tbparams[k]['federant'] = {
	1898	'name': [ { 'localname' : eid} ],
	1899	'allocID' : tbparams[k]['allocID'],
	1900	'uri': tbparams[k]['uri'],
[e83f2f2]	1901	'proof': tbparams[k]['proof'],
[cf0ff4f]	1902	}
	1903
	1904	self.state_lock.acquire()
	1905	self.state[eid]['vtopo'] = vtopo
	1906	self.state[eid]['vis'] = vis
	1907	self.state[eid]['experimentdescription'] = \
	1908	{ 'topdldescription': top.to_dict() }
	1909	self.state[eid]['federant'] = \
	1910	[ tbparams[tb]['federant'] for tb in tbparams.keys() \
	1911	if tbparams[tb].has_key('federant') ]
[e83f2f2]	1912	# Access proofs for the response message
	1913	proofs = [copy.deepcopy(p) for k in tbparams.keys()\
	1914	for p in tbparams[k]['federant']['proof']]
[cf0ff4f]	1915	if self.state_filename:
	1916	self.write_state()
	1917	self.state_lock.release()
[e83f2f2]	1918	return proofs
[cf0ff4f]	1919
	1920	def clear_placeholder(self, eid, expid, tmpdir):
	1921	"""
	1922	Clear the placeholder and remove any allocated temporary dir.
	1923	"""
	1924
	1925	self.state_lock.acquire()
	1926	del self.state[eid]
	1927	del self.state[expid]
	1928	if self.state_filename: self.write_state()
	1929	self.state_lock.release()
	1930	if tmpdir and self.cleanup:
	1931	self.remove_dirs(tmpdir)
	1932
	1933	# end of create_experiment sub-functions
[5ecb9a3]	1934
[e19b75c]	1935	def create_experiment(self, req, fid):
[db6b092]	1936	"""
	1937	The external interface to experiment creation called from the
	1938	dispatcher.
	1939
	1940	Creates a working directory, splits the incoming description using the
[43197eb]	1941	splitter script and parses out the various subsections using the
[1a4ee0f]	1942	classes above. Once each sub-experiment is created, use pooled threads
	1943	to instantiate them and start it all up.
[db6b092]	1944	"""
[7183b48]	1945
	1946	req = req.get('CreateRequestBody', None)
[5ecb9a3]	1947	if req:
[cf0ff4f]	1948	key = self.get_experiment_key(req)
[5ecb9a3]	1949	else:
[7183b48]	1950	raise service_error(service_error.req,
	1951	"Bad request format (no CreateRequestBody)")
	1952
[6e63513]	1953	# Import information from the requester
	1954	if self.auth.import_credentials(data_list=req.get('credential', [])):
	1955	self.auth.save()
	1956
[5ecb9a3]	1957	# Make sure that the caller can talk to us
[e83f2f2]	1958	proof = self.check_experiment_access(fid, key)
[db6b092]	1959
[fd07c48]	1960	# Install the testbed map entries supplied with the request into a copy
	1961	# of the testbed map.
	1962	tbmap = dict(self.tbmap)
	1963	for m in req.get('testbedmap', []):
	1964	if 'testbed' in m and 'uri' in m:
	1965	tbmap[m['testbed']] = m['uri']
	1966
[5ecb9a3]	1967	# a place to work
[db6b092]	1968	try:
	1969	tmpdir = tempfile.mkdtemp(prefix="split-")
[895a133]	1970	os.mkdir(tmpdir+"/keys")
[d3c8759]	1971	except EnvironmentError:
[db6b092]	1972	raise service_error(service_error.internal, "Cannot create tmp dir")
	1973
	1974	tbparams = { }
	1975
[5ecb9a3]	1976	eid, expid, expcert_file = \
	1977	self.get_experiment_ids_and_start(key, tmpdir)
[c573278]	1978
[5ecb9a3]	1979	# This catches exceptions to clear the placeholder if necessary
[db6b092]	1980	try:
[5ecb9a3]	1981	if not (eid and expid):
	1982	raise service_error(service_error.internal,
	1983	"Cannot find local experiment info!?")
[5f6929a]	1984
[5ecb9a3]	1985	top = self.get_topology(req, tmpdir)
[2627eb3]	1986	self.confirm_software(top)
[5ecb9a3]	1987	# Assign the IPs
[69692a9]	1988	hosts, ip_allocator = self.allocate_ips_to_topo(top)
[1a4ee0f]	1989	# Find the testbeds to look up
[5334044]	1990	tb_hosts = { }
[5ecb9a3]	1991	testbeds = [ ]
	1992	for e in top.elements:
	1993	if isinstance(e, topdl.Computer):
	1994	tb = e.get_attribute('testbed') or 'default'
	1995	if tb in tb_hosts: tb_hosts[tb].append(e.name)
	1996	else:
	1997	tb_hosts[tb] = [ e.name ]
	1998	testbeds.append(tb)
	1999
[1660f7c]	2000	masters, pmasters = self.get_testbed_services(req, testbeds)
[895a133]	2001	allocated = { } # Testbeds we can access
	2002	topo ={ } # Sub topologies
[e02cd14]	2003	connInfo = { } # Connection information
[5334044]	2004
[2627eb3]	2005	self.split_topology(top, topo, testbeds)
	2006
[5ecb9a3]	2007	self.get_access_to_testbeds(testbeds, fid, allocated,
	2008	tbparams, masters, tbmap, expid, expcert_file)
[5f96438]	2009
[cf0ff4f]	2010	attrs = self.generate_keys_and_hosts(tmpdir, expid, hosts, tbparams)
[cc8d8e9]	2011
[fd07c48]	2012	part = experiment_partition(self.auth, self.store_url, tbmap,
[175b444]	2013	self.muxmax, self.direct_transit)
[5334044]	2014	part.add_portals(top, topo, eid, pmasters, tbparams, ip_allocator,
[2761484]	2015	connInfo, expid)
[913dc7a]	2016
	2017	auth_attrs = set()
[ab847bc]	2018	# Now get access to the dynamic testbeds (those added above)
	2019	for tb in [ t for t in topo if t not in allocated]:
[cf0ff4f]	2020	self.get_access(tb, tbparams, fid, masters, tbmap,
	2021	expid, expcert_file)
[ab847bc]	2022	allocated[tb] = 1
	2023	store_keys = topo[tb].get_attribute('store_keys')
	2024	# Give the testbed access to keys it exports or imports
	2025	if store_keys:
[b16cfc0]	2026	auth_attrs.update(set([
[913dc7a]	2027	(tbparams[tb]['allocID']['fedid'], sk) \
	2028	for sk in store_keys.split(" ")]))
	2029
	2030	if auth_attrs:
	2031	self.append_experiment_authorization(expid, auth_attrs)
[69692a9]	2032
[cf0ff4f]	2033	# transit and disconnected testbeds may not have a connInfo entry.
	2034	# Fill in the blanks.
	2035	for t in allocated.keys():
	2036	if not connInfo.has_key(t):
	2037	connInfo[t] = { }
	2038
[895a133]	2039	self.wrangle_software(expid, top, topo, tbparams)
[cc8d8e9]	2040
	2041	vtopo = topdl.topology_to_vtopo(top)
	2042	vis = self.genviz(vtopo)
[e83f2f2]	2043	proofs = self.save_federant_information(allocated, tbparams,
	2044	eid, vtopo, vis, top)
[866c983]	2045	except service_error, e:
	2046	# If something goes wrong in the parse (usually an access error)
	2047	# clear the placeholder state. From here on out the code delays
[db6b092]	2048	# exceptions. Failing at this point returns a fault to the remote
	2049	# caller.
[cf0ff4f]	2050	self.clear_placeholder(eid, expid, tmpdir)
[866c983]	2051	raise e
	2052
[db6b092]	2053	# Start the background swapper and return the starting state. From
	2054	# here on out, the state will stick around a while.
[866c983]	2055
[db6b092]	2056	# Create a logger that logs to the experiment's state object as well as
	2057	# to the main log file.
	2058	alloc_log = logging.getLogger('fedd.experiment_control.%s' % eid)
[f07fa49]	2059	alloc_collector = self.list_log(self.state[eid]['log'])
	2060	h = logging.StreamHandler(alloc_collector)
[db6b092]	2061	# XXX: there should be a global one of these rather than repeating the
	2062	# code.
	2063	h.setFormatter(logging.Formatter("%(asctime)s %(name)s %(message)s",
	2064	'%d %b %y %H:%M:%S'))
	2065	alloc_log.addHandler(h)
[617592b]	2066
[db6b092]	2067	# Start a thread to do the resource allocation
[e19b75c]	2068	t = Thread(target=self.allocate_resources,
[43197eb]	2069	args=(allocated, masters, eid, expid, tbparams,
[b4b19c7]	2070	top, topo, tmpdir, alloc_log, alloc_collector, attrs,
[725c55d]	2071	connInfo, tbmap, expcert_file),
[db6b092]	2072	name=eid)
	2073	t.start()
	2074
	2075	rv = {
	2076	'experimentID': [
	2077	{'localname' : eid }, { 'fedid': copy.copy(expid) }
	2078	],
	2079	'experimentStatus': 'starting',
[e83f2f2]	2080	'proof': [ proof.to_dict() ] + proofs,
[db6b092]	2081	}
	2082
	2083	return rv
[9479343]	2084
	2085	def get_experiment_fedid(self, key):
	2086	"""
[db6b092]	2087	find the fedid associated with the localname key in the state database.
[9479343]	2088	"""
	2089
[db6b092]	2090	rv = None
	2091	self.state_lock.acquire()
	2092	if self.state.has_key(key):
	2093	if isinstance(self.state[key], dict):
	2094	try:
	2095	kl = [ f['fedid'] for f in \
	2096	self.state[key]['experimentID']\
	2097	if f.has_key('fedid') ]
	2098	except KeyError:
	2099	self.state_lock.release()
	2100	raise service_error(service_error.internal,
	2101	"No fedid for experiment %s when getting "+\
	2102	"fedid(!?)" % key)
	2103	if len(kl) == 1:
	2104	rv = kl[0]
	2105	else:
	2106	self.state_lock.release()
	2107	raise service_error(service_error.internal,
	2108	"multiple fedids for experiment %s when " +\
	2109	"getting fedid(!?)" % key)
	2110	else:
	2111	self.state_lock.release()
	2112	raise service_error(service_error.internal,
	2113	"Unexpected state for %s" % key)
	2114	self.state_lock.release()
	2115	return rv
[a97394b]	2116
[4064742]	2117	def check_experiment_access(self, fid, key):
[866c983]	2118	"""
	2119	Confirm that the fid has access to the experiment. Though a request
	2120	may be made in terms of a local name, the access attribute is always
	2121	the experiment's fedid.
	2122	"""
	2123	if not isinstance(key, fedid):
[db6b092]	2124	key = self.get_experiment_fedid(key)
[866c983]	2125
[e83f2f2]	2126	access_ok, proof = self.auth.check_attribute(fid, key, with_proof=True)
	2127
	2128	if access_ok:
	2129	return proof
[866c983]	2130	else:
[e83f2f2]	2131	raise service_error(service_error.access, "Access Denied",
	2132	proof)
[4064742]	2133
	2134
[db6b092]	2135	def get_handler(self, path, fid):
[cf0ff4f]	2136	"""
	2137	Perhaps surprisingly named, this function handles HTTP GET requests to
	2138	this server (SOAP requests are POSTs).
	2139	"""
[7183b48]	2140	self.log.info("Get handler %s %s" % (path, fid))
[e83f2f2]	2141	# XXX: log proofs?
[6c57fe9]	2142	if self.auth.check_attribute(fid, path):
	2143	return ("%s/%s" % (self.repodir, path), "application/binary")
	2144	else:
	2145	return (None, None)
[987aaa1]	2146
[e83f2f2]	2147	def clean_info_response(self, rv, proof):
[db6b092]	2148	"""
	2149	Remove the information in the experiment's state object that is not in
	2150	the info response.
	2151	"""
	2152	# Remove the owner info (should always be there, but...)
	2153	if rv.has_key('owner'): del rv['owner']
[913dc7a]	2154	if 'auth' in rv: del rv['auth']
[db6b092]	2155
	2156	# Convert the log into the allocationLog parameter and remove the
	2157	# log entry (with defensive programming)
	2158	if rv.has_key('log'):
	2159	rv['allocationLog'] = "".join(rv['log'])
	2160	del rv['log']
	2161	else:
	2162	rv['allocationLog'] = ""
	2163
	2164	if rv['experimentStatus'] != 'active':
	2165	if rv.has_key('federant'): del rv['federant']
	2166	else:
[69692a9]	2167	# remove the allocationID and uri info from each federant
[db6b092]	2168	for f in rv.get('federant', []):
	2169	if f.has_key('allocID'): del f['allocID']
[69692a9]	2170	if f.has_key('uri'): del f['uri']
[e83f2f2]	2171	rv['proof'] = proof.to_dict()
[b4b19c7]	2172
[db6b092]	2173	return rv
[65f3f29]	2174
[c52c48d]	2175	def get_info(self, req, fid):
[866c983]	2176	"""
	2177	Return all the stored info about this experiment
	2178	"""
	2179	rv = None
	2180
	2181	req = req.get('InfoRequestBody', None)
	2182	if not req:
	2183	raise service_error(service_error.req,
[65f3f29]	2184	"Bad request format (no InfoRequestBody)")
[866c983]	2185	exp = req.get('experiment', None)
	2186	if exp:
	2187	if exp.has_key('fedid'):
	2188	key = exp['fedid']
	2189	keytype = "fedid"
	2190	elif exp.has_key('localname'):
	2191	key = exp['localname']
	2192	keytype = "localname"
	2193	else:
	2194	raise service_error(service_error.req, "Unknown lookup type")
	2195	else:
	2196	raise service_error(service_error.req, "No request?")
	2197
[e83f2f2]	2198	proof = self.check_experiment_access(fid, key)
[866c983]	2199
	2200	# The state may be massaged by the service function that called
	2201	# get_info (e.g., encoded for XMLRPC transport) so send a copy of the
	2202	# state.
	2203	self.state_lock.acquire()
	2204	if self.state.has_key(key):
	2205	rv = copy.deepcopy(self.state[key])
	2206	self.state_lock.release()
	2207
[db6b092]	2208	if rv:
[e83f2f2]	2209	return self.clean_info_response(rv, proof)
[bd3e314]	2210	else:
[db6b092]	2211	raise service_error(service_error.req, "No such experiment")
[7a8d667]	2212
[65f3f29]	2213	def get_multi_info(self, req, fid):
	2214	"""
	2215	Return all the stored info that this fedid can access
	2216	"""
[e83f2f2]	2217	rv = { 'info': [ ], 'proof': [ ] }
[65f3f29]	2218
[db6b092]	2219	self.state_lock.acquire()
	2220	for key in [ k for k in self.state.keys() if isinstance(k, fedid)]:
[829246e]	2221	try:
[e83f2f2]	2222	proof = self.check_experiment_access(fid, key)
[829246e]	2223	except service_error, e:
	2224	if e.code == service_error.access:
	2225	continue
	2226	else:
	2227	self.state_lock.release()
	2228	raise e
[65f3f29]	2229
[db6b092]	2230	if self.state.has_key(key):
	2231	e = copy.deepcopy(self.state[key])
[e83f2f2]	2232	e = self.clean_info_response(e, proof)
[db6b092]	2233	rv['info'].append(e)
[e83f2f2]	2234	rv['proof'].append(proof.to_dict())
[65f3f29]	2235	self.state_lock.release()
[db6b092]	2236	return rv
[65f3f29]	2237
[cf0ff4f]	2238	def check_termination_status(self, fed_exp, force):
[e07c8f3]	2239	"""
[cf0ff4f]	2240	Confirm that the experiment is sin a valid state to stop (or force it)
	2241	return the state - invalid states for deletion and force settings cause
	2242	exceptions.
[e07c8f3]	2243	"""
[cf0ff4f]	2244	self.state_lock.acquire()
	2245	status = fed_exp.get('experimentStatus', None)
[e07c8f3]	2246
[cf0ff4f]	2247	if status:
	2248	if status in ('starting', 'terminating'):
	2249	if not force:
	2250	self.state_lock.release()
	2251	raise service_error(service_error.partial,
	2252	'Experiment still being created or destroyed')
	2253	else:
	2254	self.log.warning('Experiment in %s state ' % status + \
	2255	'being terminated by force.')
	2256	self.state_lock.release()
	2257	return status
[725c55d]	2258	else:
[cf0ff4f]	2259	# No status??? trouble
	2260	self.state_lock.release()
	2261	raise service_error(service_error.internal,
	2262	"Experiment has no status!?")
	2263
	2264
	2265	def get_termination_info(self, fed_exp):
	2266	ids = []
	2267	term_params = { }
	2268	self.state_lock.acquire()
	2269	# experimentID is a list of dicts that are self-describing
	2270	# identifiers. This finds all the fedids and localnames - the
	2271	# keys of self.state - and puts them into ids, which is used to delete
	2272	# the state after everything is swapped out.
	2273	for id in fed_exp.get('experimentID', []):
	2274	if 'fedid' in id:
	2275	ids.append(id['fedid'])
	2276	repo = "%s" % id['fedid']
	2277	if 'localname' in id: ids.append(id['localname'])
	2278
	2279	# Get the experimentAccess - the principal for this experiment. It
	2280	# is this principal to which credentials have been delegated, and
	2281	# as which the experiment controller must act.
	2282	if 'experimentAccess' in fed_exp and \
	2283	'X509' in fed_exp['experimentAccess']:
	2284	expcert = fed_exp['experimentAccess']['X509']
	2285	else:
	2286	expcert = None
	2287
	2288	# Collect the allocation/segment ids into a dict keyed by the fedid
	2289	# of the allocation (or a monotonically increasing integer) that
	2290	# contains a tuple of uri, aid (which is a dict...)
	2291	for i, fed in enumerate(fed_exp.get('federant', [])):
	2292	try:
	2293	uri = fed['uri']
	2294	aid = fed['allocID']
	2295	k = fed['allocID'].get('fedid', i)
	2296	except KeyError, e:
	2297	continue
	2298	term_params[k] = (uri, aid)
	2299	# Change the experiment state
	2300	fed_exp['experimentStatus'] = 'terminating'
	2301	if self.state_filename: self.write_state()
	2302	self.state_lock.release()
	2303
	2304	return ids, term_params, expcert, repo
	2305
	2306
	2307	def deallocate_resources(self, term_params, expcert, status, force,
	2308	dealloc_log):
	2309	tmpdir = None
	2310	# This try block makes sure the tempdir is cleared
	2311	try:
	2312	# If no expcert, try the deallocation as the experiment
	2313	# controller instance.
	2314	if expcert and self.auth_type != 'legacy':
	2315	try:
	2316	tmpdir = tempfile.mkdtemp(prefix="term-")
	2317	except EnvironmentError:
	2318	raise service_error(service_error.internal,
	2319	"Cannot create tmp dir")
	2320	cert_file = self.make_temp_certfile(expcert, tmpdir)
	2321	pw = None
	2322	else:
	2323	cert_file = self.cert_file
	2324	pw = self.cert_pwd
	2325
	2326	# Stop everyone. NB, wait_for_all waits until a thread starts
	2327	# and then completes, so we can't wait if nothing starts. So,
	2328	# no tbparams, no start.
	2329	if len(term_params) > 0:
	2330	tp = thread_pool(self.nthreads)
	2331	for k, (uri, aid) in term_params.items():
	2332	# Create and start a thread to stop the segment
	2333	tp.wait_for_slot()
	2334	t = pooled_thread(\
	2335	target=self.terminate_segment(log=dealloc_log,
	2336	testbed=uri,
	2337	cert_file=cert_file,
	2338	cert_pwd=pw,
	2339	trusted_certs=self.trusted_certs,
	2340	caller=self.call_TerminateSegment),
	2341	args=(uri, aid), name=k,
	2342	pdata=tp, trace_file=self.trace_file)
	2343	t.start()
	2344	# Wait for completions
	2345	tp.wait_for_all_done()
	2346
	2347	# release the allocations (failed experiments have done this
	2348	# already, and starting experiments may be in odd states, so we
	2349	# ignore errors releasing those allocations
	2350	try:
	2351	for k, (uri, aid) in term_params.items():
	2352	self.release_access(None, aid, uri=uri,
	2353	cert_file=cert_file, cert_pwd=pw)
	2354	except service_error, e:
	2355	if status != 'failed' and not force:
	2356	raise e
	2357
	2358	# Clean up the tmpdir no matter what
	2359	finally:
	2360	if tmpdir: self.remove_dirs(tmpdir)
	2361
[7a8d667]	2362	def terminate_experiment(self, req, fid):
[866c983]	2363	"""
	2364	Swap this experiment out on the federants and delete the shared
	2365	information
	2366	"""
	2367	tbparams = { }
	2368	req = req.get('TerminateRequestBody', None)
	2369	if not req:
	2370	raise service_error(service_error.req,
	2371	"Bad request format (no TerminateRequestBody)")
	2372
[cf0ff4f]	2373	key = self.get_experiment_key(req, 'experiment')
[e83f2f2]	2374	proof = self.check_experiment_access(fid, key)
[cf0ff4f]	2375	exp = req.get('experiment', False)
	2376	force = req.get('force', False)
[866c983]	2377
[db6b092]	2378	dealloc_list = [ ]
[46e4682]	2379
	2380
[5ae3857]	2381	# Create a logger that logs to the dealloc_list as well as to the main
	2382	# log file.
	2383	dealloc_log = logging.getLogger('fedd.experiment_control.%s' % key)
	2384	h = logging.StreamHandler(self.list_log(dealloc_list))
	2385	# XXX: there should be a global one of these rather than repeating the
	2386	# code.
	2387	h.setFormatter(logging.Formatter("%(asctime)s %(name)s %(message)s",
	2388	'%d %b %y %H:%M:%S'))
	2389	dealloc_log.addHandler(h)
	2390
	2391	self.state_lock.acquire()
	2392	fed_exp = self.state.get(key, None)
[cf0ff4f]	2393	self.state_lock.release()
[e07c8f3]	2394	repo = None
[5ae3857]	2395
	2396	if fed_exp:
[cf0ff4f]	2397	status = self.check_termination_status(fed_exp, force)
	2398	ids, term_params, expcert, repo = self.get_termination_info(fed_exp)
	2399	self.deallocate_resources(term_params, expcert, status, force,
	2400	dealloc_log)
[5ae3857]	2401
	2402	# Remove the terminated experiment
	2403	self.state_lock.acquire()
	2404	for id in ids:
[a96d946]	2405	self.clear_experiment_authorization(id, need_state_lock=False)
[cf0ff4f]	2406	if id in self.state: del self.state[id]
[5ae3857]	2407
	2408	if self.state_filename: self.write_state()
	2409	self.state_lock.release()
	2410
[2761484]	2411	# Delete any synch points associated with this experiment. All
	2412	# synch points begin with the fedid of the experiment.
	2413	fedid_keys = set(["fedid:%s" % f for f in ids \
	2414	if isinstance(f, fedid)])
	2415	for k in self.synch_store.all_keys():
	2416	try:
	2417	if len(k) > 45 and k[0:46] in fedid_keys:
	2418	self.synch_store.del_value(k)
[dadc4da]	2419	except synch_store.BadDeletionError:
[2761484]	2420	pass
	2421	self.write_store()
[e07c8f3]	2422
	2423	# Remove software and other cached stuff from the filesystem.
	2424	if repo:
	2425	self.remove_dirs("%s/%s" % (self.repodir, repo))
[e83f2f2]	2426
[5ae3857]	2427	return {
	2428	'experiment': exp ,
[cf0ff4f]	2429	'deallocationLog': string.join(dealloc_list, ''),
[e83f2f2]	2430	'proof': [proof.to_dict()],
[5ae3857]	2431	}
	2432	else:
	2433	raise service_error(service_error.req, "No saved state")
[2761484]	2434
	2435
	2436	def GetValue(self, req, fid):
	2437	"""
	2438	Get a value from the synchronized store
	2439	"""
	2440	req = req.get('GetValueRequestBody', None)
	2441	if not req:
	2442	raise service_error(service_error.req,
	2443	"Bad request format (no GetValueRequestBody)")
	2444
[cf0ff4f]	2445	name = req.get('name', None)
	2446	wait = req.get('wait', False)
[2761484]	2447	rv = { 'name': name }
	2448
[e83f2f2]	2449	if not name:
	2450	raise service_error(service_error.req, "No name?")
	2451
	2452	access_ok, proof = self.auth.check_attribute(fid, name, with_proof=True)
	2453
	2454	if access_ok:
[d8442da]	2455	self.log.debug("[GetValue] asking for %s " % name)
[dadc4da]	2456	try:
	2457	v = self.synch_store.get_value(name, wait)
	2458	except synch_store.RevokedKeyError:
	2459	# No more synch on this key
	2460	raise service_error(service_error.federant,
	2461	"Synch key %s revoked" % name)
[2761484]	2462	if v is not None:
	2463	rv['value'] = v
[e83f2f2]	2464	rv['proof'] = proof.to_dict()
[109a32a]	2465	self.log.debug("[GetValue] got %s from %s" % (v, name))
[2761484]	2466	return rv
	2467	else:
[e83f2f2]	2468	raise service_error(service_error.access, "Access Denied",
	2469	proof=proof)
[2761484]	2470
	2471
	2472	def SetValue(self, req, fid):
	2473	"""
	2474	Set a value in the synchronized store
	2475	"""
	2476	req = req.get('SetValueRequestBody', None)
	2477	if not req:
	2478	raise service_error(service_error.req,
	2479	"Bad request format (no SetValueRequestBody)")
	2480
[cf0ff4f]	2481	name = req.get('name', None)
	2482	v = req.get('value', '')
[2761484]	2483
[e83f2f2]	2484	if not name:
	2485	raise service_error(service_error.req, "No name?")
	2486
	2487	access_ok, proof = self.auth.check_attribute(fid, name, with_proof=True)
	2488
	2489	if access_ok:
[2761484]	2490	try:
	2491	self.synch_store.set_value(name, v)
	2492	self.write_store()
[109a32a]	2493	self.log.debug("[SetValue] set %s to %s" % (name, v))
[2761484]	2494	except synch_store.CollisionError:
	2495	# Translate into a service_error
	2496	raise service_error(service_error.req,
	2497	"Value already set: %s" %name)
[dadc4da]	2498	except synch_store.RevokedKeyError:
	2499	# No more synch on this key
	2500	raise service_error(service_error.federant,
	2501	"Synch key %s revoked" % name)
[e83f2f2]	2502	return { 'name': name, 'value': v, 'proof': proof.to_dict() }
[2761484]	2503	else:
[e83f2f2]	2504	raise service_error(service_error.access, "Access Denied",
	2505	proof=proof)

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

Download in other formats: