Bug 1212950 - EAP 6 start operation causes agent to run out of memory…

… due to

storing console output in an unused buffer

Now process output is captured (if captured) up to 2MB size, once output
exceeds this limit, it is ignored - so we don't run out of memory (unless
agent does not start plenty of verbose processes). Default limit can be
changed via rhq.process-execution.captured-output.limit system property.

This commit also gives more power to plugin writers about capturing process
outputs. ProcessExecution#setCaptureOutput is now deprecated in favor of new
CaptureMode setting. CaptureMode can capture to memory and/or forward to
agent.log as well as setting captured limit.

x

modules/core/native-system/src/main/java/org/rhq/core/system/JavaSystemInfo.java

@@ -23,7 +23,6 @@
 package org.rhq.core.system;
 
 import java.io.BufferedReader;
-import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.InputStreamReader;
 import java.net.InetAddress;
@@ -206,14 +205,13 @@ public ProcessExecutionResults executeProcess(ProcessExecution processExecution)
         process.setEnvironment(processExecution.getEnvironmentVariablesAsArray());
         process.setWorkingDirectory(processExecution.getWorkingDirectory());
         process.setWaitForExit(Long.valueOf(processExecution.getWaitForCompletion()));
-        process.setCaptureOutput(Boolean.valueOf(processExecution.isCaptureOutput()));
+        process.setCaptureOutput(Boolean.valueOf(processExecution.getCaptureMode().isCapture()));
         process.setKillOnTimeout(Boolean.valueOf(processExecution.isKillOnTimeout()));
 
-        if (processExecution.isCaptureOutput()) {
-            ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
-            process.setOutputStream(outputStream);
-            executionResults.setCapturedOutputStream(outputStream);
-        }
+        ProcessExecutionOutputStream outputStream = processExecution.getCaptureMode().createOutputStream();
+        process.setOutputStream(outputStream);
+        executionResults.setCapturedOutputStream(outputStream);
+
 
         ProcessExecutorResults javaExecResults = javaExec.execute(process);
         executionResults.setExitCode(javaExecResults.getExitCode());

modules/core/native-system/src/main/java/org/rhq/core/system/ProcessExecution.java

