Context Navigation

source: fedd/fedd_create_experiment.py @ d0ae12d

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

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

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

Download in other formats: