Context Navigation

source: fedd/federation/experiment_control.py @ a0c2866

axis_examplecompt_changesinfo-ops

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

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

Download in other formats: