Multiple_stdout_stderr_handlers_including_console

distro/pom.xml

@@ -75,6 +75,7 @@ atlas.graph.index.search.solr.wait-searcher=true
         <solr.embedded>false</solr.embedded>
         <cassandra.embedded>false</cassandra.embedded>
         <elasticsearch.managed>false</elasticsearch.managed>
+        <log.console>false</log.console>
 
         <entity.repository.properties>atlas.EntityAuditRepository.impl=org.apache.atlas.repository.audit.HBaseBasedAuditRepository</entity.repository.properties>
     </properties>

distro/src/bin/atlas_config.py

@@ -26,6 +26,7 @@ import errno
 import socket
 from re import split
 from time import sleep
+from distutils.util import strtobool
 
 BIN = "bin"
 LIB = "lib"
@@ -56,10 +57,11 @@ SOLR_SHARDS = "SOLR_SHARDS"
 DEFAULT_SOLR_SHARDS = "1"
 SOLR_REPLICATION_FACTOR = "SOLR_REPLICATION_FACTOR"
 DEFAULT_SOLR_REPLICATION_FACTOR = "1"
+ENABLE_LOGGING_TO_CONSOLE = "ENABLE_LOGGING_TO_CONSOLE"
 
 ENV_KEYS = ["JAVA_HOME", ATLAS_OPTS, ATLAS_SERVER_OPTS, ATLAS_SERVER_HEAP, ATLAS_LOG, ATLAS_PID, ATLAS_CONF,
             "ATLASCPPATH", ATLAS_DATA, ATLAS_HOME, ATLAS_WEBAPP, HBASE_CONF_DIR, SOLR_PORT, MANAGE_LOCAL_HBASE,
-            MANAGE_LOCAL_SOLR, MANAGE_EMBEDDED_CASSANDRA, MANAGE_LOCAL_ELASTICSEARCH]
+            MANAGE_LOCAL_SOLR, MANAGE_EMBEDDED_CASSANDRA, MANAGE_LOCAL_ELASTICSEARCH, ENABLE_LOGGING_TO_CONSOLE]
 IS_WINDOWS = platform.system() == "Windows"
 ON_POSIX = 'posix' in sys.builtin_module_names
 CONF_FILE="atlas-application.properties"
@@ -179,7 +181,7 @@ def executeEnvSh(confDir):
 
         proc.communicate()
 
-def java(classname, args, classpath, jvm_opts_list, logdir=None):
+def java(classname, args, classpath, jvm_opts_list, logdir=None, logconsole="false"):
     java_home = os.environ.get("JAVA_HOME", None)
     if java_home:
         prg = os.path.join(java_home, "bin", "java")
@@ -195,9 +197,10 @@ def java(classname, args, classpath, jvm_opts_list, logdir=None):
     commandline.append(classpath)
     commandline.append(classname)
     commandline.extend(args)
-    return runProcess(commandline, logdir)
+    return runProcess(commandline, logdir,
+                      logconsole=logconsole)
 
-def jar(path):
+def jar(path, logconsole="false"):
     java_home = os.environ.get("JAVA_HOME", None)
     if java_home:
         prg = os.path.join(java_home, "bin", "jar")
@@ -210,7 +213,8 @@ def jar(path):
     commandline = [prg]
     commandline.append("-xf")
     commandline.append(path)
-    process = runProcess(commandline)
+    process = runProcess(commandline,
+                         logconsole=logconsole)
     process.wait()
 
 def is_exe(fpath):
@@ -231,17 +235,44 @@ def which(program):
 
     return None
 
-def runProcess(commandline, logdir=None, shell=False, wait=False):
+def runProcess(commandline, logdir=None, shell=False, wait=False, logconsole="false"):
     """
     Run a process
     :param commandline: command line
+    :param: logdir: directory where logs from stdout and stderr should be saved
+    :logconsole: whether to print outout of stdout and stderr to console
     :return:the return code
     """
     global finished
     debug ("Executing : %s" % str(commandline))
     timestr = time.strftime("atlas.%Y%m%d-%H%M%S")