@@ -49,9 +49,9 @@ public class ProcessExecution {
     private Map<String, String> environmentVariables;
     private String workingDirectory;
     private long waitForCompletion = 30000L;
-    private boolean captureOutput = false;
     private boolean killOnTimeout = false;
     private boolean checkExecutableExists = true;
+    private CaptureMode captureMode = CaptureMode.none();
 
     /**
      * Constructor for {@link ProcessExecution} that defines the full path to the executable that will be run. See the
@@ -209,21 +209,38 @@ public void setWaitForCompletion(long waitForCompletion) {
         this.waitForCompletion = waitForCompletion;
     }
 
+    /**
+     * 
+     * @return whether capture process output
+     * @deprecated
+     */
     public boolean isCaptureOutput() {
-        return captureOutput;
+        return this.captureMode.isCapture();
     }
 
     /**
      * If <code>true</code>, the process's output will be captured and returned in the results. This may be ignored if
      * <code>waitForCompletion</code> is 0 or less. Be careful setting this to <code>true</code>, you must ensure that
      * the process will not write a lot of output - you might run out of memory if the process is a long-lived daemon
      * process that outputs a lot of log messages, for example. By default, output is *not* captured.
-     *
+     * @deprecated @see {@link #setCaptureMode(CaptureMode)}
      * @param captureOutput whether or not this process's output (stdout+stderr) should be captured and returned in the
      *                      results
      */
     public void setCaptureOutput(boolean captureOutput) {
-        this.captureOutput = captureOutput;
+        this.captureMode = captureOutput ? CaptureMode.memory() : CaptureMode.none();
+    }
+
+    /**
+     * get process output capture mode
+     * @return process output capture mode
+     */
+    public CaptureMode getCaptureMode() {
+        return captureMode;
+    }
+
+    public void setCaptureMode(CaptureMode captureMode) {
+        this.captureMode = captureMode;
     }
 
     public boolean isKillOnTimeout() {
@@ -267,12 +284,139 @@ public String toString() {
         buf.append("], env-vars=[").append(this.environmentVariables);
         buf.append("], working-dir=[").append(this.workingDirectory);
         buf.append("], wait=[").append(this.waitForCompletion);
-        buf.append("], capture-output=[").append(this.captureOutput);
+        buf.append("], capture-mode=[").append(this.captureMode);
         buf.append("], kill-on-timeout=[").append(this.killOnTimeout);
         buf.append("], executable-is-command=[").append(this.checkExecutableExists);
         buf.append("]");
 
         return buf.toString();
     }
 
+    /**
+     * Process output capture mode.
+     * 
+     * @author lzoubek@redhat.com
+     *
+     */
+    public static class CaptureMode {
+
+        /**
+         * The process's output is *not* captured, this is the default.
+         * @return captureMode
+         */
+        public static CaptureMode none() {
+            return new CaptureMode(false);
+        }
+
+        /**
+         * The process's output will be captured and returned in the results. This may be ignored if
+         * <code>waitForCompletion</code> is 0 or less. By default capturing to memory is limited to 2MB of
+         * process output. If the process writes more output, it will be ignored.
+         * @return captureMode
+         */
+        public static CaptureMode memory() {
+            return new CaptureMode(true);
+        }
+
+        /**
+         * The process's output will be captured and returned in the results. This may be ignored if
+         * <code>waitForCompletion</code> is 0 or less. With <code>limit</code> parameter you can set maximum captured output size. 
+         * If the process writes more output, it will be ignored.
+         * 
+         * @param limit in Bytes (if given value < 0, it's ignored and default 2MB is used instead)
+         * @return captureMode
+         */
+        public static CaptureMode memory(int limit) {
+            return new CaptureMode(true, limit);
+        }
+
+        /**
+         * The process's output will be captured and returned in the results. This may be ignored if
+         * <code>waitForCompletion</code> is 0 or less. Process output will be redirected to agent.log and at the same time
+         * captured into memory. By default capturing to memory is limited to 2MB of process output. If the process writes more output,
+         * it will only be redirected to agent.log.
+         * 
+         * @return captureMode
+         */
+        public static CaptureMode agentLog() {
+            return new CaptureMode(true, true, -1);
+        }
+
+        /**
+         * The process's output will be captured and returned in the results. This may be ignored if
+         * <code>waitForCompletion</code> is 0 or less. Process output will be logged into agent.log and at the same time
+         * captured into memory. With <code>limit</code> parameter you can set maximum memory buffer to be captured (and possibly returned)
+         * captured output size. If the process writes more output, it will only be redirected to agent.log.
+         * 
+         * @param limit in Bytes (if given value < 0, it's ignored and default 2MB is used instead)
+         * @return captureMode
+         */
+        public static CaptureMode agentLog(int limit) {
+            return new CaptureMode(true, true, limit);
+        }
+
+        private final boolean capture;
+        private final int limit;
+        private final boolean log;
+
+        private CaptureMode(boolean capture) {
+            this(capture, -1);
+        }
+
+        private CaptureMode(boolean capture, int limit) {
+            this(capture, false, limit);
+        }
+
+        private CaptureMode(boolean capture, boolean log, int limit) {
+            this.capture = capture;
+            this.log = log;
+            this.limit = limit;
+        }
+
+        /**
+         * 
+         * @return true if capturing is enabled
+         */
+        public boolean isCapture() {
+            return capture;
+        }
+
+        /**
+         * 
+         * @return captured output size limit in Bytes, -1 if default should be used 
+         */
+        public int getLimit() {
+            return limit;
+        }
+
+        /**
+         * 
+         * @return true if output should be forwarded to logging subsystem
+         */
+        public boolean isLog() {
+            return log;
+        }
+
+        ProcessExecutionOutputStream createOutputStream() {
+            if (!this.capture) {
+                // capturing is disabled still return some output stream (this instance ignores everything)
+                return new ProcessExecutionOutputStream(0, this.log);
+            }
+            if (this.limit > 0) {
+                return new ProcessExecutionOutputStream(this.limit, this.log);
+            }
+            return new ProcessExecutionOutputStream(this.log);
+        }
+
+        @Override
+        public String toString() {
+            return new StringBuilder("CaptureMode: ")
+            .append(" [capture="+isCapture())
+            .append("], [memory-limit=" + getLimit() / 1024 + "kB")
+            .append("], [log="+isLog())
+            .append("]")
+            .toString();
+        }
+    }
+
 }

modules/core/native-system/src/main/java/org/rhq/core/system/ProcessExecutionOutputStream.java

@@ -0,0 +1,90 @@
+/*
+ * RHQ Management Platform
+ * Copyright (C) 2005-2015 Red Hat, Inc.
+ * All rights reserved.
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License, version 2, as
+ * published by the Free Software Foundation, and/or the GNU Lesser
+ * General Public License, version 2.1, also as published by the Free
+ * Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License and the GNU Lesser General Public License
+ * for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * and the GNU Lesser General Public License along with this program;
+ * if not, write to the Free Software Foundation, Inc.,
+ * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
+ */
+package org.rhq.core.system;
+
+import java.io.ByteArrayOutputStream;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+/**
+ * in-memory output stream with a limited buffer size. Once limit is reached it starts to ignore
+ * {@link #write(int)} and {@link #write(byte[], int, int)} calls. Default limit is 2MB - this default can be
+ * set using  rhq.process-execution.captured-output.limit system property (in Bytes) or via {@link #ProcessExecutionOutputStream(int, boolean)}.
+ * Optionally written output can be forwarded to logging subsystem.
+ * @author lzoubek@redhat.com
+ *
+ */
+public class ProcessExecutionOutputStream extends ByteArrayOutputStream {
+
+    private static final Log LOG = LogFactory.getLog(ProcessExecutionOutputStream.class);
+
+    private int limit = Integer.getInteger("rhq.process-execution.captured-output.limit", 2 * 1024 * 1024); // default 2MB
+
+    private boolean writeToLog;
+
+    /**
+     * Creates output stream with specified limit size
+     * @param limit (in Bytes) maximum size of stream buffer, once reached, stream silently ignores any writes
+     * @param writeToLog true to forward all messages to logger
+     */
+    public ProcessExecutionOutputStream(int limit, boolean writeToLog) {
+        super();
+        this.limit = limit;
+        this.writeToLog = writeToLog;
+    }
+
+    /**
+     * Creates output stream with default stream buffer limit (2MB)
+     * @param writeToLog true to forward all messages to logger
+     */
+    public ProcessExecutionOutputStream(boolean writeToLog) {
+        super();
+        this.writeToLog = writeToLog;
+    }
+
+    @Override
+    public synchronized void write(int b) {
+        if (this.count < this.limit) {
+            super.write(b);
+        }
+
+        if (this.writeToLog) {
+            LOG.info(String.valueOf(b));
+        }
+    }
+
+    @Override
+    public synchronized void write(byte[] b, int off, int len) {
+        if (this.count < this.limit) {
+            if (this.count + len > this.limit) {
+                len = this.limit - this.count;
+            }
+            super.write(b, off, len);
+        }
+
+        if (this.writeToLog && len > 0) {
+            LOG.info(new String(this.buf, this.count - len, len));
+        }
+    }
+}