Repository: incubator-nifi Updated Branches: refs/heads/develop 315af02c5 -> 69f04cbb8
NIFI-488: Merged develop Project: http://git-wip-us.apache.org/repos/asf/incubator-nifi/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-nifi/commit/69f04cbb Tree: http://git-wip-us.apache.org/repos/asf/incubator-nifi/tree/69f04cbb Diff: http://git-wip-us.apache.org/repos/asf/incubator-nifi/diff/69f04cbb Branch: refs/heads/develop Commit: 69f04cbb8668b9c1500e23c2dbe41ebf97644bd0 Parents: 315af02 Author: Mark Payne <[email protected]> Authored: Mon Jun 8 14:13:24 2015 -0400 Committer: Mark Payne <[email protected]> Committed: Wed Jun 17 08:19:36 2015 -0400 ---------------------------------------------------------------------- .../src/main/assembly/dependencies.xml | 2 + nifi/nifi-bootstrap/pom.xml | 7 + .../org/apache/nifi/bootstrap/NiFiListener.java | 13 +- .../java/org/apache/nifi/bootstrap/RunNiFi.java | 395 +++++++++++-------- .../org/apache/nifi/bootstrap/ShutdownHook.java | 6 +- .../src/main/resources/bin/dump-nifi.bat | 4 +- .../src/main/resources/bin/nifi.sh | 21 +- .../src/main/resources/bin/run-nifi.bat | 4 +- .../src/main/resources/bin/start-nifi.bat | 47 --- .../src/main/resources/bin/stop-nifi.bat | 47 --- .../src/main/resources/conf/logback.xml | 48 +++ 11 files changed, 320 insertions(+), 274 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/69f04cbb/nifi/nifi-assembly/src/main/assembly/dependencies.xml ---------------------------------------------------------------------- diff --git a/nifi/nifi-assembly/src/main/assembly/dependencies.xml b/nifi/nifi-assembly/src/main/assembly/dependencies.xml index 27eb32d..a3e3a18 100644 --- a/nifi/nifi-assembly/src/main/assembly/dependencies.xml +++ b/nifi/nifi-assembly/src/main/assembly/dependencies.xml @@ -49,6 +49,8 @@ <useTransitiveFiltering>true</useTransitiveFiltering> <includes> <include>nifi-bootstrap</include> + <include>slf4j-api</include> + <include>logback-classic</include> </includes> </dependencySet> http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/69f04cbb/nifi/nifi-bootstrap/pom.xml ---------------------------------------------------------------------- diff --git a/nifi/nifi-bootstrap/pom.xml b/nifi/nifi-bootstrap/pom.xml index bdf5089..ff27fd3 100644 --- a/nifi/nifi-bootstrap/pom.xml +++ b/nifi/nifi-bootstrap/pom.xml @@ -21,4 +21,11 @@ </parent> <artifactId>nifi-bootstrap</artifactId> <packaging>jar</packaging> + + <dependencies> + <dependency> + <groupId>org.slf4j</groupId> + <artifactId>slf4j-api</artifactId> + </dependency> + </dependencies> </project> http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/69f04cbb/nifi/nifi-bootstrap/src/main/java/org/apache/nifi/bootstrap/NiFiListener.java ---------------------------------------------------------------------- diff --git a/nifi/nifi-bootstrap/src/main/java/org/apache/nifi/bootstrap/NiFiListener.java b/nifi/nifi-bootstrap/src/main/java/org/apache/nifi/bootstrap/NiFiListener.java index 1713478..8d74f16 100644 --- a/nifi/nifi-bootstrap/src/main/java/org/apache/nifi/bootstrap/NiFiListener.java +++ b/nifi/nifi-bootstrap/src/main/java/org/apache/nifi/bootstrap/NiFiListener.java @@ -23,6 +23,7 @@ import java.net.ServerSocket; import java.net.Socket; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; import org.apache.nifi.bootstrap.util.LimitingInputStream; @@ -40,6 +41,7 @@ public class NiFiListener { listener = new Listener(serverSocket, runner); final Thread listenThread = new Thread(listener); listenThread.setName("Listen to NiFi"); + listenThread.setDaemon(true); listenThread.start(); return localPort; } @@ -62,7 +64,16 @@ public class NiFiListener { public Listener(final ServerSocket serverSocket, final RunNiFi runner) { this.serverSocket = serverSocket; - this.executor = Executors.newFixedThreadPool(2); + this.executor = Executors.newFixedThreadPool(2, new ThreadFactory() { + @Override + public Thread newThread(final Runnable runnable) { + final Thread t = Executors.defaultThreadFactory().newThread(runnable); + t.setDaemon(true); + t.setName("NiFi Bootstrap Command Listener"); + return t; + } + }); + this.runner = runner; } http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/69f04cbb/nifi/nifi-bootstrap/src/main/java/org/apache/nifi/bootstrap/RunNiFi.java ---------------------------------------------------------------------- diff --git a/nifi/nifi-bootstrap/src/main/java/org/apache/nifi/bootstrap/RunNiFi.java b/nifi/nifi-bootstrap/src/main/java/org/apache/nifi/bootstrap/RunNiFi.java index e6b1bc5..2bc44cc 100644 --- a/nifi/nifi-bootstrap/src/main/java/org/apache/nifi/bootstrap/RunNiFi.java +++ b/nifi/nifi-bootstrap/src/main/java/org/apache/nifi/bootstrap/RunNiFi.java @@ -41,13 +41,17 @@ import java.util.List; import java.util.Map; import java.util.Properties; import java.util.Set; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; -import java.util.logging.ConsoleHandler; -import java.util.logging.Handler; -import java.util.logging.Level; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * @@ -94,19 +98,27 @@ public class RunNiFi { private final File bootstrapConfigFile; - private final java.util.logging.Logger logger; + // used for logging initial info; these will be logged to console by default when the app is started + private final Logger cmdLogger = LoggerFactory.getLogger("org.apache.nifi.bootstrap.Command"); + // used for logging all info. These by default will be written to the log file + private final Logger defaultLogger = LoggerFactory.getLogger(RunNiFi.class); + + + private final ExecutorService loggingExecutor; + private volatile Set<Future<?>> loggingFutures = new HashSet<>(2); public RunNiFi(final File bootstrapConfigFile, final boolean verbose) { this.bootstrapConfigFile = bootstrapConfigFile; - logger = java.util.logging.Logger.getLogger("Bootstrap"); - if (verbose) { - logger.info("Enabling Verbose Output"); - logger.setLevel(Level.FINE); - final Handler handler = new ConsoleHandler(); - handler.setLevel(Level.FINE); - logger.addHandler(handler); - } + loggingExecutor = Executors.newFixedThreadPool(2, new ThreadFactory() { + @Override + public Thread newThread(final Runnable runnable) { + final Thread t = Executors.defaultThreadFactory().newThread(runnable); + t.setDaemon(true); + t.setName("NiFi logging handler"); + return t; + } + }); } private static void printUsage() { @@ -185,10 +197,10 @@ public class RunNiFi { switch (cmd.toLowerCase()) { case "start": - runNiFi.start(false); + runNiFi.start(); break; case "run": - runNiFi.start(true); + runNiFi.start(); break; case "stop": runNiFi.stop(); @@ -198,7 +210,7 @@ public class RunNiFi { break; case "restart": runNiFi.stop(); - runNiFi.start(false); + runNiFi.start(); break; case "dump": runNiFi.dump(dumpFile); @@ -206,40 +218,44 @@ public class RunNiFi { } } - public File getStatusFile() { + File getStatusFile() { + return getStatusFile(defaultLogger); + } + + public File getStatusFile(final Logger logger) { final File confDir = bootstrapConfigFile.getParentFile(); final File nifiHome = confDir.getParentFile(); final File bin = new File(nifiHome, "bin"); final File statusFile = new File(bin, "nifi.pid"); - logger.log(Level.FINE, "Status File: {0}", statusFile); + logger.debug("Status File: {}", statusFile); return statusFile; } - private Properties loadProperties() throws IOException { + private Properties loadProperties(final Logger logger) throws IOException { final Properties props = new Properties(); - final File statusFile = getStatusFile(); + final File statusFile = getStatusFile(logger); if (statusFile == null || !statusFile.exists()) { - logger.fine("No status file to load properties from"); + logger.debug("No status file to load properties from"); return props; } - try (final FileInputStream fis = new FileInputStream(getStatusFile())) { + try (final FileInputStream fis = new FileInputStream(getStatusFile(logger))) { props.load(fis); } final Map<Object, Object> modified = new HashMap<>(props); modified.remove("secret.key"); - logger.log(Level.FINE, "Properties: {0}", modified); + logger.debug("Properties: {}", modified); return props; } - private synchronized void saveProperties(final Properties nifiProps) throws IOException { - final File statusFile = getStatusFile(); + private synchronized void saveProperties(final Properties nifiProps, final Logger logger) throws IOException { + final File statusFile = getStatusFile(logger); if (statusFile.exists() && !statusFile.delete()) { - logger.log(Level.WARNING, "Failed to delete {0}", statusFile); + logger.warn("Failed to delete {}", statusFile); } if (!statusFile.createNewFile()) { @@ -252,7 +268,7 @@ public class RunNiFi { perms.add(PosixFilePermission.OWNER_WRITE); Files.setPosixFilePermissions(statusFile.toPath(), perms); } catch (final Exception e) { - logger.log(Level.WARNING, "Failed to set permissions so that only the owner can read status file {0}; " + logger.warn("Failed to set permissions so that only the owner can read status file {}; " + "this may allows others to have access to the key needed to communicate with NiFi. " + "Permissions should be changed so that only the owner can read this file", statusFile); } @@ -262,23 +278,23 @@ public class RunNiFi { fos.getFD().sync(); } - logger.log(Level.FINE, "Saved Properties {0} to {1}", new Object[]{nifiProps, statusFile}); + logger.debug("Saved Properties {} to {}", new Object[]{nifiProps, statusFile}); } - private boolean isPingSuccessful(final int port, final String secretKey) { - logger.log(Level.FINE, "Pinging {0}", port); + private boolean isPingSuccessful(final int port, final String secretKey, final Logger logger) { + logger.debug("Pinging {}", port); try (final Socket socket = new Socket("localhost", port)) { final OutputStream out = socket.getOutputStream(); out.write((PING_CMD + " " + secretKey + "\n").getBytes(StandardCharsets.UTF_8)); out.flush(); - logger.fine("Sent PING command"); + logger.debug("Sent PING command"); socket.setSoTimeout(5000); final InputStream in = socket.getInputStream(); final BufferedReader reader = new BufferedReader(new InputStreamReader(in)); final String response = reader.readLine(); - logger.log(Level.FINE, "PING response: {0}", response); + logger.debug("PING response: {}", response); out.close(); reader.close(); @@ -288,27 +304,27 @@ public class RunNiFi { } } - private Integer getCurrentPort() throws IOException { - final Properties props = loadProperties(); + private Integer getCurrentPort(final Logger logger) throws IOException { + final Properties props = loadProperties(logger); final String portVal = props.getProperty("port"); if (portVal == null) { - logger.fine("No Port found in status file"); + logger.debug("No Port found in status file"); return null; } else { - logger.log(Level.FINE, "Port defined in status file: {0}", portVal); + logger.debug("Port defined in status file: {}", portVal); } final int port = Integer.parseInt(portVal); - final boolean success = isPingSuccessful(port, props.getProperty("secret.key")); + final boolean success = isPingSuccessful(port, props.getProperty("secret.key"), logger); if (success) { - logger.log(Level.FINE, "Successful PING on port {0}", port); + logger.debug("Successful PING on port {}", port); return port; } final String pid = props.getProperty("pid"); - logger.log(Level.FINE, "PID in status file is {0}", pid); + logger.debug("PID in status file is {}", pid); if (pid != null) { - final boolean procRunning = isProcessRunning(pid); + final boolean procRunning = isProcessRunning(pid, logger); if (procRunning) { return port; } else { @@ -319,7 +335,7 @@ public class RunNiFi { return null; } - private boolean isProcessRunning(final String pid) { + private boolean isProcessRunning(final String pid, final Logger logger) { try { // We use the "ps" command to check if the process is still running. final ProcessBuilder builder = new ProcessBuilder(); @@ -343,9 +359,9 @@ public class RunNiFi { // If output of the ps command had our PID, the process is running. if (running) { - logger.log(Level.FINE, "Process with PID {0} is running", pid); + logger.debug("Process with PID {} is running", pid); } else { - logger.log(Level.FINE, "Process with PID {0} is not running", pid); + logger.debug("Process with PID {} is not running", pid); } return running; @@ -355,10 +371,10 @@ public class RunNiFi { } } - private Status getStatus() { + private Status getStatus(final Logger logger) { final Properties props; try { - props = loadProperties(); + props = loadProperties(logger); } catch (final IOException ioe) { return new Status(null, null, false, false); } @@ -380,7 +396,7 @@ public class RunNiFi { if (portValue != null) { try { port = Integer.parseInt(portValue); - pingSuccess = isPingSuccessful(port, secretKey); + pingSuccess = isPingSuccessful(port, secretKey, logger); } catch (final NumberFormatException nfe) { return new Status(null, null, false, false); } @@ -390,20 +406,21 @@ public class RunNiFi { return new Status(port, pid, true, true); } - final boolean alive = (pid == null) ? false : isProcessRunning(pid); + final boolean alive = (pid == null) ? false : isProcessRunning(pid, logger); return new Status(port, pid, pingSuccess, alive); } public void status() throws IOException { - final Status status = getStatus(); + final Logger logger = cmdLogger; + final Status status = getStatus(logger); if (status.isRespondingToPing()) { - logger.log(Level.INFO, "Apache NiFi is currently running, listening to Bootstrap on port {0}, PID={1}", + logger.info("Apache NiFi is currently running, listening to Bootstrap on port {}, PID={}", new Object[]{status.getPort(), status.getPid() == null ? "unknkown" : status.getPid()}); return; } if (status.isProcessRunning()) { - logger.log(Level.INFO, "Apache NiFi is running at PID {0} but is not responding to ping requests", status.getPid()); + logger.info("Apache NiFi is running at PID {} but is not responding to ping requests", status.getPid()); return; } @@ -427,36 +444,36 @@ public class RunNiFi { * @throws IOException if any issues occur while writing the dump file */ public void dump(final File dumpFile) throws IOException { - final Integer port = getCurrentPort(); + final Logger logger = defaultLogger; // dump to bootstrap log file by default + final Integer port = getCurrentPort(logger); if (port == null) { - System.out.println("Apache NiFi is not currently running"); + logger.info("Apache NiFi is not currently running"); return; } - final Properties nifiProps = loadProperties(); + final Properties nifiProps = loadProperties(logger); final String secretKey = nifiProps.getProperty("secret.key"); final StringBuilder sb = new StringBuilder(); try (final Socket socket = new Socket()) { - logger.fine("Connecting to NiFi instance"); + logger.debug("Connecting to NiFi instance"); socket.setSoTimeout(60000); socket.connect(new InetSocketAddress("localhost", port)); - logger.fine("Established connection to NiFi instance."); + logger.debug("Established connection to NiFi instance."); socket.setSoTimeout(60000); - logger.log(Level.FINE, "Sending DUMP Command to port {0}", port); + logger.debug("Sending DUMP Command to port {}", port); final OutputStream out = socket.getOutputStream(); out.write((DUMP_CMD + " " + secretKey + "\n").getBytes(StandardCharsets.UTF_8)); out.flush(); - out.close(); final InputStream in = socket.getInputStream(); - final BufferedReader reader = new BufferedReader(new InputStreamReader(in)); - String line; - while ((line = reader.readLine()) != null) { - sb.append(line).append("\n"); + try (final BufferedReader reader = new BufferedReader(new InputStreamReader(in))) { + String line; + while ((line = reader.readLine()) != null) { + sb.append(line).append("\n"); + } } - reader.close(); } final String dump = sb.toString(); @@ -466,28 +483,35 @@ public class RunNiFi { try (final FileOutputStream fos = new FileOutputStream(dumpFile)) { fos.write(dump.getBytes(StandardCharsets.UTF_8)); } - logger.log(Level.INFO, "Successfully wrote thread dump to {0}", dumpFile.getAbsolutePath()); + // we want to log to the console (by default) that we wrote the thread dump to the specified file + cmdLogger.info("Successfully wrote thread dump to {}", dumpFile.getAbsolutePath()); } } public void stop() throws IOException { - final Integer port = getCurrentPort(); + final Logger logger = cmdLogger; + final Integer port = getCurrentPort(logger); if (port == null) { - System.out.println("Apache NiFi is not currently running"); + logger.info("Apache NiFi is not currently running"); return; } - final Properties nifiProps = loadProperties(); + final Properties nifiProps = loadProperties(logger); final String secretKey = nifiProps.getProperty("secret.key"); + final File statusFile = getStatusFile(logger); + if (statusFile.exists() && !statusFile.delete()) { + logger.error("Failed to delete status file {}; this file should be cleaned up manually", statusFile); + } + try (final Socket socket = new Socket()) { - logger.fine("Connecting to NiFi instance"); + logger.debug("Connecting to NiFi instance"); socket.setSoTimeout(60000); socket.connect(new InetSocketAddress("localhost", port)); - logger.fine("Established connection to NiFi instance."); + logger.debug("Established connection to NiFi instance."); socket.setSoTimeout(60000); - logger.log(Level.FINE, "Sending SHUTDOWN Command to port {0}", port); + logger.debug("Sending SHUTDOWN Command to port {}", port); final OutputStream out = socket.getOutputStream(); out.write((SHUTDOWN_CMD + " " + secretKey + "\n").getBytes(StandardCharsets.UTF_8)); out.flush(); @@ -501,7 +525,7 @@ public class RunNiFi { } final String response = sb.toString().trim(); - logger.log(Level.FINE, "Received response to SHUTDOWN command: {0}", response); + logger.debug("Received response to SHUTDOWN command: {}", response); if (SHUTDOWN_CMD.equals(response)) { logger.info("Apache NiFi has accepted the Shutdown Command and is shutting down now"); @@ -522,17 +546,17 @@ public class RunNiFi { } final long startWait = System.nanoTime(); - while (isProcessRunning(pid)) { + while (isProcessRunning(pid, logger)) { logger.info("Waiting for Apache NiFi to finish shutting down..."); final long waitNanos = System.nanoTime() - startWait; final long waitSeconds = TimeUnit.NANOSECONDS.toSeconds(waitNanos); if (waitSeconds >= gracefulShutdownSeconds && gracefulShutdownSeconds > 0) { - if (isProcessRunning(pid)) { - logger.log(Level.WARNING, "NiFi has not finished shutting down after {0} seconds. Killing process.", gracefulShutdownSeconds); + if (isProcessRunning(pid, logger)) { + logger.warn("NiFi has not finished shutting down after {} seconds. Killing process.", gracefulShutdownSeconds); try { - killProcessTree(pid); + killProcessTree(pid, logger); } catch (final IOException ioe) { - logger.log(Level.SEVERE, "Failed to kill Process with PID {0}", pid); + logger.error("Failed to kill Process with PID {}", pid); } } break; @@ -546,16 +570,11 @@ public class RunNiFi { logger.info("NiFi has finished shutting down."); } - - final File statusFile = getStatusFile(); - if (!statusFile.delete()) { - logger.log(Level.SEVERE, "Failed to delete status file {0}; this file should be cleaned up manually", statusFile); - } } else { - logger.log(Level.SEVERE, "When sending SHUTDOWN command to NiFi, got unexpected response {0}", response); + logger.error("When sending SHUTDOWN command to NiFi, got unexpected response {}", response); } } catch (final IOException ioe) { - logger.log(Level.SEVERE, "Failed to send shutdown command to port {0} due to {1}", new Object[]{port, ioe}); + logger.error("Failed to send shutdown command to port {} due to {}", new Object[]{port, ioe.toString(), ioe}); } } @@ -574,14 +593,14 @@ public class RunNiFi { return childPids; } - private void killProcessTree(final String pid) throws IOException { - logger.log(Level.FINE, "Killing Process Tree for PID {0}", pid); + private void killProcessTree(final String pid, final Logger logger) throws IOException { + logger.debug("Killing Process Tree for PID {}", pid); final List<String> children = getChildProcesses(pid); - logger.log(Level.FINE, "Children of PID {0}: {1}", new Object[]{pid, children}); + logger.debug("Children of PID {}: {}", new Object[]{pid, children}); for (final String childPid : children) { - killProcessTree(childPid); + killProcessTree(childPid, logger); } Runtime.getRuntime().exec(new String[]{"kill", "-9", pid}); @@ -597,10 +616,10 @@ public class RunNiFi { } @SuppressWarnings({"rawtypes", "unchecked"}) - public void start(final boolean monitor) throws IOException, InterruptedException { - final Integer port = getCurrentPort(); + public void start() throws IOException, InterruptedException { + final Integer port = getCurrentPort(cmdLogger); if (port != null) { - System.out.println("Apache NiFi is already running, listening to Bootstrap on port " + port); + cmdLogger.info("Apache NiFi is already running, listening to Bootstrap on port " + port); return; } @@ -739,122 +758,154 @@ public class RunNiFi { cmdBuilder.append(s).append(" "); } - logger.info("Starting Apache NiFi..."); - logger.log(Level.INFO, "Working Directory: {0}", workingDir.getAbsolutePath()); - logger.log(Level.INFO, "Command: {0}", cmdBuilder.toString()); + cmdLogger.info("Starting Apache NiFi..."); + cmdLogger.info("Working Directory: {}", workingDir.getAbsolutePath()); + cmdLogger.info("Command: {}", cmdBuilder.toString()); - if (monitor) { - String gracefulShutdown = props.get(GRACEFUL_SHUTDOWN_PROP); - if (gracefulShutdown == null) { - gracefulShutdown = DEFAULT_GRACEFUL_SHUTDOWN_VALUE; - } + String gracefulShutdown = props.get(GRACEFUL_SHUTDOWN_PROP); + if (gracefulShutdown == null) { + gracefulShutdown = DEFAULT_GRACEFUL_SHUTDOWN_VALUE; + } - final int gracefulShutdownSeconds; - try { - gracefulShutdownSeconds = Integer.parseInt(gracefulShutdown); - } catch (final NumberFormatException nfe) { - throw new NumberFormatException("The '" + GRACEFUL_SHUTDOWN_PROP + "' property in Bootstrap Config File " - + bootstrapConfigAbsoluteFile.getAbsolutePath() + " has an invalid value. Must be a non-negative integer"); - } + final int gracefulShutdownSeconds; + try { + gracefulShutdownSeconds = Integer.parseInt(gracefulShutdown); + } catch (final NumberFormatException nfe) { + throw new NumberFormatException("The '" + GRACEFUL_SHUTDOWN_PROP + "' property in Bootstrap Config File " + + bootstrapConfigAbsoluteFile.getAbsolutePath() + " has an invalid value. Must be a non-negative integer"); + } - if (gracefulShutdownSeconds < 0) { - throw new NumberFormatException("The '" + GRACEFUL_SHUTDOWN_PROP + "' property in Bootstrap Config File " - + bootstrapConfigAbsoluteFile.getAbsolutePath() + " has an invalid value. Must be a non-negative integer"); - } + if (gracefulShutdownSeconds < 0) { + throw new NumberFormatException("The '" + GRACEFUL_SHUTDOWN_PROP + "' property in Bootstrap Config File " + + bootstrapConfigAbsoluteFile.getAbsolutePath() + " has an invalid value. Must be a non-negative integer"); + } - Process process = builder.start(); - Long pid = getPid(process); - if (pid != null) { - nifiPid = pid; - final Properties nifiProps = new Properties(); - nifiProps.setProperty("pid", String.valueOf(nifiPid)); - saveProperties(nifiProps); - } + Process process = builder.start(); + handleLogging(process); + Long pid = getPid(process, cmdLogger); + if (pid != null) { + nifiPid = pid; + final Properties nifiProps = new Properties(); + nifiProps.setProperty("pid", String.valueOf(nifiPid)); + saveProperties(nifiProps, cmdLogger); + } - shutdownHook = new ShutdownHook(process, this, secretKey, gracefulShutdownSeconds); - final Runtime runtime = Runtime.getRuntime(); - runtime.addShutdownHook(shutdownHook); + shutdownHook = new ShutdownHook(process, this, secretKey, gracefulShutdownSeconds, loggingExecutor); + final Runtime runtime = Runtime.getRuntime(); + runtime.addShutdownHook(shutdownHook); - while (true) { - final boolean alive = isAlive(process); + while (true) { + final boolean alive = isAlive(process); - if (alive) { - try { - Thread.sleep(1000L); - } catch (final InterruptedException ie) { - } - } else { - try { - runtime.removeShutdownHook(shutdownHook); - } catch (final IllegalStateException ise) { - // happens when already shutting down + if (alive) { + try { + Thread.sleep(1000L); + } catch (final InterruptedException ie) { + } + } else { + try { + runtime.removeShutdownHook(shutdownHook); + } catch (final IllegalStateException ise) { + // happens when already shutting down + } + + if (autoRestartNiFi) { + final File statusFile = getStatusFile(defaultLogger); + if (!statusFile.exists()) { + defaultLogger.debug("Status File no longer exists. Will not restart NiFi"); + return; } - if (autoRestartNiFi) { - logger.warning("Apache NiFi appears to have died. Restarting..."); - process = builder.start(); + defaultLogger.warn("Apache NiFi appears to have died. Restarting..."); + process = builder.start(); + handleLogging(process); - pid = getPid(process); - if (pid != null) { - nifiPid = pid; - final Properties nifiProps = new Properties(); - nifiProps.setProperty("pid", String.valueOf(nifiPid)); - saveProperties(nifiProps); - } + pid = getPid(process, defaultLogger); + if (pid != null) { + nifiPid = pid; + final Properties nifiProps = new Properties(); + nifiProps.setProperty("pid", String.valueOf(nifiPid)); + saveProperties(nifiProps, defaultLogger); + } - shutdownHook = new ShutdownHook(process, this, secretKey, gracefulShutdownSeconds); - runtime.addShutdownHook(shutdownHook); + shutdownHook = new ShutdownHook(process, this, secretKey, gracefulShutdownSeconds, loggingExecutor); + runtime.addShutdownHook(shutdownHook); - final boolean started = waitForStart(); + final boolean started = waitForStart(); - if (started) { - logger.log(Level.INFO, "Successfully started Apache NiFi{0}", (pid == null ? "" : " with PID " + pid)); - } else { - logger.severe("Apache NiFi does not appear to have started"); - } + if (started) { + defaultLogger.info("Successfully started Apache NiFi{}", (pid == null ? "" : " with PID " + pid)); } else { - return; + defaultLogger.error("Apache NiFi does not appear to have started"); } + } else { + return; } } - } else { - final Process process = builder.start(); - final Long pid = getPid(process); - - if (pid != null) { - nifiPid = pid; - final Properties nifiProps = new Properties(); - nifiProps.setProperty("pid", String.valueOf(nifiPid)); - saveProperties(nifiProps); + } + } + + private void handleLogging(final Process process) { + final Set<Future<?>> existingFutures = loggingFutures; + if (existingFutures != null) { + for (final Future<?> future : existingFutures) { + future.cancel(false); } + } - boolean started = waitForStart(); + final Future<?> stdOutFuture = loggingExecutor.submit(new Runnable() { + @Override + public void run() { + final Logger stdOutLogger = LoggerFactory.getLogger("org.apache.nifi.StdOut"); + final InputStream in = process.getInputStream(); + try (final BufferedReader reader = new BufferedReader(new InputStreamReader(in))) { + String line; + while ((line = reader.readLine()) != null) { + stdOutLogger.info(line); + } + } catch (IOException e) { + defaultLogger.error("Failed to read from NiFi's Standard Out stream", e); + } + } + }); - if (started) { - logger.log(Level.INFO, "Successfully started Apache NiFi{0}", (pid == null ? "" : " with PID " + pid)); - } else { - logger.severe("Apache NiFi does not appear to have started"); + final Future<?> stdErrFuture = loggingExecutor.submit(new Runnable() { + @Override + public void run() { + final Logger stdErrLogger = LoggerFactory.getLogger("org.apache.nifi.StdErr"); + final InputStream in = process.getErrorStream(); + try (final BufferedReader reader = new BufferedReader(new InputStreamReader(in))) { + String line; + while ((line = reader.readLine()) != null) { + stdErrLogger.error(line); + } + } catch (IOException e) { + defaultLogger.error("Failed to read from NiFi's Standard Error stream", e); + } } + }); - listener.stop(); - } + final Set<Future<?>> futures = new HashSet<>(); + futures.add(stdOutFuture); + futures.add(stdErrFuture); + this.loggingFutures = futures; } - private Long getPid(final Process process) { + private Long getPid(final Process process, final Logger logger) { try { final Class<?> procClass = process.getClass(); final Field pidField = procClass.getDeclaredField("pid"); pidField.setAccessible(true); final Object pidObject = pidField.get(process); - logger.log(Level.FINE, "PID Object = {0}", pidObject); + logger.debug("PID Object = {}", pidObject); if (pidObject instanceof Number) { return ((Number) pidObject).longValue(); } return null; } catch (final IllegalAccessException | NoSuchFieldException nsfe) { - logger.log(Level.FINE, "Could not find PID for child process due to {0}", nsfe); + logger.debug("Could not find PID for child process due to {}", nsfe); return null; } } @@ -913,7 +964,7 @@ public class RunNiFi { shutdownHook.setSecretKey(secretKey); } - final File statusFile = getStatusFile(); + final File statusFile = getStatusFile(defaultLogger); final Properties nifiProps = new Properties(); if (nifiPid != -1) { @@ -923,12 +974,12 @@ public class RunNiFi { nifiProps.setProperty("secret.key", secretKey); try { - saveProperties(nifiProps); + saveProperties(nifiProps, defaultLogger); } catch (final IOException ioe) { - logger.log(Level.WARNING, "Apache NiFi has started but failed to persist NiFi Port information to {0} due to {1}", new Object[]{statusFile.getAbsolutePath(), ioe}); + defaultLogger.warn("Apache NiFi has started but failed to persist NiFi Port information to {} due to {}", new Object[]{statusFile.getAbsolutePath(), ioe}); } - logger.log(Level.INFO, "Apache NiFi now running and listening for Bootstrap requests on port {0}", port); + defaultLogger.info("Apache NiFi now running and listening for Bootstrap requests on port {}", port); } int getNiFiCommandControlPort() { http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/69f04cbb/nifi/nifi-bootstrap/src/main/java/org/apache/nifi/bootstrap/ShutdownHook.java ---------------------------------------------------------------------- diff --git a/nifi/nifi-bootstrap/src/main/java/org/apache/nifi/bootstrap/ShutdownHook.java b/nifi/nifi-bootstrap/src/main/java/org/apache/nifi/bootstrap/ShutdownHook.java index 023ab84..a594f60 100644 --- a/nifi/nifi-bootstrap/src/main/java/org/apache/nifi/bootstrap/ShutdownHook.java +++ b/nifi/nifi-bootstrap/src/main/java/org/apache/nifi/bootstrap/ShutdownHook.java @@ -21,6 +21,7 @@ import java.io.IOException; import java.io.OutputStream; import java.net.Socket; import java.nio.charset.StandardCharsets; +import java.util.concurrent.ExecutorService; import java.util.concurrent.TimeUnit; public class ShutdownHook extends Thread { @@ -28,14 +29,16 @@ public class ShutdownHook extends Thread { private final Process nifiProcess; private final RunNiFi runner; private final int gracefulShutdownSeconds; + private final ExecutorService executor; private volatile String secretKey; - public ShutdownHook(final Process nifiProcess, final RunNiFi runner, final String secretKey, final int gracefulShutdownSeconds) { + public ShutdownHook(final Process nifiProcess, final RunNiFi runner, final String secretKey, final int gracefulShutdownSeconds, final ExecutorService executor) { this.nifiProcess = nifiProcess; this.runner = runner; this.secretKey = secretKey; this.gracefulShutdownSeconds = gracefulShutdownSeconds; + this.executor = executor; } void setSecretKey(final String secretKey) { @@ -44,6 +47,7 @@ public class ShutdownHook extends Thread { @Override public void run() { + executor.shutdown(); runner.setAutoRestartNiFi(false); final int ccPort = runner.getNiFiCommandControlPort(); if (ccPort > 0) { http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/69f04cbb/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/bin/dump-nifi.bat ---------------------------------------------------------------------- diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/bin/dump-nifi.bat b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/bin/dump-nifi.bat index ab902d0..f19f029 100644 --- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/bin/dump-nifi.bat +++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/bin/dump-nifi.bat @@ -39,9 +39,9 @@ set CONF_DIR=conf set BOOTSTRAP_CONF_FILE=%CONF_DIR%\bootstrap.conf set JAVA_ARGS=-Dorg.apache.nifi.bootstrap.config.file=%BOOTSTRAP_CONF_FILE% -set JAVA_PARAMS=-cp %LIB_DIR%\* -Xms12m -Xmx24m %JAVA_ARGS% org.apache.nifi.bootstrap.RunNiFi +SET JAVA_PARAMS=-cp %CONF_DIR%;%LIB_DIR%\* -Xms12m -Xmx24m %JAVA_ARGS% org.apache.nifi.bootstrap.RunNiFi set BOOTSTRAP_ACTION=dump cmd.exe /C "%JAVA_EXE%" %JAVA_PARAMS% %BOOTSTRAP_ACTION% -popd \ No newline at end of file +popd http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/69f04cbb/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/bin/nifi.sh ---------------------------------------------------------------------- diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/bin/nifi.sh b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/bin/nifi.sh index fb0d22e..6d3191d 100644 --- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/bin/nifi.sh +++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/bin/nifi.sh @@ -159,7 +159,19 @@ run() { echo "Bootstrap Config File: $BOOTSTRAP_CONF" echo - exec "$JAVA" -cp "$NIFI_HOME"/lib/bootstrap/* -Xms12m -Xmx24m -Dorg.apache.nifi.bootstrap.config.file="$BOOTSTRAP_CONF" org.apache.nifi.bootstrap.RunNiFi $@ + # run 'start' in the background because the process will continue to run, monitoring NiFi. + # all other commands will terminate quickly so want to just wait for them + if [ "$1" = "start" ]; then + ("$JAVA" -cp "$NIFI_HOME"/conf/:"$NIFI_HOME"/lib/bootstrap/* -Xms12m -Xmx24m -Dorg.apache.nifi.bootstrap.config.file="$BOOTSTRAP_CONF" org.apache.nifi.bootstrap.RunNiFi $@ &) + else + "$JAVA" -cp "$NIFI_HOME"/conf/:"$NIFI_HOME"/lib/bootstrap/* -Xms12m -Xmx24m -Dorg.apache.nifi.bootstrap.config.file="$BOOTSTRAP_CONF" org.apache.nifi.bootstrap.RunNiFi $@ + fi + + # Wait just a bit (3 secs) to wait for the logging to finish and then echo a new-line. + # We do this to avoid having logs spewed on the console after running the command and then not giving + # control back to the user + sleep 3 + echo } main() { @@ -172,9 +184,14 @@ case "$1" in install) install "$@" ;; - start|stop|run|restart|status|dump) + start|stop|run|status|dump) main "$@" ;; + restart) + init + run "stop" + run "start" + ;; *) echo "Usage nifi {start|stop|run|restart|status|dump|install}" ;; http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/69f04cbb/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/bin/run-nifi.bat ---------------------------------------------------------------------- diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/bin/run-nifi.bat b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/bin/run-nifi.bat index 8b56968..e7708f0 100644 --- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/bin/run-nifi.bat +++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/bin/run-nifi.bat @@ -39,9 +39,9 @@ set CONF_DIR=conf set BOOTSTRAP_CONF_FILE=%CONF_DIR%\bootstrap.conf set JAVA_ARGS=-Dorg.apache.nifi.bootstrap.config.file=%BOOTSTRAP_CONF_FILE% -set JAVA_PARAMS=-cp %LIB_DIR%\* -Xms12m -Xmx24m %JAVA_ARGS% org.apache.nifi.bootstrap.RunNiFi +SET JAVA_PARAMS=-cp %CONF_DIR%;%LIB_DIR%\* -Xms12m -Xmx24m %JAVA_ARGS% org.apache.nifi.bootstrap.RunNiFi set BOOTSTRAP_ACTION=run cmd.exe /C "%JAVA_EXE%" %JAVA_PARAMS% %BOOTSTRAP_ACTION% -popd \ No newline at end of file +popd http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/69f04cbb/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/bin/start-nifi.bat ---------------------------------------------------------------------- diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/bin/start-nifi.bat b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/bin/start-nifi.bat deleted file mode 100644 index 1c95995..0000000 --- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/bin/start-nifi.bat +++ /dev/null @@ -1,47 +0,0 @@ -@echo off -rem -rem Licensed to the Apache Software Foundation (ASF) under one or more -rem contributor license agreements. See the NOTICE file distributed with -rem this work for additional information regarding copyright ownership. -rem The ASF licenses this file to You under the Apache License, Version 2.0 -rem (the "License"); you may not use this file except in compliance with -rem the License. You may obtain a copy of the License at -rem -rem http://www.apache.org/licenses/LICENSE-2.0 -rem -rem Unless required by applicable law or agreed to in writing, software -rem distributed under the License is distributed on an "AS IS" BASIS, -rem WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -rem See the License for the specific language governing permissions and -rem limitations under the License. -rem - -rem Use JAVA_HOME if it's set; otherwise, just use java - -if "%JAVA_HOME%" == "" goto noJavaHome -if not exist "%JAVA_HOME%\bin\java.exe" goto noJavaHome -set JAVA_EXE=%JAVA_HOME%\bin\java.exe -goto startNifi - -:noJavaHome -echo The JAVA_HOME environment variable is not defined correctly. -echo Instead the PATH will be used to find the java executable. -echo. -set JAVA_EXE=java -goto startNifi - -:startNifi -set NIFI_ROOT=%~dp0..\ -pushd "%NIFI_ROOT%" -set LIB_DIR=lib\bootstrap -set CONF_DIR=conf - -set BOOTSTRAP_CONF_FILE=%CONF_DIR%\bootstrap.conf -set JAVA_ARGS=-Dorg.apache.nifi.bootstrap.config.file=%BOOTSTRAP_CONF_FILE% - -set JAVA_PARAMS=-cp %LIB_DIR%\* -Xms12m -Xmx24m %JAVA_ARGS% org.apache.nifi.bootstrap.RunNiFi -set BOOTSTRAP_ACTION=start - -cmd.exe /C "%JAVA_EXE%" %JAVA_PARAMS% %BOOTSTRAP_ACTION% - -popd \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/69f04cbb/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/bin/stop-nifi.bat ---------------------------------------------------------------------- diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/bin/stop-nifi.bat b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/bin/stop-nifi.bat deleted file mode 100644 index 803b6f1..0000000 --- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/bin/stop-nifi.bat +++ /dev/null @@ -1,47 +0,0 @@ -@echo off -rem -rem Licensed to the Apache Software Foundation (ASF) under one or more -rem contributor license agreements. See the NOTICE file distributed with -rem this work for additional information regarding copyright ownership. -rem The ASF licenses this file to You under the Apache License, Version 2.0 -rem (the "License"); you may not use this file except in compliance with -rem the License. You may obtain a copy of the License at -rem -rem http://www.apache.org/licenses/LICENSE-2.0 -rem -rem Unless required by applicable law or agreed to in writing, software -rem distributed under the License is distributed on an "AS IS" BASIS, -rem WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -rem See the License for the specific language governing permissions and -rem limitations under the License. -rem - -rem Use JAVA_HOME if it's set; otherwise, just use java - -if "%JAVA_HOME%" == "" goto noJavaHome -if not exist "%JAVA_HOME%\bin\java.exe" goto noJavaHome -set JAVA_EXE=%JAVA_HOME%\bin\java.exe -goto startNifi - -:noJavaHome -echo The JAVA_HOME environment variable is not defined correctly. -echo Instead the PATH will be used to find the java executable. -echo. -set JAVA_EXE=java -goto startNifi - -:startNifi -set NIFI_ROOT=%~dp0..\ -pushd "%NIFI_ROOT%" -set LIB_DIR=lib\bootstrap -set CONF_DIR=conf - -set BOOTSTRAP_CONF_FILE=%CONF_DIR%\bootstrap.conf -set JAVA_ARGS=-Dorg.apache.nifi.bootstrap.config.file=%BOOTSTRAP_CONF_FILE% - -set JAVA_PARAMS=-cp %LIB_DIR%\* -Xms12m -Xmx24m %JAVA_ARGS% org.apache.nifi.bootstrap.RunNiFi -set BOOTSTRAP_ACTION=stop - -cmd.exe /C "%JAVA_EXE%" %JAVA_PARAMS% %BOOTSTRAP_ACTION% - -popd \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/69f04cbb/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/logback.xml ---------------------------------------------------------------------- diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/logback.xml b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/logback.xml index 296169e..36f42d7 100644 --- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/logback.xml +++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/logback.xml @@ -58,10 +58,35 @@ <pattern>%date %level [%thread] %logger{40} %msg%n</pattern> </encoder> </appender> + + <appender name="BOOTSTRAP_FILE" class="ch.qos.logback.core.rolling.RollingFileAppender"> + <file>logs/nifi-bootstrap.log</file> + <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy"> + <!-- + For daily rollover, use 'user_%d.log'. + For hourly rollover, use 'user_%d{yyyy-MM-dd_HH}.log'. + To GZIP rolled files, replace '.log' with '.log.gz'. + To ZIP rolled files, replace '.log' with '.log.zip'. + --> + <fileNamePattern>./logs/nifi-bootstrap_%d.log</fileNamePattern> + <!-- keep 5 log files worth of history --> + <maxHistory>5</maxHistory> + </rollingPolicy> + <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder"> + <pattern>%date %level [%thread] %logger{40} %msg%n</pattern> + </encoder> + </appender> + + <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender"> + <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder"> + <pattern>%date %level [%thread] %logger{40} %msg%n</pattern> + </encoder> + </appender> <!-- valid logging levels: TRACE, DEBUG, INFO, WARN, ERROR --> <logger name="org.apache.nifi" level="INFO"/> + <logger name="org.apache.nifi.controller.repository.StandardProcessSession" level="WARN" /> <!-- Logger for managing logging statements for nifi clusters. --> <logger name="org.apache.nifi.cluster" level="INFO"/> @@ -101,6 +126,29 @@ <appender-ref ref="USER_FILE"/> </logger> + + <!-- + Logger for capturing Bootstrap logs and NiFi's standard error and standard out. + --> + <logger name="org.apache.nifi.bootstrap" level="INFO" additivity="false"> + <appender-ref ref="BOOTSTRAP_FILE" /> + </logger> + <logger name="org.apache.nifi.bootstrap.Command" level="INFO" additivity="false"> + <appender-ref ref="CONSOLE" /> + <appender-ref ref="BOOTSTRAP_FILE" /> + </logger> + + <!-- Everything written to NiFi's Standard Out will be logged with the logger org.apache.nifi.StdOut at INFO level --> + <logger name="org.apache.nifi.StdOut" level="INFO" additivity="false"> + <appender-ref ref="BOOTSTRAP_FILE" /> + </logger> + + <!-- Everything written to NiFi's Standard Error will be logged with the logger org.apache.nifi.StdErr at ERROR level --> + <logger name="org.apache.nifi.StdErr" level="ERROR" additivity="false"> + <appender-ref ref="BOOTSTRAP_FILE" /> + </logger> + + <root level="INFO"> <appender-ref ref="APP_FILE"/> </root>
