Context Navigation

source: fedd/federation/experiment_control.py @ 35a4c01

axis_examplecompt_changesinfo-opsversion-3.01version-3.02

Last change on this file since 35a4c01 was 35a4c01, checked in by Ted Faber <faber@…>, 14 years ago
Access controllers manage the fedkit now. Also upped the parallelism for swapins.
Property mode set to `100644`
File size: 93.4 KB

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

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

Download in other formats: