docker setup update

Author: lylei9

example/docker/cluster_setup.py

@@ -1,117 +1,93 @@
-import paramiko
+import json
 import argparse
+import paramiko
 import traceback
+import zk
+import hdfs
+import tera
 
-ZK = 'zk'
-HDFS = 'hdfs'
+class SSH():
+	def __init__(self):
+		self.s = paramiko.SSHClient()
+		self.s.load_system_host_keys()
+		self.s.set_missing_host_key_policy(paramiko.AutoAddPolicy())
 
-def parse_input():
-        parser = argparse.ArgumentParser()
-        parser.add_argument('file', type=str, help='A file describes the zk cluster')
-        args = parser.parse_args()
-        return args
-
-def read_ip_list(args):
-	try:
-		fp = open(args.file, 'r')
-	except:
-		traceback.print_exc()
-	
-	zk_ip_dicts = []
-	hdfs_ip_dicts = {'master':[], 'slave':[]}
-	start_port = 2888
-	end_port = 3888
-	client_port = 2181
-	myid = 1
-	while True:
+	def run_cmd(self, ip, cmd):
 		try:
-			comp = fp.readline().split(' ')
-			if comp[0].startswith(ZK):
-				zk_ip_dicts.append([comp[1], {'start_port': str(start_port), 'end_port': str(end_port), 'client_port': str(client_port), 'myid': str(myid), 'path': comp[2][:-1]}])
-				start_port += 1
-				end_port += 1
-				client_port += 1
-				myid += 1
-			elif comp[0].startswith(HDFS):
-				if comp[3].startswith('master'):
-					hdfs_ip_dicts['master'].append([comp[1], {'path': comp[2]}]) 
-				else:
-					hdfs_ip_dicts['slave'].append([comp[1], {'path': comp[2]}])
-			else:
-				break
+			self.s.connect(ip)
+			stdin, stdout, stderr = self.s.exec_command(cmd)
+			self.s.close()
 		except:
-			break
-	if hdfs_ip_dicts['slave'] != [] and hdfs_ip_dicts == []:
-		print 'must have a master!'
-		return
-	return zk_ip_dicts, hdfs_ip_dicts
+			traceback.print_exc()
+		return stdin, stdout, stderr
+
+def parse_input():
+	parser = argparse.ArgumentParser()
+	parser.add_argument('file', type=str, help='A file describes the zk cluster')
+	parser.add_argument('--docker', type=str, required=True, help='ID of the docker image')
+	parser.add_argument('--zk', action='store_true', help='Launch zk')
+	parser.add_argument('--hdfs', action='store_true', help='Launch hdfs')
+	parser.add_argument('--tera', action='store_true', help='Launch tera')
+	args = parser.parse_args()
+	return args
+
+def config(args):
+	config = json.load(open(args.file, 'r'))
+	ip_list = config['ip'].split(':')
+	log_prefix = config['log_prefix']
+	zk_cluster = zk.ZkCluster(ip_list, config['zk'], log_prefix)
+	zk_cluster.populate_zk_cluster()
+	for z in zk_cluster.cluster:
+		print z.to_string()
+
+	hdfs_cluster = hdfs.HdfsCluster(ip_list, config['hdfs'], log_prefix)
+	ret = hdfs_cluster.populate_hdfs_cluster()
+	if ret is False:
+		exit(1)
+	for h in hdfs_cluster.cluster:
+		print h.to_string()
 
-def start_zks(ip_dicts):
-	if ip_dicts == []:
+	tera_cluster = tera.TeraCluster(ip_list, config['tera'], log_prefix)
+	tera_cluster.populate_tera_cluster()
+	for t in tera_cluster.cluster:
+		print t.to_string()
+
+	return zk_cluster, hdfs_cluster, tera_cluster
+
+def start_zk(args, zk_cluster, s):
+	if (args.hdfs or args.tera) and not args.zk:
 		return