+
+    logconsole = bool(strtobool(logconsole))
+
+    if logconsole:
+        stdoutTargetsList = [sys.stdout]
+        stderrTargetsList = [sys.stderr]
+
+        if logdir:
+            stdoutFile = open(os.path.join(logdir, timestr + ".out"), "w")
+            stderrFile = open(os.path.join(logdir,timestr + ".err"), "w")
+
+            stdoutTargetsList.append(stdoutFile)
+            stderrTargetsList.append(stderrFile)
+
+        p = subprocess.Popen(commandline, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=shell)
+
+        for line in p.stdout:
+            for target in stdoutTargetsList:
+                target.write(line)
+
+        for line in p.stderr:
+            for target in stderrTargetsList:
+                target.write(line)
+    else:
         stdoutFile = None
         stderrFile = None
+
         if logdir:
             stdoutFile = open(os.path.join(logdir, timestr + ".out"), "w")
             stderrFile = open(os.path.join(logdir,timestr + ".err"), "w")
@@ -418,7 +449,7 @@ def is_hbase_local(confdir):
     confFile = os.path.join(confdir, CONF_FILE)
     return is_hbase(confdir) and grep(confFile, HBASE_STORAGE_LOCAL_CONF_ENTRY) is not None
 
-def run_hbase_action(dir, action, hbase_conf_dir = None, logdir = None, wait=True):
+def run_hbase_action(dir, action, hbase_conf_dir = None, logdir = None, wait=True, logconsole="false"):
     if IS_WINDOWS:
         if action == 'start':
             hbaseScript = 'start-hbase.cmd'
