You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tika.apache.org by ta...@apache.org on 2021/02/16 16:53:10 UTC
[tika] branch main updated: TIKA-3301 -- simplify forking and
monitoring in tika-server
This is an automated email from the ASF dual-hosted git repository.
tallison pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/tika.git
The following commit(s) were added to refs/heads/main by this push:
new 28be37e TIKA-3301 -- simplify forking and monitoring in tika-server
28be37e is described below
commit 28be37e68db8c5b749ed08b4b14d02b14b24c9f6
Author: tballison <ta...@apache.org>
AuthorDate: Tue Feb 16 11:52:58 2021 -0500
TIKA-3301 -- simplify forking and monitoring in tika-server
---
.../tika/server/core/ServerStatusWatcher.java | 110 ++++-----
.../org/apache/tika/server/core/TikaServerCli.java | 6 +-
.../apache/tika/server/core/TikaServerConfig.java | 58 ++---
.../apache/tika/server/core/TikaServerProcess.java | 21 +-
.../tika/server/core/TikaServerWatchDog.java | 269 ++++++++++-----------
.../tika/server/core/IntegrationTestBase.java | 7 +-
.../core/TikaServerAsyncIntegrationTest.java | 4 +-
.../core/TikaServerEmitterIntegrationTest.java | 2 +-
.../server/core/TikaServerIntegrationTest.java | 78 +++---
.../configs/tika-config-server-emitter.xml | 2 +-
.../configs/tika-config-server-timeout-10000.xml | 4 +-
.../test/resources/configs/tika-config-server.xml | 2 +
12 files changed, 252 insertions(+), 311 deletions(-)
diff --git a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/ServerStatusWatcher.java b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/ServerStatusWatcher.java
index aaff0a1..968a41a 100644
--- a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/ServerStatusWatcher.java
+++ b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/ServerStatusWatcher.java
@@ -38,23 +38,20 @@ public class ServerStatusWatcher implements Runnable {
private static final Logger LOG = LoggerFactory.getLogger(ServerStatusWatcher.class);
private final ServerStatus serverStatus;
- private final DataInputStream fromParent;
+ private final InputStream fromParent;
private final TikaServerConfig tikaServerConfig;
private final Path forkedStatusPath;
private final ByteBuffer statusBuffer = ByteBuffer.allocate(16);
private volatile boolean shuttingDown = false;
-
- private volatile Instant lastPing = null;
-
public ServerStatusWatcher(ServerStatus serverStatus,
InputStream inputStream, Path forkedStatusPath,
- TikaServerConfig tikaServerConfig) throws IOException {
+ TikaServerConfig tikaServerConfig) throws InterruptedException {
this.serverStatus = serverStatus;
this.tikaServerConfig = tikaServerConfig;
this.forkedStatusPath = forkedStatusPath;
serverStatus.setStatus(ServerStatus.STATUS.OPERATING);
- this.fromParent = new DataInputStream(inputStream);
+ this.fromParent = inputStream;
Thread statusWatcher = new Thread(new StatusWatcher());
statusWatcher.setDaemon(true);
statusWatcher.start();
@@ -64,45 +61,26 @@ public class ServerStatusWatcher implements Runnable {
@Override
public void run() {
- byte directive = (byte)-1;
- while (true) {
- try {
- directive = fromParent.readByte();
- lastPing = Instant.now();
- } catch (Exception e) {
- LOG.warn("Exception reading from parent", e);
- serverStatus.setStatus(ServerStatus.STATUS.PARENT_EXCEPTION);
- shutdown(ServerStatus.STATUS.PARENT_EXCEPTION);
- }
- if (directive == ServerStatus.DIRECTIVES.PING.getByte()) {
- if (serverStatus.getStatus().equals(ServerStatus.STATUS.OPERATING)) {
- checkForHitMaxFiles();
- checkForTaskTimeouts();
- }
- try {
- writeStatus(false);
- } catch (Exception e) {
- LOG.warn("Exception writing to parent", e);
- serverStatus.setStatus(ServerStatus.STATUS.PARENT_EXCEPTION);
- shutdown(ServerStatus.STATUS.PARENT_EXCEPTION);
- }
- } else if (directive == ServerStatus.DIRECTIVES.SHUTDOWN.getByte()) {
- LOG.info("Parent requested shutdown");
- serverStatus.setStatus(ServerStatus.STATUS.PARENT_REQUESTED_SHUTDOWN);
- shutdown(ServerStatus.STATUS.PARENT_REQUESTED_SHUTDOWN);
- } else if (directive == ServerStatus.DIRECTIVES.PING_ACTIVE_SERVER_TASKS.getByte()) {
- try {
- writeStatus(false);
- } catch (Exception e) {
- LOG.warn("Exception writing to parent", e);
- serverStatus.setStatus(ServerStatus.STATUS.PARENT_EXCEPTION);
- shutdown(ServerStatus.STATUS.PARENT_EXCEPTION);
- }
+
+ try {
+ //this should block forever until the parent dies
+ int directive = fromParent.read();
+ if (directive != -1) {
+ LOG.debug("Read byte ({}) from forking process. Shouldn't have received anything",
+ directive);
}
+ } catch (Exception e) {
+ LOG.debug("Exception reading from parent", e);
+ } finally {
+ LOG.info("Forking process signalled I should shutdown.");
+ serverStatus.setStatus(ServerStatus.STATUS.PARENT_EXCEPTION);
+ shutdown(ServerStatus.STATUS.PARENT_EXCEPTION);
}
+
}
- private synchronized void writeStatus(boolean shuttingDown) throws IOException {
+ private synchronized void writeStatus(boolean shuttingDown)
+ throws InterruptedException {
if (this.shuttingDown == true) {
return;
}
@@ -115,7 +93,7 @@ public class ServerStatusWatcher implements Runnable {
try (FileChannel channel = FileChannel.open(forkedStatusPath,
StandardOpenOption.CREATE,
StandardOpenOption.WRITE)) {
- while (elapsed < tikaServerConfig.getPingTimeoutMillis()) {
+ while (elapsed < tikaServerConfig.getTaskTimeoutMillis()) {
try (FileLock lock = channel.tryLock()) {
if (lock != null) {
((Buffer) statusBuffer).position(0);
@@ -126,9 +104,14 @@ public class ServerStatusWatcher implements Runnable {
channel.force(true);
return;
}
+ } catch (IOException e) {
+ LOG.warn("Problem writing to status file", e);
}
+ Thread.sleep(100);
elapsed = Duration.between(started, Instant.now()).toMillis();
}
+ } catch (IOException e) {
+ LOG.warn("Couldn't open forked status file for writing", e);
}
throw new FatalException("Couldn't write to status file after trying for " + elapsed + " millis.");
}
@@ -152,12 +135,12 @@ public class ServerStatusWatcher implements Runnable {
if (status.fileName.isPresent()) {
LOG.error("Timeout task {}, millis elapsed {}, file {}" +
"consider increasing the allowable time with the " +
- "-taskTimeoutMillis flag",
+ "<taskTimeoutMillis/> parameter",
status.task.toString(), millisElapsed, status.fileName.get());
} else {
LOG.error("Timeout task {}, millis elapsed {}; " +
"consider increasing the allowable time with the " +
- "-taskTimeoutMillis flag",
+ "<taskTimeoutMillis/> parameter",
status.task.toString(), millisElapsed);
}
}
@@ -165,12 +148,6 @@ public class ServerStatusWatcher implements Runnable {
}
private void shutdown(ServerStatus.STATUS status) {
- try {
- writeStatus(true);
- } catch (Exception e) {
- LOG.debug("problem writing status before shutdown", e);
- }
-
//if something went wrong with the parent,
//the forked process should try to delete the tmp file
if (status == ServerStatus.STATUS.PARENT_EXCEPTION) {
@@ -179,7 +156,14 @@ public class ServerStatusWatcher implements Runnable {
} catch (IOException e) {
//swallow
}
+ } else {
+ try {
+ writeStatus(true);
+ } catch (Exception e) {
+ LOG.debug("problem writing status before shutdown", e);
+ }
}
+
LOG.info("Shutting down forked process with status: {}", status.name());
System.exit(status.getShutdownCode());
}
@@ -191,24 +175,30 @@ public class ServerStatusWatcher implements Runnable {
@Override
public void run() {
+ Instant lastWrite = Instant.now();
while (true) {
+ checkForHitMaxFiles();
+ checkForTaskTimeouts();
ServerStatus.STATUS currStatus = serverStatus.getStatus();
-
if (currStatus != ServerStatus.STATUS.OPERATING) {
- LOG.warn("forked process observed "+currStatus.name()+ " and is shutting down.");
+ LOG.warn("forked process observed " + currStatus.name() + " and is shutting down.");
shutdown(currStatus);
- }
-
- if (lastPing != null) {
- long elapsed = Duration.between(lastPing, Instant.now()).toMillis();
- if (elapsed > tikaServerConfig.getPingTimeoutMillis()) {
- serverStatus.setStatus(ServerStatus.STATUS.PARENT_EXCEPTION);
- shutdown(ServerStatus.STATUS.PARENT_EXCEPTION);
+ } else {
+ long elapsed = Duration.between(lastWrite, Instant.now()).toMillis();
+ if (elapsed > tikaServerConfig.getTaskPulseMillis()) {
+ try {
+ writeStatus(false);
+ lastWrite = Instant.now();
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
}
}
+
try {
- Thread.sleep(tikaServerConfig.getPingPulseMillis());
+ Thread.sleep(100);
} catch (InterruptedException e) {
+ LOG.warn("status watcher sees interrupted exception");
return;
}
}
diff --git a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TikaServerCli.java b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TikaServerCli.java
index d1dcc48..b42f50d 100644
--- a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TikaServerCli.java
+++ b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TikaServerCli.java
@@ -44,7 +44,10 @@ import static org.apache.tika.server.core.TikaServerConfig.DEFAULT_PORT;
public class TikaServerCli {
private static final Logger LOG = LoggerFactory.getLogger(TikaServerCli.class);
-
+ /**
+ * This value is set to the server's id in the forked process.
+ */
+ public static String TIKA_SERVER_ID_ENV = "tika.server.id";
private static Options getOptions() {
Options options = new Options();
options.addOption("h", "host", true, "host name (default = "
@@ -65,7 +68,6 @@ public class TikaServerCli {
}
public static void main(String[] args) {
- LOG.info("Starting {} server", new Tika());
try {
execute(args);
} catch (Exception e) {
diff --git a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TikaServerConfig.java b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TikaServerConfig.java
index 332f428..2726e8e 100644
--- a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TikaServerConfig.java
+++ b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TikaServerConfig.java
@@ -147,16 +147,6 @@ public class TikaServerConfig {
- /**
- * If the forked process doesn't receive a ping or the parent doesn't
- * hear back from a ping in this amount of time, terminate and restart the forked process.
- */
- public static final long DEFAULT_PING_TIMEOUT_MILLIS = 30000;
-
- /**
- * How often should the parent try to ping the forked process to check status
- */
- public static final long DEFAULT_PING_PULSE_MILLIS = 100;
/**
* Number of milliseconds to wait per server task (parse, detect, unpack, translate,
@@ -164,7 +154,10 @@ public class TikaServerConfig {
*/
public static final long DEFAULT_TASK_TIMEOUT_MILLIS = 120000;
- public static final long DEFAULT_TASK_PULSE_MILLIS = 100;
+ /**
+ * How often to check to see that the task hasn't timed out
+ */
+ public static final long DEFAULT_TASK_PULSE_MILLIS = 10000;
/**
* Number of milliseconds to wait for forked process to startup
@@ -175,8 +168,6 @@ public class TikaServerConfig {
private long maxFiles = 100000;
private long taskTimeoutMillis = DEFAULT_TASK_TIMEOUT_MILLIS;
private long taskPulseMillis = DEFAULT_TASK_PULSE_MILLIS;
- private long pingTimeoutMillis = DEFAULT_PING_TIMEOUT_MILLIS;
- private long pingPulseMillis = DEFAULT_PING_PULSE_MILLIS;
private long maxforkedStartupMillis = DEFAULT_FORKED_STARTUP_MILLIS;
private boolean enableUnsecureFeatures = false;
private String cors = "";
@@ -228,6 +219,7 @@ public class TikaServerConfig {
public void setPort(int port) {
this.port = port;
}
+
/**
* How long to wait for a task before shutting down the forked server process
* and restarting it.
@@ -238,39 +230,21 @@ public class TikaServerConfig {
}
/**
- *
- * @param taskTimeoutMillis number of milliseconds to allow per task
- * (parse, detection, unzipping, etc.)
- */
- public void setTaskTimeoutMillis(long taskTimeoutMillis) {
- this.taskTimeoutMillis = taskTimeoutMillis;
- }
-
- public long getPingTimeoutMillis() {
- return pingTimeoutMillis;
- }
-
- /**
- *
- * @param pingTimeoutMillis if the parent doesn't receive a response
- * in this amount of time, or
- * if the forked doesn't receive a ping
- * in this amount of time, restart the forked process
+ * How often to check to see that a task has timed out
+ * @return
*/
- public void setPingTimeoutMillis(long pingTimeoutMillis) {
- this.pingTimeoutMillis = pingTimeoutMillis;
+ public long getTaskPulseMillis() {
+ return taskPulseMillis;
}
- public long getPingPulseMillis() {
- return pingPulseMillis;
- }
/**
*
- * @param pingPulseMillis how often to test that the parent and/or forked is alive
+ * @param taskTimeoutMillis number of milliseconds to allow per task
+ * (parse, detection, unzipping, etc.)
*/
- public void setPingPulseMillis(long pingPulseMillis) {
- this.pingPulseMillis = pingPulseMillis;
+ public void setTaskTimeoutMillis(long taskTimeoutMillis) {
+ this.taskTimeoutMillis = taskTimeoutMillis;
}
public int getMaxRestarts() {
@@ -330,7 +304,8 @@ public class TikaServerConfig {
}
public List<String> getForkedJvmArgs() {
- return forkedJvmArgs;
+ //defensively copy
+ return new ArrayList<>(forkedJvmArgs);
}
public String getTempFilePrefix() {
@@ -515,8 +490,7 @@ public class TikaServerConfig {
"maxRestarts=" + maxRestarts +
", maxFiles=" + maxFiles +
", taskTimeoutMillis=" + taskTimeoutMillis +
- ", pingTimeoutMillis=" + pingTimeoutMillis +
- ", pingPulseMillis=" + pingPulseMillis +
+ ", taskPulseMillis=" + taskPulseMillis +
", maxforkedStartupMillis=" + maxforkedStartupMillis +
", enableUnsecureFeatures=" + enableUnsecureFeatures +
", cors='" + cors + '\'' +
diff --git a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TikaServerProcess.java b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TikaServerProcess.java
index a9afecd..e2dc773 100644
--- a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TikaServerProcess.java
+++ b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TikaServerProcess.java
@@ -29,6 +29,7 @@ import org.apache.cxf.jaxrs.JAXRSServerFactoryBean;
import org.apache.cxf.jaxrs.lifecycle.ResourceProvider;
import org.apache.cxf.jaxrs.lifecycle.SingletonResourceProvider;
import org.apache.cxf.rs.security.cors.CrossOriginResourceSharingFilter;
+import org.apache.cxf.service.factory.ServiceConstructionException;
import org.apache.cxf.transport.common.gzip.GZIPInInterceptor;
import org.apache.cxf.transport.common.gzip.GZIPOutInterceptor;
import org.apache.tika.Tika;
@@ -90,14 +91,9 @@ import java.util.concurrent.TimeUnit;
public class TikaServerProcess {
- //used in fork mode -- restart after processing this many files
- private static final long DEFAULT_MAX_FILES = 100000;
- private static String ENABLE_UNSECURE_FEATURES = "enableUnsecureFeatures";
-
- private static final int DEFAULT_DIGEST_MARK_LIMIT = 20 * 1024 * 1024;
public static final Set<String> LOG_LEVELS = new HashSet<>(Arrays.asList("debug", "info"));
private static final Logger LOG = LoggerFactory.getLogger(TikaServerProcess.class);
-
+ public static int DO_NOT_RESTART_EXIT_VALUE = -100;
private static Options getOptions() {
Options options = new Options();
@@ -126,7 +122,6 @@ public class TikaServerProcess {
LOG.debug("forked config: {}", tikaServerConfig);
mainLoop(tikaServerConfig);
} catch (Exception e) {
- e.printStackTrace();
LOG.error("Can't start: ", e);
System.exit(-1);
}
@@ -153,8 +148,14 @@ public class TikaServerProcess {
executorCompletionService.submit(new AsyncParser(asyncFetchEmitQueue, asyncEmitData));
}
}
- //start the server
- Server server = serverDetails.sf.create();
+ try {
+ //start the server
+ Server server = serverDetails.sf.create();
+ } catch (ServiceConstructionException e) {
+ LOG.warn("exception starting server", e);
+ System.exit(DO_NOT_RESTART_EXIT_VALUE);
+ }
+
LOG.info("Started Apache Tika server {} at {}",
serverDetails.serverId,
serverDetails.url);
@@ -162,7 +163,7 @@ public class TikaServerProcess {
while (true) {
Future<Integer> future = executorCompletionService.poll(1, TimeUnit.MINUTES);
if (future != null) {
- LOG.warn("future val: " + future.get());
+ LOG.warn("Daemon should not stop: " + future.get());
}
}
}
diff --git a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TikaServerWatchDog.java b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TikaServerWatchDog.java
index ddddd88..4c4337c 100644
--- a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TikaServerWatchDog.java
+++ b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TikaServerWatchDog.java
@@ -17,7 +17,10 @@
package org.apache.tika.server.core;
+import org.apache.tika.config.TikaConfig;
+import org.apache.tika.exception.TikaException;
import org.apache.tika.utils.ProcessUtils;
+import org.apache.tika.utils.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -27,6 +30,8 @@ import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.PrintStream;
+import java.net.URL;
+import java.net.URLClassLoader;
import java.nio.Buffer;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
@@ -39,21 +44,21 @@ import java.nio.file.Paths;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.List;
-import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import static java.nio.file.StandardOpenOption.READ;
import static java.nio.file.StandardOpenOption.WRITE;
+import static org.apache.tika.server.core.TikaServerCli.TIKA_SERVER_ID_ENV;
public class TikaServerWatchDog implements Callable<WatchDogResult> {
private enum FORKED_STATUS {
INITIALIZING,
RUNNING,
- SHUTTING_DOWN
+ SHUTTING_DOWN,
+ FAILED_COMMUNICATION
}
private static final Logger LOG = LoggerFactory.getLogger(TikaServerWatchDog.class);
@@ -73,89 +78,76 @@ public class TikaServerWatchDog implements Callable<WatchDogResult> {
TikaServerWatchDog(int port, String id,
TikaServerConfig tikaServerConfig) {
this.port = port;
- this.id = id;
+ this.id = id;
this.tikaServerConfig = tikaServerConfig;
}
@Override
public WatchDogResult call() throws Exception {
- while (true) {
- if (tikaServerConfig.getMaxRestarts() > 0 && restarts >=
- tikaServerConfig.getMaxRestarts()) {
- LOG.warn("hit max restarts ({}). Ending processing for {} {}",
- restarts, id, port);
- return new WatchDogResult(port, id, restarts);
- }
-
- try {
- forkedProcess = new ForkedProcess(restarts++);
- setForkedStatus(FORKED_STATUS.RUNNING);
- startPingTimer();
- while (forkedProcess.ping()) {
- Thread.sleep(tikaServerConfig.getPingPulseMillis());
- }
- } catch (InterruptedException e) {
- return new WatchDogResult(port, id,restarts);
- } finally {
- setForkedStatus(FORKED_STATUS.SHUTTING_DOWN);
- LOG.debug("about to shutdown");
- if (forkedProcess != null) {
- LOG.info("about to shutdown process");
- forkedProcess.close();
+ boolean mustRestart = true;
+ try {
+ while (true) {
+ if (tikaServerConfig.getMaxRestarts() > 0
+ && restarts >= tikaServerConfig.getMaxRestarts()) {
+ LOG.warn("hit max restarts ({}). Ending processing for {} {}",
+ restarts, id, port);
+ return new WatchDogResult(port, id, restarts);
}
- }
- }
- }
- public void shutDown() {
- shutDown = true;
- }
-
- private void startPingTimer() {
- //if the forked thread is in stop-the-world mode, and isn't
- //reading the ping, this thread checks to make sure
- //that the parent ping is sent often enough.
- //The write() in ping() could block.
- //If there isn't a successful ping often enough,
- //this force destroys the forked process.
- Thread pingTimer = new Thread(new Runnable() {
- @Override
- public void run() {
- while (true) {
- long tmpLastPing = -1L;
- synchronized (forkedStatusLock) {
- if (forkedStatus == FORKED_STATUS.RUNNING) {
- tmpLastPing = lastPing.toEpochMilli();
- }
+ try {
+ if (mustRestart) {
+ forkedProcess = new ForkedProcess(restarts++);
+ setForkedStatus(FORKED_STATUS.RUNNING);
+ mustRestart = false;
}
- if (tmpLastPing > 0) {
- long elapsed = Duration.between(Instant.ofEpochMilli(tmpLastPing), Instant.now()).toMillis();
- if (elapsed > tikaServerConfig.getPingTimeoutMillis()) {
- Process processToDestroy = null;
- try {
- processToDestroy = forkedProcess.process;
- LOG.warn("{} ms have elapsed since last successful ping. Destroying forked now",
+ boolean exited = forkedProcess.process.waitFor(tikaServerConfig.getTaskPulseMillis(),
+ TimeUnit.MILLISECONDS);
+ if (exited) {
+ LOG.info("forked process exited with exit value {}",
+ forkedProcess.process.exitValue());
+ forkedProcess.close();
+ mustRestart = true;
+ } else {
+ ForkedStatus status = forkedProcess.readStatus();
+ if (status.status == FORKED_STATUS.FAILED_COMMUNICATION.ordinal()) {
+ LOG.info("failed to read from status file. Restarting now.");
+ forkedProcess.close();
+ mustRestart = true;
+ } else if (status.status == FORKED_STATUS.SHUTTING_DOWN.ordinal()) {
+ LOG.info("Forked process is in shutting down mode. Will wait a bit");
+ forkedProcess.process.waitFor(tikaServerConfig.getTaskTimeoutMillis(),
+ TimeUnit.MILLISECONDS);
+ forkedProcess.close();
+ mustRestart = true;
+ } else {
+ Long elapsed = Duration.between(Instant.ofEpochMilli(status.timestamp), Instant.now()).toMillis();
+ if (elapsed > tikaServerConfig.getTaskTimeoutMillis()) {
+ LOG.info("{} ms have elapsed since forked process last updated status. " +
+ "Shutting down and restarting.",
elapsed);
- destroyForkedForcibly(processToDestroy);
forkedProcess.close();
- } catch (NullPointerException e) {
- //ignore
+ mustRestart = true;
}
}
}
- try {
- Thread.sleep(tikaServerConfig.getPingPulseMillis());
- } catch (InterruptedException e) {
- //swallow
- }
+ } catch (InterruptedException e) {
+ return new WatchDogResult(port, id, restarts);
}
}
+ } finally {
+ setForkedStatus(FORKED_STATUS.SHUTTING_DOWN);
+ LOG.debug("about to shutdown");
+ if (forkedProcess != null) {
+ LOG.info("about to shutdown process");
+ forkedProcess.close();
+ }
}
- );
- pingTimer.setDaemon(true);
- pingTimer.start();
+ }
+
+ public void shutDown() {
+ shutDown = true;
}
private void setForkedStatus(FORKED_STATUS status) {
@@ -168,7 +160,7 @@ public class TikaServerWatchDog implements Callable<WatchDogResult> {
private Thread SHUTDOWN_HOOK = null;
private final Process process;
- private final DataOutputStream toForked;
+// private final DataOutputStream toForked;
private final Path forkedStatusFile;
private final ByteBuffer statusBuffer = ByteBuffer.allocate(16);
@@ -195,7 +187,7 @@ public class TikaServerWatchDog implements Callable<WatchDogResult> {
if (elapsed > tikaServerConfig.getMaxForkedStartupMillis()) {
close();
- throw new RuntimeException("Forked process failed to start after "+elapsed + " (ms)");
+ throw new RuntimeException("Forked process failed to start after " + elapsed + " (ms)");
}
if (!process.isAlive()) {
close();
@@ -206,51 +198,7 @@ public class TikaServerWatchDog implements Callable<WatchDogResult> {
throw new RuntimeException("Failed to start forked process -- forked status file does not exist");
}
- this.toForked = new DataOutputStream(process.getOutputStream());
- lastPing = Instant.now();
- }
-
- public boolean ping() {
- if (!process.isAlive()) {
- LOG.debug("process is not alive");
- return false;
- }
- try {
- toForked.writeByte(ServerStatus.DIRECTIVES.PING.getByte());
- toForked.flush();
- } catch (Exception e) {
- LOG.warn("Exception pinging forked process", e);
- return false;
- }
- ForkedStatus forkedStatus = null;
- try {
- forkedStatus = readStatus();
- } catch (Exception e) {
- LOG.warn("Exception reading status from forked", e);
- return false;
- }
-
- if (forkedStatus.status != ServerStatus.STATUS.OPERATING.getInt()) {
- LOG.warn("Received non-operating status from forked: {}",
- ServerStatus.STATUS.lookup(forkedStatus.status));
- return false;
- }
-
- long elapsedSinceLastUpdate =
- Duration.between(Instant.ofEpochMilli(forkedStatus.timestamp), Instant.now()).toMillis();
- LOG.debug("last update: {}, elapsed:{}, status:{}", forkedStatus.timestamp, elapsedSinceLastUpdate,
- forkedStatus.status);
-
- if (elapsedSinceLastUpdate >
- tikaServerConfig.getPingTimeoutMillis()) {
- //forked hasn't written a status update in a longer time than allowed
- LOG.warn("Forked's last update exceeded ping timeout: {} (ms) with status {}",
- elapsedSinceLastUpdate, forkedStatus.status);
- return false;
- }
-
lastPing = Instant.now();
- return true;
}
private ForkedStatus readStatus() throws Exception {
@@ -259,10 +207,10 @@ public class TikaServerWatchDog implements Callable<WatchDogResult> {
//only reading, but need to include write to allow for locking
try (FileChannel fc = FileChannel.open(forkedStatusFile, READ, WRITE)) {
- while (elapsed < tikaServerConfig.getPingTimeoutMillis()) {
+ while (elapsed < tikaServerConfig.getTaskTimeoutMillis()) {
try (FileLock lock = fc.tryLock(0, 16, true)) {
if (lock != null) {
- ((Buffer)statusBuffer).position(0);
+ ((Buffer) statusBuffer).position(0);
fc.read(statusBuffer);
long timestamp = statusBuffer.getLong(0);
int status = statusBuffer.getInt(8);
@@ -272,45 +220,42 @@ public class TikaServerWatchDog implements Callable<WatchDogResult> {
} catch (OverlappingFileLockException e) {
//swallow
}
+ Thread.sleep(100);
elapsed = Duration.between(started, Instant.now()).toMillis();
}
}
- throw new RuntimeException("couldn't read from status file after "+elapsed +" millis");
+ return new ForkedStatus(-1, FORKED_STATUS.FAILED_COMMUNICATION.ordinal(), -1);
}
- private void close() {
- try {
- if (toForked != null) {
- toForked.writeByte(ServerStatus.DIRECTIVES.SHUTDOWN.getByte());
- toForked.flush();
- }
- } catch (IOException e) {
- LOG.debug("Exception asking forked process to shutdown", e);
- }
+ private void close() throws DoNotRestartException {
try {
- if (toForked != null) {
- toForked.close();
- }
- } catch (IOException e) {
- LOG.debug("Problem shutting down writer to forked process", e);
- }
- destroyForkedForcibly(process);
+ if (!process.isAlive()) {
+ try {
+ int exit = process.exitValue();
+ if (exit == TikaServerProcess.DO_NOT_RESTART_EXIT_VALUE) {
+ throw new DoNotRestartException("Forked exited with: " + exit);
+ }
+ } catch (IllegalThreadStateException e) {
- if (forkedStatusFile != null) {
- try {
- if (Files.isRegularFile(forkedStatusFile)) {
- Files.delete(forkedStatusFile);
}
- LOG.debug("deleted "+forkedStatusFile);
- } catch (IOException e) {
- LOG.warn("problem deleting forked process status file", e);
+ }
+ destroyForkedForcibly(process);
+ } finally {
+ if (forkedStatusFile != null) {
+ try {
+ if (Files.isRegularFile(forkedStatusFile)) {
+ Files.delete(forkedStatusFile);
+ }
+ LOG.debug("deleted " + forkedStatusFile);
+ } catch (IOException e) {
+ LOG.warn("problem deleting forked process status file", e);
+ }
}
}
}
private Process startProcess(int numRestarts, Path forkedStatusFile) throws IOException {
-
ProcessBuilder builder = new ProcessBuilder();
builder.redirectError(ProcessBuilder.Redirect.INHERIT);
@@ -318,31 +263,51 @@ public class TikaServerWatchDog implements Callable<WatchDogResult> {
String javaPath = tikaServerConfig.getJavaPath();
List<String> jvmArgs = tikaServerConfig.getForkedJvmArgs();
List<String> forkedArgs = tikaServerConfig.getForkedProcessArgs(port, id);
-
forkedArgs.add("-forkedStatusFile");
forkedArgs.add(ProcessUtils.escapeCommandLine(forkedStatusFile.toAbsolutePath().toString()));
argList.add(javaPath);
- if (! jvmArgs.contains("-cp") && ! jvmArgs.contains("--classpath")) {
+ if (!jvmArgs.contains("-cp") && !jvmArgs.contains("--classpath")) {
String cp = System.getProperty("java.class.path");
jvmArgs.add("-cp");
jvmArgs.add(cp);
}
+ //this is mostly for log4j 1.x so that different processes
+ //can log to different log files
+ jvmArgs.add("-Dtika.server.id=" + tikaServerConfig.getId());
argList.addAll(jvmArgs);
+
argList.add("org.apache.tika.server.core.TikaServerProcess");
argList.addAll(forkedArgs);
+
argList.add("-numRestarts");
argList.add(Integer.toString(numRestarts));
- LOG.debug("forked process commandline: " +argList.toString());
+ LOG.debug("forked process commandline: " + argList.toString());
builder.command(argList);
+ //this is mostly for log4j 2.x so that different processes
+ //can log to different log files via {env:tika.server.id}
+ builder.environment().put(TIKA_SERVER_ID_ENV, id);
+ //pass through from forking process
+ String tikaConfigEnv = System.getenv("TIKA_CONFIG");
+ if (tikaConfigEnv != null) {
+ builder.environment().put("TIKA_CONFIG", tikaConfigEnv);
+ }
Process process = builder.start();
//redirect stdout to parent stderr to avoid error msgs
//from maven during build: Corrupted STDOUT by directly writing to native stream in forked
redirectIO(process.getInputStream(), System.err);
+ redirectIO(process.getErrorStream(), System.err);
if (SHUTDOWN_HOOK != null) {
Runtime.getRuntime().removeShutdownHook(SHUTDOWN_HOOK);
}
- SHUTDOWN_HOOK = new Thread(() -> this.close());
+ SHUTDOWN_HOOK = new Thread(() -> {
+ try {
+ this.close();
+ } catch (DoNotRestartException e) {
+ LOG.error("should not restart", e);
+ shutDown();
+ }
+ });
Runtime.getRuntime().addShutdownHook(SHUTDOWN_HOOK);
return process;
@@ -373,9 +338,9 @@ public class TikaServerWatchDog implements Callable<WatchDogResult> {
process = process.destroyForcibly();
try {
boolean destroyed = process.waitFor(60, TimeUnit.SECONDS);
- if (! destroyed) {
+ if (!destroyed) {
LOG.error("Forked process still alive after 60 seconds. " +
- "Shutting down the parent.");
+ "Shutting down the forking process.");
System.exit(1);
}
} catch (InterruptedException e) {
@@ -404,4 +369,14 @@ public class TikaServerWatchDog implements Callable<WatchDogResult> {
}
}
+ private static class DoNotRestartException extends TikaException {
+
+ public DoNotRestartException(String msg) {
+ super(msg);
+ }
+
+ public DoNotRestartException(String msg, Throwable cause) {
+ super(msg, cause);
+ }
+ }
}
diff --git a/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/IntegrationTestBase.java b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/IntegrationTestBase.java
index fff4c0e..36bd53b 100644
--- a/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/IntegrationTestBase.java
+++ b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/IntegrationTestBase.java
@@ -127,8 +127,9 @@ public class IntegrationTestBase extends TikaTest {
List<String> args = new ArrayList<>(Arrays.asList(base));
args.addAll(Arrays.asList(extraArgs));
ProcessBuilder pb = new ProcessBuilder(args);
- pb.redirectInput(Files.createTempFile(STREAMS_DIR, "tika-stream-out", ".log").toFile());
- pb.redirectError(Files.createTempFile(STREAMS_DIR, "tika-stream-err", ".log").toFile());
+ pb.inheritIO();
+// pb.redirectInput(Files.createTempFile(STREAMS_DIR, "tika-stream-out", ".log").toFile());
+ // pb.redirectError(Files.createTempFile(STREAMS_DIR, "tika-stream-err", ".log").toFile());
return pb.start();
}
@@ -151,7 +152,7 @@ public class IntegrationTestBase extends TikaTest {
LOG.debug("tika test client failed to connect to server", e);
}
- Thread.sleep(100);
+ Thread.sleep(1000);
elapsed = Duration.between(started, Instant.now()).toMillis();
}
throw new TimeoutException("couldn't connect to server after " +
diff --git a/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerAsyncIntegrationTest.java b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerAsyncIntegrationTest.java
index 580cf81..fc30eab 100644
--- a/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerAsyncIntegrationTest.java
+++ b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerAsyncIntegrationTest.java
@@ -113,6 +113,7 @@ public class TikaServerAsyncIntegrationTest extends IntegrationTestBase {
"</params>"+
"</emitter>"+
"</emitters>"+
+ "<server><enableUnsecureFeatures>true</enableUnsecureFeatures></server>"+
"</properties>";
FileUtils.write(TIKA_CONFIG.toFile(), TIKA_CONFIG_XML, UTF_8);
@@ -144,10 +145,7 @@ public class TikaServerAsyncIntegrationTest extends IntegrationTestBase {
public void run() {
TikaServerCli.main(
new String[]{
- "-enableUnsecureFeatures",
- "-maxFiles", "10000",
"-p", INTEGRATION_TEST_PORT,
- "-tmpFilePrefix", "basic-",
"-config", TIKA_CONFIG.toAbsolutePath().toString()
});
}
diff --git a/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerEmitterIntegrationTest.java b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerEmitterIntegrationTest.java
index 15dfea6..913465d 100644
--- a/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerEmitterIntegrationTest.java
+++ b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerEmitterIntegrationTest.java
@@ -121,6 +121,7 @@ public class TikaServerEmitterIntegrationTest extends IntegrationTestBase {
FileUtils.write(TIKA_CONFIG.toFile(), tikaConfigXML, UTF_8);
String tikaConfigTimeoutXML = xml1+
+ "<taskPulseMillis>100</taskPulseMillis>"+
"<taskTimeoutMillis>10000</taskTimeoutMillis>"+xml2;
FileUtils.write(TIKA_CONFIG_TIMEOUT.toFile(), tikaConfigTimeoutXML, UTF_8);
@@ -243,7 +244,6 @@ public class TikaServerEmitterIntegrationTest extends IntegrationTestBase {
@Test(expected = ProcessingException.class)
public void testTimeout() throws Exception {
-
Process p = null;
try {
p = startProcess( new String[]{ "-config",
diff --git a/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerIntegrationTest.java b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerIntegrationTest.java
index 11b6f2a..514de76 100644
--- a/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerIntegrationTest.java
+++ b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerIntegrationTest.java
@@ -405,25 +405,20 @@ public class TikaServerIntegrationTest extends IntegrationTestBase {
//this isn't a real regression test yet.
//Can watch logs at least for confirmation of behavior
//TODO: convert to real test
- Thread serverThread = new Thread() {
- @Override
- public void run() {
- TikaServerCli.main(
- new String[]{
- "-maxFiles", "10",
- "-taskTimeoutMillis", "10000", "-taskPulseMillis", "500",
- "-p", INTEGRATION_TEST_PORT
- });
- }
- };
- serverThread.start();
- awaitServerStartup();
- Random r = new Random();
- for (int i = 0; i < 100; i++) {
- boolean ex = false;
- Response response = null;
- String file = TEST_HELLO_WORLD;
- try {
+
+ Process p = null;
+ try {
+ p = startProcess(new String[]{"-config",
+ getConfig("tika-config-server-timeout-10000.xml")});
+ awaitServerStartup();
+
+
+ Random r = new Random();
+ for (int i = 0; i < 100; i++) {
+ boolean ex = false;
+ Response response = null;
+ String file = TEST_HELLO_WORLD;
+ try {
if (r.nextFloat() < 0.01) {
file = TEST_SYSTEM_EXIT;
} else if (r.nextFloat() < 0.015) {
@@ -431,30 +426,33 @@ public class TikaServerIntegrationTest extends IntegrationTestBase {
} else if (r.nextFloat() < 0.02) {
file = TEST_HEAVY_HANG;
}
- response = WebClient
- .create(endPoint + META_PATH)
- .accept("application/json")
- .put(ClassLoader
- .getSystemResourceAsStream(file));
- } catch (Exception e) {
- ex = true;
- }
+ System.out.println("writing file "+i + " : " +file);
+ response = WebClient
+ .create(endPoint + META_PATH)
+ .accept("application/json")
+ .put(ClassLoader
+ .getSystemResourceAsStream(file));
+ } catch (Exception e) {
+ ex = true;
+ }
- if (ex || response.getStatus() != 200) {
- i--;
- awaitServerStartup();
- continue;
- }
- if (file.equals(TEST_HELLO_WORLD)) {
- Reader reader = new InputStreamReader((InputStream) response.getEntity(), UTF_8);
- List<Metadata> metadataList = JsonMetadataList.fromJson(reader);
- assertEquals(1, metadataList.size());
- assertEquals("Nikolai Lobachevsky", metadataList.get(0).get("author"));
- assertContains("hello world", metadataList.get(0).get("X-TIKA:content"));
+ if (ex || response.getStatus() != 200) {
+ i--;
+ awaitServerStartup();
+ continue;
+ }
+ if (file.equals(TEST_HELLO_WORLD)) {
+ Reader reader = new InputStreamReader((InputStream) response.getEntity(), UTF_8);
+ List<Metadata> metadataList = JsonMetadataList.fromJson(reader);
+ assertEquals(1, metadataList.size());
+ assertEquals("Nikolai Lobachevsky", metadataList.get(0).get("author"));
+ assertContains("hello world", metadataList.get(0).get("X-TIKA:content"));
+ }
+ //assertEquals("a38e6c7b38541af87148dee9634cb811", metadataList.get(10).get("X-TIKA:digest:MD5"));
}
- //assertEquals("a38e6c7b38541af87148dee9634cb811", metadataList.get(10).get("X-TIKA:digest:MD5"));
+ } finally {
+ p.destroyForcibly();
}
- serverThread.interrupt();
}
diff --git a/tika-server/tika-server-core/src/test/resources/configs/tika-config-server-emitter.xml b/tika-server/tika-server-core/src/test/resources/configs/tika-config-server-emitter.xml
index 9f30f20..7cf0a29 100644
--- a/tika-server/tika-server-core/src/test/resources/configs/tika-config-server-emitter.xml
+++ b/tika-server/tika-server-core/src/test/resources/configs/tika-config-server-emitter.xml
@@ -23,7 +23,7 @@
<forkedJVMArgs>
<arg>-Xmx512m</arg>
</forkedJVMArgs>
- <enableUnsecure
+ <enableUnsecure>true</enableUnsecure>
<endpoints>
<endpoint>emit</endpoint>
<endpoint>async</endpoint>
diff --git a/tika-server/tika-server-core/src/test/resources/configs/tika-config-server-timeout-10000.xml b/tika-server/tika-server-core/src/test/resources/configs/tika-config-server-timeout-10000.xml
index 6ff6280..6d59747 100644
--- a/tika-server/tika-server-core/src/test/resources/configs/tika-config-server-timeout-10000.xml
+++ b/tika-server/tika-server-core/src/test/resources/configs/tika-config-server-timeout-10000.xml
@@ -19,8 +19,8 @@
<server>
<port>9999</port>
<taskTimeoutMillis>10000</taskTimeoutMillis>
- <pingTimeoutMillis>30000</pingTimeoutMillis>
- <pingPulseMillis>100</pingPulseMillis>
+ <taskPulseMillis>100</taskPulseMillis>
+ <maxFiles>20</maxFiles>
<forkedJVMArgs>
<arg>-Xmx512m</arg>
</forkedJVMArgs>
diff --git a/tika-server/tika-server-core/src/test/resources/configs/tika-config-server.xml b/tika-server/tika-server-core/src/test/resources/configs/tika-config-server.xml
index 50a751e..0ccbb75 100644
--- a/tika-server/tika-server-core/src/test/resources/configs/tika-config-server.xml
+++ b/tika-server/tika-server-core/src/test/resources/configs/tika-config-server.xml
@@ -17,8 +17,10 @@
-->
<properties>
<server>
+ <port>9999</port>
<taskTimeoutMillis>54321</taskTimeoutMillis>
<enableUnsecureFeatures>true</enableUnsecureFeatures>
+ <maxFiles>20</maxFiles>
<forkedJVMArgs>
<arg>-Xmx2g</arg>
</forkedJVMArgs>