-	s=paramiko.SSHClient()
-	s.load_system_host_keys()
-	s.set_missing_host_key_policy(paramiko.AutoAddPolicy())
-	ips = []
-	for item in ip_dicts:
-		ips.append(item[0])
-	ips = ' '.join(ips)
-	for details in ip_dicts:
-		ip = details[0]
-		details = details[1]
-		try:
-			s.connect(ip)
-			cmd = 'docker run -t -d -v {dir}:/opt/share -p {cport}:{cport} -p {sport}:{sport} -p {eport}:{eport} --net=host d8 /usr/bin/python /opt/zk_setup.py --servers {ip} --port {cport} --myid {myid}'.\
-				format(dir=details['path'], cport=details['client_port'], sport=details['start_port'], eport=details['end_port'], ip=ips, myid=details['myid'])
-			stdin, stdout, stderr = s.exec_command(cmd)
-			print cmd
-			print stdout.read()
-			print '\n', stderr.read()
-			s.close()
-		except:
-			traceback.print_exc()
-	
-def start_hdfs(ip_dicts):
-	if ip_dicts == {}:
+	for zk_instance in zk_cluster.cluster:
+		#print zk_instance.to_string()
+		cmd = zk_instance.to_cmd(' '.join(zk_cluster.ip_zk), args.docker)
+		print cmd
+		s.run_cmd(zk_instance.ip, cmd)
+
+def start_hdfs(args, hdfs_cluster, s):
+	if (args.zk or args.tera) and not args.hdfs:
 		return
-	s = paramiko.SSHClient()
-	s.load_system_host_keys()
-	s.set_missing_host_key_policy(paramiko.AutoAddPolicy())
-	ips = []
-	master = ip_dicts['master'][0]
-	master_ip = master[0]
-	master_details = master[1]
-	slave_list = ip_dicts['slave']
-	for item in slave_list:
-		ips.append(item[0])
-	ips = ' '.join(ips)
-	
-	cmd = 'docker run -t -d -v {dir}:/opt/share -p 9000:9000 -p 9001:9001 --net=host 67 /usr/bin/python /opt/hdfs_setup.py --masters {master} --slaves {slaves} --mode master'.\
-		format(dir=master_details['path'], master=master_ip, slaves=ips)
-	print cmd
-	s.connect(master_ip)
-	#stdin, stdout, stderr = s.exec_command(cmd)
-	#print stdout.read()
-	#print '\n', stderr.read()
-	s.close()
+	for hdfs_instance in hdfs_cluster.cluster:
+		#print hdfs_instance.to_string()
+		cmd = hdfs_instance.to_cmd(args.docker, hdfs_cluster.master_ip, ' '.join(hdfs_cluster.slave_ip))
+		print cmd
+		s.run_cmd(hdfs_instance.ip, cmd)
 
-	for slave in slave_list:
-		slave_ip = slave[0]
-		slave_details = slave[1]
-		cmd = 'docker run -t -d -v {dir}:/opt/share -p 9000:9000 -p 9001:9001 --net=host 67 /usr/bin/python /opt/hdfs_setup.py --masters {master} --slaves {slaves} --mode slave'.\
-			format(dir=slave_details['path'], master=master_ip, slaves=ips)
+def start_tera(args, tera_cluster, zk_cluster, s):
+	if (args.zk or args.hdfs) and not args.tera:
+		return
+	for tera_instance in tera_cluster.cluster:
+		#print tera_instance.to_string()
+		cmd = tera_instance.to_cmd(args.docker, ','.join(zk_cluster.ip_tera))
 		print cmd
-		s.connect(slave_ip)
-		#stdin, stdout, stderr = s.exec_command(cmd)
-		#print stdout.read()
-		#print '\n', stderr.read()
-		s.close()
+		s.run_cmd(tera_instance.ip, cmd)
 
 def main():
 	args = parse_input()
-	zk_ip_dicts, hdfs_ip_dicts = read_ip_list(args)
-	start_zks(zk_ip_dicts)
-	start_hdfs(hdfs_ip_dicts)
+	zk_cluster, hdfs_cluster, tera_cluster = config(args)
+	s = SSH()
+	start_zk(args, zk_cluster, s)
+	start_hdfs(args, hdfs_cluster, s)
+	start_tera(args, tera_cluster, zk_cluster, s)
 
 if __name__ == '__main__':
 	main()

example/docker/dockerfile