@@ -436,7 +467,8 @@ def run_hbase_action(dir, action, hbase_conf_dir = None, logdir = None, wait=Tru
             cmd = [os.path.join(dir, hbaseScript), action, 'master']
 
 
-    return runProcess(cmd, logdir, False, wait)
+    return runProcess(cmd, logdir, False, wait,
+                      logconsole=logconsole)
 
 def is_solr(confdir):
     confdir = os.path.join(confdir, CONF_FILE)
@@ -523,7 +555,7 @@ def wait_for_startup(confdir, wait):
 
     sys.stdout.write('\n')
 
-def run_zookeeper(dir, action, logdir = None, wait=True):
+def run_zookeeper(dir, action, logdir = None, wait=True, logconsole="false"):
     zookeeperScript = "zkServer.sh"
 
     if IS_WINDOWS:
@@ -531,9 +563,10 @@ def run_zookeeper(dir, action, logdir = None, wait=True):
 
     cmd = [os.path.join(dir, zookeeperScript), action, os.path.join(dir, '../../conf/zookeeper/zoo.cfg')]
 
-    return runProcess(cmd, logdir, False, wait)
+    return runProcess(cmd, logdir, False, wait,
+                      logconsole=logconsole)
 
-def start_elasticsearch(dir, logdir = None, wait=True):
+def start_elasticsearch(dir, logdir = None, wait=True, logconsole="false"):
 
     elasticsearchScript = "elasticsearch"
 
@@ -542,11 +575,12 @@ def start_elasticsearch(dir, logdir = None, wait=True):
 
     cmd = [os.path.join(dir, elasticsearchScript), '-d', '-p', os.path.join(logdir, 'elasticsearch.pid')]
 
-    processVal = runProcess(cmd, logdir, False, wait)
+    processVal = runProcess(cmd, logdir, False, wait,
+                            logconsole=logconsole)
     sleep(6)
     return processVal
 
-def run_solr(dir, action, zk_url = None, port = None, logdir = None, wait=True):
+def run_solr(dir, action, zk_url = None, port = None, logdir = None, wait=True, logconsole="false"):
 
     solrScript = "solr"
 
@@ -564,9 +598,10 @@ def run_solr(dir, action, zk_url = None, port = None, logdir = None, wait=True):
         else:
             cmd = [os.path.join(dir, solrScript), action, '-z', zk_url, '-p', port]
 
-    return runProcess(cmd, logdir, False, wait)
+    return runProcess(cmd, logdir, False, wait,
+                      logconsole=logconsole)
 
-def create_solr_collection(dir, confdir, index, logdir = None, wait=True):
+def create_solr_collection(dir, confdir, index, logdir = None, wait=True, logconsole="false"):
     solrScript = "solr"
 
     if IS_WINDOWS:
@@ -574,7 +609,8 @@ def create_solr_collection(dir, confdir, index, logdir = None, wait=True):
 
     cmd = [os.path.join(dir, solrScript), 'create', '-c', index, '-d', confdir,  '-shards',  solrShards(),  '-replicationFactor', solrReplicationFactor()]
 
-    return runProcess(cmd, logdir, False, wait)
+    return runProcess(cmd, logdir, False, wait,
+                      logconsole=logconsole)
 
 def configure_hbase(dir):
     env_conf_dir = os.environ.get(HBASE_CONF_DIR)

distro/src/bin/atlas_start.py

@@ -21,6 +21,8 @@ import traceback
 
 import atlas_config as mc
 
+from distutils.util import strtobool
+
 ATLAS_LOG_OPTS="-Datlas.log.dir=%s -Datlas.log.file=%s.log"
 ATLAS_COMMAND_OPTS="-Datlas.home=%s"
 ATLAS_CONFIG_OPTS="-Datlas.conf=%s"
@@ -35,6 +37,9 @@ def main():
     confdir = mc.dirMustExist(mc.confDir(atlas_home))
     mc.executeEnvSh(confdir)
     logdir = mc.dirMustExist(mc.logDir(atlas_home))
+    console_logging = "enabled" if bool(strtobool(os.environ.get(mc.ENABLE_LOGGING_TO_CONSOLE)))  else "disabled"
+    print "Logging to console is %s." % console_logging
+
     mc.dirMustExist(mc.dataDir(atlas_home))
     if mc.isCygwin():
         # Pathnames that are passed to JVM must be converted to Windows format.
@@ -112,7 +117,9 @@ def main():
     if is_hbase and mc.is_hbase_local(confdir):
         print "configured for local hbase."
         mc.configure_hbase(atlas_home)
-        mc.run_hbase_action(mc.hbaseBinDir(atlas_home), "start", hbase_conf_dir, logdir)
+
+        mc.run_hbase_action(mc.hbaseBinDir(atlas_home), "start", hbase_conf_dir, logdir,
+                            logconsole=os.environ.get(mc.ENABLE_LOGGING_TO_CONSOLE))
         print "hbase started."
 
     #solr setup
@@ -123,39 +130,54 @@ def main():
             print "Cassandra embedded configured."
             mc.configure_cassandra(atlas_home)
             mc.configure_zookeeper(atlas_home)
-            mc.run_zookeeper(mc.zookeeperBinDir(atlas_home), "start", logdir)
+
+            mc.run_zookeeper(mc.zookeeperBinDir(atlas_home), "start", logdir,
+                             logconsole=os.environ.get(mc.ENABLE_LOGGING_TO_CONSOLE))
             print "zookeeper started."
 
-        mc.run_solr(mc.solrBinDir(atlas_home), "start", mc.get_solr_zk_url(confdir), mc.solrPort(), logdir)
+        mc.run_solr(mc.solrBinDir(atlas_home), "start", mc.get_solr_zk_url(confdir), mc.solrPort(), logdir,
+                    logconsole=os.environ.get(mc.ENABLE_LOGGING_TO_CONSOLE))
+
         print "solr started."
 
         print "setting up solr collections..."
-        mc.create_solr_collection(mc.solrBinDir(atlas_home), mc.solrConfDir(atlas_home), "vertex_index", logdir)
-        mc.create_solr_collection(mc.solrBinDir(atlas_home), mc.solrConfDir(atlas_home), "edge_index", logdir)
-        mc.create_solr_collection(mc.solrBinDir(atlas_home), mc.solrConfDir(atlas_home), "fulltext_index", logdir)
+        mc.create_solr_collection(mc.solrBinDir(atlas_home), mc.solrConfDir(atlas_home), "vertex_index", logdir,
+                                  logconsole=os.environ.get(mc.ENABLE_LOGGING_TO_CONSOLE))
+        mc.create_solr_collection(mc.solrBinDir(atlas_home), mc.solrConfDir(atlas_home), "edge_index", logdir,
+                                  logconsole=os.environ.get(mc.ENABLE_LOGGING_TO_CONSOLE))
+        mc.create_solr_collection(mc.solrBinDir(atlas_home), mc.solrConfDir(atlas_home), "fulltext_index", logdir,
+                                  logconsole=os.environ.get(mc.ENABLE_LOGGING_TO_CONSOLE))
 
     #elasticsearch setup
     if mc.is_elasticsearch_local():
         print "configured for local elasticsearch."
-        mc.start_elasticsearch(mc.elasticsearchBinDir(atlas_home), logdir)
+
+        mc.start_elasticsearch(mc.elasticsearchBinDir(atlas_home), logdir,
+                               logconsole=os.environ.get(mc.ENABLE_LOGGING_TO_CONSOLE))
         print "elasticsearch started."
 
     web_app_path = os.path.join(web_app_dir, "atlas")
     if (mc.isCygwin()):
         web_app_path = mc.convertCygwinPath(web_app_path)
     if not is_setup:
-        start_atlas_server(atlas_classpath, atlas_pid_file, jvm_logdir, jvm_opts_list, web_app_path)
+
+        start_atlas_server(atlas_classpath, atlas_pid_file, jvm_logdir, jvm_opts_list, web_app_path,
+                           logconsole=os.environ.get(mc.ENABLE_LOGGING_TO_CONSOLE))
         mc.wait_for_startup(confdir, 300)
         print "Apache Atlas Server started!!!\n"
     else:
-        process = mc.java("org.apache.atlas.web.setup.AtlasSetup", [], atlas_classpath, jvm_opts_list, jvm_logdir)
+        process = mc.java("org.apache.atlas.web.setup.AtlasSetup", [], atlas_classpath, jvm_opts_list, jvm_logdir,
+                          logconsole=os.environ.get(mc.ENABLE_LOGGING_TO_CONSOLE))
+
         return process.wait()
 
 
-def start_atlas_server(atlas_classpath, atlas_pid_file, jvm_logdir, jvm_opts_list, web_app_path):
+def start_atlas_server(atlas_classpath, atlas_pid_file, jvm_logdir, jvm_opts_list, web_app_path, logconsole="false"):
     args = ["-app", web_app_path]
     args.extend(sys.argv[1:])
-    process = mc.java("org.apache.atlas.Atlas", args, atlas_classpath, jvm_opts_list, jvm_logdir)
+    process = mc.java("org.apache.atlas.Atlas", args, atlas_classpath, jvm_opts_list, jvm_logdir,
+                      logconsole=logconsole)
+
     mc.writePid(atlas_pid_file, process)
 
 if __name__ == '__main__':

distro/src/bin/cputil.py

@@ -45,7 +45,9 @@ def main():
                        + os.path.join(web_app_dir, "atlas", "WEB-INF", "lib", "*" )  + p \
                        + os.path.join(atlas_home, "libext", "*")
 
-    process = mc.java("org.apache.atlas.util.CredentialProviderUtility", sys.argv[1:], atlas_classpath, jvm_opts_list)
+    process = mc.java("org.apache.atlas.util.CredentialProviderUtility", sys.argv[1:], atlas_classpath, jvm_opts_list,
+                      logconsole=os.environ.get(mc.ENABLE_LOGGING_TO_CONSOLE))
+
     process.wait()
 
 if __name__ == '__main__':

distro/src/conf/atlas-env.sh

@@ -66,3 +66,6 @@ export MANAGE_EMBEDDED_CASSANDRA=${cassandra.embedded}
 
 # indicates whether or not a local instance of Elasticsearch should be started for Atlas
 export MANAGE_LOCAL_ELASTICSEARCH=${elasticsearch.managed}
+
+# Indicates whether or not allow printing logs to stdout. To be used only with processes not requiring user input.
+export ENABLE_LOGGING_TO_CONSOLE=${log.console}