ProcessUtils.java

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.tika.utils;


import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.tika.config.ParseTimeout;
import org.apache.tika.parser.ParseContext;

public class ProcessUtils {

    private static final Logger LOG = LoggerFactory.getLogger(ProcessUtils.class);

    // How often a bounded subprocess wait checkpoints the task's ParseTimeout -- must be
    // finer-grained than progressTimeoutMillis or a long wait looks like a false "hung" kill.
    public static final long HEARTBEAT_INTERVAL_MILLIS = 1000;

    // Timeout for checkCommand's binary-existence probe (e.g. "myapp --version") -- kept
    // short since an unresponsive binary isn't usable and a full minute would delay
    // every task that happens to probe first.
    public static final long DEFAULT_CHECK_COMMAND_TIMEOUT_MILLIS = 5000;

    private static final ConcurrentHashMap<String, Process> PROCESS_MAP = new ConcurrentHashMap<>();

    static {
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            PROCESS_MAP.forEachValue(1, Process::destroyForcibly);
        }));
    }

    private static String register(Process p) {
        String id = UUID.randomUUID().toString();
        PROCESS_MAP.put(id, p);
        return id;
    }

    private static Process release(String id) {
        return PROCESS_MAP.remove(id);
    }

    /**
     * This should correctly put double-quotes around an argument if
     * ProcessBuilder doesn't seem to work (as it doesn't
     * on paths with spaces on Windows)
     *
     * @param arg
     * @return
     */
    public static String escapeCommandLine(String arg) {
        if (arg == null) {
            return arg;
        }
        //need to test for " " on windows, can't just add double quotes
        //across platforms.
        if (arg.contains(" ") && SystemUtils.IS_OS_WINDOWS &&
                (!arg.startsWith("\"") && !arg.endsWith("\""))) {
            arg = "\"" + arg + "\"";
        }
        return arg;
    }

    public static String unescapeCommandLine(String arg) {
        if (arg.contains(" ") && SystemUtils.IS_OS_WINDOWS &&
                (arg.startsWith("\"") && arg.endsWith("\""))) {
            arg = arg.substring(1, arg.length() - 1);
        }
        return arg;
    }

    /**
     * This writes stdout and stderr to the FileProcessResult.
     * <p>
     * Equivalent to {@link #execute(ProcessBuilder, ParseContext, long, int, int)} with a
     * null context: {@code requestedTimeoutMillis} is granted unclipped, no checkpointing.
     *
     * @param pb
     * @param requestedTimeoutMillis
     * @param maxStdoutBuffer
     * @param maxStdErrBuffer
     * @return
     * @throws IOException
     */
    public static FileProcessResult execute(ProcessBuilder pb,
                                            long requestedTimeoutMillis,
                                            int maxStdoutBuffer, int maxStdErrBuffer)
            throws IOException {
        return execute(pb, null, requestedTimeoutMillis, maxStdoutBuffer, maxStdErrBuffer);
    }

    /**
     * Same as {@link #execute(ProcessBuilder, long, int, int)}, but bounds the wait to
     * {@code min(requestedTimeoutMillis, ParseTimeout.remainingMillis())} (see
     * {@link ParseTimeout#budgetFor(long)}) so no single call can outlast the task's
     * total timeout regardless of its own configuration. The granted budget and original
     * request are both recorded on the result (see
     * {@link FileProcessResult#getRequestedTimeoutMillis()},
     * {@link FileProcessResult#isClippedByRemaining()}).
     * <p>
     * While waiting, checkpoints the {@link ParseTimeout} in {@code context} (if any)
     * every {@value #HEARTBEAT_INTERVAL_MILLIS} ms, so a bounded external call can run
     * longer than the progress (stall-detection) timeout without looking like a hang --
     * the wait itself is progress. A null {@code context} behaves like the
     * four-argument overload.
     *
     * @param pb
     * @param context                may be null
     * @param requestedTimeoutMillis the timeout the caller's own configuration asks for
     * @param maxStdoutBuffer
     * @param maxStdErrBuffer
     * @return
     * @throws IOException
     */
    public static FileProcessResult execute(ProcessBuilder pb, ParseContext context,
                                            long requestedTimeoutMillis,
                                            int maxStdoutBuffer, int maxStdErrBuffer)
            throws IOException {
        // A null context has no task to clip against -- getOrCreate(null) would otherwise
        // hand back a detached ParseTimeout built from *default* TimeoutLimits (1 hour),
        // silently capping any request above that and breaking the "granted unclipped"
        // contract documented above for null-context callers (e.g. a one-off version-check
        // process run outside any parse).
        long grantedTimeoutMillis = context == null
                ? requestedTimeoutMillis
                : ParseTimeout.getOrCreate(context).budgetFor(requestedTimeoutMillis);
        Process p = null;
        String id = null;
        try {
            p = pb.start();
            id = register(p);
            long elapsed = -1;
            long start = System.currentTimeMillis();
            StreamGobbler outGobbler = new StreamGobbler(p.getInputStream(), maxStdoutBuffer);
            StreamGobbler errGobbler = new StreamGobbler(p.getErrorStream(), maxStdErrBuffer);

            Thread outThread = new Thread(outGobbler);
            outThread.start();

            Thread errThread = new Thread(errGobbler);
            errThread.start();
            int exitValue = -1;
            boolean complete = false;
            try {
                complete = waitForWithHeartbeat(p, context, grantedTimeoutMillis);
                elapsed = System.currentTimeMillis() - start;
                if (complete) {
                    exitValue = p.exitValue();
                    outThread.join(1000);
                    errThread.join(1000);
                } else {
                    p.destroyForcibly();
                    outThread.join(1000);
                    errThread.join(1000);
                    boolean completed = p.waitFor(500, TimeUnit.MILLISECONDS);
                    if (completed) {
                        try {
                            exitValue = p.exitValue();
                        } catch (IllegalThreadStateException e) {
                            //not finished!
                        }
                    }
                }
            } catch (InterruptedException e) {
                // restore the flag or the interrupt is misread as a subprocess timeout
                Thread.currentThread().interrupt();
                exitValue = -1000;
            } finally {
                outThread.interrupt();
                errThread.interrupt();
            }
            FileProcessResult result = new FileProcessResult();
            result.processTimeMillis = elapsed;
            result.stderrLength = errGobbler.getStreamLength();
            result.stdoutLength = outGobbler.getStreamLength();
            result.isTimeout = ! complete;
            result.exitValue = exitValue;
            result.stdout = StringUtils.joinWith("\n", outGobbler.getLines());
            result.stderr = StringUtils.joinWith("\n", errGobbler.getLines());
            result.stdoutTruncated = outGobbler.getIsTruncated();
            result.stderrTruncated = errGobbler.getIsTruncated();
            result.requestedTimeoutMillis = requestedTimeoutMillis;
            result.grantedTimeoutMillis = grantedTimeoutMillis;
            return result;
        } finally {
            if (p != null) {
                p.destroyForcibly();
            }
            if (id != null) {
                release(id);
            }
        }
    }

    /**
     * This redirects stdout to stdoutRedirect path.
     * <p>
     * Equivalent to {@link #execute(ProcessBuilder, ParseContext, long, Path, int)} with
     * a null context, i.e. {@code requestedTimeoutMillis} is granted unclipped and the
     * wait does not checkpoint any task's progress timeout.
     *
     * @param pb
     * @param requestedTimeoutMillis
     * @param stdoutRedirect
     * @param maxStdErrBuffer
     * @return
     * @throws IOException
     */
    public static FileProcessResult execute(ProcessBuilder pb,
                                            long requestedTimeoutMillis,
                                            Path stdoutRedirect, int maxStdErrBuffer) throws IOException {
        return execute(pb, null, requestedTimeoutMillis, stdoutRedirect, maxStdErrBuffer);
    }

    /**
     * Same as {@link #execute(ProcessBuilder, long, Path, int)}, but bounds the wait to
     * {@code min(requestedTimeoutMillis, ParseTimeout.remainingMillis())} and checkpoints
     * while waiting -- see {@link #execute(ProcessBuilder, ParseContext, long, int, int)}.
     *
     * @param pb
     * @param context                may be null
     * @param requestedTimeoutMillis the timeout the caller's own configuration asks for
     * @param stdoutRedirect
     * @param maxStdErrBuffer
     * @return
     * @throws IOException
     */
    public static FileProcessResult execute(ProcessBuilder pb, ParseContext context,
                                            long requestedTimeoutMillis,
                                            Path stdoutRedirect, int maxStdErrBuffer) throws IOException {

        if (!Files.isDirectory(stdoutRedirect.getParent())) {
            Files.createDirectories(stdoutRedirect.getParent());
        }

        // See the (ProcessBuilder, ParseContext, long, int, int) overload above: a null
        // context must grant requestedTimeoutMillis unclipped, not silently cap it against
        // a detached, default-TimeoutLimits ParseTimeout.
        long grantedTimeoutMillis = context == null
                ? requestedTimeoutMillis
                : ParseTimeout.getOrCreate(context).budgetFor(requestedTimeoutMillis);
        pb.redirectOutput(stdoutRedirect.toFile());
        Process p = null;
        String id = null;
        try {
            p = pb.start();
            id = register(p);
            long elapsed = -1;
            long start = System.currentTimeMillis();
            StreamGobbler errGobbler = new StreamGobbler(p.getErrorStream(), maxStdErrBuffer);

            Thread errThread = new Thread(errGobbler);
            errThread.start();
            int exitValue = -1;
            boolean complete = false;
            try {
                complete = waitForWithHeartbeat(p, context, grantedTimeoutMillis);
                elapsed = System.currentTimeMillis() - start;
                if (complete) {
                    exitValue = p.exitValue();
                    errThread.join(1000);
                } else {
                    p.destroyForcibly();
                    errThread.join(1000);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                exitValue = -1000;
            }
            FileProcessResult result = new FileProcessResult();
            result.processTimeMillis = elapsed;
            result.stderrLength = errGobbler.getStreamLength();
            result.stdoutLength = Files.size(stdoutRedirect);
            result.isTimeout = !complete;
            result.exitValue = exitValue;
            result.stdout = "";
            result.stderr = StringUtils.joinWith("\n", errGobbler.getLines());
            result.stdoutTruncated = false;
            result.stderrTruncated = errGobbler.getIsTruncated();
            result.requestedTimeoutMillis = requestedTimeoutMillis;
            result.grantedTimeoutMillis = grantedTimeoutMillis;
            return result;
        } finally {
            if (p != null) {
                p.destroyForcibly();
            }
            if (id != null) {
                release(id);
            }
        }

    }

    /**
     * Checks to see if the command can be run. Typically used with
     * something like "myapp --version" to check to see if "myapp"
     * is installed and on the path.
     * <p>
     * Equivalent to {@link #checkCommandWithTimeout(String[], long, int...)} with
     * {@link #DEFAULT_CHECK_COMMAND_TIMEOUT_MILLIS}.
     *
     * @param checkCmd   The check command to run
     * @param errorValue What is considered an error value? Default is 127 (command not found).
     * @return true if the command ran successfully (exit code not in errorValue list)
     */
    public static boolean checkCommand(String checkCmd, int... errorValue) {
        return checkCommandWithTimeout(new String[]{checkCmd}, DEFAULT_CHECK_COMMAND_TIMEOUT_MILLIS, errorValue);
    }

    /**
     * Checks to see if the command can be run. Typically used with
     * something like {@code new String[]{"myapp", "--version"}} to check to see if "myapp"
     * is installed and on the path.
     * <p>
     * Equivalent to {@link #checkCommandWithTimeout(String[], long, int...)} with
     * {@link #DEFAULT_CHECK_COMMAND_TIMEOUT_MILLIS}.
     *
     * @param checkCmd   The check command to run
     * @param errorValue What is considered an error value? Default is 127 (command not found).
     * @return true if the command ran successfully (exit code not in errorValue list)
     */
    public static boolean checkCommand(String[] checkCmd, int... errorValue) {
        return checkCommandWithTimeout(checkCmd, DEFAULT_CHECK_COMMAND_TIMEOUT_MILLIS, errorValue);
    }

    /**
     * Same as {@link #checkCommand(String[], int...)}, but with a caller-specified timeout
     * instead of the {@value #DEFAULT_CHECK_COMMAND_TIMEOUT_MILLIS}ms default.
     * <p>
     * Deliberately a distinct method name, not a same-named overload: {@code checkCommand(cmd, 500)}
     * would silently resolve to {@code checkCommand(String[], int...)} with
     * {@code errorValue={500}} at the default timeout -- Java prefers the varargs-only
     * overload over widening {@code int} to this method's {@code long} parameter, with no
     * compile error to catch the mistake.
     *
     * @param timeoutMillis how long to wait for the command to exit
     * @param errorValue    What is considered an error value? Default is 127 (command not found).
     * @return true if the command ran successfully (exit code not in errorValue list)
     */
    public static boolean checkCommandWithTimeout(String[] checkCmd, long timeoutMillis, int... errorValue) {
        if (errorValue.length == 0) {
            errorValue = new int[]{127};
        }

        Process process = null;
        try {
            process = Runtime.getRuntime().exec(checkCmd);
            StreamGobbler outGobbler = new StreamGobbler(process.getInputStream(), 0);
            StreamGobbler errGobbler = new StreamGobbler(process.getErrorStream(), 0);
            Thread outThread = new Thread(outGobbler);
            Thread errThread = new Thread(errGobbler);
            outThread.start();
            errThread.start();
            boolean finished = process.waitFor(timeoutMillis, TimeUnit.MILLISECONDS);
            if (!finished) {
                throw new TimeoutException();
            }
            outThread.join(1000);
            errThread.join(1000);
            int result = process.exitValue();
            LOG.debug("exit value for {}: {}", checkCmd[0], result);
            for (int err : errorValue) {
                if (result == err) {
                    return false;
                }
            }
            return true;
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            LOG.debug("interrupted trying to run " + checkCmd[0], e);
            return false;
        } catch (IOException | TimeoutException e) {
            LOG.debug("exception trying to run " + checkCmd[0], e);
            return false;
        } catch (SecurityException se) {
            throw se;
        } catch (Error err) {
            if (err.getMessage() != null && (err.getMessage().contains("posix_spawn") ||
                    err.getMessage().contains("UNIXProcess"))) {
                LOG.debug("(TIKA-1526): exception trying to run: " + checkCmd[0], err);
                return false;
            }
            throw err;
        } finally {
            if (process != null) {
                process.destroyForcibly();
            }
        }
    }

    /**
     * Waits for the process to exit, like {@link Process#waitFor(long, TimeUnit)}, but polls
     * in {@value #HEARTBEAT_INTERVAL_MILLIS} ms increments and checkpoints {@code context}'s
     * {@link ParseTimeout} after each increment that doesn't complete -- this is what lets a
     * bounded external call run longer than the progress timeout without tripping the stall
     * detector: the wait itself is progress.
     * <p>
     * Public so callers managing their own {@link Process} (not going through
     * {@link #execute(ProcessBuilder, ParseContext, long, int, int)}) can still checkpoint
     * while waiting.
     *
     * @param context       may be null, in which case no checkpoint is recorded
     * @param timeoutMillis total wait time in ms; zero or negative checks once without waiting
     * @return true if the process exited before the timeout elapsed
     */
    public static boolean waitForWithHeartbeat(Process p, ParseContext context, long timeoutMillis)
            throws InterruptedException {
        long startNanos = System.nanoTime();
        while (true) {
            long elapsedMillis = (System.nanoTime() - startNanos) / 1_000_000L;
            long remaining = timeoutMillis - elapsedMillis;
            long pollMillis = remaining <= 0 ? 0 : Math.min(remaining, HEARTBEAT_INTERVAL_MILLIS);
            if (p.waitFor(pollMillis, TimeUnit.MILLISECONDS)) {
                return true;
            }
            if (remaining <= 0) {
                return false;
            }
            ParseTimeout.checkpoint(context);
        }
    }

}