@@ -0,0 +1,41 @@
+FROM debian:jessie  
+
+RUN apt-get update && apt-get install -y openjdk-7-jre-headless wget
+RUN wget -q -O - http://apache.mirrors.pair.com/zookeeper/zookeeper-3.4.6/zookeeper-3.4.6.tar.gz | tar -xzf - -C /opt \
+    && mv /opt/zookeeper-3.4.6 /opt/zookeeper \
+    && cp /opt/zookeeper/conf/zoo_sample.cfg /opt/zookeeper/conf/zoo.cfg \
+    && mkdir -p /tmp/zookeeper
+
+ENV JAVA_HOME /usr/lib/jvm/java-7-openjdk-amd64
+
+RUN apt-get -y install python
+RUN apt-get -y install python-dev
+RUN apt-get -y install gcc
+RUN apt-get -y install make
+RUN cd /opt/zookeeper/src/c && ./configure && make && make install
+
+ADD workspace /opt/workspace
+
+RUN wget https://bootstrap.pypa.io/ez_setup.py -O - | python
+RUN cd /opt/workspace/argparse-1.3.0/ && python setup.py install
+RUN cd /opt/workspace/zkpython-0.4/ && python setup.py install
+RUN cp /opt/workspace/zk_setup.py /opt/
+RUN mkdir /opt/share
+
+ENV LD_LIBRARY_PATH /usr/local/lib/
+
+WORKDIR /opt/
+
+ADD hadoop-1.2.1 /opt/hadoop-1.2.1
+RUN cp /opt/hadoop-1.2.1/hdfs_setup.py /opt
+RUN /opt/hadoop-1.2.1/bin/hadoop-config.sh
+
+RUN apt-get -y install build-essential
+RUN cd /opt/workspace/zlib-1.2.8 && ./configure && make && make install
+RUN apt-get -y install libssl-dev
+ENV PATH /usr/bin:/bin:/opt/hadoop-1.2.1/bin:/usr/local/sbin:/usr/sbin:/sbin
+ENV LD_LIBRARY_PATH $JAVA_HOME/jre/lib/amd64/server:$JAVA_HOME/jre/lib/amd64/:/opt/hadoop-1.2.1/c++/Linux-amd64-64/lib:/usr/local/lib/
+RUN cp /opt/workspace/ifconfig /sbin/
+
+ADD terawork /opt/tera
+RUN mv /opt/tera/bin/tera_setup.py /opt

example/docker/hdfs.py

@@ -0,0 +1,47 @@
+import time
+import os
+
+class Hdfs:
+	def __init__(self, ip, mode, log_prefix):
+		self.ip = ip
+		self.mode = mode
+		self.path = self.get_log_path(log_prefix)
+
+	def get_log_path(self, log_prefix):
+		path = '{pre}/hdfs/{ip}-{mode}-{time}'.format(pre=log_prefix, ip=self.ip, mode=self.mode, time=time.strftime('%Y%m%d%H%M%S'))
+		os.makedirs(path)
+		return path
+
+	def to_string(self):
+		info = 'hdfs\t{ip}:\t{mode}\tlog:{log}'.format(ip=self.ip, mode=self.mode, log=self.path)
+		return info
+
+	def to_cmd(self, docker, masters, slaves):
+		cmd = 'docker run -t -d -v {dir}:/opt/share -p 9000:9000 -p 9001:9001 --net=host {docker} /usr/bin/python /opt/hdfs_setup.py --masters {master} --slaves {slaves} --mode {mode}'.\
+			format(dir=self.path, docker=docker, master=masters, slaves=slaves, mode=self.mode)
+		return cmd
+
+class HdfsCluster:
+	def __init__(self, ip_list, num_of_hdfs, log_prefix):
+		self.ip_list = ip_list
+		self.ip_index = 0
+		self.num_of_hdfs = num_of_hdfs
+		self.cluster = []
+		self.log_prefix = log_prefix
+		self.master_ip = self.ip_list[0]
+		self.slave_ip = []
+
+	def add_hdfs(self):
+		hdfs = Hdfs(self.ip_list[self.ip_index], 'slave', self.log_prefix)
+		self.cluster.append(hdfs)
+		self.slave_ip.append(hdfs.ip)
+		self.ip_index += 1
+
+	def populate_hdfs_cluster(self):
+		if self.num_of_hdfs > len(self.ip_list):
+			print 'not enough ip address for hdfs!!'
+			return False
+		master = Hdfs(self.ip_list[0], 'master', self.log_prefix)
+		self.cluster.append(master)
+		for i in range(self.num_of_hdfs):
+			self.add_hdfs()

example/docker/hdfs_setup.py

@@ -1,9 +1,10 @@
 import subprocess
 import argparse
+import time
 from xml.dom.minidom import Document
 
-conf_path_prefix = '/home/ubuntu/leiliyuan/hadoop-1.2.1/conf/'
-share_path_prefix = '/home/ubuntu/leiliyuan/share/'
+conf_path_prefix = '/opt/hadoop-1.2.1/conf/'
+share_path_prefix = '/opt/share/'
 
 def parse_input():
 	parser = argparse.ArgumentParser()
@@ -54,7 +55,7 @@ def write_core_site_xml(args):
 
 def write_hdfs_site_xml():
 	doc, hdfssite = get_doc()	
-	prop = write_property(doc, 'dfs.data.dir', share_path_prefix + 'name', True)
+	prop = write_property(doc, 'dfs.name.dir', share_path_prefix + 'name', True)
 	hdfssite.appendChild(prop)
 	prop = write_property(doc, 'dfs.data.dir', share_path_prefix + 'data', True)
 	hdfssite.appendChild(prop)
