Context Navigation

source: fedd/fedd_create_experiment.py @ 1af38d6

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

Last change on this file since 1af38d6 was 1af38d6, checked in by Ted Faber <faber@…>, 16 years ago
multithreading for startup. Needs more testing
Property mode set to `100644`
File size: 33.9 KB

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

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

Download in other formats: