You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@storm.apache.org by priyank5485 <gi...@git.apache.org> on 2015/07/31 20:24:36 UTC
[GitHub] storm pull request: STORM-904: Move bin/storm command line to java...
GitHub user priyank5485 opened a pull request:
https://github.com/apache/storm/pull/662
STORM-904: Move bin/storm command line to java.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/priyank5485/storm storm-904
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/storm/pull/662.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #662
----
commit 193f59e027a8ab95d72c8e386d66f0b90f9b86f9
Author: Priyank <ps...@hortonworks.com>
Date: 2015-07-17T23:45:13Z
Remove python and add a java class to support storm cli.
Add a java class that encapsulates storm cli functionality currently
coded in two different places. storm, storm.py for bash and storm.cmd
for windows. Now both storm and storm.cmd will call the java class
eliminating python code and putting all the code in one place for bash
and windows respectively.
commit 793781cfd55c56f7184910fd744341351f6cc4bf
Author: Priyank <ps...@hortonworks.com>
Date: 2015-07-20T23:04:48Z
Add apace commons cli library for command line parsing
commit a425fc50ab9b518337d0c095ae80e9d78221e3fd
Author: Priyank <ps...@hortonworks.com>
Date: 2015-07-21T17:44:30Z
Adding version for dependency in pom file.
commit 5a4ec36ac17af69297dfc73aaa682b8e43edf5de
Author: Priyank <ps...@hortonworks.com>
Date: 2015-07-21T21:23:54Z
Sample code to test for apache commons cli.
commit 1a8240a558516110a5c14d3a4bb0b177ab76090b
Author: Priyank <ps...@hortonworks.com>
Date: 2015-07-22T23:40:40Z
Removing apache commons cli, renaming and some intiialization code.
commit b69e295e81216e992187737cc069ff66d2e74687
Author: Priyank <ps...@hortonworks.com>
Date: 2015-07-23T10:14:48Z
Some redesigning/refactoring.
commit 8c2fbeadbf699b5b286479649d133623c92bc85c
Author: Priyank <ps...@hortonworks.com>
Date: 2015-07-23T21:49:43Z
Add printUsage mainly.
commit 74f45c7c9d6ae813635c91ba14fdb1cd3dd24c74
Author: Priyank <ps...@hortonworks.com>
Date: 2015-07-24T18:49:15Z
Call storm command methods using reflection.
commit 10ba25f8c81e2fdc34f4ae3e0e2cdf9ecd142a4a
Author: Priyank <ps...@hortonworks.com>
Date: 2015-07-24T19:25:01Z
Add method boxes without code for other storm commands.
commit fb2022db8dda7a5183ee5a2cc64a7ef035006c80
Author: Priyank <ps...@hortonworks.com>
Date: 2015-07-27T17:59:20Z
Initial executeStorm method creating a new process.
Need to test storm daemons if they can be killed using shutdown hooks.
commit 493b0dfeb7be2daa0a54e7dec2f3f716d0af0e26
Author: Priyank <ps...@hortonworks.com>
Date: 2015-07-29T07:02:49Z
Ship some more code over from python.
commit 3082d4f5ea2d66babe232e5ff49506b322385473
Author: Priyank <ps...@hortonworks.com>
Date: 2015-07-29T18:49:57Z
Fix an exception which failed to start the nimbus command.
commit c616068d90dd47fa61fbb670c2853805d7ccb324
Author: Priyank <ps...@hortonworks.com>
Date: 2015-07-29T21:53:27Z
Use exec to replace current process rather than creating a new one.
commit 06f4dcc17d8dcf3c81094557ed97ff117d6410ed
Author: Priyank <ps...@hortonworks.com>
Date: 2015-07-30T08:37:08Z
Add some commented code.
commit da0ad60438841a1f874c115cef27190a59401f4f
Author: Priyank <ps...@hortonworks.com>
Date: 2015-07-30T15:45:52Z
Add nil check for strings returned from conclave
commit 0b242adf0bc261e9140c3587b3eedd2141e31718
Author: Priyank <ps...@hortonworks.com>
Date: 2015-07-30T19:04:35Z
Fix an issue with passing options to storm command,
commit 0bb15782c171d336be905d2f5143865e288a8239
Author: Priyank <ps...@hortonworks.com>
Date: 2015-07-31T00:37:07Z
STORM-904: Add shutdown hook thread to kill subprocess.
Does not work for kill -9. Need to debug.
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-904: Move bin/storm command line to java...
Posted by caofangkun <gi...@git.apache.org>.
Github user caofangkun commented on a diff in the pull request:
https://github.com/apache/storm/pull/662#discussion_r36284000
--- Diff: storm-core/src/jvm/backtype/storm/utils/StormCommandExecutor.java ---
@@ -0,0 +1,785 @@
+package backtype.storm.utils;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.nio.charset.StandardCharsets;
+import java.util.*;
+
+import clojure.lang.IFn;
+import org.apache.commons.lang.StringUtils;
+import org.apache.commons.lang.SystemUtils;
+
+/**
+ * Created by pshah on 7/17/15.
+ */
+abstract class StormCommandExecutor {
+ final String NIMBUS_CLASS = "backtype.storm.daemon.nimbus";
+ final String SUPERVISOR_CLASS = "backtype.storm.daemon.supervisor";
+ final String UI_CLASS = "backtype.storm.ui.core";
+ final String LOGVIEWER_CLASS = "backtype.storm.daemon.logviewer";
+ final String DRPC_CLASS = "backtype.storm.daemon.drpc";
+ final String REPL_CLASS = "clojure.main";
+ final String ACTIVATE_CLASS = "backtype.storm.command.activate";
+ final String DEACTIVATE_CLASS = "backtype.storm.command.deactivate";
+ final String REBALANCE_CLASS = "backtype.storm.command.rebalance";
+ final String LIST_CLASS = "backtype.storm.command.list";
+ String stormHomeDirectory;
+ String userConfDirectory;
+ String stormConfDirectory;
+ String clusterConfDirectory;
+ String stormLibDirectory;
+ String stormBinDirectory;
+ String stormLog4jConfDirectory;
+ String configFile = "";
+ String javaCommand;
+ List<String> configOptions = new ArrayList<String>();
+ String stormExternalClasspath;
+ String stormExternalClasspathDaemon;
+ String fileSeparator;
+ final List<String> COMMANDS = Arrays.asList("jar", "kill", "shell",
+ "nimbus", "ui", "logviewer", "drpc", "supervisor",
+ "localconfvalue", "remoteconfvalue", "repl", "classpath",
+ "activate", "deactivate", "rebalance", "help", "list",
+ "dev-zookeeper", "version", "monitor", "upload-credentials",
+ "get-errors");
+
+ public static void main (String[] args) {
+ for (String arg : args) {
+ System.out.println("Argument ++ is " + arg);
+ }
+ StormCommandExecutor stormCommandExecutor;
+ if (System.getProperty("os.name").startsWith("Windows")) {
+ stormCommandExecutor = new WindowsStormCommandExecutor();
+ } else {
+ stormCommandExecutor = new UnixStormCommandExecutor();
+ }
+ stormCommandExecutor.initialize();
+ stormCommandExecutor.execute(args);
+ }
+
+ StormCommandExecutor () {
+
+ }
+
+ abstract void initialize ();
+
+ abstract void execute (String[] args);
+
+ void callMethod (String command, List<String> args) {
+ Class implementation = this.getClass();
+ String methodName = command.replace("-", "") + "Command";
+ try {
+ Method method = implementation.getDeclaredMethod(methodName, List
+ .class);
+ method.invoke(this, args);
+ } catch (NoSuchMethodException ex) {
+ System.out.println("No such method exception occured while trying" +
+ " to run storm method " + command);
+ } catch (IllegalAccessException ex) {
+ System.out.println("Illegal access exception occured while trying" +
+ " to run storm method " + command);
+ } catch (IllegalArgumentException ex) {
+ System.out.println("Illegal argument exception occured while " +
+ "trying" + " to run storm method " + command);
+ } catch (InvocationTargetException ex) {
+ System.out.println("Invocation target exception occured while " +
+ "trying" + " to run storm method " + command);
+ }
+ }
+}
+
+class UnixStormCommandExecutor extends StormCommandExecutor {
+
+ UnixStormCommandExecutor () {
+
+ }
+
+ void initialize () {
+ Collections.sort(this.COMMANDS);
+ this.fileSeparator = System .getProperty ("file.separator");
+ this.stormHomeDirectory = System.getenv("STORM_BASE_DIR");
+ this.userConfDirectory = System.getProperty("user.home") +
+ this.fileSeparator + "" +
+ ".storm";
+ this.stormConfDirectory = System.getenv("STORM_CONF_DIR");
+ this.clusterConfDirectory = this.stormConfDirectory == null ? (this
+ .stormHomeDirectory + this.fileSeparator + "conf") : this
+ .stormConfDirectory;
+ File f = new File(this.userConfDirectory + this.fileSeparator +
+ "storm.yaml");
+ if (!f.isFile()) {
+ this.userConfDirectory = this.clusterConfDirectory;
+ }
+ this.stormLibDirectory = this.stormHomeDirectory + this.fileSeparator +
+ "lib";
+ this.stormBinDirectory = this.stormHomeDirectory + this.fileSeparator +
+ "bin";
+ this.stormLog4jConfDirectory = this.stormHomeDirectory +
+ this.fileSeparator + "log4j2";
+ if (System.getenv("JAVA_HOME") != null) {
+ this.javaCommand = System.getenv("JAVA_HOME") + this.fileSeparator +
+ "bin" + this.fileSeparator + "java";
+ if (!(new File(this.javaCommand).exists())) {
+ System.out.println("ERROR: JAVA_HOME is invalid. Could not " +
+ "find " + this.javaCommand);
+ System.exit(1);
+ }
+ } else {
+ this.javaCommand = "java";
+ }
+ this.stormExternalClasspath = System.getenv("STORM_EXT_CLASSPATH");
+ this.stormExternalClasspathDaemon = System.getenv
+ ("STORM_EXT_CLASSPATH_DAEMON");
+ if (!(new File(this.stormLibDirectory).exists())) {
+ System.out.println("******************************************");
+ System.out.println("The storm client can only be run from within " +
+ "a release. " + "You appear to be trying to run the client" +
+ " from a checkout of Storm's source code.");
+ System.out.println("You can download a Storm release at " +
+ "http://storm-project.net/downloads.html");
+ System.out.println("******************************************");
+ System.exit(1);
+ }
+ //System.getProperties().list(System.out);
+ }
+
+ void execute (String[] args) {
+ if (args.length == 0) {
+ this.printUsage();
+ System.exit(-1);
+ }
+ List<String> commandArgs = new ArrayList<String>();
+ for (int i = 0; i < args.length; ++i) {
+ if (args[i] == "-c") {
+ this.configOptions.add(args[++i]);
+ } else if (args[i] == "--config") {
+ this.configFile = args[++i];
+ } else {
+ commandArgs.add(args[i]);
+ }
+ }
+ if ((commandArgs.size() == 0) || (!this.COMMANDS.contains
+ (commandArgs.get(0)))) {
+ System.out.println("Unknown command: [storm " + StringUtils.join
+ (args, " ") + "]");
+ this.printUsage();
+ System.exit(254);
+
+ }
+ this.callMethod(commandArgs.get(0), commandArgs.subList(1,
+ commandArgs.size()));
+
+ }
+
+ String getConfigOptions() {
+ String configOptions = "-Dstorm.options=";
+ //TODO - do urlencode here. python does quote_plus to each configoption
+ return configOptions + StringUtils.join(this.configOptions, ',');
+
+ }
+
+ List<String> getJarsFull (String directory) {
+ List<String> fullJarFiles = new ArrayList<String>();
+ File file = new File(directory);
+ File[] files = file.listFiles();
+ if (files != null) {
+ for (File f : files) {
+ if (f.getName().endsWith(".jar")) {
+ fullJarFiles.add(f.getPath());
+ }
+ }
+ }
+ return fullJarFiles;
+ }
+
+ String getClassPath (List<String> extraJars, boolean daemon) {
+ List<String> classPaths = this.getJarsFull(this.stormHomeDirectory);
+ classPaths.addAll(this.getJarsFull(this.stormLibDirectory));
+ classPaths.addAll(this.getJarsFull(this.stormHomeDirectory + this
+ .fileSeparator + "extlib"));
+ if (daemon == true) {
+ classPaths.addAll(this.getJarsFull(this.stormHomeDirectory + this
+ .fileSeparator + "extlib-daemon"));
+ }
+ if (this.stormExternalClasspath != null) {
+ classPaths.add(this.stormExternalClasspath);
+ }
+ if (this.stormExternalClasspathDaemon != null) {
+ classPaths.add(this.stormExternalClasspathDaemon);
+ }
+ classPaths.addAll(extraJars);
+ return StringUtils.join(classPaths, System.getProperty("path" +
+ ".separator"));
+ }
+
+ String confValue (String name, List<String> extraPaths, boolean daemon) {
+ // The original code from python started a process that started a jvm
+ // with backtype.storm.command.config_value main method that would
+ // read the conf value and print it out to an output stream. python
+ // tapped on to the output stream of that subprocess and returned the
+ // confvalue for the name. Because the pythong code has been shipped
+ // to java now it should not spawn a new process which is a jvm since
+ // we are already in jvm. Instead it should just be doing as the code
+ // commeneted below.
+ // However looking at the pythong code it was
+ // starting a jvm with -cp argument that had classpaths which might
+ // not be available to this java process. Hence there is a chance
+ // that the below code might break existing scripts. As a result I
+ // have decided to still spawn a new process from java just like
+ // python with similar classpaths being constructed for the jvm
+ // execution
+ /*IFn fn = Utils.loadClojureFn("backtype.storm.config",
+ "read-storm-config");
+ Object o = fn.invoke();
+ return ((Map) o).get(name).toString();*/
+ String confValue = "";
+ ProcessBuilder processBuilder = new ProcessBuilder(this.javaCommand,
+ "-client", this.getConfigOptions(), "-Dstorm.conf.file=" +
+ this.configFile, "-cp", this.getClassPath(extraPaths, daemon),
+ "backtype.storm.command.config_value", name);
+ BufferedReader br;
+ try {
+ Process process = processBuilder.start();
+ br = new BufferedReader(new InputStreamReader(process
+ .getInputStream(), StandardCharsets.UTF_8));
+ process.waitFor();
+ String line;
+ while ((line = br.readLine()) != null) {
+ String[] tokens = line.split(" ");
+ if ("VALUE:".equals(tokens[0])) {
+ confValue = StringUtils.join(Arrays.copyOfRange(tokens, 1,
+ tokens.length), " ");
+ break;
+ }
+ }
+ br.close();
+ } catch (Exception ex) {
+ System.out.println("Exception occured while starting process via " +
+ "processbuilder " + ex.getMessage());
+ }
+ return confValue;
+ }
+
+ void executeStormClass (String className, String jvmType, List<String>
+ jvmOptions, List<String> extraJars, List<String> args, boolean
+ fork, boolean daemon, String daemonName) {
+ List<String> extraPaths = new ArrayList<>();
+ extraPaths.add(this.clusterConfDirectory);
+ String stormLogDirectory = this.confValue("storm.log.dir",
+ extraPaths, daemon);
+ if ((stormLogDirectory == null) || ("".equals(stormLogDirectory)) ||
+ ("nil".equals(stormLogDirectory))) {
+ stormLogDirectory = this.stormHomeDirectory + this.fileSeparator
+ + "logs";
+ }
+ List<String> commandList = new ArrayList<String>();
+ commandList.add(this.javaCommand);
+ commandList.add(jvmType);
+ commandList.add("-Ddaemon.name=" + daemonName);
+ commandList.add(this.getConfigOptions());
+ commandList.add("-Dstorm.home=" + this.stormHomeDirectory);
+ commandList.add("-Dstorm.log.dir=" + stormLogDirectory);
+ commandList.add("-Djava.library.path=" + this
+ .confValue("java.library.path", extraJars, daemon));
+ commandList.add("-Dstorm.conf.file=" + this.configFile);
+ commandList.add("-cp");
+ commandList.add(this.getClassPath(extraJars, daemon));
+ commandList.addAll(jvmOptions);
+ commandList.add(className);
+ commandList.addAll(args);
+ ProcessBuilder processBuilder = new ProcessBuilder(commandList);
+ processBuilder.inheritIO();
+ try {
+ Process process = processBuilder.start();
+ System.out.println("Executing the command: ");
+ String commandLine = StringUtils.join(commandList, " ");
+ System.out.println(commandLine);
+ if (daemon == true) {
+ Runtime.getRuntime().addShutdownHook(new ShutdownHookThread
+ (process, commandLine));
+ }
+ System.out.println("Waiting for subprocess to finish");
+ process.waitFor();
+ System.out.println("subprocess finished");
+ System.out.println("Exit value from subprocess is :" + process
+ .exitValue());
+ } catch (Exception ex) {
+ System.out.println("Exception occured while starting process via " +
+ "processbuilder " + ex.getMessage());
+ }
+ }
+
+ void jarCommand (List<String> args) {
+ System.out.println("Called jarCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ if ((args == null) || (args.size() < 2)) {
+ System.out.println("Not enough arguments for storm jar command");
+ System.out.println("Please pass a jar file location and the " +
+ "topology class for jar command");
+ //TODO print usage for jar command here
+ System.exit(-1);
+ }
+ String jarJvmOptions = System.getenv("STORM_JAR_JVM_OPTS");
+ List<String> jvmOptions = new ArrayList<String>();
+ if (jarJvmOptions != null) {
+ //TODO the python code to parse STORM_JAR_JVM_OPTIONS uses shlex
+ // .split to get the different jvm options for the jar. For now
+ // keeping it simple and splitting on space. Need to be in synch
+ // with python. Not sure though if we really need to use a
+ // lexical parser
+ jvmOptions.addAll(Arrays.asList(jarJvmOptions.split(" ")));
+ }
+ jvmOptions.add("-Dstorm.jar=" + args.get(0));
+ List<String> extraPaths = new ArrayList<String>();
+ extraPaths.add(args.get(0));
+ extraPaths.add(this.userConfDirectory);
+ extraPaths.add(this.stormBinDirectory);
+ this.executeStormClass(args.get(1), "-client", jvmOptions,
+ extraPaths, args.subList(2, args.size()), false, false, "");
+ return;
+ }
+
+ void killCommand (List<String> args) {
+ System.out.println("Called killCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ if ((args == null) || (args.size() < 1)) {
+ System.out.println("Not enough arguments for storm kill command");
+ //TODO print usage for kill command here
+ System.exit(2);
+ }
+ List<String> extraPaths = new ArrayList<String>();
+ extraPaths.add(this.userConfDirectory);
+ extraPaths.add(this.stormBinDirectory);
+ this.executeStormClass("backtype.storm.command.kill_topology",
+ "-client", new ArrayList<String>(), extraPaths, args, false,
+ false, "");
+ return;
+ }
+
+ //TODO implement shell command after understanding more about it
+ void shellCommand (List<String> args) {
+ System.out.println("Called shellCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ return;
+ }
+
+ void nimbusCommand (List<String> args) {
+ System.out.println("Called nimbusCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ List<String> jvmOptions = new ArrayList<String>();
+ List<String> extraPaths = new ArrayList<String>();
+ extraPaths.add(this.clusterConfDirectory);
+ String nimbusOptions = this.confValue("nimbus.childopts", extraPaths,
+ true);
+ // below line is different from original python script storm.py where
+ // it called parse_args method on nimbusOptions. Now we just call a
+ // split with a space. Hence this will have different behavior and
+ // a buggy one if the nimbusOptions string in the config file has a
+ // space. TODO need to fix this
+ jvmOptions.addAll(Arrays.asList(nimbusOptions.split(" ")));
+ jvmOptions.add("-Dlogfile.name=nimbus.log");
+ jvmOptions.add("-Dlog4j.configurationFile=" + this
+ .getLog4jConfigDirectory() + this.fileSeparator + "cluster" +
+ ".xml");
+ this.executeStormClass(this.NIMBUS_CLASS, "-server", jvmOptions,
+ extraPaths, new ArrayList<String>(), false, true, "nimbus");
+ return;
+ }
+
+ void uiCommand (List<String> args) {
+ System.out.println("Called uiCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ List<String> jvmOptions = new ArrayList<String>();
+ List<String> extraPaths = new ArrayList<String>();
+ extraPaths.add(this.clusterConfDirectory);
+ String uiOptions = this.confValue("ui.childopts", extraPaths,
+ true);
+ // below line is different from original python script storm.py where
+ // it called parse_args method on nimbusOptions. Now we just call a
+ // split with a space. Hence this will have different behavior and
+ // a buggy one if the nimbusOptions string in the config file has a
+ // space. TODO need to fix this
+ jvmOptions.addAll(Arrays.asList(uiOptions.split(" ")));
+ jvmOptions.add("-Dlogfile.name=ui.log");
+ jvmOptions.add("-Dlog4j.configurationFile=" + this
+ .getLog4jConfigDirectory() + this.fileSeparator + "cluster" +
+ ".xml");
+ extraPaths.add(0, this.stormHomeDirectory);
+ this.executeStormClass(this.UI_CLASS, "-server", jvmOptions,
+ extraPaths, new ArrayList<String>(), false, true, "ui");
+ return;
+ }
+
+ void logviewerCommand (List<String> args) {
+ System.out.println("Called logviewerCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ List<String> jvmOptions = new ArrayList<String>();
+ List<String> extraPaths = new ArrayList<String>();
+ extraPaths.add(this.clusterConfDirectory);
+ String logviewerOptions = this.confValue("logviewer.childopts",
+ extraPaths, true);
+ // below line is different from original python script storm.py where
+ // it called parse_args method on nimbusOptions. Now we just call a
+ // split with a space. Hence this will have different behavior and
+ // a buggy one if the nimbusOptions string in the config file has a
+ // space. TODO need to fix this
+ jvmOptions.addAll(Arrays.asList(logviewerOptions.split(" ")));
+ jvmOptions.add("-Dlogfile.name=logviewer.log");
+ jvmOptions.add("-Dlog4j.configurationFile=" + this
+ .getLog4jConfigDirectory() + this.fileSeparator + "cluster" +
+ ".xml");
+ extraPaths.add(0, this.stormHomeDirectory);
+ this.executeStormClass(this.LOGVIEWER_CLASS, "-server", jvmOptions,
+ extraPaths, new ArrayList<String>(), false, true, "logviewer");
+ return;
+ }
+
+ void drpcCommand (List<String> args) {
+ System.out.println("Called drpcCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ List<String> jvmOptions = new ArrayList<String>();
+ List<String> extraPaths = new ArrayList<String>();
+ extraPaths.add(this.clusterConfDirectory);
+ String drpcOptions = this.confValue("drpc.childopts", extraPaths,
+ true);
+ // below line is different from original python script storm.py where
+ // it called parse_args method on nimbusOptions. Now we just call a
+ // split with a space. Hence this will have different behavior and
+ // a buggy one if the nimbusOptions string in the config file has a
+ // space. TODO need to fix this
+ jvmOptions.addAll(Arrays.asList(drpcOptions.split(" ")));
+ jvmOptions.add("-Dlogfile.name=drpc.log");
+ jvmOptions.add("-Dlog4j.configurationFile=" + this
+ .getLog4jConfigDirectory() + this.fileSeparator + "cluster" +
+ ".xml");
+ this.executeStormClass(this.DRPC_CLASS, "-server", jvmOptions,
+ extraPaths, new ArrayList<String>(), false, true, "drpc");
+ return;
+ }
+
+ void supervisorCommand (List<String> args) {
+ System.out.println("Called supervisorCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ List<String> jvmOptions = new ArrayList<String>();
+ List<String> extraPaths = new ArrayList<String>();
+ extraPaths.add(this.clusterConfDirectory);
+ String supervisorOptions = this.confValue("supervisor.childopts",
+ extraPaths,
+ true);
+ // below line is different from original python script storm.py where
+ // it called parse_args method on nimbusOptions. Now we just call a
+ // split with a space. Hence this will have different behavior and
+ // a buggy one if the nimbusOptions string in the config file has a
+ // space. TODO need to fix this
+ jvmOptions.addAll(Arrays.asList(supervisorOptions.split(" ")));
+ jvmOptions.add("-Dlogfile.name=supervisor.log");
+ jvmOptions.add("-Dlog4j.configurationFile=" + this
+ .getLog4jConfigDirectory() + this.fileSeparator + "cluster" +
+ ".xml");
+ this.executeStormClass(this.SUPERVISOR_CLASS, "-server", jvmOptions,
+ extraPaths, new ArrayList<String>(), false, true, "supervisor");
+ return;
+ }
+
+ void localconfvalueCommand (List<String> args) {
+ System.out.println("Called localconfvalueCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ if ((args == null) || (args.size() == 0)) {
+ System.out.println("Not enough arguments for localconfvalue " +
+ "command");
+ System.out.println("Please pass the name of the config value you " +
+ "want to be printed");
+ //TODO print command help for localconfvalue command
+ System.exit(2);
+ }
+ List<String> extraPaths = new ArrayList<String>();
+ extraPaths.add(this.userConfDirectory);
+ System.out.println(args.get(0) + ": " + this.confValue(args.get(0),
+ extraPaths,
+ true));
+ return;
+ }
+
+ void remoteconfvalueCommand (List<String> args) {
+ System.out.println("Called remoteconfvalueCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ if ((args == null) || (args.size() == 0)) {
+ System.out.println("Not enough arguments for remoteconfvalue " +
+ "command");
+ System.out.println("Please pass the name of the config value you " +
+ "want to be printed");
+ //TODO print command help for remoteconfvalue command
+ System.exit(2);
+ }
+ List<String> extraPaths = new ArrayList<String>();
+ extraPaths.add(this.clusterConfDirectory);
+ System.out.println(args.get(0) + ": " + this.confValue(args.get(0),
+ extraPaths,
+ true));
+ return;
+ }
+
+ void replCommand (List<String> args) {
+ System.out.println("Called replCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ List<String> extraPaths = new ArrayList<String>();
+ extraPaths.add(this.clusterConfDirectory);
+ this.executeStormClass(this.REPL_CLASS, "-client", new
+ ArrayList<String>(), extraPaths, new ArrayList<String>(),
+ false, true, "");
+ return;
+ }
+
+ void classpathCommand (List<String> args) {
+ System.out.println("Called classpathCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ System.out.println(this.getClassPath(new ArrayList<String>(), true));
+ return;
+ }
+
+ void activateCommand (List<String> args) {
+ System.out.println("Called activateCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ if ((args == null) || (args.size() < 1)) {
+ System.out.println("Not enough arguments for activate command");
+ System.out.println("Please pass the topology name to activate");
+ //TODO print usage for activate command here
+ System.exit(2);
+ }
+ List<String> extraPaths = new ArrayList<String>();
+ extraPaths.add(this.userConfDirectory);
+ extraPaths.add(this.stormBinDirectory);
+ this.executeStormClass(this.ACTIVATE_CLASS, "-client", new
+ ArrayList<String>(), extraPaths, args, false, false, "");
+ return;
+ }
+
+ void deactivateCommand (List<String> args) {
+ System.out.println("Called deactivateCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ if ((args == null) || (args.size() < 1)) {
+ System.out.println("Not enough arguments for deactivate command");
+ System.out.println("Please pass the topology name to deactivate");
+ //TODO print usage for deactivate command here
+ System.exit(2);
+ }
+ List<String> extraPaths = new ArrayList<String>();
+ extraPaths.add(this.userConfDirectory);
+ extraPaths.add(this.stormBinDirectory);
+ this.executeStormClass(this.DEACTIVATE_CLASS, "-client", new
+ ArrayList<String>(), extraPaths, args, false, false, "");
+ return;
+ }
+
+ void rebalanceCommand (List<String> args) {
+ System.out.println("Called rebalanceCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ if ((args == null) || (args.size() < 1)) {
+ System.out.println("Not enough arguments for rebalance command");
+ System.out.println("Please pass the topology name to rebalance");
+ //TODO print usage for rebalance command here
+ System.exit(2);
+ }
+ List<String> extraPaths = new ArrayList<String>();
+ extraPaths.add(this.userConfDirectory);
+ extraPaths.add(this.stormBinDirectory);
+ this.executeStormClass(this.REBALANCE_CLASS, "-client", new
+ ArrayList<String>(), extraPaths, args, false, false, "");
+ return;
+ }
+
+ void helpCommand (List<String> args) {
+ System.out.println("Called helpCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ if ((args == null) || (args.size() == 0)) {
+ this.printUsage();
+ } else {
+ if ((!this.COMMANDS.contains(args.get(0)))) {
+ System.out.println(args.get(0) + " is not a valid command");
+ } else {
+ //TODO print indivudual commands help here
+ System.out.println("Print command specific help here");
+ }
+ }
+ return;
+ }
+
+ void listCommand (List<String> args) {
+ System.out.println("Called listCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ List<String> extraPaths = new ArrayList<String>();
+ extraPaths.add(this.userConfDirectory);
+ extraPaths.add(this.stormBinDirectory);
+ this.executeStormClass(this.LIST_CLASS, "-client", new
+ ArrayList<String>(), extraPaths, args, false, false, "");
+ return;
+ }
+
+ void devzookeeperCommand (List<String> args) {
+ System.out.println("Called devzookeeperCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ return;
+ }
+
+ void versionCommand (List<String> args) {
+ System.out.println("Called versionCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ return;
+ }
+
+ void monitorCommand (List<String> args) {
+ System.out.println("Called monitorCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ return;
+ }
+
+ void uploadcredentialsCommand (List<String> args) {
+ System.out.println("Called uploadcredentialsCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ return;
+ }
+
+ void geterrorsCommand (List<String> args) {
+ System.out.println("Called geterrorsCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ return;
+ }
+
+ String getLog4jConfigDirectory () {
+ List<String> extraPaths = new ArrayList<String>();
+ extraPaths.add(this.clusterConfDirectory);
+ String log4jDirectory = this.confValue("storm.logback.conf.dir",
+ extraPaths, true);
+ if ((log4jDirectory == null) || ("".equals(log4jDirectory)) ||
+ ("nil".equals(log4jDirectory))) {
+ log4jDirectory = this.stormLog4jConfDirectory;
+ }
+ return log4jDirectory;
+ }
+
+ private void printUsage () {
+ String commands = StringUtils.join(this.COMMANDS, "\n\t");
+ System.out.println("Commands:\n\t" + commands);
+ System.out.println("\nHelp: \n\thelp \n\thelp <command>\n");
+ System.out.println("Documentation for the storm client can be found" +
+ " at " +
+ "http://storm.incubator.apache" +
+ ".org/documentation/Command-line-client.html\n");
+ System.out.println("Configs can be overridden using one or more -c " +
+ "flags, e.g. " +
+ "\"storm list -c nimbus.host=nimbus.mycompany.com\"\n");
+ }
+
+ private void executeHelpCommand () {
+ System.out.println("Print storm help here");
+ }
+
+}
+
+class WindowsStormCommandExecutor extends StormCommandExecutor {
+
+ WindowsStormCommandExecutor () {
+
+ }
+
+ void initialize () {
+ return;
+ }
+
+ void execute (String[] args) {
--- End diff --
If run on Windows ,do nothing but return ？
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-904: Move bin/storm command line to java...
Posted by longdafeng <gi...@git.apache.org>.
Github user longdafeng commented on the pull request:
https://github.com/apache/storm/pull/662#issuecomment-157686065
@priyank5485
I don't prefer switch storm.py from python to java. Because I often add debug point in the storm.py, it is very easy to debug in online system(especially when environment is bad or jvm parameter is wrong). if we change to java version, it is very hard to debug.
Another point is that python's code is more easy to add new functions in the script.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-904: Move bin/storm command line to java...
Posted by priyank5485 <gi...@git.apache.org>.
Github user priyank5485 commented on a diff in the pull request:
https://github.com/apache/storm/pull/662#discussion_r36324914
--- Diff: storm-core/src/jvm/backtype/storm/utils/StormCommandExecutor.java ---
@@ -0,0 +1,785 @@
+package backtype.storm.utils;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.nio.charset.StandardCharsets;
+import java.util.*;
+
+import clojure.lang.IFn;
+import org.apache.commons.lang.StringUtils;
+import org.apache.commons.lang.SystemUtils;
+
+/**
+ * Created by pshah on 7/17/15.
+ */
+abstract class StormCommandExecutor {
+ final String NIMBUS_CLASS = "backtype.storm.daemon.nimbus";
+ final String SUPERVISOR_CLASS = "backtype.storm.daemon.supervisor";
+ final String UI_CLASS = "backtype.storm.ui.core";
+ final String LOGVIEWER_CLASS = "backtype.storm.daemon.logviewer";
+ final String DRPC_CLASS = "backtype.storm.daemon.drpc";
+ final String REPL_CLASS = "clojure.main";
+ final String ACTIVATE_CLASS = "backtype.storm.command.activate";
+ final String DEACTIVATE_CLASS = "backtype.storm.command.deactivate";
+ final String REBALANCE_CLASS = "backtype.storm.command.rebalance";
+ final String LIST_CLASS = "backtype.storm.command.list";
+ String stormHomeDirectory;
+ String userConfDirectory;
+ String stormConfDirectory;
+ String clusterConfDirectory;
+ String stormLibDirectory;
+ String stormBinDirectory;
+ String stormLog4jConfDirectory;
+ String configFile = "";
+ String javaCommand;
+ List<String> configOptions = new ArrayList<String>();
+ String stormExternalClasspath;
+ String stormExternalClasspathDaemon;
+ String fileSeparator;
+ final List<String> COMMANDS = Arrays.asList("jar", "kill", "shell",
+ "nimbus", "ui", "logviewer", "drpc", "supervisor",
+ "localconfvalue", "remoteconfvalue", "repl", "classpath",
+ "activate", "deactivate", "rebalance", "help", "list",
+ "dev-zookeeper", "version", "monitor", "upload-credentials",
+ "get-errors");
+
+ public static void main (String[] args) {
+ for (String arg : args) {
+ System.out.println("Argument ++ is " + arg);
+ }
+ StormCommandExecutor stormCommandExecutor;
+ if (System.getProperty("os.name").startsWith("Windows")) {
+ stormCommandExecutor = new WindowsStormCommandExecutor();
+ } else {
+ stormCommandExecutor = new UnixStormCommandExecutor();
+ }
+ stormCommandExecutor.initialize();
+ stormCommandExecutor.execute(args);
+ }
+
+ StormCommandExecutor () {
+
+ }
+
+ abstract void initialize ();
+
+ abstract void execute (String[] args);
+
+ void callMethod (String command, List<String> args) {
+ Class implementation = this.getClass();
+ String methodName = command.replace("-", "") + "Command";
+ try {
+ Method method = implementation.getDeclaredMethod(methodName, List
+ .class);
+ method.invoke(this, args);
+ } catch (NoSuchMethodException ex) {
+ System.out.println("No such method exception occured while trying" +
+ " to run storm method " + command);
+ } catch (IllegalAccessException ex) {
+ System.out.println("Illegal access exception occured while trying" +
+ " to run storm method " + command);
+ } catch (IllegalArgumentException ex) {
+ System.out.println("Illegal argument exception occured while " +
+ "trying" + " to run storm method " + command);
+ } catch (InvocationTargetException ex) {
+ System.out.println("Invocation target exception occured while " +
+ "trying" + " to run storm method " + command);
+ }
+ }
+}
+
+class UnixStormCommandExecutor extends StormCommandExecutor {
+
+ UnixStormCommandExecutor () {
+
+ }
+
+ void initialize () {
+ Collections.sort(this.COMMANDS);
+ this.fileSeparator = System .getProperty ("file.separator");
+ this.stormHomeDirectory = System.getenv("STORM_BASE_DIR");
+ this.userConfDirectory = System.getProperty("user.home") +
+ this.fileSeparator + "" +
+ ".storm";
+ this.stormConfDirectory = System.getenv("STORM_CONF_DIR");
+ this.clusterConfDirectory = this.stormConfDirectory == null ? (this
+ .stormHomeDirectory + this.fileSeparator + "conf") : this
+ .stormConfDirectory;
+ File f = new File(this.userConfDirectory + this.fileSeparator +
+ "storm.yaml");
+ if (!f.isFile()) {
+ this.userConfDirectory = this.clusterConfDirectory;
+ }
+ this.stormLibDirectory = this.stormHomeDirectory + this.fileSeparator +
+ "lib";
+ this.stormBinDirectory = this.stormHomeDirectory + this.fileSeparator +
+ "bin";
+ this.stormLog4jConfDirectory = this.stormHomeDirectory +
+ this.fileSeparator + "log4j2";
+ if (System.getenv("JAVA_HOME") != null) {
+ this.javaCommand = System.getenv("JAVA_HOME") + this.fileSeparator +
+ "bin" + this.fileSeparator + "java";
+ if (!(new File(this.javaCommand).exists())) {
+ System.out.println("ERROR: JAVA_HOME is invalid. Could not " +
+ "find " + this.javaCommand);
+ System.exit(1);
+ }
+ } else {
+ this.javaCommand = "java";
+ }
+ this.stormExternalClasspath = System.getenv("STORM_EXT_CLASSPATH");
+ this.stormExternalClasspathDaemon = System.getenv
+ ("STORM_EXT_CLASSPATH_DAEMON");
+ if (!(new File(this.stormLibDirectory).exists())) {
+ System.out.println("******************************************");
+ System.out.println("The storm client can only be run from within " +
+ "a release. " + "You appear to be trying to run the client" +
+ " from a checkout of Storm's source code.");
+ System.out.println("You can download a Storm release at " +
+ "http://storm-project.net/downloads.html");
+ System.out.println("******************************************");
+ System.exit(1);
+ }
+ //System.getProperties().list(System.out);
+ }
+
+ void execute (String[] args) {
+ if (args.length == 0) {
+ this.printUsage();
+ System.exit(-1);
+ }
+ List<String> commandArgs = new ArrayList<String>();
+ for (int i = 0; i < args.length; ++i) {
+ if (args[i] == "-c") {
+ this.configOptions.add(args[++i]);
+ } else if (args[i] == "--config") {
+ this.configFile = args[++i];
+ } else {
+ commandArgs.add(args[i]);
+ }
+ }
+ if ((commandArgs.size() == 0) || (!this.COMMANDS.contains
+ (commandArgs.get(0)))) {
+ System.out.println("Unknown command: [storm " + StringUtils.join
+ (args, " ") + "]");
+ this.printUsage();
+ System.exit(254);
+
+ }
+ this.callMethod(commandArgs.get(0), commandArgs.subList(1,
+ commandArgs.size()));
+
+ }
+
+ String getConfigOptions() {
+ String configOptions = "-Dstorm.options=";
+ //TODO - do urlencode here. python does quote_plus to each configoption
+ return configOptions + StringUtils.join(this.configOptions, ',');
+
+ }
+
+ List<String> getJarsFull (String directory) {
+ List<String> fullJarFiles = new ArrayList<String>();
+ File file = new File(directory);
+ File[] files = file.listFiles();
+ if (files != null) {
+ for (File f : files) {
+ if (f.getName().endsWith(".jar")) {
+ fullJarFiles.add(f.getPath());
+ }
+ }
+ }
+ return fullJarFiles;
+ }
+
+ String getClassPath (List<String> extraJars, boolean daemon) {
+ List<String> classPaths = this.getJarsFull(this.stormHomeDirectory);
+ classPaths.addAll(this.getJarsFull(this.stormLibDirectory));
+ classPaths.addAll(this.getJarsFull(this.stormHomeDirectory + this
+ .fileSeparator + "extlib"));
+ if (daemon == true) {
+ classPaths.addAll(this.getJarsFull(this.stormHomeDirectory + this
+ .fileSeparator + "extlib-daemon"));
+ }
+ if (this.stormExternalClasspath != null) {
+ classPaths.add(this.stormExternalClasspath);
+ }
+ if (this.stormExternalClasspathDaemon != null) {
+ classPaths.add(this.stormExternalClasspathDaemon);
+ }
+ classPaths.addAll(extraJars);
+ return StringUtils.join(classPaths, System.getProperty("path" +
+ ".separator"));
+ }
+
+ String confValue (String name, List<String> extraPaths, boolean daemon) {
+ // The original code from python started a process that started a jvm
+ // with backtype.storm.command.config_value main method that would
+ // read the conf value and print it out to an output stream. python
+ // tapped on to the output stream of that subprocess and returned the
+ // confvalue for the name. Because the pythong code has been shipped
+ // to java now it should not spawn a new process which is a jvm since
+ // we are already in jvm. Instead it should just be doing as the code
+ // commeneted below.
+ // However looking at the pythong code it was
+ // starting a jvm with -cp argument that had classpaths which might
+ // not be available to this java process. Hence there is a chance
+ // that the below code might break existing scripts. As a result I
+ // have decided to still spawn a new process from java just like
+ // python with similar classpaths being constructed for the jvm
+ // execution
+ /*IFn fn = Utils.loadClojureFn("backtype.storm.config",
+ "read-storm-config");
+ Object o = fn.invoke();
+ return ((Map) o).get(name).toString();*/
+ String confValue = "";
+ ProcessBuilder processBuilder = new ProcessBuilder(this.javaCommand,
+ "-client", this.getConfigOptions(), "-Dstorm.conf.file=" +
+ this.configFile, "-cp", this.getClassPath(extraPaths, daemon),
+ "backtype.storm.command.config_value", name);
+ BufferedReader br;
+ try {
+ Process process = processBuilder.start();
+ br = new BufferedReader(new InputStreamReader(process
+ .getInputStream(), StandardCharsets.UTF_8));
+ process.waitFor();
+ String line;
+ while ((line = br.readLine()) != null) {
+ String[] tokens = line.split(" ");
+ if ("VALUE:".equals(tokens[0])) {
+ confValue = StringUtils.join(Arrays.copyOfRange(tokens, 1,
+ tokens.length), " ");
+ break;
+ }
+ }
+ br.close();
+ } catch (Exception ex) {
+ System.out.println("Exception occured while starting process via " +
+ "processbuilder " + ex.getMessage());
+ }
+ return confValue;
+ }
+
+ void executeStormClass (String className, String jvmType, List<String>
+ jvmOptions, List<String> extraJars, List<String> args, boolean
+ fork, boolean daemon, String daemonName) {
+ List<String> extraPaths = new ArrayList<>();
+ extraPaths.add(this.clusterConfDirectory);
+ String stormLogDirectory = this.confValue("storm.log.dir",
+ extraPaths, daemon);
+ if ((stormLogDirectory == null) || ("".equals(stormLogDirectory)) ||
+ ("nil".equals(stormLogDirectory))) {
+ stormLogDirectory = this.stormHomeDirectory + this.fileSeparator
+ + "logs";
+ }
+ List<String> commandList = new ArrayList<String>();
+ commandList.add(this.javaCommand);
+ commandList.add(jvmType);
+ commandList.add("-Ddaemon.name=" + daemonName);
+ commandList.add(this.getConfigOptions());
+ commandList.add("-Dstorm.home=" + this.stormHomeDirectory);
+ commandList.add("-Dstorm.log.dir=" + stormLogDirectory);
+ commandList.add("-Djava.library.path=" + this
+ .confValue("java.library.path", extraJars, daemon));
+ commandList.add("-Dstorm.conf.file=" + this.configFile);
+ commandList.add("-cp");
+ commandList.add(this.getClassPath(extraJars, daemon));
+ commandList.addAll(jvmOptions);
+ commandList.add(className);
+ commandList.addAll(args);
+ ProcessBuilder processBuilder = new ProcessBuilder(commandList);
+ processBuilder.inheritIO();
+ try {
+ Process process = processBuilder.start();
+ System.out.println("Executing the command: ");
+ String commandLine = StringUtils.join(commandList, " ");
+ System.out.println(commandLine);
+ if (daemon == true) {
+ Runtime.getRuntime().addShutdownHook(new ShutdownHookThread
+ (process, commandLine));
+ }
+ System.out.println("Waiting for subprocess to finish");
+ process.waitFor();
+ System.out.println("subprocess finished");
+ System.out.println("Exit value from subprocess is :" + process
+ .exitValue());
+ } catch (Exception ex) {
+ System.out.println("Exception occured while starting process via " +
+ "processbuilder " + ex.getMessage());
+ }
+ }
+
+ void jarCommand (List<String> args) {
+ System.out.println("Called jarCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ if ((args == null) || (args.size() < 2)) {
+ System.out.println("Not enough arguments for storm jar command");
+ System.out.println("Please pass a jar file location and the " +
+ "topology class for jar command");
+ //TODO print usage for jar command here
+ System.exit(-1);
+ }
+ String jarJvmOptions = System.getenv("STORM_JAR_JVM_OPTS");
+ List<String> jvmOptions = new ArrayList<String>();
+ if (jarJvmOptions != null) {
+ //TODO the python code to parse STORM_JAR_JVM_OPTIONS uses shlex
+ // .split to get the different jvm options for the jar. For now
+ // keeping it simple and splitting on space. Need to be in synch
+ // with python. Not sure though if we really need to use a
+ // lexical parser
+ jvmOptions.addAll(Arrays.asList(jarJvmOptions.split(" ")));
+ }
+ jvmOptions.add("-Dstorm.jar=" + args.get(0));
+ List<String> extraPaths = new ArrayList<String>();
+ extraPaths.add(args.get(0));
+ extraPaths.add(this.userConfDirectory);
+ extraPaths.add(this.stormBinDirectory);
+ this.executeStormClass(args.get(1), "-client", jvmOptions,
+ extraPaths, args.subList(2, args.size()), false, false, "");
+ return;
+ }
+
+ void killCommand (List<String> args) {
+ System.out.println("Called killCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ if ((args == null) || (args.size() < 1)) {
+ System.out.println("Not enough arguments for storm kill command");
+ //TODO print usage for kill command here
+ System.exit(2);
+ }
+ List<String> extraPaths = new ArrayList<String>();
+ extraPaths.add(this.userConfDirectory);
+ extraPaths.add(this.stormBinDirectory);
+ this.executeStormClass("backtype.storm.command.kill_topology",
+ "-client", new ArrayList<String>(), extraPaths, args, false,
+ false, "");
+ return;
+ }
+
+ //TODO implement shell command after understanding more about it
+ void shellCommand (List<String> args) {
+ System.out.println("Called shellCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ return;
+ }
+
+ void nimbusCommand (List<String> args) {
+ System.out.println("Called nimbusCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ List<String> jvmOptions = new ArrayList<String>();
+ List<String> extraPaths = new ArrayList<String>();
+ extraPaths.add(this.clusterConfDirectory);
+ String nimbusOptions = this.confValue("nimbus.childopts", extraPaths,
+ true);
+ // below line is different from original python script storm.py where
+ // it called parse_args method on nimbusOptions. Now we just call a
+ // split with a space. Hence this will have different behavior and
+ // a buggy one if the nimbusOptions string in the config file has a
+ // space. TODO need to fix this
+ jvmOptions.addAll(Arrays.asList(nimbusOptions.split(" ")));
+ jvmOptions.add("-Dlogfile.name=nimbus.log");
+ jvmOptions.add("-Dlog4j.configurationFile=" + this
+ .getLog4jConfigDirectory() + this.fileSeparator + "cluster" +
+ ".xml");
+ this.executeStormClass(this.NIMBUS_CLASS, "-server", jvmOptions,
+ extraPaths, new ArrayList<String>(), false, true, "nimbus");
+ return;
+ }
+
+ void uiCommand (List<String> args) {
+ System.out.println("Called uiCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ List<String> jvmOptions = new ArrayList<String>();
+ List<String> extraPaths = new ArrayList<String>();
+ extraPaths.add(this.clusterConfDirectory);
+ String uiOptions = this.confValue("ui.childopts", extraPaths,
+ true);
+ // below line is different from original python script storm.py where
+ // it called parse_args method on nimbusOptions. Now we just call a
+ // split with a space. Hence this will have different behavior and
+ // a buggy one if the nimbusOptions string in the config file has a
+ // space. TODO need to fix this
+ jvmOptions.addAll(Arrays.asList(uiOptions.split(" ")));
+ jvmOptions.add("-Dlogfile.name=ui.log");
+ jvmOptions.add("-Dlog4j.configurationFile=" + this
+ .getLog4jConfigDirectory() + this.fileSeparator + "cluster" +
+ ".xml");
+ extraPaths.add(0, this.stormHomeDirectory);
+ this.executeStormClass(this.UI_CLASS, "-server", jvmOptions,
+ extraPaths, new ArrayList<String>(), false, true, "ui");
+ return;
+ }
+
+ void logviewerCommand (List<String> args) {
+ System.out.println("Called logviewerCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ List<String> jvmOptions = new ArrayList<String>();
+ List<String> extraPaths = new ArrayList<String>();
+ extraPaths.add(this.clusterConfDirectory);
+ String logviewerOptions = this.confValue("logviewer.childopts",
+ extraPaths, true);
+ // below line is different from original python script storm.py where
+ // it called parse_args method on nimbusOptions. Now we just call a
+ // split with a space. Hence this will have different behavior and
+ // a buggy one if the nimbusOptions string in the config file has a
+ // space. TODO need to fix this
+ jvmOptions.addAll(Arrays.asList(logviewerOptions.split(" ")));
+ jvmOptions.add("-Dlogfile.name=logviewer.log");
+ jvmOptions.add("-Dlog4j.configurationFile=" + this
+ .getLog4jConfigDirectory() + this.fileSeparator + "cluster" +
+ ".xml");
+ extraPaths.add(0, this.stormHomeDirectory);
+ this.executeStormClass(this.LOGVIEWER_CLASS, "-server", jvmOptions,
+ extraPaths, new ArrayList<String>(), false, true, "logviewer");
+ return;
+ }
+
+ void drpcCommand (List<String> args) {
+ System.out.println("Called drpcCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ List<String> jvmOptions = new ArrayList<String>();
+ List<String> extraPaths = new ArrayList<String>();
+ extraPaths.add(this.clusterConfDirectory);
+ String drpcOptions = this.confValue("drpc.childopts", extraPaths,
+ true);
+ // below line is different from original python script storm.py where
+ // it called parse_args method on nimbusOptions. Now we just call a
+ // split with a space. Hence this will have different behavior and
+ // a buggy one if the nimbusOptions string in the config file has a
+ // space. TODO need to fix this
+ jvmOptions.addAll(Arrays.asList(drpcOptions.split(" ")));
+ jvmOptions.add("-Dlogfile.name=drpc.log");
+ jvmOptions.add("-Dlog4j.configurationFile=" + this
+ .getLog4jConfigDirectory() + this.fileSeparator + "cluster" +
+ ".xml");
+ this.executeStormClass(this.DRPC_CLASS, "-server", jvmOptions,
+ extraPaths, new ArrayList<String>(), false, true, "drpc");
+ return;
+ }
+
+ void supervisorCommand (List<String> args) {
+ System.out.println("Called supervisorCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ List<String> jvmOptions = new ArrayList<String>();
+ List<String> extraPaths = new ArrayList<String>();
+ extraPaths.add(this.clusterConfDirectory);
+ String supervisorOptions = this.confValue("supervisor.childopts",
+ extraPaths,
+ true);
+ // below line is different from original python script storm.py where
+ // it called parse_args method on nimbusOptions. Now we just call a
+ // split with a space. Hence this will have different behavior and
+ // a buggy one if the nimbusOptions string in the config file has a
+ // space. TODO need to fix this
+ jvmOptions.addAll(Arrays.asList(supervisorOptions.split(" ")));
+ jvmOptions.add("-Dlogfile.name=supervisor.log");
+ jvmOptions.add("-Dlog4j.configurationFile=" + this
+ .getLog4jConfigDirectory() + this.fileSeparator + "cluster" +
+ ".xml");
+ this.executeStormClass(this.SUPERVISOR_CLASS, "-server", jvmOptions,
+ extraPaths, new ArrayList<String>(), false, true, "supervisor");
+ return;
+ }
+
+ void localconfvalueCommand (List<String> args) {
+ System.out.println("Called localconfvalueCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ if ((args == null) || (args.size() == 0)) {
+ System.out.println("Not enough arguments for localconfvalue " +
+ "command");
+ System.out.println("Please pass the name of the config value you " +
+ "want to be printed");
+ //TODO print command help for localconfvalue command
+ System.exit(2);
+ }
+ List<String> extraPaths = new ArrayList<String>();
+ extraPaths.add(this.userConfDirectory);
+ System.out.println(args.get(0) + ": " + this.confValue(args.get(0),
+ extraPaths,
+ true));
+ return;
+ }
+
+ void remoteconfvalueCommand (List<String> args) {
+ System.out.println("Called remoteconfvalueCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ if ((args == null) || (args.size() == 0)) {
+ System.out.println("Not enough arguments for remoteconfvalue " +
+ "command");
+ System.out.println("Please pass the name of the config value you " +
+ "want to be printed");
+ //TODO print command help for remoteconfvalue command
+ System.exit(2);
+ }
+ List<String> extraPaths = new ArrayList<String>();
+ extraPaths.add(this.clusterConfDirectory);
+ System.out.println(args.get(0) + ": " + this.confValue(args.get(0),
+ extraPaths,
+ true));
+ return;
+ }
+
+ void replCommand (List<String> args) {
+ System.out.println("Called replCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ List<String> extraPaths = new ArrayList<String>();
+ extraPaths.add(this.clusterConfDirectory);
+ this.executeStormClass(this.REPL_CLASS, "-client", new
+ ArrayList<String>(), extraPaths, new ArrayList<String>(),
+ false, true, "");
+ return;
+ }
+
+ void classpathCommand (List<String> args) {
+ System.out.println("Called classpathCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ System.out.println(this.getClassPath(new ArrayList<String>(), true));
+ return;
+ }
+
+ void activateCommand (List<String> args) {
+ System.out.println("Called activateCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ if ((args == null) || (args.size() < 1)) {
+ System.out.println("Not enough arguments for activate command");
+ System.out.println("Please pass the topology name to activate");
+ //TODO print usage for activate command here
+ System.exit(2);
+ }
+ List<String> extraPaths = new ArrayList<String>();
+ extraPaths.add(this.userConfDirectory);
+ extraPaths.add(this.stormBinDirectory);
+ this.executeStormClass(this.ACTIVATE_CLASS, "-client", new
+ ArrayList<String>(), extraPaths, args, false, false, "");
+ return;
+ }
+
+ void deactivateCommand (List<String> args) {
+ System.out.println("Called deactivateCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ if ((args == null) || (args.size() < 1)) {
+ System.out.println("Not enough arguments for deactivate command");
+ System.out.println("Please pass the topology name to deactivate");
+ //TODO print usage for deactivate command here
+ System.exit(2);
+ }
+ List<String> extraPaths = new ArrayList<String>();
+ extraPaths.add(this.userConfDirectory);
+ extraPaths.add(this.stormBinDirectory);
+ this.executeStormClass(this.DEACTIVATE_CLASS, "-client", new
+ ArrayList<String>(), extraPaths, args, false, false, "");
+ return;
+ }
+
+ void rebalanceCommand (List<String> args) {
+ System.out.println("Called rebalanceCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ if ((args == null) || (args.size() < 1)) {
+ System.out.println("Not enough arguments for rebalance command");
+ System.out.println("Please pass the topology name to rebalance");
+ //TODO print usage for rebalance command here
+ System.exit(2);
+ }
+ List<String> extraPaths = new ArrayList<String>();
+ extraPaths.add(this.userConfDirectory);
+ extraPaths.add(this.stormBinDirectory);
+ this.executeStormClass(this.REBALANCE_CLASS, "-client", new
+ ArrayList<String>(), extraPaths, args, false, false, "");
+ return;
+ }
+
+ void helpCommand (List<String> args) {
+ System.out.println("Called helpCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ if ((args == null) || (args.size() == 0)) {
+ this.printUsage();
+ } else {
+ if ((!this.COMMANDS.contains(args.get(0)))) {
+ System.out.println(args.get(0) + " is not a valid command");
+ } else {
+ //TODO print indivudual commands help here
+ System.out.println("Print command specific help here");
+ }
+ }
+ return;
+ }
+
+ void listCommand (List<String> args) {
+ System.out.println("Called listCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ List<String> extraPaths = new ArrayList<String>();
+ extraPaths.add(this.userConfDirectory);
+ extraPaths.add(this.stormBinDirectory);
+ this.executeStormClass(this.LIST_CLASS, "-client", new
+ ArrayList<String>(), extraPaths, args, false, false, "");
+ return;
+ }
+
+ void devzookeeperCommand (List<String> args) {
+ System.out.println("Called devzookeeperCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ return;
+ }
+
+ void versionCommand (List<String> args) {
+ System.out.println("Called versionCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ return;
+ }
+
+ void monitorCommand (List<String> args) {
+ System.out.println("Called monitorCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ return;
+ }
+
+ void uploadcredentialsCommand (List<String> args) {
+ System.out.println("Called uploadcredentialsCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ return;
+ }
+
+ void geterrorsCommand (List<String> args) {
+ System.out.println("Called geterrorsCommand using reflection");
+ System.out.println("Arguments are : ");
+ for (String s: args) {
+ System.out.println(s);
+ }
+ return;
+ }
+
+ String getLog4jConfigDirectory () {
+ List<String> extraPaths = new ArrayList<String>();
+ extraPaths.add(this.clusterConfDirectory);
+ String log4jDirectory = this.confValue("storm.logback.conf.dir",
+ extraPaths, true);
+ if ((log4jDirectory == null) || ("".equals(log4jDirectory)) ||
+ ("nil".equals(log4jDirectory))) {
+ log4jDirectory = this.stormLog4jConfDirectory;
+ }
+ return log4jDirectory;
+ }
+
+ private void printUsage () {
+ String commands = StringUtils.join(this.COMMANDS, "\n\t");
+ System.out.println("Commands:\n\t" + commands);
+ System.out.println("\nHelp: \n\thelp \n\thelp <command>\n");
+ System.out.println("Documentation for the storm client can be found" +
+ " at " +
+ "http://storm.incubator.apache" +
+ ".org/documentation/Command-line-client.html\n");
+ System.out.println("Configs can be overridden using one or more -c " +
+ "flags, e.g. " +
+ "\"storm list -c nimbus.host=nimbus.mycompany.com\"\n");
+ }
+
+ private void executeHelpCommand () {
+ System.out.println("Print storm help here");
+ }
+
+}
+
+class WindowsStormCommandExecutor extends StormCommandExecutor {
+
+ WindowsStormCommandExecutor () {
+
+ }
+
+ void initialize () {
+ return;
+ }
+
+ void execute (String[] args) {
--- End diff --
@caofangkun thanks for checking this out. The code is still not complete. I need to put the windows implementation as well. Most of the code in the UnixStormCommandExecutor will move to abstract base class so that WindowsStormCommandLineExecutor can reuse. It will only have code that is specific to windows.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-904: Move bin/storm command line to java...
Posted by knusbaum <gi...@git.apache.org>.
Github user knusbaum commented on the pull request:
https://github.com/apache/storm/pull/662#issuecomment-130832806
I have a little confusion about why we're moving the python to java instead of just having the storm.cmd run the existing storm.py.
I'm all in favor of de-duplicating the storm.py and storm.cmd commands, but rather than reimplementing storm.py, I'd rather reuse it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-904: Move bin/storm command line to java...
Posted by priyank5485 <gi...@git.apache.org>.
Github user priyank5485 commented on the pull request:
https://github.com/apache/storm/pull/662#issuecomment-126779134
This PR aims at factoring out the logic for storm command line client which currently resides in two places. One in storm bash script and storm.py for Unix and storm.cmd for Windows. Idea is to create a java program that will be called from storm.cmd and storm bash script which will have all the logic from storm.py and storm.cmd in one place. This will help us maintain the storm client better since we will have only one file to change and in most cases only the base class that both Windows and Unix classes inherit from. I have already removed storm.py. I am working on finishing up the java file and also removing the storm.cmd file. Just putting it out there to elicit response to this approach from the community.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-904: Move bin/storm command line to java...
Posted by hustfxj <gi...@git.apache.org>.
Github user hustfxj commented on a diff in the pull request:
https://github.com/apache/storm/pull/662#discussion_r45188898
--- Diff: storm-core/src/jvm/backtype/storm/utils/StormCommandExecutor.java ---
@@ -0,0 +1,868 @@
+package backtype.storm.utils;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.nio.charset.StandardCharsets;
+import java.util.*;
+
+import clojure.lang.IFn;
+import org.apache.commons.lang.StringUtils;
+import org.apache.commons.lang.SystemUtils;
+
+/**
+ * Created by pshah on 7/17/15.
+ */
+abstract class StormCommandExecutor {
+ final String NIMBUS_CLASS = "backtype.storm.daemon.nimbus";
+ final String SUPERVISOR_CLASS = "backtype.storm.daemon.supervisor";
+ final String UI_CLASS = "backtype.storm.ui.core";
+ final String LOGVIEWER_CLASS = "backtype.storm.daemon.logviewer";
+ final String DRPC_CLASS = "backtype.storm.daemon.drpc";
+ final String REPL_CLASS = "clojure.main";
+ final String ACTIVATE_CLASS = "backtype.storm.command.activate";
+ final String DEACTIVATE_CLASS = "backtype.storm.command.deactivate";
+ final String REBALANCE_CLASS = "backtype.storm.command.rebalance";
+ final String LIST_CLASS = "backtype.storm.command.list";
+ final String DEVZOOKEEPER_CLASS = "backtype.storm.command.dev_zookeeper";
+ final String VERSION_CLASS = "backtype.storm.utils.VersionInfo";
+ final String MONITOR_CLASS = "backtype.storm.command.monitor";
+ final String UPLOADCREDENTIALS_CLASS = "backtype.storm.command" +
+ ".upload_credentials";
+ final String GETERRORS_CLASS = "backtype.storm.command.get_errors";
+ final String SHELL_CLASS = "backtype.storm.command.shell_submission";
+ String stormHomeDirectory;
+ String userConfDirectory;
+ String stormConfDirectory;
+ String clusterConfDirectory;
+ String stormLibDirectory;
+ String stormBinDirectory;
+ String stormLog4jConfDirectory;
+ String configFile = "";
+ String javaCommand;
+ List<String> configOptions = new ArrayList<String>();
+ String stormExternalClasspath;
+ String stormExternalClasspathDaemon;
+ String fileSeparator;
+ final List<String> COMMANDS = Arrays.asList("jar", "kill", "shell",
+ "nimbus", "ui", "logviewer", "drpc", "supervisor",
+ "localconfvalue", "remoteconfvalue", "repl", "classpath",
+ "activate", "deactivate", "rebalance", "help", "list",
+ "dev-zookeeper", "version", "monitor", "upload-credentials",
+ "get-errors");
+
+ public static void main (String[] args) {
+ for (String arg : args) {
+ System.out.println("Argument ++ is " + arg);
+ }
+ StormCommandExecutor stormCommandExecutor;
+ if (System.getProperty("os.name").startsWith("Windows")) {
+ stormCommandExecutor = new WindowsStormCommandExecutor();
+ } else {
+ stormCommandExecutor = new UnixStormCommandExecutor();
+ }
+ stormCommandExecutor.initialize();
+ stormCommandExecutor.execute(args);
+ }
+
+ StormCommandExecutor () {
+
+ }
+
+ abstract void initialize ();
+
+ abstract void execute (String[] args);
+
+ void callMethod (String command, List<String> args) {
+ Class implementation = this.getClass();
+ String methodName = command.replace("-", "") + "Command";
+ try {
+ Method method = implementation.getDeclaredMethod(methodName, List
+ .class);
+ method.invoke(this, args);
+ } catch (NoSuchMethodException ex) {
+ System.out.println("No such method exception occured while trying" +
+ " to run storm method " + command);
+ } catch (IllegalAccessException ex) {
+ System.out.println("Illegal access exception occured while trying" +
+ " to run storm method " + command);
+ } catch (IllegalArgumentException ex) {
+ System.out.println("Illegal argument exception occured while " +
+ "trying" + " to run storm method " + command);
+ } catch (InvocationTargetException ex) {
+ System.out.println("Invocation target exception occured while " +
+ "trying" + " to run storm method " + command);
+ }
+ }
+}
+
+class UnixStormCommandExecutor extends StormCommandExecutor {
+
+ UnixStormCommandExecutor () {
+
+ }
+
+ void initialize () {
+ Collections.sort(this.COMMANDS);
+ this.fileSeparator = System .getProperty ("file.separator");
+ this.stormHomeDirectory = System.getenv("STORM_BASE_DIR");
+ this.userConfDirectory = System.getProperty("user.home") +
+ this.fileSeparator + "" +
+ ".storm";
+ this.stormConfDirectory = System.getenv("STORM_CONF_DIR");
+ this.clusterConfDirectory = this.stormConfDirectory == null ? (this
+ .stormHomeDirectory + this.fileSeparator + "conf") : this
+ .stormConfDirectory;
+ File f = new File(this.userConfDirectory + this.fileSeparator +
+ "storm.yaml");
+ if (!f.isFile()) {
+ this.userConfDirectory = this.clusterConfDirectory;
+ }
+ this.stormLibDirectory = this.stormHomeDirectory + this.fileSeparator +
+ "lib";
+ this.stormBinDirectory = this.stormHomeDirectory + this.fileSeparator +
+ "bin";
+ this.stormLog4jConfDirectory = this.stormHomeDirectory +
+ this.fileSeparator + "log4j2";
+ if (System.getenv("JAVA_HOME") != null) {
+ this.javaCommand = System.getenv("JAVA_HOME") + this.fileSeparator +
+ "bin" + this.fileSeparator + "java";
+ if (!(new File(this.javaCommand).exists())) {
+ System.out.println("ERROR: JAVA_HOME is invalid. Could not " +
+ "find " + this.javaCommand);
+ System.exit(1);
+ }
+ } else {
+ this.javaCommand = "java";
+ }
+ this.stormExternalClasspath = System.getenv("STORM_EXT_CLASSPATH");
+ this.stormExternalClasspathDaemon = System.getenv
+ ("STORM_EXT_CLASSPATH_DAEMON");
+ if (!(new File(this.stormLibDirectory).exists())) {
+ System.out.println("******************************************");
+ System.out.println("The storm client can only be run from within " +
+ "a release. " + "You appear to be trying to run the client" +
+ " from a checkout of Storm's source code.");
+ System.out.println("You can download a Storm release at " +
+ "http://storm-project.net/downloads.html");
+ System.out.println("******************************************");
+ System.exit(1);
+ }
+ //System.getProperties().list(System.out);
+ }
+
+ void execute (String[] args) {
+ if (args.length == 0) {
+ this.printUsage();
+ System.exit(-1);
+ }
+ List<String> commandArgs = new ArrayList<String>();
+ for (int i = 0; i < args.length; ++i) {
+ if (args[i] == "-c") {
+ this.configOptions.add(args[++i]);
+ } else if (args[i] == "--config") {
+ this.configFile = args[++i];
+ } else {
+ commandArgs.add(args[i]);
+ }
+ }
+ if ((commandArgs.size() == 0) || (!this.COMMANDS.contains
+ (commandArgs.get(0)))) {
+ System.out.println("Unknown command: [storm " + StringUtils.join
+ (args, " ") + "]");
+ this.printUsage();
+ System.exit(254);
+
+ }
+ this.callMethod(commandArgs.get(0), commandArgs.subList(1,
+ commandArgs.size()));
+
+ }
+
+ String getConfigOptions() {
+ String configOptions = "-Dstorm.options=";
+ //TODO - do urlencode here. python does quote_plus to each configoption
+ return configOptions + StringUtils.join(this.configOptions, ',');
+
+ }
+
+ List<String> getJarsFull (String directory) {
+ List<String> fullJarFiles = new ArrayList<String>();
+ File file = new File(directory);
+ File[] files = file.listFiles();
+ if (files != null) {
+ for (File f : files) {
+ if (f.getName().endsWith(".jar")) {
+ fullJarFiles.add(f.getPath());
+ }
+ }
+ }
+ return fullJarFiles;
+ }
+
+ String getClassPath (List<String> extraJars, boolean daemon) {
+ List<String> classPaths = this.getJarsFull(this.stormHomeDirectory);
+ classPaths.addAll(this.getJarsFull(this.stormLibDirectory));
+ classPaths.addAll(this.getJarsFull(this.stormHomeDirectory + this
+ .fileSeparator + "extlib"));
+ if (daemon == true) {
+ classPaths.addAll(this.getJarsFull(this.stormHomeDirectory + this
+ .fileSeparator + "extlib-daemon"));
+ }
+ if (this.stormExternalClasspath != null) {
+ classPaths.add(this.stormExternalClasspath);
+ }
+ if (this.stormExternalClasspathDaemon != null) {
+ classPaths.add(this.stormExternalClasspathDaemon);
+ }
+ classPaths.addAll(extraJars);
+ return StringUtils.join(classPaths, System.getProperty("path" +
+ ".separator"));
+ }
+
+ String confValue (String name, List<String> extraPaths, boolean daemon) {
+ // The original code from python started a process that started a jvm
+ // with backtype.storm.command.config_value main method that would
+ // read the conf value and print it out to an output stream. python
+ // tapped on to the output stream of that subprocess and returned the
+ // confvalue for the name. Because the pythong code has been shipped
+ // to java now it should not spawn a new process which is a jvm since
+ // we are already in jvm. Instead it should just be doing as the code
+ // commeneted below.
+ // However looking at the pythong code it was
+ // starting a jvm with -cp argument that had classpaths which might
+ // not be available to this java process. Hence there is a chance
+ // that the below code might break existing scripts. As a result I
+ // have decided to still spawn a new process from java just like
+ // python with similar classpaths being constructed for the jvm
+ // execution
+ /*IFn fn = Utils.loadClojureFn("backtype.storm.config",
+ "read-storm-config");
+ Object o = fn.invoke();
+ return ((Map) o).get(name).toString();*/
+ String confValue = "";
+ ProcessBuilder processBuilder = new ProcessBuilder(this.javaCommand,
+ "-client", this.getConfigOptions(), "-Dstorm.conf.file=" +
+ this.configFile, "-cp", this.getClassPath(extraPaths, daemon),
+ "backtype.storm.command.config_value", name);
+ BufferedReader br;
+ try {
+ Process process = processBuilder.start();
+ br = new BufferedReader(new InputStreamReader(process
+ .getInputStream(), StandardCharsets.UTF_8));
+ process.waitFor();
+ String line;
+ while ((line = br.readLine()) != null) {
+ String[] tokens = line.split(" ");
+ if ("VALUE:".equals(tokens[0])) {
+ confValue = StringUtils.join(Arrays.copyOfRange(tokens, 1,
+ tokens.length), " ");
+ break;
+ }
+ }
+ br.close();
+ } catch (Exception ex) {
+ System.out.println("Exception occured while starting process via " +
+ "processbuilder " + ex.getMessage());
+ }
+ return confValue;
+ }
+
+ void executeStormClass (String className, String jvmType, List<String>
+ jvmOptions, List<String> extraJars, List<String> args, boolean
+ fork, boolean daemon, String daemonName) {
+ List<String> extraPaths = new ArrayList<>();
+ extraPaths.add(this.clusterConfDirectory);
+ String stormLogDirectory = this.confValue("storm.log.dir",
+ extraPaths, daemon);
+ if ((stormLogDirectory == null) || ("".equals(stormLogDirectory)) ||
+ ("nil".equals(stormLogDirectory))) {
+ stormLogDirectory = this.stormHomeDirectory + this.fileSeparator
+ + "logs";
+ }
+ List<String> commandList = new ArrayList<String>();
+ commandList.add(this.javaCommand);
+ commandList.add(jvmType);
+ commandList.add("-Ddaemon.name=" + daemonName);
+ commandList.add(this.getConfigOptions());
+ commandList.add("-Dstorm.home=" + this.stormHomeDirectory);
+ commandList.add("-Dstorm.log.dir=" + stormLogDirectory);
+ commandList.add("-Djava.library.path=" + this
+ .confValue("java.library.path", extraJars, daemon));
+ commandList.add("-Dstorm.conf.file=" + this.configFile);
+ commandList.add("-cp");
+ commandList.add(this.getClassPath(extraJars, daemon));
+ commandList.addAll(jvmOptions);
+ commandList.add(className);
+ commandList.addAll(args);
+ ProcessBuilder processBuilder = new ProcessBuilder(commandList);
+ processBuilder.inheritIO();
+ try {
+ Process process = processBuilder.start();
+ System.out.println("Executing the command: ");
+ String commandLine = StringUtils.join(commandList, " ");
+ System.out.println(commandLine);
+ if (daemon == true) {
+ Runtime.getRuntime().addShutdownHook(new ShutdownHookThread
+ (process, commandLine));
+ }
+ System.out.println("Waiting for subprocess to finish");
+ process.waitFor();
+ System.out.println("subprocess finished");
+ System.out.println("Exit value from subprocess is :" + process
+ .exitValue());
+ } catch (Exception ex) {
+ System.out.println("Exception occured while starting process via " +
+ "processbuilder " + ex.getMessage());
+ }
--- End diff --
not all process need waitfor(). maybe you can add option. Like this:
public static java.lang.Process launch_process(final String command, final Map<String, String> environment, boolean backend) throws IOException {
if (backend == true) {
new Thread(new Runnable() {
@Override
public void run() {
String[] cmdlist = (new String("nohup " + command + " &")).split(" ");
try {
launchProcess(cmdlist, environment);
} catch (IOException e) {
LOG.error("Failed to run " + command + ":" + e.getCause(), e);
}
}
}).start();
return null;
} else {
String[] cmdlist = command.split(" ");
return launchProcess(cmdlist, environment);
}
}
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-904: Move bin/storm command line to java...
Posted by priyank5485 <gi...@git.apache.org>.
Github user priyank5485 commented on the pull request:
https://github.com/apache/storm/pull/662#issuecomment-157804563
@hustfxj I might have missed something and its been a long time since i did this. I could not find any storm command that would not need a waitFor. Can you give an example? If we do need that flag it should be easy to add and I can change that.
@longdafeng I am not sure I understand the point about debugging. You can still put a break point in this java version of storm command line. I agree with you that its easier to write new functions in script. But the whole point of this JIRA is to have one client that works for Unix based systems and Windows platform. The last time I was working on it we had storm.py for Unix and storm batch client for windows. Both of them also had a separate file for environment variables. It is hard to maintain two scripts anytime we want to change the storm command. We can ask others about their opinion and what they think is better between the two approaches.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-904: Move bin/storm command line to java...
Posted by priyank5485 <gi...@git.apache.org>.
Github user priyank5485 commented on the pull request:
https://github.com/apache/storm/pull/662#issuecomment-157808005
@longdafeng Also please see one of the previous comments about why not to use python.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #662: STORM-904: Move bin/storm command line to java.
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/storm/pull/662
---
[GitHub] storm pull request: STORM-904: Move bin/storm command line to java...
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on the pull request:
https://github.com/apache/storm/pull/662#issuecomment-130838110
@knusbaum it will make lot easier for commands to be in java and we can have just wrappers for windows & nix* os. We don't need to have python dependency and in our experience of running python to execute java command to start a jvm takes lot of time and in larger clusters Ambari times out these commands.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-904: Move bin/storm command line to java...
Posted by ptgoetz <gi...@git.apache.org>.
Github user ptgoetz commented on the pull request:
https://github.com/apache/storm/pull/662#issuecomment-157395059
Any update on this PR? It would seem it at least needs an upmerge and additional review.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---