@@ -91,7 +92,7 @@ def write_maters_slaves(args):
 	f.close()
 
 def start_hdfs(args):
-	cmd_prefix = '/home/ubuntu/leiliyuan/hadoop-1.2.1/bin/hadoop-daemon.sh --config /home/ubuntu/leiliyuan/hadoop-1.2.1/'
+	cmd_prefix = '/opt/hadoop-1.2.1/bin/hadoop-daemon.sh --config /opt/hadoop-1.2.1/'
 	if args.mode == 'master':
 		p = subprocess.Popen('/opt/hadoop-1.2.1/bin/hadoop namenode -format', stdout=subprocess.PIPE, shell=True)
 		print p.stdout.read()
@@ -103,14 +104,19 @@ def start_hdfs(args):
 		p = subprocess.Popen(cmd_prefix + 'conf start datanode', stdout=subprocess.PIPE, shell=True)
 		print p.stdout.read()
 
+def doing_nothing():
+	while True:
+		time.sleep(1000)
+
 def main():
 	args = parse_input()
 	write_core_site_xml(args)
 	write_hdfs_site_xml()
 	#write_mapred_site_xml(args)
 	write_hadoop_env()
 	write_maters_slaves(args)
-	#start_hdfs(args)
+	start_hdfs(args)
+	doing_nothing()
 
 if __name__ == '__main__':
 	main()

example/docker/master

@@ -0,0 +1,5 @@
+#!/bin/bash
+
+export CLASSPATH=.:$CLASSPATH:$JAVA_HOME/lib:$JAVA_HOME/jre/lib:$JAVA_HOME/jre/lib/amd64/:/home/ubuntu/leiliyuan/hadoop-1.2.1/lib:$(hadoop classpath)
+cd /opt/tera/bin
+nohup ./tera_main --flagfile=../conf/tera.flag --tera_role=master &> ../../share/log/master.stderr &

example/docker/tabletnode

@@ -0,0 +1,5 @@
+#!/bin/bash
+
+export CLASSPATH=.:$CLASSPATH:$JAVA_HOME/lib:$JAVA_HOME/jre/lib:$JAVA_HOME/jre/lib/amd64/:/home/ubuntu/leiliyuan/hadoop-1.2.1/lib:$(hadoop classpath)
+cd /opt/tera/bin
+nohup ./tera_main --flagfile=../conf/tera.flag --tera_role=tabletnode &> ../../share/log/tabletserver.stderr &

example/docker/tera.py

@@ -0,0 +1,45 @@
+import time
+import os
+
+class Tera:
+	def __init__(self, ip, port, mode, log_prefix):
+		self.ip = ip
+		self.port = port
+		self.mode = mode
+		self.path = self.get_log_path(log_prefix)
+
+	def get_log_path(self, log_prefix):
+		path = '{pre}/tera/{ip}:{port}-{mode}-{time}'.format(pre=log_prefix, ip=self.ip, port=self.port,
+																												 mode=self.mode, time=time.strftime('%Y%m%d%H%M%S'))
+		os.makedirs(path)
+		return path
+
+	def to_string(self):
+		info = 'tera\t{ip}:{port}\t{mode}\tlog:{log}'.format(ip=self.ip, port=self.port, mode=self.mode, log=self.path)
+		return info
+
+	def to_cmd(self, docker, zk):
+		cmd = 'docker run -t -d -v {dir}:/opt/share -p {port}:{port} --net=host {docker} /usr/bin/python /opt/tera_setup.py --zk {zk} --port {port} --mode {mode}'.\
+			format(dir=self.path, port=self.port, zk=zk, docker=docker, mode=self.mode)
+		return cmd
+
+class TeraCluster():
+	def __init__(self, ip_list, num_of_tera, log_prefix):
+		self.ip_list = ip_list
+		self.ip_index = 0
+		self.port = 2200
+		self.num_of_tera = num_of_tera
+		self.log_prefix = log_prefix
+		self.cluster = []
+
+	def add_tera(self):
+		tera = Tera(self.ip_list[self.ip_index % len(self.ip_list)], str(self.port), 'tabletnode', self.log_prefix)
+		self.cluster.append(tera)
+		self.ip_index += 1
+		self.port += 1
+
+	def populate_tera_cluster(self):
+		for i in range(self.num_of_tera):
+			self.add_tera()
+		master = Tera(self.ip_list[0], '1100', 'master', self.log_prefix)
+		self.cluster.append(master)