Context Navigation

source: fedd/fedd_create_experiment.py @ 6546868

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

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

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

Download in other formats: