Repository navigation
Expand file tree
/
Copy pathstorm
More file actions
executable file
·93 lines (74 loc) · 3.55 KB
/
Copy pathstorm
File metadata and controls
executable file
·93 lines (74 loc) · 3.55 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
#!/usr/bin/python
import os
import sys
import random
import subprocess as sub
CONF_DIR = os.path.expanduser("~/.storm")
STORM_DIR = "/".join(os.path.abspath( __file__ ).split("/")[:-2])
def get_jars_full(adir):
files = os.listdir(adir)
ret = []
for f in files:
if f.endswith(".jar"):
ret.append(adir + "/" + f)
return ret
def get_classpath(extrajars):
ret = get_jars_full(STORM_DIR)
ret.extend(get_jars_full(STORM_DIR + "/lib"))
ret.extend(extrajars)
return ":".join(ret)
def confvalue(name, extrapaths):
cp = get_classpath(extrapaths)
command = ["java", "-client", "-cp", cp, "backtype.storm.command.config_value", name]
p = sub.Popen(command,stdout=sub.PIPE)
output, errors = p.communicate()
lines = output.split("\n")
for line in lines:
tokens = line.split(" ")
if tokens[0] == "VALUE:":
return tokens[1]
def print_localconfvalue(name):
print name + ": " + confvalue(name, [CONF_DIR])
def print_remoteconfvalue(name):
print name + ": " + confvalue(name, [STORM_DIR + "/conf"])
def exec_storm_class(klass, jvmtype="-server", childopts="", extrajars=[], args=[], prefix=""):
nativepath = confvalue("java.library.path", extrajars)
command = prefix + " java " + jvmtype + " -Djava.library.path=" + nativepath + " " + childopts + " -cp " + get_classpath(extrajars) + " " + klass + " " + " ".join(args)
print "Running: " + command
os.system(command)
COMMAND = sys.argv[1]
ARGS = sys.argv[2:]
def jar(jarfile, klass, *args):
exec_storm_class(
klass,
jvmtype="-client",
extrajars=[jarfile, CONF_DIR, STORM_DIR + "/bin"],
args=args,
prefix="export STORM_JAR=" + jarfile + ";")
def kill(name):
exec_storm_class("backtype.storm.command.kill_topology", args=[name], jvmtype="-client", extrajars=[CONF_DIR, STORM_DIR + "/bin"])
def shell(resourcesdir, command, *args):
tmpjarpath = "stormshell" + str(random.randint(0, 10000000)) + ".jar"
os.system("jar cf %s %s" % (tmpjarpath, resourcesdir))
runnerargs = [tmpjarpath, command]
runnerargs.extend(args)
exec_storm_class("backtype.storm.command.shell_submission", args=runnerargs, jvmtype="-client", extrajars=[CONF_DIR])
os.system("rm " + tmpjarpath)
def repl():
cppaths = [STORM_DIR + "/conf"]
exec_storm_class("clojure.lang.Repl", jvmtype="-client", extrajars=cppaths)
def nimbus():
cppaths = [STORM_DIR + "/log4j", STORM_DIR + "/conf"]
childopts = confvalue("nimbus.childopts", cppaths) + " -Dlogfile.name=nimbus.log -Dlog4j.configuration=storm.log.properties"
exec_storm_class("backtype.storm.daemon.nimbus", jvmtype="-server", extrajars=cppaths, childopts=childopts)
def supervisor():
cppaths = [STORM_DIR + "/log4j", STORM_DIR + "/conf"]
childopts = confvalue("nimbus.childopts", cppaths) + " -Dlogfile.name=supervisor.log -Dlog4j.configuration=storm.log.properties"
exec_storm_class("backtype.storm.daemon.supervisor", jvmtype="-server", extrajars=cppaths, childopts=childopts)
def ui():
childopts = "-Xmx768m -Dlogfile.name=ui.log -Dlog4j.configuration=storm.log.properties"
exec_storm_class("backtype.storm.ui.core", jvmtype="-server", childopts=childopts, extrajars=[STORM_DIR + "/log4j", STORM_DIR, STORM_DIR + "/conf"])
def print_classpath():
print get_classpath([])
COMMANDS = {"jar": jar, "kill": kill, "shell": shell, "nimbus": nimbus, "ui": ui, "supervisor": supervisor, "localconfvalue": print_localconfvalue, "remoteconfvalue": print_remoteconfvalue, "repl": repl, "classpath": print_classpath}
COMMANDS[COMMAND](*ARGS)