Context Navigation

source: fedd/fedd_create_experiment.py @ 0d830de

axis_examplecompt_changesinfo-opsversion-1.30version-2.00version-3.01version-3.02

Last change on this file since 0d830de was 0d830de, checked in by Ted Faber <faber@…>, 16 years ago
topology and visualization into real data structures
Property mode set to `100644`
File size: 37.9 KB

Line
1	#!/usr/local/bin/python
2
3	import os,sys
4
5	from ZSI import *
6	from M2Crypto import SSL
7	from M2Crypto.SSL.SSLServer import SSLServer
8	import M2Crypto.httpslib
9
10	import xml.parsers.expat
11
12	import re
13	import random
14	import string
15	import subprocess
16	import tempfile
17	import copy
18
19	import traceback
20
21	from threading import *
22
23	from subprocess import *
24
25	from fedd_services import *
26	from fedd_internal_services import *
27	from fedd_util import *
28	import parse_detail
29	from service_error import *
30
31	class fedd_create_experiment_local:
32	scripts = ["fed_bootstrap", "federate.sh", "smbmount.FreeBSD.pl",
33	"smbmount.Linux.pl", "make_hosts", "fed-tun.pl", "fed-tun.ucb.pl",
34	"fed_evrepeater", "rc.accounts.patch"]
35
36	def __init__(self,
37	cert_file=None,
38	cert_pwd=None,
39	exp_stem="faber-splitter",
40	debug=False,
41	muxmax=2,
42	nthreads=2,
43	randomize_experiments=False,
44	scp_exec="/usr/bin/scp",
45	scripts_dir="./",
46	splitter=None,
47	ssh_exec="/usr/bin/ssh",
48	ssh_identity_file=None,
49	ssh_keygen="/usr/bin/ssh-keygen",
50	ssh_pubkey_file=None,
51	ssh_type="rsa",
52	tbmap=None,
53	tclsh="/usr/local/bin/otclsh",
54	tcl_splitter="/usr/testbed/lib/ns2ir/parse.tcl",
55	trace_file=None,
56	trusted_certs=None,
57	):
58	self.scripts = fedd_create_experiment_local.scripts
59	self.thread_with_rv = fedd_create_experiment_local.pooled_thread
60	self.thread_pool = fedd_create_experiment_local.thread_pool
61
62	self.cert_file = cert_file
63	self.cert_pwd = cert_pwd
64	self.exp_stem = exp_stem
65	self.debug = debug
66	self.muxmax = muxmax
67	self.nthreads = nthreads
68	self.randomize_experiments = randomize_experiments
69	self.scp_exec = scp_exec
70	self.scripts_dir = scripts_dir
71	self.splitter = splitter
72	self.ssh_exec=ssh_exec
73	self.ssh_keygen = ssh_keygen
74	self.ssh_identity_file = ssh_identity_file
75	self.ssh_type = ssh_type
76	self.tclsh = tclsh
77	self.tcl_splitter = tcl_splitter
78	self.tbmap = tbmap
79	self.trace_file = trace_file
80	self.trusted_certs=trusted_certs
81
82	self.def_expstart = \
83	"sudo -H /bin/sh FEDDIR/fed_bootstrap >& /tmp/federate";
84	self.def_mexpstart = "sudo -H FEDDIR/make_hosts FEDDIR/hosts";
85	self.def_gwstart = \
86	"sudo -H FEDDIR/fed-tun.pl -f GWCONF>& /tmp/bridge.log";
87	self.def_mgwstart = \
88	"sudo -H FEDDIR/fed-tun.pl -f GWCONF >& /tmp/bridge.log";
89	self.def_gwimage = "FBSD61-TUNNEL2";
90	self.def_gwtype = "pc";
91
92
93	if ssh_pubkey_file:
94	try:
95	f = open(ssh_pubkey_file, 'r')
96	self.ssh_pubkey = f.read()
97	f.close()
98	except IOError:
99	raise service_error(service_error.internal,
100	"Cannot read sshpubkey")
101
102	# Confirm federation scripts in the right place
103	for s in self.scripts:
104	if not os.path.exists(self.scripts_dir + "/" + s):
105	raise service_error(service_error.server_config,
106	"%s/%s not in local script dir" % (self.scripts_dir, s))
107	class thread_pool:
108	def __init__(self):
109	self.changed = Condition()
110	self.started = 0
111	self.terminated = 0
112
113	def acquire(self):
114	self.changed.acquire()
115
116	def release(self):
117	self.changed.release()
118
119	def wait(self, timeout = None):
120	self.changed.wait(timeout)
121
122	def start(self):
123	self.changed.acquire()
124	self.started += 1
125	self.changed.notifyAll()
126	self.changed.release()
127
128	def terminate(self):
129	self.changed.acquire()
130	self.terminated += 1
131	self.changed.notifyAll()
132	self.changed.release()
133
134	def clear(self):
135	self.changed.acquire()
136	self.started = 0
137	self.terminated =0
138	self.changed.notifyAll()
139	self.changed.release()
140
141
142
143	class pooled_thread(Thread):
144	def __init__(self, group=None, target=None, name=None, args=(),
145	kwargs={}, pdata=None, trace_file=None):
146	Thread.__init__(self, group, target, name, args, kwargs)
147	self.rv = None
148	self.exception = None
149	self.target=target
150	self.args = args
151	self.kwargs = kwargs
152	self.pdata = pdata
153	self.trace_file = trace_file
154
155	def run(self):
156	if self.pdata:
157	self.pdata.start()
158
159	if self.target:
160	try:
161	self.rv = self.target(self.args, *self.kwargs)
162	except service_error, s:
163	self.exception = s
164	if self.trace_file:
165	print >>self.trace_file, "Thread exception: %s %s" % \
166	(s.code_string(), s.desc)
167
168	except:
169	self.exception = sys.exc_info()[1]
170	if self.trace_file:
171	print >>self.trace_file, \
172	"Unexpected thread exception: %s" % \
173	self.exception
174	print >>self.trace_file, "Trace: %s" % \
175	traceback.format_exc()
176	if self.pdata:
177	self.pdata.terminate()
178
179	def copy_file(self, src, dest, size=1024):
180	"""
181	Exceedingly simple file copy.
182	"""
183	s = open(src,'r')
184	d = open(dest, 'w')
185
186	buf = "x"
187	while buf != "":
188	buf = s.read(size)
189	d.write(buf)
190	s.close()
191	d.close()
192
193	def scp_file(self, file, user, host, dest=""):
194	"""
195	scp a file to the remote host.
196	"""
197
198	scp_cmd = [self.scp_exec, file, "%s@%s:%s" % (user, host, dest)]
199
200	trace = self.trace_file
201	if not trace:
202	try:
203	trace = open("/dev/null", "w")
204	except IOError:
205	raise service_error(service_error.internal,
206	"Cannot open /dev/null??");
207
208	if not self.debug:
209	rv = call(scp_cmd, stdout=trace, stderr=trace)
210	else:
211	if self.trace_file:
212	print >>self.trace_file, "debug [scp_file]: %s" % \
213	" ".join(scp_cmd)
214	rv = 0
215
216	return rv == 0
217
218	def ssh_cmd(self, user, host, cmd, wname=None):
219	sh_str = "%s %s@%s %s" % (self.ssh_exec, user, host, cmd)
220
221	trace = self.trace_file
222	if not trace:
223	try:
224	trace = open("/dev/null", "w")
225	except IOError:
226	raise service_error(service_error.internal,
227	"Cannot open /dev/null??");
228
229	if not self.debug:
230	sub = Popen(sh_str, shell=True, stdout=trace, stderr=trace)
231	return sub.wait() == 0
232	else:
233	if self.trace_file:
234	print >>self.trace_file,"debug [ssh_cmd]: %s" % sh_str
235	return True
236
237	def ship_scripts(self, host, user, dest_dir):
238	if self.ssh_cmd(user, host, "mkdir -p %s" % dest_dir):
239	for s in self.scripts:
240	if not self.scp_file("%s/%s" % (self.scripts_dir, s),
241	user, host, dest_dir):
242	return False
243	return True
244	else:
245	return False
246
247	def ship_configs(self, host, user, src_dir, dest_dir):
248	if not self.ssh_cmd(user, host, "mkdir -p %s" % dest_dir):
249	return False
250	if not self.ssh_cmd(user, host, "chmod 770 %s" % dest_dir):
251	return False
252
253	for f in os.listdir(src_dir):
254	if os.path.isdir(f):
255	if not self.ship_configs(host, user, "%s/%s" % (src_dir, f),
256	"%s/%s" % (dest_dir, f)):
257	return False
258	else:
259	if not self.scp_file("%s/%s" % (src_dir, f),
260	user, host, dest_dir):
261	return False
262	return True
263
264	def start_segment(self, tb, eid, tbparams, tmpdir, timeout=0):
265	host = "%s%s" % (tbparams[tb]['host'], tbparams[tb]['domain'])
266	user = tbparams[tb]['user']
267	pid = tbparams[tb]['project']
268	# XXX
269	base_confs = ( "hosts",)
270	tclfile = "%s.%s.tcl" % (eid, tb)
271	expinfo_exec = "/usr/testbed/bin/expinfo"
272	proj_dir = "/proj/%s/exp/%s/tmp" % (pid, eid)
273	tarfiles_dir = "/proj/%s/tarfiles/%s" % (pid, eid)
274	rpms_dir = "/proj/%s/rpms/%s" % (pid, eid)
275	state_re = re.compile("State:\s+(\w+)")
276	no_exp_re = re.compile("^No\s+such\s+experiment")
277	state = None
278	cmd = [self.ssh_exec, "%s@%s" % (user, host), expinfo_exec, pid, eid]
279
280
281	if self.trace_file:
282	print >>self.trace_file, "status request: %s" % " ".join(cmd)
283
284	if not self.trace_file:
285	try:
286	st_file = open("/dev/null", "w")
287	except IOError:
288	raise service_error(service_error.internal,
289	"Cannot open /dev/null!?")
290	else:
291	st_file = self.trace_file
292
293	status = Popen(cmd, stdout=PIPE, stderr=st_file)
294	for line in status.stdout:
295	m = state_re.match(line)
296	if m: state = m.group(1)
297	else:
298	m = no_exp_re.match(line)
299	if m: state = "none"
300	rv = status.wait()
301	# If the experiment is not present the subcommand returns a non-zero
302	# return value. If we successfully parsed a "none" outcome, ignore the
303	# return code.
304	if rv != 0 and state != "none":
305	raise service_error(service_error.internal,
306	"Cannot get status of segment %s:%s/%s" % (tb, pid, eid))
307	if self.trace_file:
308	print >>self.trace_file, "%s: %s" % (tb, state)
309	print >>self.trace_file, "transferring experiment to %s" % tb
310
311	if not self.scp_file("%s/%s/%s" % (tmpdir, tb, tclfile), user, host):
312	return False
313	# Clear and create the tarfiles and rpm directories
314	for d in (tarfiles_dir, rpms_dir):
315	if not self.ssh_cmd(user, host,
316	"/bin/sh -c \"'/bin/rm -rf %s/*'\"" % d):
317	return False
318	if not self.ssh_cmd(user, host, "mkdir -p %s" % d,
319	"create tarfiles"):
320	return False
321
322	if state == 'active':
323	# Remote experiment is active. Modify it.
324	for f in base_confs:
325	if not self.scp_file("%s/%s" % (tmpdir, f), user, host,
326	"%s/%s" % (proj_dir, f)):
327	return False
328	if not self.ship_scripts(host, user, proj_dir):
329	return False
330	if not self.ship_configs(host, user, "%s/%s" % (tmpdir, tb),
331	proj_dir):
332	return False
333	if os.path.isdir("%s/tarfiles" % tmpdir):
334	if not self.ship_configs(host, user,
335	"%s/tarfiles" % tmpdir, tarfiles_dir):
336	return False
337	if os.path.isdir("%s/rpms" % tmpdir):
338	if not self.ship_configs(host, user,
339	"%s/rpms" % tmpdir, tarfiles_dir):
340	return False
341	if self.trace_file:
342	print >>self.trace_file, "Modifying %s on %s" % (eid, tb)
343	if not self.ssh_cmd(user, host,
344	"/usr/testbed/bin/modexp -r -s -w %s %s %s" % \
345	(pid, eid, tclfile), "modexp"):
346	return False
347	return True
348	elif state == "swapped":
349	# Remote experiment swapped out. Modify it and swap it in.
350	for f in base_confs:
351	if not self.scp_file("%s/%s" % (tmpdir, f), user, host,
352	"%s/%s" % (proj_dir, f)):
353	return False
354	if not self.ship_scripts(host, user, proj_dir):
355	return False
356	if not self.ship_configs(host, user, "%s/%s" % (tmpdir, tb),
357	proj_dir):
358	return False
359	if os.path.isdir("%s/tarfiles" % tmpdir):
360	if not self.ship_configs(host, user,
361	"%s/tarfiles" % tmpdir, tarfiles_dir):
362	return False
363	if os.path.isdir("%s/rpms" % tmpdir):
364	if not self.ship_configs(host, user,
365	"%s/rpms" % tmpdir, tarfiles_dir):
366	return False
367	if self.trace_file:
368	print >>self.trace_file, "Modifying %s on %s" % (eid, tb)
369	if not self.ssh_cmd(user, host,
370	"/usr/testbed/bin/modexp -w %s %s %s" % (pid, eid, tclfile),
371	"modexp"):
372	return False
373	if self.trace_file:
374	print >>self.trace_file, "Swapping %s in on %s" % (eid, tb)
375	if not self.ssh_cmd(user, host,
376	"/usr/testbed/bin/swapexp -w %s %s in" % (pid, eid),
377	"swapexp"):
378	return False
379	return True
380	elif state == "none":
381	# No remote experiment. Create one. We do this in 2 steps so we
382	# can put the configuration files and scripts into the new
383	# experiment directories.
384
385	# Tarfiles must be present for creation to work
386	if os.path.isdir("%s/tarfiles" % tmpdir):
387	if not self.ship_configs(host, user,
388	"%s/tarfiles" % tmpdir, tarfiles_dir):
389	return False
390	if os.path.isdir("%s/rpms" % tmpdir):
391	if not self.ship_configs(host, user,
392	"%s/rpms" % tmpdir, tarfiles_dir):
393	return False
394	if self.trace_file:
395	print >>self.trace_file, "Creating %s on %s" % (eid, tb)
396	if not self.ssh_cmd(user, host,
397	"/usr/testbed/bin/startexp -i -f -w -p %s -e %s %s" % \
398	(pid, eid, tclfile), "startexp"):
399	return False
400	# After startexp the per-experiment directories exist
401	for f in base_confs:
402	if not self.scp_file("%s/%s" % (tmpdir, f), user, host,
403	"%s/%s" % (proj_dir, f)):
404	return False
405	if not self.ship_scripts(host, user, proj_dir):
406	return False
407	if not self.ship_configs(host, user, "%s/%s" % (tmpdir, tb),
408	proj_dir):
409	return False
410	if self.trace_file:
411	print >>self.trace_file, "Swapping %s in on %s" % (eid, tb)
412	if not self.ssh_cmd(user, host,
413	"/usr/testbed/bin/swapexp -w %s %s in" % (pid, eid),
414	"swapexp"):
415	return False
416	return True
417	else:
418	if self.trace_file:
419	print >>self.trace_file, "unknown state %s" % state
420	return False
421
422	def stop_segment(self, tb, eid, tbparams):
423	user = tbparams[tb]['user']
424	host = "%s%s" % (tbparams[tb]['host'], tbparams[tb]['domain'])
425	pid = tbparams[tb]['project']
426
427	if self.trace_file:
428	print >>self.trace_file, "Stopping %s on %s" % (eid, tb)
429	return self.ssh_cmd(user, host,
430	"/usr/testbed/bin/swapexp -w %s %s out" % (pid, eid))
431
432
433	def generate_ssh_keys(self, dest, type="rsa" ):
434	"""
435	Generate a set of keys for the gateways to use to talk.
436
437	Keys are of type type and are stored in the required dest file.
438	"""
439	valid_types = ("rsa", "dsa")
440	t = type.lower();
441	if t not in valid_types: raise ValueError
442
443	trace = self.trace_file
444	if not trace:
445	try:
446	trace = open("/dev/null", "w")
447	except IOError:
448	raise service_error(service_error.internal,
449	"Cannot open /dev/null??");
450
451	# May raise CalledProcessError
452	rv = call([self.ssh_keygen, '-t', t, '-N', '', '-f', dest],
453	stdout=trace, stderr=trace)
454	if rv != 0:
455	raise service_error(service_error.internal,
456	"Cannot generate nonce ssh keys. %s return code %d" \
457	% (self.ssh_keygen, rv))
458
459	def gentopo(self, str):
460	class topo_parse:
461	def __init__(self):
462	self.str_subelements = ('vname', 'vnode', 'ips', 'ip', 'member')
463	self.int_subelements = ( 'bandwidth',)
464	self.float_subelements = ( 'delay',)
465	self.nodes = [ ]
466	self.lans = [ ]
467	self.element = { }
468	self.topo = { \
469	'nodes': { 'node': self.nodes },\
470	'lans' : { 'lan' : self.lans },\
471	}
472	self.chars = ""
473
474	def end_element(self, name):
475	if name == 'node':
476	self.nodes.append(self.element)
477	self.element = { }
478	elif name == 'lan':
479	self.lans.append(self.element)
480	self.element = { }
481	elif name in self.str_subelements:
482	self.element[name] = self.chars
483	self.chars = ""
484	elif name in self.int_subelements:
485	self.element[name] = int(self.chars)
486	self.chars = ""
487	elif name in self.float_subelements:
488	self.element[name] = float(self.chars)
489	self.chars = ""
490
491	def found_chars(self, data):
492	self.chars += data.rstrip()
493
494
495	tp = topo_parse();
496	parser = xml.parsers.expat.ParserCreate()
497	parser.EndElementHandler = tp.end_element
498	parser.CharacterDataHandler = tp.found_chars
499
500	parser.Parse(str)
501
502	return tp.topo
503
504
505	def genviz(self, topo):
506	"""
507	Generate the visualization the virtual topology
508	"""
509
510	neato = "/usr/local/bin/neato"
511	# These are used to parse neato output and to create the visualization
512	# file.
513	vis_re = re.compile('^\s"?([\w\-]+)"?\s+\[.pos="(\d+),(\d+)"')
514	vis_fmt = "<node><name>%s</name><x>%s</x><y>%s</y><type>" + \
515	"%s</type></node>"
516
517	try:
518	# Node names
519	nodes = [ n['vname'] for n in topo['nodes']['node'] ]
520	topo_lans = topo['lans']['lan']
521	except KeyError:
522	raise service_error(service_error.internal, "Bad topology")
523
524	lans = { }
525	links = { }
526
527	# Walk through the virtual topology, organizing the connections into
528	# 2-node connections (links) and more-than-2-node connections (lans).
529	# When a lan is created, it's added to the list of nodes (there's a
530	# node in the visualization for the lan).
531	for l in topo_lans:
532	if links.has_key(l['vname']):
533	if len(links[l['vname']]) < 2:
534	links[l['vname']].append(l['vnode'])
535	else:
536	nodes.append(l['vname'])
537	lans[l['vname']] = links[l['vname']]
538	del links[l['vname']]
539	lans[l['vname']].append(l['vnode'])
540	elif lans.has_key(l['vname']):
541	lans[l['vname']].append(l['vnode'])
542	else:
543	links[l['vname']] = [ l['vnode'] ]
544
545
546	# Open up a temporary file for dot to turn into a visualization
547	try:
548	df, dotname = tempfile.mkstemp()
549	dotfile = os.fdopen(df, 'w')
550	except IOError:
551	raise service_error(service_error.internal,
552	"Failed to open file in genviz")
553
554	# Generate a dot/neato input file from the links, nodes and lans
555	try:
556	print >>dotfile, "graph G {"
557	for n in nodes:
558	print >>dotfile, '\t"%s"' % n
559	for l in links.keys():
560	print >>dotfile, '\t"%s" -- "%s"' % tuple(links[l])
561	for l in lans.keys():
562	for n in lans[l]:
563	print >>dotfile, '\t "%s" -- "%s"' % (n,l)
564	print >>dotfile, "}"
565	dotfile.close()
566	except TypeError:
567	raise service_error(service_error.internal,
568	"Single endpoint link in vtopo")
569	except IOError:
570	raise service_error(service_error.internal, "Cannot write dot file")
571
572	# Use dot to create a visualization
573	dot = Popen([neato, '-Gstart=rand', '-Gepsilon=0.005', '-Gmaxiter=2000',
574	'-Gpack=true', dotname], stdout=PIPE)
575
576	# Translate dot to vis format
577	vis_nodes = [ ]
578	vis = { 'node': vis_nodes }
579	for line in dot.stdout:
580	m = vis_re.match(line)
581	if m:
582	vn = m.group(1)
583	vis_node = {'name': vn, \
584	'x': int(m.group(2)),\
585	'y' : int(m.group(3)),\
586	}
587	if vn in links.keys() or vn in lans.keys():
588	vis_node['type'] = 'lan'
589	else:
590	vis_node['type'] = 'node'
591	vis_nodes.append(vis_node)
592	rv = dot.wait()
593
594	os.remove(dotname)
595	if rv == 0 : return vis
596	else: return None
597
598
599	def get_access(self, tb, nodes, user, tbparam):
600	"""
601	Get access to testbed through fedd and set the parameters for that tb
602	"""
603
604	translate_attr = {
605	'slavenodestartcmd': 'expstart',
606	'slaveconnectorstartcmd': 'gwstart',
607	'masternodestartcmd': 'mexpstart',
608	'masterconnectorstartcmd': 'mgwstart',
609	'connectorimage': 'gwimage',
610	'connectortype': 'gwtype',
611	'tunnelcfg': 'tun',
612	'smbshare': 'smbshare',
613	}
614
615	# XXX multi-level access
616	uri = self.tbmap.get(tb, None)
617	if not uri:
618	raise service_error(serice_error.server_config,
619	"Unknown testbed: %s" % tb)
620
621	# The basic request
622	req = {\
623	'destinationTestbed' : { 'uri' : uri },
624	'user': user,
625	'allocID' : { 'username': 'test' },
626	'access' : [ { 'sshPubkey' : self.ssh_pubkey } ]
627	}
628
629	# node resources if any
630	if nodes != None and len(nodes) > 0:
631	rnodes = [ ]
632	for n in nodes:
633	rn = { }
634	image, hw, count = n.split(":")
635	if image: rn['image'] = [ image ]
636	if hw: rn['hardware'] = [ hw ]
637	if count: rn['count'] = int(count)
638	rnodes.append(rn)
639	req['resources']= { }
640	req['resources']['node'] = rnodes
641
642	# No retry loop here. Proxy servers must correctly authenticate
643	# themselves without help
644	try:
645	ctx = fedd_ssl_context(self.cert_file,
646	self.trusted_certs, password=self.cert_pwd)
647	except SSL.SSLError:
648	raise service_error(service_error.server_config,
649	"Server certificates misconfigured")
650
651	loc = feddServiceLocator();
652	port = loc.getfeddPortType(uri,
653	transport=M2Crypto.httpslib.HTTPSConnection,
654	transdict={ 'ssl_context' : ctx })
655
656	# Reconstruct the full request message
657	msg = RequestAccessRequestMessage()
658	msg.set_element_RequestAccessRequestBody(
659	pack_soap(msg, "RequestAccessRequestBody", req))
660	try:
661	resp = port.RequestAccess(msg)
662	except ZSI.ParseException, e:
663	raise service_error(service_error.req,
664	"Bad format message (XMLRPC??): %s" %
665	str(e))
666	r = unpack_soap(resp)
667
668	if r.has_key('RequestAccessResponseBody'):
669	r = r['RequestAccessResponseBody']
670	else:
671	raise service_error(service_error.proxy,
672	"Bad proxy response")
673
674
675	e = r['emulab']
676	p = e['project']
677	tbparam[tb] = {
678	"boss": e['boss'],
679	"host": e['ops'],
680	"domain": e['domain'],
681	"fs": e['fileServer'],
682	"eventserver": e['eventServer'],
683	"project": unpack_id(p['name']),
684	"emulab" : e
685	}
686
687	for u in p['user']:
688	tbparam[tb]['user'] = unpack_id(u['userID'])
689
690	for a in e['fedAttr']:
691	if a['attribute']:
692	key = translate_attr.get(a['attribute'].lower(), None)
693	if key:
694	tbparam[tb][key]= a['value']
695
696	class current_testbed:
697	def __init__(self, eid, tmpdir):
698	self.begin_testbed = re.compile("^#\s+Begin\s+Testbed\s+\((\w+)\)")
699	self.end_testbed = re.compile("^#\s+End\s+Testbed\s+\((\w+)\)")
700	self.current_testbed = None
701	self.testbed_file = None
702
703	self.def_expstart = \
704	"sudo -H /bin/sh FEDDIR/fed_bootstrap >& /tmp/federate";
705	self.def_mexpstart = "sudo -H FEDDIR/make_hosts FEDDIR/hosts";
706	self.def_gwstart = \
707	"sudo -H FEDDIR/fed-tun.pl -f GWCONF>& /tmp/bridge.log";
708	self.def_mgwstart = \
709	"sudo -H FEDDIR/fed-tun.pl -f GWCONF >& /tmp/bridge.log";
710	self.def_gwimage = "FBSD61-TUNNEL2";
711	self.def_gwtype = "pc";
712
713	self.eid = eid
714	self.tmpdir = tmpdir
715
716	def __call__(self, line, master, allocated, tbparams):
717	# Capture testbed topology descriptions
718	if self.current_testbed == None:
719	m = self.begin_testbed.match(line)
720	if m != None:
721	self.current_testbed = m.group(1)
722	if self.current_testbed == None:
723	raise service_error(service_error.req,
724	"Bad request format (unnamed testbed)")
725	allocated[self.current_testbed] = \
726	allocated.get(self.current_testbed,0) + 1
727	tb_dir = "%s/%s" % (self.tmpdir, self.current_testbed)
728	if not os.path.exists(tb_dir):
729	try:
730	os.mkdir(tb_dir)
731	except IOError:
732	raise service_error(service_error.internal,
733	"Cannot create %s" % tb_dir)
734	try:
735	self.testbed_file = open("%s/%s.%s.tcl" %
736	(tb_dir, self.eid, self.current_testbed), 'w')
737	except IOError:
738	self.testbed_file = None
739	return True
740	else: return False
741	else:
742	m = self.end_testbed.match(line)
743	if m != None:
744	if m.group(1) != self.current_testbed:
745	raise service_error(service_error.internal,
746	"Mismatched testbed markers!?")
747	if self.testbed_file != None:
748	self.testbed_file.close()
749	self.testbed_file = None
750	self.current_testbed = None
751	elif self.testbed_file:
752	# Substitute variables and put the line into the local
753	# testbed file.
754	gwtype = tbparams[self.current_testbed].get('gwtype',
755	self.def_gwtype)
756	gwimage = tbparams[self.current_testbed].get('gwimage',
757	self.def_gwimage)
758	mgwstart = tbparams[self.current_testbed].get('mgwstart',
759	self.def_mgwstart)
760	mexpstart = tbparams[self.current_testbed].get('mexpstart',
761	self.def_mexpstart)
762	gwstart = tbparams[self.current_testbed].get('gwstart',
763	self.def_gwstart)
764	expstart = tbparams[self.current_testbed].get('expstart',
765	self.def_expstart)
766	project = tbparams[self.current_testbed].get('project')
767	line = re.sub("GWTYPE", gwtype, line)
768	line = re.sub("GWIMAGE", gwimage, line)
769	if self.current_testbed == master:
770	line = re.sub("GWSTART", mgwstart, line)
771	line = re.sub("EXPSTART", mexpstart, line)
772	else:
773	line = re.sub("GWSTART", gwstart, line)
774	line = re.sub("EXPSTART", expstart, line)
775	# XXX: does `` embed without doing enything else?
776	line = re.sub("GWCONF", "FEDDIR`hostname`.gw.conf", line)
777	line = re.sub("PROJDIR", "/proj/%s/" % project, line)
778	line = re.sub("EID", self.eid, line)
779	line = re.sub("FEDDIR", "/proj/%s/exp/%s/tmp/" % \
780	(project, self.eid), line)
781	print >>self.testbed_file, line
782	return True
783
784	class allbeds:
785	def __init__(self, get_access):
786	self.begin_allbeds = re.compile("^#\s+Begin\s+Allbeds")
787	self.end_allbeds = re.compile("^#\s+End\s+Allbeds")
788	self.in_allbeds = False
789	self.get_access = get_access
790
791	def __call__(self, line, user, tbparams):
792	# Testbed access parameters
793	if not self.in_allbeds:
794	if self.begin_allbeds.match(line):
795	self.in_allbeds = True
796	return True
797	else:
798	return False
799	else:
800	if self.end_allbeds.match(line):
801	self.in_allbeds = False
802	else:
803	nodes = line.split('\|')
804	tb = nodes.pop(0)
805	self.get_access(tb, nodes, user, tbparams)
806	return True
807
808	class gateways:
809	def __init__(self, eid, master, tmpdir, gw_pubkey,
810	gw_secretkey, copy_file):
811	self.begin_gateways = \
812	re.compile("^#\s+Begin\s+gateways\s+\((\w+)\)")
813	self.end_gateways = re.compile("^#\s+End\s+gateways\s+\((\w+)\)")
814	self.current_gateways = None
815	self.control_gateway = None
816	self.active_end = { }
817
818	self.eid = eid
819	self.master = master
820	self.tmpdir = tmpdir
821	self.gw_pubkey_base = gw_pubkey
822	self.gw_secretkey_base = gw_secretkey
823
824	self.copy_file = copy_file
825
826
827	def gateway_conf_file(self, gw, master, eid, pubkey, privkey,
828	active_end, tbparams, dtb, myname, desthost, type):
829	"""
830	Produce a gateway configuration file from a gateways line.
831	"""
832
833	sproject = tbparams[gw].get('project', 'project')
834	dproject = tbparams[dtb].get('project', 'project')
835	sdomain = ".%s.%s%s" % (eid, sproject,
836	tbparams[gw].get('domain', ".example.com"))
837	ddomain = ".%s.%s%s" % (eid, dproject,
838	tbparams[dtb].get('domain', ".example.com"))
839	boss = tbparams[master].get('boss', "boss")
840	fs = tbparams[master].get('fs', "fs")
841	event_server = "%s%s" % \
842	(tbparams[master].get('eventserver', "event_server"),
843	tbparams[master].get('domain', "example.com"))
844	remote_event_server = "%s%s" % \
845	(tbparams[dtb].get('eventserver', "event_server"),
846	tbparams[dtb].get('domain', "example.com"))
847
848	remote_script_dir = "/proj/%s/exp/%s/tmp" % ( dproject, eid)
849	local_script_dir = "/proj/%s/exp/%s/tmp" % ( sproject, eid)
850	tunnel_cfg = tbparams[gw].get("tun", "false")
851
852	conf_file = "%s%s.gw.conf" % (myname, sdomain)
853	remote_conf_file = "%s%s.gw.conf" % (desthost, ddomain)
854
855	# translate to lower case so the `hostname` hack for specifying
856	# configuration files works.
857	conf_file = conf_file.lower();
858	remote_conf_file = remote_conf_file.lower();
859
860	if dtb == master:
861	active = "false"
862	elif gw == master:
863	active = "true"
864	elif active_end.has_key['%s-%s' % (dtb, gw)]:
865	active = "false"
866	else:
867	active_end['%s-%s' % (gw, dtb)] = 1
868	active = "true"
869
870	gwconfig = open("%s/%s/%s" % (self.tmpdir, gw, conf_file), "w")
871	print >>gwconfig, "Active: %s" % active
872	print >>gwconfig, "TunnelCfg: %s" % tunnel_cfg
873	print >>gwconfig, "BossName: %s" % boss
874	print >>gwconfig, "FsName: %s" % fs
875	print >>gwconfig, "EventServerName: %s" % event_server
876	print >>gwconfig, "RemoteEventServerName: %s" % remote_event_server
877	print >>gwconfig, "Type: %s" % type
878	print >>gwconfig, "RemoteScriptDir: %s" % remote_script_dir
879	print >>gwconfig, "EventRepeater: %s/fed_evrepeater" % \
880	local_script_dir
881	print >>gwconfig, "RemoteExperiment: %s/%s" % (dproject, eid)
882	print >>gwconfig, "LocalExperiment: %s/%s" % (sproject, eid)
883	print >>gwconfig, "RemoteConfigFile: %s/%s" % \
884	(remote_script_dir, remote_conf_file)
885	print >>gwconfig, "Peer: %s%s" % (desthost, ddomain)
886	print >>gwconfig, "Pubkeys: %s/%s" % (local_script_dir, pubkey)
887	print >>gwconfig, "Privkeys: %s/%s" % (local_script_dir, privkey)
888	gwconfig.close()
889
890	return active == "true"
891
892	def __call__(self, line, allocated, tbparams):
893	# Process gateways
894	if not self.current_gateways:
895	m = self.begin_gateways.match(line)
896	if m:
897	self.current_gateways = m.group(1)
898	if allocated.has_key(self.current_gateways):
899	# This test should always succeed
900	tb_dir = "%s/%s" % (self.tmpdir, self.current_gateways)
901	if not os.path.exists(tb_dir):
902	try:
903	os.mkdir(tb_dir)
904	except IOError:
905	raise service_error(service_error.internal,
906	"Cannot create %s" % tb_dir)
907	else:
908	# XXX
909	print >>sys.stderr, \
910	"Ignoring gateways for unknown testbed %s" \
911	% self.current_gateways
912	self.current_gateways = None
913	return True
914	else:
915	return False
916	else:
917	m = self.end_gateways.match(line)
918	if m :
919	if m.group(1) != self.current_gateways:
920	raise service_error(service_error.internal,
921	"Mismatched gateway markers!?")
922	if self.control_gateway:
923	try:
924	cc = open("%s/%s/client.conf" %
925	(self.tmpdir, self.current_gateways), 'w')
926	print >>cc, "ControlGateway: %s" % \
927	self.control_gateway
928	if tbparams[self.master].has_key('smbshare'):
929	print >>cc, "SMBSHare: %s" % \
930	tbparams[self.master]['smbshare']
931	print >>cc, "ProjectUser: %s" % \
932	tbparams[self.master]['user']
933	print >>cc, "ProjectName: %s" % \
934	tbparams[self.master]['project']
935	cc.close()
936	except IOError:
937	raise service_error(service_error.internal,
938	"Error creating client config")
939	else:
940	if self.trace_file:
941	print >>sys.stderr, "No control gateway for %s" %\
942	self.current_gateways
943	self.current_gateways = None
944	else:
945	dtb, myname, desthost, type = line.split(" ")
946
947	if type == "control" or type == "both":
948	self.control_gateway = "%s.%s.%s%s" % (myname,
949	self.eid,
950	tbparams[self.current_gateways]['project'],
951	tbparams[self.current_gateways]['domain'])
952	try:
953	active = self.gateway_conf_file(self.current_gateways,
954	self.master, self.eid, self.gw_pubkey_base,
955	self.gw_secretkey_base,
956	self.active_end, tbparams, dtb, myname,
957	desthost, type)
958	except IOError, e:
959	raise service_error(service_error.internal,
960	"Failed to write config file for %s" % \
961	self.current_gateway)
962
963	gw_pubkey = "%s/keys/%s" % \
964	(self.tmpdir, self.gw_pubkey_base)
965	gw_secretkey = "%s/keys/%s" % \
966	(self.tmpdir, self.gw_secretkey_base)
967
968	pkfile = "%s/%s/%s" % \
969	( self.tmpdir, self.current_gateways,
970	self.gw_pubkey_base)
971	skfile = "%s/%s/%s" % \
972	( self.tmpdir, self.current_gateways,
973	self.gw_secretkey_base)
974
975	if not os.path.exists(pkfile):
976	try:
977	self.copy_file(gw_pubkey, pkfile)
978	except IOError:
979	service_error(service_error.internal,
980	"Failed to copy pubkey file")
981
982	if active and not os.path.exists(skfile):
983	try:
984	self.copy_file(gw_secretkey, skfile)
985	except IOError:
986	service_error(service_error.internal,
987	"Failed to copy secretkey file")
988	return True
989
990	class shunt_to_file:
991	def __init__(self, begin, end, filename):
992	self.begin = re.compile(begin)
993	self.end = re.compile(end)
994	self.in_shunt = False
995	self.file = None
996	self.filename = filename
997
998	def __call__(self, line):
999	if not self.in_shunt:
1000	if self.begin.match(line):
1001	self.in_shunt = True
1002	try:
1003	self.file = open(self.filename, "w")
1004	except:
1005	self.file = None
1006	raise
1007	return True
1008	else:
1009	return False
1010	else:
1011	if self.end.match(line):
1012	if self.file:
1013	self.file.close()
1014	self.file = None
1015	self.in_shunt = False
1016	else:
1017	if self.file:
1018	print >>self.file, line
1019	return True
1020
1021	class shunt_to_list:
1022	def __init__(self, begin, end):
1023	self.begin = re.compile(begin)
1024	self.end = re.compile(end)
1025	self.in_shunt = False
1026	self.list = [ ]
1027
1028	def __call__(self, line):
1029	if not self.in_shunt:
1030	if self.begin.match(line):
1031	self.in_shunt = True
1032	return True
1033	else:
1034	return False
1035	else:
1036	if self.end.match(line):
1037	self.in_shunt = False
1038	else:
1039	self.list.append(line)
1040	return True
1041
1042	class shunt_to_string:
1043	def __init__(self, begin, end):
1044	self.begin = re.compile(begin)
1045	self.end = re.compile(end)
1046	self.in_shunt = False
1047	self.str = ""
1048
1049	def __call__(self, line):
1050	if not self.in_shunt:
1051	if self.begin.match(line):
1052	self.in_shunt = True
1053	return True
1054	else:
1055	return False
1056	else:
1057	if self.end.match(line):
1058	self.in_shunt = False
1059	else:
1060	self.str += line
1061	return True
1062
1063	def create_experiment(self, req, fid):
1064	try:
1065	tmpdir = tempfile.mkdtemp(prefix="split-")
1066	except IOError:
1067	raise service_error(service_error.internal, "Cannot create tmp dir")
1068
1069	gw_pubkey_base = "fed.%s.pub" % self.ssh_type
1070	gw_secretkey_base = "fed.%s" % self.ssh_type
1071	gw_pubkey = tmpdir + "/keys/" + gw_pubkey_base
1072	gw_secretkey = tmpdir + "/keys/" + gw_secretkey_base
1073	tclfile = tmpdir + "/experiment.tcl"
1074	tbparams = { }
1075
1076	pid = "dummy"
1077	gid = "dummy"
1078	eid = self.exp_stem
1079	if self.randomize_experiments:
1080	for i in range(0,5):
1081	eid += random.choice(string.ascii_letters)
1082	# XXX
1083	fail_soft = False
1084
1085	try:
1086	os.mkdir(tmpdir+"/keys")
1087	except OSError:
1088	raise service_error(service_error.internal,
1089	"Can't make temporary dir")
1090
1091	# The tcl parser needs to read a file so put the content into that file
1092	file_content=req.get('experimentdescription', None)
1093	if file_content != None:
1094	try:
1095	f = open(tclfile, 'w')
1096	f.write(file_content)
1097	f.close()
1098	except IOError:
1099	raise service_error(service_error.internal,
1100	"Cannot write temp experiment description")
1101	else:
1102	raise service_error(service_error.req, "No experiment description")
1103
1104	try:
1105	self.generate_ssh_keys(gw_secretkey, self.ssh_type)
1106	except ValueError:
1107	raise service_error(service_error.server_config,
1108	"Bad key type (%s)" % self.ssh_type)
1109
1110	user = req.get('user', None)
1111	if user == None:
1112	raise service_error(service_error.req, "No user")
1113
1114	master = req.get('master', None)
1115	if master == None:
1116	raise service_error(service_error.req, "No master testbed label")
1117
1118
1119	tclcmd = [self.tclsh, self.tcl_splitter, '-s', '-x',
1120	str(self.muxmax), '-m', master, pid, gid, eid, tclfile]
1121	tclparser = Popen(tclcmd, stdout=PIPE)
1122
1123	allocated = { }
1124	started = { }
1125
1126	parse_current_testbed = self.current_testbed(eid, tmpdir)
1127	parse_allbeds = self.allbeds(self.get_access)
1128	parse_gateways = self.gateways(eid, master, tmpdir,
1129	gw_pubkey_base, gw_secretkey_base, self.copy_file)
1130	parse_vtopo = self.shunt_to_string("^#\s+Begin\s+Vtopo",
1131	"^#\s+End\s+Vtopo")
1132	parse_hostnames = self.shunt_to_file("^#\s+Begin\s+hostnames",
1133	"^#\s+End\s+hostnames", tmpdir + "/hosts")
1134	parse_tarfiles = self.shunt_to_list("^#\s+Begin\s+tarfiles",
1135	"^#\s+End\s+tarfiles")
1136	parse_rpms = self.shunt_to_list("^#\s+Begin\s+rpms",
1137	"^#\s+End\s+rpms")
1138
1139	for line in tclparser.stdout:
1140	line = line.rstrip()
1141	if parse_current_testbed(line, master, allocated, tbparams):
1142	continue
1143	elif parse_allbeds(line, user, tbparams):
1144	continue
1145	elif parse_gateways(line, allocated, tbparams):
1146	continue
1147	elif parse_vtopo(line):
1148	continue
1149	elif parse_hostnames(line):
1150	continue
1151	elif parse_tarfiles(line):
1152	continue
1153	elif parse_rpms(line):
1154	continue
1155	else:
1156	raise service_error(service_error.internal,
1157	"Bad tcl parse? %s" % line)
1158
1159	vtopo = self.gentopo(parse_vtopo.str)
1160	if not vtopo:
1161	raise service_error(service_error.internal,
1162	"Failed to generate virtual topology")
1163
1164	vis = self.genviz(vtopo)
1165	if not vis:
1166	raise service_error(service_error.internal,
1167	"Failed to generate visualization")
1168
1169	# Copy tarfiles and rpms needed at remote sites into a staging area
1170	try:
1171	for t in parse_tarfiles.list:
1172	if not os.path.exists("%s/tarfiles" % tmpdir):
1173	os.mkdir("%s/tarfiles" % tmpdir)
1174	self.copy_file(t, "%s/tarfiles/%s" % \
1175	(tmpdir, os.path.basename(t)))
1176	for r in parse_rpms.list:
1177	if not os.path.exists("%s/rpms" % tmpdir):
1178	os.mkdir("%s/rpms" % tmpdir)
1179	self.copy_file(r, "%s/rpms/%s" % \
1180	(tmpdir, os.path.basename(r)))
1181	except IOError, e:
1182	raise service_error(service_error.internal,
1183	"Cannot stage tarfile/rpm: %s" % e.strerror)
1184
1185	thread_pool_info = self.thread_pool()
1186	threads = [ ]
1187
1188	for tb in [ k for k in allocated.keys() if k != master]:
1189	# Wait until we have a free slot to start the next testbed load
1190	thread_pool_info.acquire()
1191	while thread_pool_info.started - \
1192	thread_pool_info.terminated >= self.nthreads:
1193	thread_pool_info.wait()
1194	thread_pool_info.release()
1195
1196	# Create and start a thread to start the segment, and save it to
1197	# get the return value later
1198	t = self.pooled_thread(target=self.start_segment,
1199	args=(tb, eid, tbparams, tmpdir, 0), name=tb,
1200	pdata=thread_pool_info, trace_file=self.trace_file)
1201	threads.append(t)
1202	t.start()
1203
1204	# Wait until all finish (the first clause of the while is to make sure
1205	# one starts)
1206	thread_pool_info.acquire()
1207	while thread_pool_info.started == 0 or \
1208	thread_pool_info.started > thread_pool_info.terminated:
1209	thread_pool_info.wait()
1210	thread_pool_info.release()
1211
1212	# If none failed, start the master
1213	failed = [ t.getName() for t in threads if not t.rv ]
1214
1215	if len(failed) == 0:
1216	if not self.start_segment(master, eid, tbparams, tmpdir):
1217	failed.append(master)
1218
1219	# If one failed clean up
1220	if len(failed) > 0:
1221	succeeded = [tb for tb in allocated.keys() if tb not in failed]
1222	if fail_soft:
1223	raise service_error(service_error.partial, \
1224	"Partial swap in on %s" % ",".join(succeeded))
1225	else:
1226	for tb in succeeded:
1227	self.stop_segment(tb, eid, tbparams)
1228	raise service_error(service_error.federant,
1229	"Swap in failed on %s" % ",".join(failed))
1230	else:
1231	if self.trace_file:
1232	print >>self.trace_file, "Experiment started"
1233
1234	return { 'emulab' : [ tbparams[tb]['emulab'] \
1235	for tb in tbparams.keys() \
1236	if tbparams[tb].has_key('emulab') ],\
1237	'experiment': vtopo,\
1238	'vis' : vis }
1239
1240	if __name__ == '__main__':
1241	from optparse import OptionParser
1242
1243	parser = OptionParser()
1244	parser.add_option('-d', '--debug', dest='debug', default=False,
1245	action='store_true', help='print actions rather than take them')
1246	parser.add_option('-f', '--file', dest='tcl', help='tcl file to parse')
1247	parser.add_option('-m', '--master', dest='master',
1248	help='testbed label for matster testbd')
1249	parser.add_option('-t', '--trace', dest='trace', default=None,
1250	help='file to print intermediate messages to')
1251	parser.add_option('-T', '--trace-stderr', dest='trace',
1252	action='store_const',const=sys.stderr,
1253	help='file to print intermediate messages to')
1254	opts, args = parser.parse_args()
1255
1256	trace_file = None
1257	if opts.trace:
1258	try:
1259	trace_file = open(opts.trace, 'w')
1260	except IOError:
1261	print >>sys.stderr, "Can't open trace file"
1262
1263	if opts.debug:
1264	if not trace_file:
1265	trace_file = sys.stderr
1266
1267	if opts.tcl != None:
1268	try:
1269	f = open(opts.tcl, 'r')
1270	content = ''.join(f)
1271	f.close()
1272	except IOError, e:
1273	sys.exit("Can't read %s: %s" % (opts.tcl, e))
1274	else:
1275	sys.exit("Must specify a file name")
1276
1277	if not opts.master:
1278	sys.exit("Must supply master tb label (--master)");
1279
1280	obj = fedd_create_experiment_local(
1281	debug=opts.debug,
1282	scripts_dir="/users/faber/testbed/federation",
1283	cert_file="./fedd_client.pem", cert_pwd="faber",
1284	ssh_pubkey_file='/users/faber/.ssh/id_rsa.pub',
1285	trusted_certs="./cacert.pem",
1286	tbmap = {
1287	'deter':'https://users.isi.deterlab.net:23235',
1288	'emulab':'https://users.isi.deterlab.net:23236',
1289	'ucb':'https://users.isi.deterlab.net:23237',
1290	},
1291	trace_file=trace_file
1292	)
1293	rv = obj.create_experiment( {\
1294	'experimentdescription' : content,
1295	'master' : opts.master,
1296	'user': [ {'userID' : { 'username' : 'faber' } } ],
1297	},
1298	None)
1299
1300	print rv

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

Download in other formats: