This is an automated email from the ASF dual-hosted git repository.
tballison 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 d016f8aaed TIKA-4839 - add micrometer to tika-server (#3064)
d016f8aaed is described below
commit d016f8aaedd1f265e0dabb30994daef0b5194b71
Author: Tim Allison <[email protected]>
AuthorDate: Wed Aug 26 19:45:04 2026 -0400
TIKA-4839 - add micrometer to tika-server (#3064)
---
CHANGES.txt | 2 +
docs/modules/ROOT/nav.adoc | 1 +
.../ROOT/pages/using-tika/server/index.adoc | 16 +-
.../ROOT/pages/using-tika/server/monitoring.adoc | 207 ++++++++++++++++
tika-parent/pom.xml | 16 ++
.../tika/pipes/core/PerClientServerManager.java | 50 +++-
.../org/apache/tika/pipes/core/PipesClient.java | 24 +-
.../org/apache/tika/pipes/core/PipesParser.java | 14 ++
.../org/apache/tika/pipes/core/RestartCounter.java | 105 ++++++++
.../org/apache/tika/pipes/core/RestartReason.java | 36 +++
.../org/apache/tika/pipes/core/ServerManager.java | 76 ++++--
.../tika/pipes/core/SharedServerManager.java | 103 +++++---
.../tika/pipes/core/async/AsyncProcessor.java | 11 +
.../apache/tika/pipes/core/server/PipesServer.java | 6 +-
.../apache/tika/pipes/core/RestartCounterTest.java | 125 ++++++++++
.../pipes/core/ServerManagerMarkContractTest.java | 150 ++++++++++++
.../tika/pipes/core/SharedServerModeTest.java | 26 ++
tika-server/tika-server-core/pom.xml | 12 +
.../org/apache/tika/server/core/ServerStatus.java | 5 +
.../org/apache/tika/server/core/TikaServerCli.java | 3 +
.../apache/tika/server/core/TikaServerConfig.java | 29 +++
.../apache/tika/server/core/TikaServerProcess.java | 69 +++++-
.../tika/server/core/metrics/MetricsServer.java | 113 +++++++++
.../server/core/metrics/TikaMetricsFilter.java | 95 ++++++++
.../server/core/metrics/TikaServerMetrics.java | 204 ++++++++++++++++
.../tika/server/core/resource/AsyncResource.java | 4 +
.../tika/server/core/TikaServerConfigTest.java | 56 +++++
.../core/TikaServerMetricsIntegrationTest.java | 247 +++++++++++++++++++
.../server/core/metrics/MetricsServerTest.java | 96 ++++++++
.../server/core/metrics/RejectedReasonTest.java | 65 +++++
.../server/core/metrics/TikaMetricsFilterTest.java | 269 +++++++++++++++++++++
.../configs/tika-config-server-async-metrics.json | 24 ++
.../tika-config-server-metrics-restarts.json | 18 ++
.../configs/tika-config-server-metrics-shared.json | 17 ++
.../tika-config-server-metrics-timeout.json | 21 ++
.../configs/tika-config-server-metrics.json | 10 +
.../src/main/appended-resources/META-INF/LICENSE | 84 +++++++
37 files changed, 2327 insertions(+), 82 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index b7514b3cc7..d517042ccb 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,5 +1,7 @@
Release 4.1.0 - unreleased
+ * Add Micrometer reporting and opt-in endpoint for tika-server (TIKA-4839).
+
* Improve spooling/decrease number of spills to disk (TIKA-4835).
* Fixed a bug that made per-request (parse-context) configuration unusable
diff --git a/docs/modules/ROOT/nav.adoc b/docs/modules/ROOT/nav.adoc
index fe2b5105f7..fe609b2e9a 100644
--- a/docs/modules/ROOT/nav.adoc
+++ b/docs/modules/ROOT/nav.adoc
@@ -20,6 +20,7 @@
** xref:using-tika/cli/index.adoc[Command Line]
** xref:using-tika/server/index.adoc[Tika Server]
*** xref:using-tika/server/tls.adoc[TLS/SSL Configuration]
+*** xref:using-tika/server/monitoring.adoc[Monitoring with Prometheus]
** xref:using-tika/grpc/index.adoc[gRPC]
** xref:using-tika/docker.adoc[Running Tika in Docker]
* xref:pipes/index.adoc[Pipes]
diff --git a/docs/modules/ROOT/pages/using-tika/server/index.adoc
b/docs/modules/ROOT/pages/using-tika/server/index.adoc
index a156fa187f..4d291ad840 100644
--- a/docs/modules/ROOT/pages/using-tika/server/index.adoc
+++ b/docs/modules/ROOT/pages/using-tika/server/index.adoc
@@ -56,6 +56,9 @@ moment" apart from "this document broke a fork" and act
accordingly: back off an
their fetch rate, and autoscalers have a signal to scale on — none of which is
possible against a
server that can only get slower or fail blind.
+The same signal is available as metrics -- busy workers, `429` rate, worker
restarts by reason --
+when a metrics port is set; see
xref:using-tika/server/monitoring.adoc[Monitoring with Prometheus].
+
The cost is real and lands on you: backpressure only reports the capacity you
configured.
`numClients` is now an operational obligation, sized against both your request
volume and your
host's core count, as the note above says — undersized shows up as `429`s
under load that should
@@ -206,11 +209,15 @@ The server starts on `localhost:9998` by default.
|`-i <id>`, `--id <id>`
|Server ID, written to the startup log. Defaults to a random UUID.
+|`--metricsPort <port>`
+|Enable Prometheus metrics on this port. See
+xref:using-tika/server/monitoring.adoc[Monitoring with Prometheus].
+
|`-?`, `--help`
|Print the usage message.
|===
-NOTE: `-h`, `-p` and `-i` override the JSON config. Everything else —
`allowPipes`,
+NOTE: `-h`, `-p`, `-i` and `--metricsPort` override the JSON config.
Everything else — `allowPipes`,
`allowPerRequestConfig`, CORS, TLS, timeouts — is JSON-only.
== Endpoints
@@ -516,6 +523,11 @@ changes no other log level.
|`tlsConfig`
|_TLS off_
|Nested TLS/mTLS settings. See xref:using-tika/server/tls.adoc[TLS/SSL
Configuration].
+
+|`metricsPort`
+|_metrics off_
+|Serve Prometheus metrics on this port (same setting as `--metricsPort`). See
+xref:using-tika/server/monitoring.adoc[Monitoring with Prometheus].
|===
NOTE: Digests are configured in `parse-context`, not in `server`. See
@@ -574,5 +586,7 @@ explicitly with the combined total in mind).
== Topics
* xref:using-tika/server/tls.adoc[TLS/SSL Configuration] — TLS and mutual
authentication
+* xref:using-tika/server/monitoring.adoc[Monitoring with Prometheus] — metrics
for
+autoscaling and alerting
* xref:migration-to-4x/migrating-tika-server-4x.adoc[Migrating Tika Server to
4.x] — breaking
changes from 3.x
diff --git a/docs/modules/ROOT/pages/using-tika/server/monitoring.adoc
b/docs/modules/ROOT/pages/using-tika/server/monitoring.adoc
new file mode 100644
index 0000000000..6374aee862
--- /dev/null
+++ b/docs/modules/ROOT/pages/using-tika/server/monitoring.adoc
@@ -0,0 +1,207 @@
+//
+// 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.
+//
+
+= Monitoring with Prometheus
+:toc:
+:toclevels: 3
+
+Tika Server can publish operational metrics in the Prometheus text exposition
format,
+via https://micrometer.io[Micrometer]. The metrics are built for two questions
an operator
+of a parse fleet actually has: _is this server saturated_ (scale on that, not
on CPU) and
+_are its forked workers dying_ (alert on that).
+
+== Enabling
+
+Metrics are off by default. Setting a metrics port turns them on; nothing else
is needed.
+
+[source,bash]
+----
+java -jar tika-server-standard-X.Y.Z.jar --metricsPort 9404
+----
+
+or in `tika-config.json`:
+
+[source,json]
+----
+{
+ "server": {
+ "port": 9998,
+ "metricsPort": 9404
+ }
+}
+----
+
+`--metricsPort` on the command line overrides the JSON value. The port must
differ from
+the server port; `0` picks a free port, which is written to the startup log.
+
+The scrape listener binds to the server's own host (`-h`), so a server started
with
+`-h 0.0.0.0` in a container exposes its metrics on the pod IP with no extra
+configuration. Labels such as a cluster or tier name belong in your scrape
configuration
+(`relabel_configs`, or the `ServiceMonitor` under the Prometheus Operator),
not in Tika.
+
+Metrics are served on a *separate port* from the parse endpoints, on purpose.
The parse
+port receives untrusted documents; the metrics port should be reachable only
by your
+scraper. Keep them apart at the network layer (a Kubernetes `NetworkPolicy`, a
security
+group). The scrape listener serves only `/metrics` (`GET` or `HEAD`) -- every
parse
+endpoint is `404` there, and `/metrics` is `404` on the parse port. It also
has its own small thread pool
+and a cap of 64 open connections, so a scrape never waits behind a slow parse.
+
+The scrape listener is plain HTTP even when the parse port uses
+xref:using-tika/server/tls.adoc[TLS].
+
+== Probes
+
+Point Kubernetes liveness and readiness probes at the parse port (for example
+`GET /version`, or `GET /status` when that endpoint is enabled), never at
`/metrics`. A
+scrape target is not a health check: the listener stays up while the parser is
failing,
+and a metrics misconfiguration must not take a healthy parser out of rotation.
+
+== Scrape configuration
+
+[source,yaml]
+----
+scrape_configs:
+ - job_name: tika-server
+ static_configs:
+ - targets: ['tika-1:9404', 'tika-2:9404']
+----
+
+For the Prometheus Operator, a `ServiceMonitor` selecting a Service that
exposes the
+metrics port works the same way.
+
+== Meters
+
+Durations are Micrometer timers exported in seconds with a fixed histogram of
twelve
+buckets (10ms, 50ms, 100ms, 250ms, 500ms, 1s, 2s, 5s, 10s, 30s, 60s, 120s), so
+`histogram_quantile` works and the series count per pod stays small. Every
label value is
+drawn from a fixed set -- endpoint names, status classes, enum names -- never
from the
+request, so cardinality cannot grow with traffic.
+
+=== HTTP (parse port)
+
+[cols="2,1,2,3"]
+|===
+|Meter |Type |Labels |Meaning
+
+|`tika_server_requests_seconds`
+|timer
+|`endpoint`, `method`, `status`
+|Every request on the parse port, timed from before routing until the response
is
+returned -- the parse is inside that window, streaming the response body to
the client is
+not. `endpoint` is the first path segment when it is one of the server's
endpoints
+(`tika`, `rmeta`, `meta`, `unpack`, `detect`, `language`, `mime`, `mime-types`,
+`detectors`, `parsers`, `version`, `status`, `pipes`, `async`), `other` for
any other
+resource, `unmatched` for a request answered before routing (a `404`).
`method` is the
+HTTP verb when it is a standard one, else `other`. `status` is the class:
`2xx`, `3xx`,
+`4xx`, `5xx`, or `other` for a status outside 200-599.
+
+|`tika_server_request_size_bytes`
+|summary
+|`endpoint`
+|Request body size as declared by the client's `Content-Length`, not bytes
actually read;
+chunked uploads have none and are not recorded. Bucketed at 1KB..1GB by
decades.
+
+|`tika_server_rejected_total`
+|counter
+|`reason`
+|Requests refused for capacity reasons, by the status the server already uses
to encode
+them: `busy_429` (the `/async` queue was full, or no fork was free within
+`maxWaitForClientMillis`), `crash_503` (the fork serving the request OOM'd,
timed out or
+crashed), `payload_413` (body over `maxRequestSizeBytes` or the IPC payload
limit).
+
+|`tika_server_tasks_active`
+|gauge
+|
+|Sync parse/detect tasks (`/tika`, `/rmeta`, `/meta`, `/detect`) in flight
right now;
+`/unpack`, `/pipes` and `/async` work is not included.
+|===
+
+=== Forked workers
+
+[cols="2,1,2,3"]
+|===
+|Meter |Type |Labels |Meaning
+
+|`tika_pipes_workers`
+|gauge
+|`pool`, `state`
+|Pipes worker slots `busy` and `idle`. Only `pool="sync"` has these: busy/idle
needs a
+borrowable client queue, which the `/async` pool does not have. Under
`useSharedServer`
+these are client slots against one forked JVM, not JVMs. `busy / (busy +
idle)` sustained
+near 1 means the server is at capacity; see
xref:pipes/cpu-sizing.adoc[Forked-JVM CPU and
+Heap Sizing] before raising `numClients`.
+
+|`tika_pipes_worker_restarts_total`
+|counter
+|`pool`, `reason`
+|Forked workers restarted, by why: `oom`, `timeout`, `crash` (any other
failure,
+including an IPC error the server could not attribute), `max_files` (routine
recycling
+after `maxFilesProcessedPerProcess`), `idle` (the worker shut itself down after
+`socketTimeoutMillis` without a request -- exit code 24 -- and was started
again on the
+next one; per-client mode only, a shared server stays up when idle),
+`connection_abandoned` (the client dropped the connection: a request was
interrupted, or a
+worker reply exceeded `maxIpcPayloadBytes`), `shutdown` (the parent asked the
worker to stop --
+usually after a failed health check prompted a reconnect -- and it exited
cleanly).
+Alert on `oom`, `timeout` and `crash`; the rest are expected.
+
+`pool` separates two independent sets of forks: `sync` serves `/tika`,
`/rmeta`, `/meta`,
+`/unpack`, `/detect` and `/pipes`; `async` serves `/async`. Each pool is sized
by its own
+`numClients` and is present only when its endpoints are. Sum over `pool`
unless you mean
+one of them specifically.
+
+|`tika_pipes_queue_depth`
+|gauge
+|`pool`
+|Tuples accepted by `/async` and not yet picked up by a worker. Only
`pool="async"`
+exists today; present only when the `async` endpoint is enabled.
+|===
+
+=== JVM and process
+
+`jvm_memory_*`, `jvm_buffer_*`, `jvm_threads_*`, `process_files_*`,
+`process_start_time_seconds` and `process_uptime_seconds`.
+
+These describe *this* JVM, which routes requests and holds the results coming
back over
+IPC. It is not where documents are parsed: that happens in forked workers this
server
+starts and restarts, and no meter on this page except `tika_pipes_*` sees
inside them.
+A near-idle heap here is not evidence of headroom.
+
+There is deliberately no GC or CPU binder. Both describe a process that does
not parse,
+and both invite that false read. For CPU that actually covers the workers, use
the
+container/node metrics your cluster already collects (cAdvisor,
node-exporter); for
+worker health use `tika_pipes_worker_restarts_total` and the saturation
signals below.
+
+== What to scale and alert on
+
+* Saturation, for an autoscaler: on a sync workload,
`tika_pipes_workers{state="busy"}` as
+ a ratio of the total and the rate of
`tika_server_rejected_total{reason="busy_429"}`.
+ On an `/async` workload use `tika_pipes_queue_depth` instead -- there is no
busy/idle
+ gauge for that pool, and the sync gauge sits at 0 while `/async` saturates.
These move
+ before latency does, which CPU does not.
+* Failure, for alerting:
+ `sum by (reason)
(rate(tika_pipes_worker_restarts_total{reason=~"oom|timeout|crash"}[5m]))`
+ -- summed over `pool`, so async workers are included -- and
+ `tika_server_rejected_total{reason="crash_503"}`. A `503` tells the client
the
+ document broke a fork; the restart counter tells you how often that is
happening.
+* Latency: `histogram_quantile(0.95, sum by (le, endpoint)
(rate(tika_server_requests_seconds_bucket[5m])))`.
+
+== Not counted
+
+An exception that no JAX-RS `ExceptionMapper` handles is answered by the
servlet
+container's own error path, which bypasses the response filter that records
+`tika_server_requests_seconds`. Tika Server maps its own parse and pipes
failures, so this
+only affects genuine server bugs.
diff --git a/tika-parent/pom.xml b/tika-parent/pom.xml
index 1fd35ecfe3..673c84cb95 100644
--- a/tika-parent/pom.xml
+++ b/tika-parent/pom.xml
@@ -404,6 +404,7 @@
<lucene.version>9.12.3</lucene.version>
<maven.plugin.annotations.version>3.15.2</maven.plugin.annotations.version>
<metadata.extractor.version>2.21.0</metadata.extractor.version>
+ <micrometer.version>1.17.1</micrometer.version>
<microsoft.translator.version>0.6.2</microsoft.translator.version>
<!-- can't update to 4 because Apache Ignite needs
io.micronaut.core.convert.DefaultConversionService
which no longer exists in 4.0:
https://stackoverflow.com/questions/79937141/ -->
@@ -593,6 +594,16 @@
<artifactId>jetty-http2-server</artifactId>
<version>${jetty.http2.version}</version>
</dependency>
+ <dependency>
+ <groupId>io.micrometer</groupId>
+ <artifactId>micrometer-core</artifactId>
+ <version>${micrometer.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>io.micrometer</groupId>
+ <artifactId>micrometer-registry-prometheus</artifactId>
+ <version>${micrometer.version}</version>
+ </dependency>
<dependency>
<groupId>org.jsoup</groupId>
<artifactId>jsoup</artifactId>
@@ -1248,6 +1259,11 @@
<artifactId>junit-jupiter-engine</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-params</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PerClientServerManager.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PerClientServerManager.java
index e5e6cc83eb..b381251d8c 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PerClientServerManager.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PerClientServerManager.java
@@ -180,6 +180,7 @@ public class PerClientServerManager implements
ServerManager {
private volatile int port = -1;
private long filesProcessed = 0;
private volatile boolean pendingRestart = false;
+ private final RestartCounter restarts = new RestartCounter();
// Set once by shutdown()/close(); guards a request thread from starting a
fresh
// process after the manager has been torn down (which would leak the
child).
private volatile boolean closed = false;
@@ -281,7 +282,7 @@ public class PerClientServerManager implements
ServerManager {
if (filesProcessed >= maxFilesPerProcess) {
LOG.info("clientId={}: reached max files limit ({}/{}), marking
for restart",
clientId, filesProcessed, maxFilesPerProcess);
- pendingRestart = true;
+ markForRestart(RestartReason.MAX_FILES);
}
}
@@ -292,18 +293,34 @@ public class PerClientServerManager implements
ServerManager {
@Override
public void markServerForRestart() {
- LOG.info("clientId={}: marking server for restart", clientId);
+ markServerForRestart(RestartReason.CRASH);
+ }
+
+ @Override
+ public void markServerForRestart(RestartReason reason) {
+ LOG.info("clientId={}: marking server for restart ({})", clientId,
reason);
+ markForRestart(reason);
+ }
+
+ private void markForRestart(RestartReason reason) {
+ restarts.mark(reason);
pendingRestart = true;
}
+ @Override
+ public long getRestartCount(RestartReason reason) {
+ return restarts.count(reason);
+ }
+
@Override
public void connectionAbandoned() {
LOG.info("clientId={}: connection abandoned, worker will be recycled
on next use", clientId);
- pendingRestart = true;
+ markForRestart(RestartReason.CONNECTION_ABANDONED);
}
@Override
public int handleCrashAndGetExitCode() {
+ // Not marked: RestartCounter attributes by exit code; the caller
refines OOM/TIMEOUT.
pendingRestart = true;
if (process != null) {
try {
@@ -311,7 +328,10 @@ public class PerClientServerManager implements
ServerManager {
if (!process.isAlive()) {
int exitValue = process.exitValue();
if (exitValue == 0) {
- LOG.info("clientId={}: process exited cleanly",
clientId);
+ LOG.warn("clientId={}: process exited with code 0
without a shutdown request; " +
+ "counted as a crash restart", clientId);
+ } else if (exitValue == PipesServer.IDLE_EXIT_CODE) {
+ LOG.info("clientId={}: process exited after idle
timeout", clientId);
} else {
LOG.warn("clientId={}: process exited with code {}",
clientId, exitValue);
}
@@ -335,7 +355,15 @@ public class PerClientServerManager implements
ServerManager {
if (isRunning() && !pendingRestart) {
return;
}
- startServer();
+ Process previous = process;
+ try {
+ startServer();
+ } finally {
+ // Count the old process once it is really gone, even if the new
start failed.
+ if (process != previous) {
+ restarts.restarted(previous);
+ }
+ }
}
@Override
@@ -366,9 +394,15 @@ public class PerClientServerManager implements
ServerManager {
}
if (!p.isAlive()) {
int exitValue = p.exitValue();
- LOG.error("clientId={}: Process exited with code {} before
connecting to socket",
- clientId, exitValue);
- ServerProcessIO.surfaceCrashDiagnostics(LOG, "clientId=" +
clientId, tmpDir);
+ if (exitValue == 0 || exitValue ==
PipesServer.IDLE_EXIT_CODE) {
+ // Idle/SHUT_DOWN exit raced the reconnect; not a
crash.
+ LOG.info("clientId={}: process exited with code {}
before connecting",
+ clientId, exitValue);
+ } else {
+ LOG.error("clientId={}: Process exited with code {}
before connecting to socket",
+ clientId, exitValue);
+ ServerProcessIO.surfaceCrashDiagnostics(LOG,
"clientId=" + clientId, tmpDir);
+ }
// Always treat pre-connect death as retryable.
// The only non-retryable paths are:
// 1. pb.start() fails (can't launch process) - handled in
startServer()
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesClient.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesClient.java
index c4743e1a65..77339145f5 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesClient.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesClient.java
@@ -85,8 +85,8 @@ public class PipesClient implements Closeable {
/**
* The server generation this client's connection belongs to, captured in
{@link #reconnect()}.
* Starts at MAX_VALUE so a report made before we have ever connected is
never mistaken for a
- * stale one: dropping a legitimate report can wedge the pool, while an
extra one only costs a
- * restart, so the un-connected case must fail toward reporting.
+ * stale one: dropping a legitimate mark can wedge the pool, while an
extra mark only costs a
+ * restart, so the un-connected case must fail toward marking.
*/
private volatile long connectionGeneration = Long.MAX_VALUE;
private int filesProcessed = 0;
@@ -225,11 +225,13 @@ public class PipesClient implements Closeable {
throw e;
} catch (ServerInitializationException e) {
LOG.error("server initialization failed: {} ", t.getId(), e);
+ serverManager.markServerForRestart(RestartReason.CRASH,
connectionGeneration);
closeConnection();
return buildFatalResult(t.getId(), t.getEmitKey(),
PipesResult.RESULT_STATUS.FAILED_TO_INITIALIZE,
intermediateResult.get(), e.getMessage());
} catch (SecurityException e) {
LOG.error("security exception during initialization: {} ",
t.getId());
+ serverManager.markServerForRestart(RestartReason.CRASH,
connectionGeneration);
closeConnection();
return buildFatalResult(t.getId(), t.getEmitKey(),
PipesResult.RESULT_STATUS.FAILED_TO_INITIALIZE,
intermediateResult.get());
@@ -258,6 +260,7 @@ public class PipesClient implements Closeable {
intermediateResult.get(), e.getMessage());
} catch (Exception e) {
LOG.error("exception waiting for server to complete task: {} ",
t.getId(), e);
+ serverManager.markServerForRestart(RestartReason.CRASH,
connectionGeneration);
closeConnection();
return buildFatalResult(t.getId(), t.getEmitKey(),
UNSPECIFIED_CRASH, intermediateResult.get());
}
@@ -332,8 +335,8 @@ public class PipesClient implements Closeable {
// Get port after ensureRunning - this is the port we'll connect to
int port = serverManager.getPort();
- // Captured with the port: every report we make below is about THIS
process, and must be
- // dropped if a sibling has already replaced it.
+ // Captured with the port: every report we make below is about THIS
process, and must
+ // be dropped if a sibling has already replaced it.
connectionGeneration = serverManager.getGeneration();
LOG.debug("pipesClientId={}: connecting to server", pipesClientId);
@@ -414,7 +417,7 @@ public class PipesClient implements Closeable {
LOG.warn("clientId={}: client-side backstop timeout: id={}
elapsed={}ms limit={}ms " +
"-- server should have self-terminated well
before this", pipesClientId,
t.getId(), totalElapsed, clientBackstopMillis);
- serverManager.markServerForRestart(connectionGeneration);
+ serverManager.markServerForRestart(RestartReason.TIMEOUT,
connectionGeneration);
closeConnection();
return buildFatalResult(t.getId(), t.getEmitKey(), TIMEOUT,
intermediateResult.get());
}
@@ -430,19 +433,19 @@ public class PipesClient implements Closeable {
switch (msg.type()) {
case OOM:
String oomMsg = JsonPipesIpc.fromBytes(msg.payload(),
String.class);
-
serverManager.markServerForRestart(connectionGeneration);
+ serverManager.markServerForRestart(RestartReason.OOM,
connectionGeneration);
closeConnection();
return buildFatalResult(t.getId(), t.getEmitKey(),
PipesResult.RESULT_STATUS.OOM,
intermediateResult.get(), oomMsg);
case TIMEOUT:
String timeoutMsg =
JsonPipesIpc.fromBytes(msg.payload(), String.class);
-
serverManager.markServerForRestart(connectionGeneration);
+
serverManager.markServerForRestart(RestartReason.TIMEOUT, connectionGeneration);
closeConnection();
return buildFatalResult(t.getId(), t.getEmitKey(),
TIMEOUT,
intermediateResult.get(), timeoutMsg);
case UNSPECIFIED_CRASH:
String crashMsg =
JsonPipesIpc.fromBytes(msg.payload(), String.class);
-
serverManager.markServerForRestart(connectionGeneration);
+
serverManager.markServerForRestart(RestartReason.CRASH, connectionGeneration);
closeConnection();
return buildFatalResult(t.getId(), t.getEmitKey(),
UNSPECIFIED_CRASH,
intermediateResult.get(), crashMsg);
@@ -476,13 +479,14 @@ public class PipesClient implements Closeable {
} catch (SocketTimeoutException e) {
LOG.warn("clientId={}: Socket timeout exception while waiting
for server", pipesClientId, e);
// Mark for restart - server is stuck on current request and
needs to be restarted
- serverManager.markServerForRestart(connectionGeneration);
+ serverManager.markServerForRestart(RestartReason.TIMEOUT,
connectionGeneration);
closeConnection();
return buildFatalResult(t.getId(), t.getEmitKey(), TIMEOUT,
intermediateResult.get(),
ExceptionUtils.getStackTrace(e));
} catch (PayloadLimitExceededException e) {
// Stream is desynchronized (payload bytes were not consumed);
close the connection.
LOG.warn("clientId={}: payload too large for id={}: {}",
pipesClientId, t.getId(), e.getMessage());
+ serverManager.connectionAbandoned();
closeConnection();
return buildFatalResult(t.getId(), t.getEmitKey(),
PipesResult.RESULT_STATUS.PAYLOAD_LIMIT_EXCEEDED,
@@ -496,8 +500,10 @@ public class PipesClient implements Closeable {
PipesResult.RESULT_STATUS status = UNSPECIFIED_CRASH;
if (exitCode == PipesMessageType.OOM.getExitCode().orElse(-1))
{
status = PipesResult.RESULT_STATUS.OOM;
+ serverManager.markServerForRestart(RestartReason.OOM,
connectionGeneration);
} else if (exitCode ==
PipesMessageType.TIMEOUT.getExitCode().orElse(-1)) {
status = PipesResult.RESULT_STATUS.TIMEOUT;
+ serverManager.markServerForRestart(RestartReason.TIMEOUT,
connectionGeneration);
}
closeConnection();
return buildFatalResult(t.getId(), t.getEmitKey(), status,
intermediateResult.get(),
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesParser.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesParser.java
index a03e24efe5..d154a5962e 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesParser.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesParser.java
@@ -168,6 +168,20 @@ public class PipesParser implements Closeable {
return
serverManagers.stream().filter(ServerManager::isRunning).count();
}
+ public int getNumClients() {
+ return pipesConfig.getNumClients();
+ }
+
+ /** Clients not currently borrowed by a {@link #parse} call. */
+ public int getIdleClientCount() {
+ return clientQueue.size();
+ }
+
+ /** Forked-server restarts performed so far for {@code reason}, summed
over all servers. */
+ public long getRestartCount(RestartReason reason) {
+ return serverManagers.stream().mapToLong(m ->
m.getRestartCount(reason)).sum();
+ }
+
/**
* Returns whether this parser is using shared server mode.
*
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/RestartCounter.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/RestartCounter.java
new file mode 100644
index 0000000000..3c3979a728
--- /dev/null
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/RestartCounter.java
@@ -0,0 +1,105 @@
+/*
+ * 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.pipes.core;
+
+import java.util.EnumMap;
+import java.util.concurrent.atomic.LongAdder;
+
+import org.apache.tika.pipes.core.protocol.PipesMessageType;
+import org.apache.tika.pipes.core.server.PipesServer;
+
+/**
+ * Reason marked while a restart is pending, plus counts of restarts
performed, by reason.
+ * Last mark wins; an unmarked restart is attributed by the old process's exit
code.
+ */
+final class RestartCounter {
+
+ private final EnumMap<RestartReason, LongAdder> counts = new
EnumMap<>(RestartReason.class);
+ private volatile RestartReason pending;
+
+ RestartCounter() {
+ for (RestartReason r : RestartReason.values()) {
+ counts.put(r, new LongAdder());
+ }
+ }
+
+ void mark(RestartReason reason) {
+ pending = reason;
+ }
+
+ /** Marks {@code reason} unless a more specific one is already pending. */
+ void markIfUnmarked(RestartReason reason) {
+ if (pending == null) {
+ pending = reason;
+ }
+ }
+
+ /**
+ * Records a restart of {@code previous}. Null means there was no process
to restart
+ * (first start, or the last start failed): nothing is counted and any
mark is dropped,
+ * since it referred to a restart that never happened.
+ */
+ void restarted(Process previous) {
+ if (previous == null) {
+ pending = null;
+ return;
+ }
+ restarted(previous.isAlive() ? -1 : previous.exitValue());
+ }
+
+ /** Unmarked restarts are attributed by the exit code the child chose. */
+ void restarted(int exitCode) {
+ RestartReason reason = pending;
+ pending = null;
+ if (reason == null) {
+ reason = fromExitCode(exitCode);
+ }
+ counts.get(reason).increment();
+ }
+
+ /**
+ * The child reports why it died in its exit status, and the parent may
reach a restart
+ * without ever having read the corresponding frame (the socket can break
first, or the
+ * child can outlive the parent's one-second wait). Every code the child
can deliberately
+ * choose is honoured here; only genuinely unexplained deaths fall through
to CRASH.
+ */
+ private static RestartReason fromExitCode(int exitCode) {
+ if (exitCode == PipesServer.IDLE_EXIT_CODE) {
+ return RestartReason.IDLE;
+ }
+ if (exitCode == 0) {
+ // Only PipesServer's SHUT_DOWN handler exits 0, and the parent is
the only sender:
+ // a worker we asked to stop is not a crash, however we noticed it
was gone.
+ return RestartReason.SHUTDOWN;
+ }
+ if (matches(PipesMessageType.OOM, exitCode)) {
+ return RestartReason.OOM;
+ }
+ if (matches(PipesMessageType.TIMEOUT, exitCode)) {
+ return RestartReason.TIMEOUT;
+ }
+ return RestartReason.CRASH;
+ }
+
+ private static boolean matches(PipesMessageType type, int exitCode) {
+ return type.getExitCode().orElse(Integer.MIN_VALUE) == exitCode;
+ }
+
+ long count(RestartReason reason) {
+ return counts.get(reason).sum();
+ }
+}
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/RestartReason.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/RestartReason.java
new file mode 100644
index 0000000000..e093296007
--- /dev/null
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/RestartReason.java
@@ -0,0 +1,36 @@
+/*
+ * 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.pipes.core;
+
+/**
+ * Why a forked pipes server was (or will be) restarted. Bounded on purpose:
these
+ * become metric tag values.
+ */
+public enum RestartReason {
+ OOM,
+ TIMEOUT,
+ CRASH,
+ MAX_FILES,
+ CONNECTION_ABANDONED,
+ /** The worker exited on its own (exit 24) after sitting idle for
socketTimeoutMillis. */
+ IDLE,
+ /**
+ * The worker exited cleanly (exit 0) because the parent sent it SHUT_DOWN
-- typically after a
+ * failed health check prompted a reconnect. The replacement is
deliberate, not a failure.
+ */
+ SHUTDOWN
+}
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/ServerManager.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/ServerManager.java
index b40d8c4757..f620cda3c0 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/ServerManager.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/ServerManager.java
@@ -103,11 +103,57 @@ public interface ServerManager extends Closeable {
* might still return true briefly. The next call to {@link
#ensureRunning()} will
* wait for the process to fully exit and then restart.
* <p>
- * In per-client mode, this is typically a no-op since the client owns the
server.
- * In shared mode, this is important for coordinating restarts among
multiple clients.
+ * The reason form below defaults to this one, so this must NOT default to
the reason form:
+ * an implementation overriding neither would recurse until the stack
blew. Concrete managers
+ * in tika-pipes override both, so callers of either spelling reach a real
implementation.
*/
default void markServerForRestart() {
- // Default no-op for backward compatibility
+ // Default no-op: preserves implementations written before
RestartReason existed.
+ }
+
+ /** As {@link #markServerForRestart()}, attributing the restart to {@code
reason}. Override this one. */
+ default void markServerForRestart(RestartReason reason) {
+ markServerForRestart();
+ }
+
+ /**
+ * The generation of the currently running process: a counter incremented
every time this
+ * manager forks a replacement. A client captures it when it connects and
hands it back with
+ * every report, so a report about a process that has already been
replaced can be recognised
+ * and dropped rather than being applied to its healthy successor.
+ */
+ default long getGeneration() {
+ return 0;
+ }
+
+ /**
+ * As {@link #markServerForRestart(RestartReason)}, but only if {@code
generation} is still
+ * current. Reports about a superseded process are dropped.
+ */
+ default void markServerForRestart(RestartReason reason, long generation) {
+ markServerForRestart(reason);
+ }
+
+ /**
+ * The reasonless spelling of the above, kept for callers that cannot
attribute the failure.
+ * Routed through the reason form rather than the bare no-arg default:
that default exists
+ * only to keep pre-RestartReason implementations working, and delegating
here would leave
+ * this silently inert for any implementation that overrides only the
reason form.
+ */
+ default void markServerForRestart(long generation) {
+ markServerForRestart(RestartReason.CRASH, generation);
+ }
+
+ /**
+ * As {@link #handleCrashAndGetExitCode()}, but only if {@code generation}
is still current.
+ */
+ default int handleCrashAndGetExitCode(long generation) {
+ return handleCrashAndGetExitCode();
+ }
+
+ /** Restarts performed so far for {@code reason}; monotonic, never reset.
*/
+ default long getRestartCount(RestartReason reason) {
+ return 0;
}
/**
@@ -160,30 +206,8 @@ public interface ServerManager extends Closeable {
* @return the exit code if available, or -1 if the process is still
running or unavailable
*/
default int handleCrashAndGetExitCode() {
- markServerForRestart();
+ markServerForRestart(RestartReason.CRASH);
return -1;
}
- /**
- * The generation of the currently running process: a counter incremented
every time this
- * manager forks a replacement. A client captures it when it connects and
hands it back with
- * every report, so a report about a process that has already been
replaced can be recognised
- * and dropped rather than being applied to its healthy successor.
- */
- default long getGeneration() {
- return 0;
- }
-
- /**
- * As {@link #markServerForRestart()}, but only if {@code generation} is
still current.
- * Reports about a superseded process are dropped.
- */
- default void markServerForRestart(long generation) {
- markServerForRestart();
- }
-
- /** As {@link #handleCrashAndGetExitCode()}, but only if {@code
generation} is still current. */
- default int handleCrashAndGetExitCode(long generation) {
- return handleCrashAndGetExitCode();
- }
}
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/SharedServerManager.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/SharedServerManager.java
index 93b4f03828..17cdf4f6ff 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/SharedServerManager.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/SharedServerManager.java
@@ -76,10 +76,11 @@ public class SharedServerManager implements ServerManager {
private final Object lock = new Object();
private final AtomicLong filesProcessed = new AtomicLong(0);
private volatile boolean restarting = false;
- private volatile boolean pendingRestart = false;
- private volatile long generation;
- private volatile boolean closed; // Set when a client reports fatal error
or max files reached
+ private volatile boolean pendingRestart = false; // Set when a client
reports fatal error or max files reached
+ private final RestartCounter restarts = new RestartCounter();
private volatile byte[] currentToken;
+ private volatile long generation;
+ private volatile boolean closed;
private Process process;
private Path tmpDir;
private int serverPort = -1;
@@ -147,12 +148,17 @@ public class SharedServerManager implements ServerManager
{
throw new IllegalStateException("shared server manager is
closed");
}
restarting = true;
+ Process previous = process;
try {
startServer();
generation++; // supersedes every report still in flight
against the old process
pendingRestart = false; // Clear the flag after successful
restart
filesProcessed.set(0); // Reset file counter on restart
} finally {
+ // Count the old process once it is really gone, even if the
new start failed.
+ if (process != previous) {
+ restarts.restarted(previous);
+ }
restarting = false;
lock.notifyAll(); // Wake up any threads waiting in getPort()
}
@@ -160,21 +166,53 @@ public class SharedServerManager implements ServerManager
{
}
/**
- * Marks the server for restart due to a fatal error (OOM, timeout).
- * <p>
- * This is called by clients when they receive OOM or TIMEOUT status.
- * It signals that the server process is stopping (System.exit was called),
- * even if isRunning() might still return true briefly.
- * <p>
- * The next call to ensureRunning() will wait for the process to fully
- * exit and then restart the server.
+ * Called by a client that received OOM or TIMEOUT: the process is exiting
even if
+ * isRunning() still says otherwise; the next ensureRunning() restarts it.
*/
@Override
public void markServerForRestart() {
+ markServerForRestart(RestartReason.CRASH);
+ }
+
+ @Override
+ public void markServerForRestart(RestartReason reason) {
synchronized (lock) {
- LOG.debug("Server marked for restart - will restart on next
ensureRunning()");
- pendingRestart = true;
+ LOG.debug("Server marked for restart ({}) - will restart on next
ensureRunning()", reason);
+ markForRestart(reason);
+ }
+ }
+
+ @Override
+ public void markServerForRestart(RestartReason reason, long generation) {
+ synchronized (lock) {
+ if (isSuperseded(generation, reason)) {
+ return;
+ }
+ LOG.debug("Server marked for restart ({}) - will restart on next
ensureRunning()", reason);
+ markForRestart(reason);
+ }
+ }
+
+ /**
+ * A restart kills the shared JVM out from under every sibling still
parsing on it, and
+ * {@code ensureRunning} holds {@code lock} for the whole fork, so those
siblings cannot
+ * report until after the replacement is up. Their reports describe the
process that was
+ * already destroyed and already counted; applying them would destroy the
healthy
+ * replacement and book a second restart for a single death.
+ */
+ private boolean isSuperseded(long reportedGeneration, RestartReason
reason) {
+ if (reportedGeneration >= generation) {
+ return false;
}
+ LOG.debug("dropping stale {} report for generation {}; current
generation is {}",
+ reason, reportedGeneration, generation);
+ return true;
+ }
+
+ /** Callers hold {@code lock}. */
+ private void markForRestart(RestartReason reason) {
+ restarts.mark(reason);
+ pendingRestart = true;
}
@Override
@@ -185,43 +223,32 @@ public class SharedServerManager implements ServerManager
{
}
@Override
- public void markServerForRestart(long generation) {
+ public long getRestartCount(RestartReason reason) {
+ return restarts.count(reason);
+ }
+
+ /** Another client may already have attributed this crash (OOM/TIMEOUT);
don't overwrite it. */
+ @Override
+ public int handleCrashAndGetExitCode() {
synchronized (lock) {
- if (isSuperseded(generation)) {
- return;
- }
- LOG.debug("Server marked for restart - will restart on next
ensureRunning()");
+ restarts.markIfUnmarked(RestartReason.CRASH);
pendingRestart = true;
}
+ return -1;
}
@Override
public int handleCrashAndGetExitCode(long generation) {
synchronized (lock) {
- if (isSuperseded(generation)) {
+ if (isSuperseded(generation, RestartReason.CRASH)) {
return -1;
}
+ restarts.markIfUnmarked(RestartReason.CRASH);
pendingRestart = true;
}
return -1;
}
- /**
- * A restart kills the shared JVM out from under every sibling still
parsing on it, and
- * {@code ensureRunning} holds {@code lock} for the whole fork, so those
siblings cannot
- * report until after the replacement is up. Their reports describe the
process that was
- * already destroyed; applying them would destroy the healthy replacement
too. Callers
- * hold {@code lock}.
- */
- private boolean isSuperseded(long reportedGeneration) {
- if (reportedGeneration >= generation) {
- return false;
- }
- LOG.debug("dropping stale restart report for generation {}; current
generation is {}",
- reportedGeneration, generation);
- return true;
- }
-
/**
* Increments the count of files processed and marks for restart if limit
reached.
*/
@@ -235,6 +262,9 @@ public class SharedServerManager implements ServerManager {
synchronized (lock) {
LOG.info("Shared server reached max files limit ({}/{}),
marking for restart",
count, maxFilesPerProcess);
+ // A fatal reason already recorded for this same pending
restart outranks a
+ // scheduled recycle: the boundary file may be the one that
killed the worker.
+ restarts.markIfUnmarked(RestartReason.MAX_FILES);
pendingRestart = true;
}
}
@@ -461,7 +491,8 @@ public class SharedServerManager implements ServerManager {
}
} finally {
// An interrupt here must not leave the field pointing at a
SIGKILLed process:
- // startServer() would then try to reap it again and tmpDir
would never be deleted.
+ // ensureRunning would then see process == previous and skip
counting the restart,
+ // startServer() would try to reap it again, and tmpDir would
never be deleted.
process = null;
}
}
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/async/AsyncProcessor.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/async/AsyncProcessor.java
index be61563d12..dbb8336e71 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/async/AsyncProcessor.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/async/AsyncProcessor.java
@@ -50,6 +50,7 @@ import org.apache.tika.pipes.core.PipesClient;
import org.apache.tika.pipes.core.PipesConfig;
import org.apache.tika.pipes.core.PipesException;
import org.apache.tika.pipes.core.PipesResults;
+import org.apache.tika.pipes.core.RestartReason;
import org.apache.tika.pipes.core.ServerManager;
import org.apache.tika.pipes.core.SharedServerManager;
import org.apache.tika.pipes.core.emitter.EmitterManager;
@@ -247,6 +248,16 @@ public class AsyncProcessor implements Closeable {
return fetchEmitTuples.remainingCapacity();
}
+ /** Tuples accepted but not yet picked up by a worker. */
+ public int getQueueDepth() {
+ return fetchEmitTuples.size();
+ }
+
+ /** Restarts of this processor's own forks, summed over all servers, for
{@code reason}. */
+ public long getRestartCount(RestartReason reason) {
+ return serverManagers.stream().mapToLong(m ->
m.getRestartCount(reason)).sum();
+ }
+
/**
* Long-running callers (tika-server) set true so one bad tuple
* (unknown fetcher, init failure) can't halt the shared workers;
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesServer.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesServer.java
index 3bcbe0cb0a..a92602242d 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesServer.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesServer.java
@@ -162,6 +162,10 @@ public class PipesServer implements AutoCloseable {
* tell the difference between "I crashed" and "my parent went away". */
public static final int PARENT_GONE_EXIT_CODE = 23;
+ /** Idle self-exit (no request within socketTimeoutMillis). Not 0: that
means a
+ * parent-requested SHUT_DOWN. Other codes: {@link
PipesMessageType#getExitCode()}. */
+ public static final int IDLE_EXIT_CODE = 24;
+
private final long heartbeatIntervalMillis;
private final String pipesClientId;
@@ -416,7 +420,7 @@ public class PipesServer implements AutoCloseable {
} catch (Exception ex) {
//swallow
}
- System.exit(0);
+ System.exit(IDLE_EXIT_CODE);
return; // unreachable, but needed for compilation
}
LOG.trace("received message type={}", msg.type());
diff --git
a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/RestartCounterTest.java
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/RestartCounterTest.java
new file mode 100644
index 0000000000..0bb633dbe6
--- /dev/null
+++
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/RestartCounterTest.java
@@ -0,0 +1,125 @@
+/*
+ * 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.pipes.core;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.stream.Stream;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import org.apache.tika.pipes.core.protocol.PipesMessageType;
+import org.apache.tika.pipes.core.server.PipesServer;
+
+public class RestartCounterTest {
+
+ @Test
+ public void testMarkedReasonWins() {
+ RestartCounter c = new RestartCounter();
+ c.mark(RestartReason.MAX_FILES);
+ c.restarted(PipesServer.IDLE_EXIT_CODE);
+ assertEquals(1, c.count(RestartReason.MAX_FILES));
+ assertEquals(0, c.count(RestartReason.IDLE));
+ }
+
+ @Test
+ public void testLastMarkWins() {
+ RestartCounter c = new RestartCounter();
+ c.mark(RestartReason.CRASH);
+ c.mark(RestartReason.OOM);
+ c.restarted(-1);
+ assertEquals(1, c.count(RestartReason.OOM));
+ assertEquals(0, c.count(RestartReason.CRASH));
+ }
+
+ @Test
+ public void testMarkIfUnmarkedKeepsSpecificReason() {
+ RestartCounter c = new RestartCounter();
+ c.mark(RestartReason.OOM);
+ c.markIfUnmarked(RestartReason.CRASH);
+ c.restarted(-1);
+ assertEquals(1, c.count(RestartReason.OOM));
+ c.markIfUnmarked(RestartReason.CRASH);
+ c.restarted(-1);
+ assertEquals(1, c.count(RestartReason.CRASH));
+ }
+
+ @ParameterizedTest
+ @MethodSource("exitCodes")
+ public void testUnmarkedAttributedByExitCode(int exitCode, RestartReason
expected) {
+ RestartCounter c = new RestartCounter();
+ c.restarted(exitCode);
+ for (RestartReason r : RestartReason.values()) {
+ assertEquals(r == expected ? 1 : 0, c.count(r), r.name());
+ }
+ }
+
+ /**
+ * Enumerated from the codes the child can deliberately choose ({@link
PipesMessageType} and
+ * {@link PipesServer}), not from RestartCounter's own branches: a table
derived from the
+ * implementation cannot catch a case the implementation forgot.
+ */
+ static Stream<Arguments> exitCodes() {
+ return Stream.of(
+ Arguments.of(PipesServer.IDLE_EXIT_CODE, RestartReason.IDLE),
+ Arguments.of(PipesMessageType.OOM.getExitCode().getAsInt(),
RestartReason.OOM),
+
Arguments.of(PipesMessageType.TIMEOUT.getExitCode().getAsInt(),
RestartReason.TIMEOUT),
+
Arguments.of(PipesMessageType.UNSPECIFIED_CRASH.getExitCode().getAsInt(),
RestartReason.CRASH),
+ Arguments.of(0, RestartReason.SHUTDOWN),
+ Arguments.of(1, RestartReason.CRASH),
+ Arguments.of(-1, RestartReason.CRASH));
+ }
+
+ @Test
+ public void testFatalReasonOutranksScheduledRecycle() {
+ // The file-limit boundary document may be the one that killed the
worker; a scheduled
+ // recycle must not overwrite the fatal reason recorded for the same
pending restart.
+ RestartCounter c = new RestartCounter();
+ c.mark(RestartReason.OOM);
+ c.markIfUnmarked(RestartReason.MAX_FILES);
+ c.restarted(-1);
+ assertEquals(1, c.count(RestartReason.OOM));
+ assertEquals(0, c.count(RestartReason.MAX_FILES));
+ }
+
+ @Test
+ public void testPendingClearsAfterRestart() {
+ RestartCounter c = new RestartCounter();
+ c.mark(RestartReason.TIMEOUT);
+ c.restarted(-1);
+ c.restarted(PipesServer.IDLE_EXIT_CODE);
+ assertEquals(1, c.count(RestartReason.TIMEOUT));
+ assertEquals(1, c.count(RestartReason.IDLE));
+ }
+
+ /** A mark left over from a start that never produced a process must not
bleed into the next restart. */
+ @Test
+ public void testFirstStartNotCountedAndClearsMark() {
+ RestartCounter c = new RestartCounter();
+ c.mark(RestartReason.CRASH);
+ c.restarted((Process) null);
+ for (RestartReason r : RestartReason.values()) {
+ assertEquals(0, c.count(r), r.name());
+ }
+ c.restarted(PipesServer.IDLE_EXIT_CODE);
+ assertEquals(1, c.count(RestartReason.IDLE));
+ assertEquals(0, c.count(RestartReason.CRASH));
+ }
+}
diff --git
a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/ServerManagerMarkContractTest.java
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/ServerManagerMarkContractTest.java
new file mode 100644
index 0000000000..72bd63e7df
--- /dev/null
+++
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/ServerManagerMarkContractTest.java
@@ -0,0 +1,150 @@
+/*
+ * 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.pipes.core;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.net.Socket;
+import java.nio.file.Path;
+
+import org.junit.jupiter.api.Test;
+
+/**
+ * {@link ServerManager} offers two spellings of "recycle this worker" and
every in-repo caller
+ * uses the reason form, so a regression in the no-arg form is invisible to
every other test.
+ * These pin the contract from the <em>caller's</em> side: whichever spelling
a downstream
+ * integration picked up, the worker must actually be marked.
+ */
+public class ServerManagerMarkContractTest {
+
+ /** Overrides only the no-arg form -- what a pre-RestartReason
implementation would have. */
+ private static class NoArgOnly implements ServerManager {
+ private int marks;
+
+ @Override
+ public void markServerForRestart() {
+ marks++;
+ }
+
+ @Override
+ public int getPort() {
+ return -1;
+ }
+
+ @Override
+ public void ensureRunning() {
+ }
+
+ @Override
+ public Socket connect(int socketTimeoutMillis) {
+ return null;
+ }
+
+ @Override
+ public void shutdown() {
+ }
+
+ @Override
+ public boolean isRunning() {
+ return false;
+ }
+
+ @Override
+ public Path getTempDirectory() {
+ return null;
+ }
+
+ @Override
+ public void close() {
+ }
+ }
+
+ private static PipesConfig config() {
+ return new PipesConfig();
+ }
+
+ @Test
+ public void testPerClientHonoursBothSpellings() {
+ PerClientServerManager sm = new PerClientServerManager(config(), null,
0);
+ sm.markServerForRestart();
+ assertTrue(sm.needsRestart(), "no-arg markServerForRestart() must
recycle the worker");
+
+ PerClientServerManager other = new PerClientServerManager(config(),
null, 1);
+ other.markServerForRestart(RestartReason.OOM);
+ assertTrue(other.needsRestart(), "reason form must recycle the
worker");
+ }
+
+ @Test
+ public void testSharedHonoursBothSpellings() {
+ SharedServerManager sm = new SharedServerManager(config(), null, 2);
+ sm.markServerForRestart();
+ assertTrue(sm.needsRestart(), "no-arg markServerForRestart() must
recycle the worker");
+
+ SharedServerManager other = new SharedServerManager(config(), null, 2);
+ other.markServerForRestart(RestartReason.OOM);
+ assertTrue(other.needsRestart(), "reason form must recycle the
worker");
+ }
+
+ @Test
+ public void testReasonFormReachesANoArgOnlyImplementation() {
+ NoArgOnly sm = new NoArgOnly();
+ sm.markServerForRestart(RestartReason.OOM);
+ assertEquals(1, sm.marks, "reason form must fall back to an older
no-arg implementation");
+ }
+
+ @Test
+ public void testDefaultsDoNotRecurse() {
+ // markServerForRestart(reason) defaults to the no-arg form, so the
no-arg form must not
+ // default back to it: an implementation overriding neither would blow
the stack.
+ ServerManager sm = new ServerManager() {
+ @Override
+ public int getPort() {
+ return -1;
+ }
+
+ @Override
+ public void ensureRunning() {
+ }
+
+ @Override
+ public Socket connect(int socketTimeoutMillis) {
+ return null;
+ }
+
+ @Override
+ public void shutdown() {
+ }
+
+ @Override
+ public boolean isRunning() {
+ return false;
+ }
+
+ @Override
+ public Path getTempDirectory() {
+ return null;
+ }
+
+ @Override
+ public void close() {
+ }
+ };
+ sm.markServerForRestart();
+ sm.markServerForRestart(RestartReason.CRASH);
+ }
+}
diff --git
a/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/SharedServerModeTest.java
b/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/SharedServerModeTest.java
index 0e5a2d2e23..b7f0e15bc6 100644
---
a/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/SharedServerModeTest.java
+++
b/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/SharedServerModeTest.java
@@ -623,6 +623,7 @@ public class SharedServerModeTest {
assertTrue(verifyResult.isSuccess(),
"After concurrent OOM, server should restart and process
new request. Got: " + verifyResult.status());
+ assertRestartsAttributedToOom(pipesParser);
assertOneForkPerDeath(pipesParser, 1);
}
}
@@ -768,6 +769,7 @@ public class SharedServerModeTest {
"Phase 1 had: " + phase1OomCount + " OOMs, " +
phase1SuccessCount + " successes, " +
phase1CrashCount + " crashes");
+ assertRestartsAttributedToOom(pipesParser);
assertOneForkPerDeath(pipesParser, 1);
}
}
@@ -863,6 +865,30 @@ public class SharedServerModeTest {
}
}
+ /**
+ * The shared JVM died once, from OOM. Siblings whose in-flight parses
were killed by that
+ * death report a crash <em>result</em>, which is correct; none of them is
a second process
+ * death, so nothing may be booked as an extra restart.
+ */
+ private void assertRestartsAttributedToOom(PipesParser pipesParser) {
+ long oom = pipesParser.getRestartCount(RestartReason.OOM);
+ long crash = pipesParser.getRestartCount(RestartReason.CRASH);
+ String counts = " (oom=" + oom + ", crash=" + crash + ", timeout=" +
+ pipesParser.getRestartCount(RestartReason.TIMEOUT) + ",
maxFiles=" +
+ pipesParser.getRestartCount(RestartReason.MAX_FILES) + ")";
+ assertTrue(oom >= 1, "the OOM death should be attributed to OOM" +
counts);
+ assertEquals(0, crash,
+ "no restart may be attributed to CRASH: the only process death
was the OOM" + counts);
+ // Reason tags alone cannot distinguish "the phantom restart is gone"
from "the phantom
+ // restart is now labelled OOM". Total restarts is the honest
observable: one death, one
+ // restart, whatever it is called.
+ long total = 0;
+ for (RestartReason r : RestartReason.values()) {
+ total += pipesParser.getRestartCount(r);
+ }
+ assertEquals(1, total, "exactly one restart may follow a single
process death" + counts);
+ }
+
private Path setupInputDir(Path tmp) throws Exception {
Path inputDir = tmp.resolve("input");
Files.createDirectories(inputDir);
diff --git a/tika-server/tika-server-core/pom.xml
b/tika-server/tika-server-core/pom.xml
index b030d37f70..fd60838c42 100644
--- a/tika-server/tika-server-core/pom.xml
+++ b/tika-server/tika-server-core/pom.xml
@@ -88,6 +88,18 @@
<groupId>jakarta.servlet</groupId>
<artifactId>jakarta.servlet-api</artifactId>
</dependency>
+ <dependency>
+ <groupId>jakarta.annotation</groupId>
+ <artifactId>jakarta.annotation-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>io.micrometer</groupId>
+ <artifactId>micrometer-core</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>io.micrometer</groupId>
+ <artifactId>micrometer-registry-prometheus</artifactId>
+ </dependency>
<dependency>
<groupId>org.apache.cxf</groupId>
<artifactId>cxf-rt-rs-security-cors</artifactId>
diff --git
a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/ServerStatus.java
b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/ServerStatus.java
index 6219a72839..14329693b4 100644
---
a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/ServerStatus.java
+++
b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/ServerStatus.java
@@ -69,6 +69,11 @@ public class ServerStatus {
return new HashMap<>(tasks);
}
+ /** Count only; avoids copying the task map (and its filenames) on every
metrics scrape. */
+ public synchronized int getNumTasks() {
+ return tasks.size();
+ }
+
/**
* Returns the total number of tasks started since server startup.
*/
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 6e5da93fd5..49dab1b970 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
@@ -39,6 +39,9 @@ public class TikaServerCli {
options.addOption("?", "help", false, "this help message");
options.addOption("c", "config", true, "tika-config file");
options.addOption("i", "id", true, "id for this server, written to the
startup log");
+ options.addOption(null, "metricsPort", true,
+ "serve Prometheus metrics on this port (9404 by convention); "
+ + "unset means metrics are off\n");
return options;
}
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 eb8912eb85..b99a234406 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
@@ -74,6 +74,7 @@ public class TikaServerConfig {
private ArrayList<String> endpoints = new ArrayList<>();
private TlsConfig tlsConfig = new TlsConfig();
+ private Integer metricsPort;
/**
* Config with only the defaults
@@ -109,6 +110,16 @@ public class TikaServerConfig {
settings.add("id");
}
+ if (commandLine.hasOption("metricsPort")) {
+ String value = commandLine.getOptionValue("metricsPort");
+ try {
+ config.setMetricsPort(Integer.parseInt(value));
+ } catch (NumberFormatException e) {
+ throw new TikaConfigException("--metricsPort must be an
integer, got '" + value + "'");
+ }
+ settings.add("metricsPort");
+ }
+
config.validateConsistency(settings);
return config;
}
@@ -180,6 +191,15 @@ public class TikaServerConfig {
}
}
tlsConfig.checkInitialization();
+ if (metricsPort != null) {
+ if (metricsPort < 0 || metricsPort > 65535) {
+ throw new TikaConfigException("metricsPort must be 0-65535,
got " + metricsPort);
+ }
+ if (metricsPort != 0 && metricsPort == port) {
+ throw new TikaConfigException("metricsPort (" + metricsPort
+ + ") must differ from the server port");
+ }
+ }
}
public String getHost() {
@@ -269,6 +289,15 @@ public class TikaServerConfig {
this.tlsConfig = tlsConfig;
}
+ /** Prometheus scrape port; null means metrics are off. The listener binds
to {@link #getHost()}. */
+ public Integer getMetricsPort() {
+ return metricsPort;
+ }
+
+ public void setMetricsPort(Integer metricsPort) {
+ this.metricsPort = metricsPort;
+ }
+
public ArrayList<String> getEndpoints() {
return endpoints;
}
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 0e99ec7c71..e83b8a67cb 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
@@ -70,6 +70,9 @@ import org.apache.tika.pipes.core.PipesConfig;
import org.apache.tika.pipes.core.PipesParser;
import org.apache.tika.pipes.core.config.ConfigMerger;
import org.apache.tika.pipes.core.config.ConfigOverrides;
+import org.apache.tika.server.core.metrics.MetricsServer;
+import org.apache.tika.server.core.metrics.TikaMetricsFilter;
+import org.apache.tika.server.core.metrics.TikaServerMetrics;
import org.apache.tika.server.core.resource.AsyncResource;
import org.apache.tika.server.core.resource.DetectorResource;
import org.apache.tika.server.core.resource.LanguageResource;
@@ -118,6 +121,9 @@ public class TikaServerProcess {
options.addOption("p", "port", true, "listen port");
options.addOption("c", "config", true, "Tika Configuration xml file to
override default config with.");
options.addOption("i", "id", true, "id for this server, written to the
startup log");
+ options.addOption(null, "metricsPort", true,
+ "serve Prometheus metrics on this port (9404 by convention); "
+ + "unset means metrics are off\n");
options.addOption("?", "help", false, "this help message");
return options;
}
@@ -137,8 +143,8 @@ public class TikaServerProcess {
LOG.debug("forked config: {}", tikaServerConfig);
ServerDetails serverDetails = initServer(tikaServerConfig);
- startServer(serverDetails);
registerOrderedShutdown(serverDetails);
+ startServer(serverDetails);
} catch (Exception e) {
LOG.error("Can't start: ", e);
@@ -157,6 +163,10 @@ public class TikaServerProcess {
}
private static void startServer(ServerDetails serverDetails) {
+ // Metrics first: a scrape-port clash must not leave the parse port
briefly open.
+ if (serverDetails.metricsServer != null) {
+ startMetricsServer(serverDetails);
+ }
try {
//start the server
serverDetails.server = serverDetails.sf.create();
@@ -170,6 +180,21 @@ public class TikaServerProcess {
LOG.info("Started Apache Tika server {} at {}",
serverDetails.serverId, serverDetails.url);
}
+ private static void startMetricsServer(ServerDetails serverDetails) {
+ try {
+ serverDetails.metricsServer.start();
+ } catch (Exception e) {
+ LOG.warn("exception starting metrics server", e);
+ if (isBindException(e)) {
+ System.exit(BIND_EXCEPTION);
+ }
+ System.exit(DO_NOT_RESTART_EXIT_VALUE);
+ }
+ LOG.info("Started metrics listener (plain HTTP) on http://{}:{}{}";,
+ serverDetails.metricsServer.getHost(),
serverDetails.metricsServer.getPort(),
+ MetricsServer.PATH);
+ }
+
/**
* One ordered shutdown: stop the HTTP endpoint first so no new request
can arrive, then
* tear down the pipes workers and delete their temp directories. Replaces
the previously
@@ -185,9 +210,19 @@ public class TikaServerProcess {
LOG.warn("Error stopping HTTP server", e);
}
}
+ if (serverDetails.metricsServer != null) {
+ try {
+ serverDetails.metricsServer.close();
+ } catch (Exception e) {
+ LOG.warn("Error stopping metrics server", e);
+ }
+ }
if (serverDetails.pipesParsingHelper != null) {
serverDetails.pipesParsingHelper.shutdown();
}
+ if (serverDetails.metrics != null) {
+ serverDetails.metrics.close();
+ }
}));
}
@@ -249,6 +284,24 @@ public class TikaServerProcess {
List<Object> providers = new ArrayList<>();
loadAllProviders(tikaServerConfig, serverStatus, tikaResource,
resourceProviders, providers);
+ TikaServerMetrics metrics = null;
+ MetricsServer metricsServer = null;
+ if (tikaServerConfig.getMetricsPort() != null) {
+ metrics = new TikaServerMetrics();
+ metrics.bindJvm();
+ metrics.bindServerStatus(serverStatus);
+ if (pipesParsingHelper != null) {
+ metrics.bindSyncPool(pipesParsingHelper.getPipesParser());
+ }
+ AsyncResource asyncResource = findAsyncResource(resourceProviders);
+ if (asyncResource != null) {
+ metrics.bindAsyncPool(asyncResource.getAsyncProcessor());
+ }
+ providers.add(new TikaMetricsFilter(metrics,
Set.copyOf(VALID_ENDPOINTS)));
+ metricsServer = new MetricsServer(tikaServerConfig.getHost(),
+ tikaServerConfig.getMetricsPort(), metrics);
+ }
+
sf.setResourceProviders(resourceProviders);
sf.setProviders(providers);
@@ -290,9 +343,21 @@ public class TikaServerProcess {
details.serverId = tikaServerConfig.getId();
details.serverStatus = serverStatus;
details.pipesParsingHelper = pipesParsingHelper;
+ details.metrics = metrics;
+ details.metricsServer = metricsServer;
return details;
}
+ private static AsyncResource findAsyncResource(List<ResourceProvider>
resourceProviders) {
+ for (ResourceProvider p : resourceProviders) {
+ if (p instanceof SingletonResourceProvider s
+ && s.getInstance(null) instanceof AsyncResource
asyncResource) {
+ return asyncResource;
+ }
+ }
+ return null;
+ }
+
private static TLSServerParameters getTlsParams(TlsConfig tlsConfig)
throws GeneralSecurityException, IOException, TikaConfigException {
// Also checked in TlsConfig.checkInitialization() at config-load
time; kept here too
@@ -753,5 +818,7 @@ public class TikaServerProcess {
String url;
ServerStatus serverStatus;
PipesParsingHelper pipesParsingHelper;
+ TikaServerMetrics metrics;
+ MetricsServer metricsServer;
}
}
diff --git
a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/metrics/MetricsServer.java
b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/metrics/MetricsServer.java
new file mode 100644
index 0000000000..2ab2c1ed1d
--- /dev/null
+++
b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/metrics/MetricsServer.java
@@ -0,0 +1,113 @@
+/*
+ * 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.server.core.metrics;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+
+import org.eclipse.jetty.http.HttpHeader;
+import org.eclipse.jetty.http.HttpMethod;
+import org.eclipse.jetty.http.HttpStatus;
+import org.eclipse.jetty.server.ConnectionLimit;
+import org.eclipse.jetty.server.Handler;
+import org.eclipse.jetty.server.HttpConfiguration;
+import org.eclipse.jetty.server.HttpConnectionFactory;
+import org.eclipse.jetty.server.Request;
+import org.eclipse.jetty.server.Response;
+import org.eclipse.jetty.server.Server;
+import org.eclipse.jetty.server.ServerConnector;
+import org.eclipse.jetty.util.Callback;
+import org.eclipse.jetty.util.thread.QueuedThreadPool;
+
+/**
+ * Minimal standalone Jetty answering only {@code GET /metrics}. Separate from
the CXF
+ * server so the parse endpoints are unreachable here and a slow parse cannot
starve a scrape.
+ */
+public final class MetricsServer implements AutoCloseable {
+
+ public static final String PATH = "/metrics";
+ static final String CONTENT_TYPE = "text/plain; version=0.0.4;
charset=utf-8";
+ private static final int MAX_THREADS = 4;
+ // Unauthenticated port: cap what idle sockets can take from the process
FD budget.
+ private static final int MAX_CONNECTIONS = 64;
+ private static final long IDLE_TIMEOUT_MS = 10_000;
+
+ private final Server server;
+ private final ServerConnector connector;
+
+ public MetricsServer(String host, int port, TikaServerMetrics metrics) {
+ QueuedThreadPool pool = new QueuedThreadPool(MAX_THREADS, 1);
+ pool.setName("tika-metrics");
+ server = new Server(pool);
+ HttpConfiguration httpConfig = new HttpConfiguration();
+ httpConfig.setSendServerVersion(false);
+ connector = new ServerConnector(server, new
HttpConnectionFactory(httpConfig));
+ connector.setHost(host);
+ connector.setPort(port);
+ connector.setIdleTimeout(IDLE_TIMEOUT_MS);
+ server.addConnector(connector);
+ server.addBean(new ConnectionLimit(MAX_CONNECTIONS, connector));
+ server.setHandler(new ScrapeHandler(metrics));
+ }
+
+ public void start() throws Exception {
+ server.start();
+ }
+
+ public String getHost() {
+ return connector.getHost();
+ }
+
+ /** The bound port; meaningful after {@link #start()} (port 0 picks a free
one). */
+ public int getPort() {
+ return connector.getLocalPort();
+ }
+
+ @Override
+ public void close() throws Exception {
+ server.stop();
+ }
+
+ private static final class ScrapeHandler extends Handler.Abstract {
+
+ private final TikaServerMetrics metrics;
+
+ private ScrapeHandler(TikaServerMetrics metrics) {
+ this.metrics = metrics;
+ }
+
+ @Override
+ public boolean handle(Request request, Response response, Callback
callback) {
+ String path = Request.getPathInContext(request);
+ if (!PATH.equals(path) && !(PATH + "/").equals(path)) {
+ Response.writeError(request, response, callback,
HttpStatus.NOT_FOUND_404);
+ return true;
+ }
+ boolean head = HttpMethod.HEAD.is(request.getMethod());
+ if (!head && !HttpMethod.GET.is(request.getMethod())) {
+ Response.writeError(request, response, callback,
HttpStatus.METHOD_NOT_ALLOWED_405);
+ return true;
+ }
+ byte[] body = metrics.scrape().getBytes(StandardCharsets.UTF_8);
+ response.setStatus(HttpStatus.OK_200);
+ response.getHeaders().put(HttpHeader.CONTENT_TYPE, CONTENT_TYPE);
+ response.getHeaders().put(HttpHeader.CONTENT_LENGTH, body.length);
+ response.write(true, head ? null : ByteBuffer.wrap(body),
callback);
+ return true;
+ }
+ }
+}
diff --git
a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/metrics/TikaMetricsFilter.java
b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/metrics/TikaMetricsFilter.java
new file mode 100644
index 0000000000..c288588fee
--- /dev/null
+++
b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/metrics/TikaMetricsFilter.java
@@ -0,0 +1,95 @@
+/*
+ * 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.server.core.metrics;
+
+import java.util.List;
+import java.util.Set;
+
+import jakarta.annotation.Priority;
+import jakarta.ws.rs.container.ContainerRequestContext;
+import jakarta.ws.rs.container.ContainerRequestFilter;
+import jakarta.ws.rs.container.ContainerResponseContext;
+import jakarta.ws.rs.container.ContainerResponseFilter;
+import jakarta.ws.rs.container.PreMatching;
+import jakarta.ws.rs.ext.Provider;
+
+/**
+ * Records {@code tika_server_requests} and the request-size summary for every
request on
+ * the parse-side listener. Timing starts pre-matching so unmatched (404) and
aborted
+ * requests are counted too.
+ * <p>
+ * Not counted: an exception no {@code ExceptionMapper} handles. CXF answers
those from
+ * its fault chain, which does not run response filters.
+ */
+@Provider
+@PreMatching
+// Ahead of CXF's CORS filter (999) so a preflight abort is timed, not
recorded as 0.
+@Priority(100)
+public class TikaMetricsFilter implements ContainerRequestFilter,
ContainerResponseFilter {
+
+ static final String START_NANOS = TikaMetricsFilter.class.getName() +
".start";
+ static final String RECORDED = TikaMetricsFilter.class.getName() +
".recorded";
+ static final String UNMATCHED = "unmatched";
+ static final String OTHER = "other";
+
+ private final TikaServerMetrics metrics;
+ private final Set<String> knownEndpoints;
+
+ public TikaMetricsFilter(TikaServerMetrics metrics, Set<String>
knownEndpoints) {
+ this.metrics = metrics;
+ this.knownEndpoints = knownEndpoints;
+ }
+
+ @Override
+ public void filter(ContainerRequestContext requestContext) {
+ requestContext.setProperty(START_NANOS, System.nanoTime());
+ }
+
+ @Override
+ public void filter(ContainerRequestContext requestContext,
+ ContainerResponseContext responseContext) {
+ // CXF re-runs response filters when a mapped exception is thrown
mid-write.
+ if (requestContext.getProperty(RECORDED) != null) {
+ return;
+ }
+ requestContext.setProperty(RECORDED, Boolean.TRUE);
+ String endpoint = endpoint(requestContext);
+ Object start = requestContext.getProperty(START_NANOS);
+ long nanos = start instanceof Long s ? System.nanoTime() - s : 0L;
+ metrics.recordRequest(endpoint, requestContext.getMethod(),
responseContext.getStatus(),
+ nanos);
+ int length = requestContext.getLength();
+ if (length >= 0) {
+ metrics.recordRequestSize(endpoint, length);
+ }
+ }
+
+ /**
+ * First path segment when it names a known endpoint; otherwise {@code
other} for a
+ * matched resource and {@code unmatched} for a request answered before
routing.
+ */
+ private String endpoint(ContainerRequestContext requestContext) {
+ String path = requestContext.getUriInfo().getPath();
+ int slash = path.indexOf('/');
+ String root = slash < 0 ? path : path.substring(0, slash);
+ if (knownEndpoints.contains(root)) {
+ return root;
+ }
+ List<Object> matched =
requestContext.getUriInfo().getMatchedResources();
+ return matched == null || matched.isEmpty() ? UNMATCHED : OTHER;
+ }
+}
diff --git
a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/metrics/TikaServerMetrics.java
b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/metrics/TikaServerMetrics.java
new file mode 100644
index 0000000000..bd98312302
--- /dev/null
+++
b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/metrics/TikaServerMetrics.java
@@ -0,0 +1,204 @@
+/*
+ * 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.server.core.metrics;
+
+import java.time.Duration;
+import java.util.Locale;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.function.ToLongFunction;
+
+import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.DistributionSummary;
+import io.micrometer.core.instrument.FunctionCounter;
+import io.micrometer.core.instrument.Gauge;
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Timer;
+import io.micrometer.core.instrument.binder.jvm.JvmMemoryMetrics;
+import io.micrometer.core.instrument.binder.jvm.JvmThreadMetrics;
+import io.micrometer.core.instrument.binder.system.FileDescriptorMetrics;
+import io.micrometer.core.instrument.binder.system.UptimeMetrics;
+import io.micrometer.prometheusmetrics.PrometheusConfig;
+import io.micrometer.prometheusmetrics.PrometheusMeterRegistry;
+
+import org.apache.tika.pipes.core.PipesParser;
+import org.apache.tika.pipes.core.RestartReason;
+import org.apache.tika.pipes.core.async.AsyncProcessor;
+import org.apache.tika.server.core.ServerStatus;
+
+/**
+ * Owns the process's {@link PrometheusMeterRegistry} and every meter recorded
against it.
+ * Every tag value comes from a bounded set; nothing request-derived becomes a
tag.
+ */
+public final class TikaServerMetrics implements AutoCloseable {
+
+ /**
+ * One bucket set for every duration timer. Twelve explicit boundaries
instead of
+ * Micrometer's percentile histogram, which would emit roughly 70 per
series.
+ */
+ public static final Duration[] DURATION_SLOS = {
+ Duration.ofMillis(10), Duration.ofMillis(50),
Duration.ofMillis(100),
+ Duration.ofMillis(250), Duration.ofMillis(500),
Duration.ofSeconds(1),
+ Duration.ofSeconds(2), Duration.ofSeconds(5),
Duration.ofSeconds(10),
+ Duration.ofSeconds(30), Duration.ofSeconds(60),
Duration.ofSeconds(120)
+ };
+
+ /** Bucket boundaries for byte-size summaries: 1KB .. 1GB by decades. */
+ private static final double[] SIZE_SLOS = {
+ 1_000, 10_000, 100_000, 1_000_000, 10_000_000, 100_000_000,
1_000_000_000
+ };
+
+ /**
+ * Jetty passes any RFC-token method through to the filter, so the raw
verb would let a
+ * client mint meters without limit. Anything outside this set is tagged
{@code other}.
+ */
+ private static final Set<String> KNOWN_METHODS =
+ Set.of("GET", "POST", "PUT", "DELETE", "HEAD", "OPTIONS", "PATCH",
"TRACE", "CONNECT");
+
+ static final String REQUESTS = "tika_server_requests";
+ static final String REQUEST_SIZE = "tika_server_request_size_bytes";
+ static final String REJECTED = "tika_server_rejected_total";
+ static final String TASKS_ACTIVE = "tika_server_tasks_active";
+ static final String PIPES_WORKERS = "tika_pipes_workers";
+ static final String PIPES_RESTARTS = "tika_pipes_worker_restarts_total";
+ static final String PIPES_QUEUE_DEPTH = "tika_pipes_queue_depth";
+
+ static final String TAG_ENDPOINT = "endpoint";
+ static final String TAG_METHOD = "method";
+ static final String TAG_STATUS = "status";
+ static final String TAG_REASON = "reason";
+ static final String TAG_POOL = "pool";
+ static final String TAG_STATE = "state";
+
+ /** Worker pools a tika-server can run at once; each has its own forks. */
+ static final String POOL_SYNC = "sync";
+ static final String POOL_ASYNC = "async";
+
+ private final PrometheusMeterRegistry registry = new
PrometheusMeterRegistry(PrometheusConfig.DEFAULT);
+
+ MeterRegistry getRegistry() {
+ return registry;
+ }
+
+ /** Prometheus text exposition (format 0.0.4). */
+ public String scrape() {
+ return registry.scrape();
+ }
+
+ /** No GC/CPU binders: this JVM does not parse, so they would suggest
headroom that is not there. */
+ public void bindJvm() {
+ new JvmMemoryMetrics().bindTo(registry);
+ new JvmThreadMetrics().bindTo(registry);
+ new UptimeMetrics().bindTo(registry);
+ new FileDescriptorMetrics().bindTo(registry);
+ }
+
+ public void bindServerStatus(ServerStatus serverStatus) {
+ Gauge.builder(TASKS_ACTIVE, serverStatus, ServerStatus::getNumTasks)
+ .description("Sync parse/detect tasks currently running in
this server")
+ .register(registry);
+ }
+
+ /** The sync endpoints' forks: restart counters plus busy/idle slots. */
+ public void bindSyncPool(PipesParser pipesParser) {
+ bindRestarts(POOL_SYNC, pipesParser::getRestartCount);
+ Gauge.builder(PIPES_WORKERS, pipesParser, p -> p.getNumClients() -
p.getIdleClientCount())
+ .tag(TAG_POOL, POOL_SYNC)
+ .tag(TAG_STATE, "busy")
+ .description("Pipes worker slots by state")
+ .register(registry);
+ Gauge.builder(PIPES_WORKERS, pipesParser,
PipesParser::getIdleClientCount)
+ .tag(TAG_POOL, POOL_SYNC)
+ .tag(TAG_STATE, "idle")
+ .description("Pipes worker slots by state")
+ .register(registry);
+ }
+
+ /** {@code /async}'s own forks: restart counters plus queue depth. */
+ public void bindAsyncPool(AsyncProcessor asyncProcessor) {
+ bindRestarts(POOL_ASYNC, asyncProcessor::getRestartCount);
+ Gauge.builder(PIPES_QUEUE_DEPTH, asyncProcessor,
AsyncProcessor::getQueueDepth)
+ .tag(TAG_POOL, POOL_ASYNC)
+ .description("Tuples waiting in the /async queue")
+ .register(registry);
+ }
+
+ private void bindRestarts(String pool, ToLongFunction<RestartReason>
restartCount) {
+ for (RestartReason reason : RestartReason.values()) {
+ FunctionCounter.builder(PIPES_RESTARTS, reason,
restartCount::applyAsLong)
+ .tag(TAG_POOL, pool)
+ .tag(TAG_REASON, reason.name().toLowerCase(Locale.ROOT))
+ .description("Forked pipes worker restarts by reason")
+ .register(registry);
+ }
+ }
+
+ void recordRequest(String endpoint, String method, int status, long nanos)
{
+ Timer.builder(REQUESTS)
+ .tag(TAG_ENDPOINT, endpoint)
+ .tag(TAG_METHOD, methodTag(method))
+ .tag(TAG_STATUS, statusClass(status))
+ .description("HTTP requests handled by the parse-side
listener")
+ .serviceLevelObjectives(DURATION_SLOS)
+ .register(registry)
+ .record(nanos, TimeUnit.NANOSECONDS);
+ String rejected = rejectedReason(status);
+ if (rejected != null) {
+ Counter.builder(REJECTED)
+ .tag(TAG_REASON, rejected)
+ .description("Requests refused for capacity reasons")
+ .register(registry)
+ .increment();
+ }
+ }
+
+ void recordRequestSize(String endpoint, long bytes) {
+ DistributionSummary.builder(REQUEST_SIZE)
+ .tag(TAG_ENDPOINT, endpoint)
+ .baseUnit("bytes")
+ .description("Request body bytes (Content-Length)")
+ .serviceLevelObjectives(SIZE_SLOS)
+ .register(registry)
+ .record(bytes);
+ }
+
+ static String methodTag(String method) {
+ return method != null && KNOWN_METHODS.contains(method) ? method :
"other";
+ }
+
+ static String statusClass(int status) {
+ if (status < 200 || status >= 600) {
+ return "other";
+ }
+ return (status / 100) + "xx";
+ }
+
+ /** By status, as tika-server already maps them: 429 capacity, 503 fork
OOM/timeout/crash, 413 body limit. */
+ static String rejectedReason(int status) {
+ return switch (status) {
+ case 429 -> "busy_429";
+ case 413 -> "payload_413";
+ case 503 -> "crash_503";
+ default -> null;
+ };
+ }
+
+ @Override
+ public void close() {
+ registry.close();
+ }
+}
diff --git
a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/AsyncResource.java
b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/AsyncResource.java
index 1244f5ba31..a5de82d0e6 100644
---
a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/AsyncResource.java
+++
b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/AsyncResource.java
@@ -81,6 +81,10 @@ public class AsyncResource {
this.maxQueuePauseMillis = maxQueuePauseMillis;
}
+ public AsyncProcessor getAsyncProcessor() {
+ return asyncProcessor;
+ }
+
public ArrayBlockingQueue<FetchEmitTuple> getFetchEmitQueue(int queueSize)
{
this.queue = new ArrayBlockingQueue<>(queueSize);
return queue;
diff --git
a/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerConfigTest.java
b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerConfigTest.java
index 0f044ead39..6f884c48e7 100644
---
a/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerConfigTest.java
+++
b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerConfigTest.java
@@ -18,6 +18,7 @@ package org.apache.tika.server.core;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -31,6 +32,8 @@ import org.apache.commons.cli.DefaultParser;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.Options;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import org.apache.tika.TikaTest;
import org.apache.tika.exception.TikaConfigException;
@@ -238,4 +241,57 @@ public class TikaServerConfigTest extends TikaTest {
TikaServerConfig config = TikaServerConfig.load(commandLine);
assertTrue(config.getTlsConfig().isClientAuthenticationRequired());
}
+ private static Options metricsOptions() {
+ return new Options()
+
.addOption(Option.builder("c").longOpt("config").hasArg().get())
+ .addOption(Option.builder("p").longOpt("port").hasArg().get())
+
.addOption(Option.builder().longOpt("metricsPort").hasArg().get());
+ }
+
+ @Test
+ public void testMetricsOffByDefault() throws Exception {
+ TikaServerConfig config = TikaServerConfig.load(
+ new DefaultParser().parse(metricsOptions(), new String[]{}));
+ assertNull(config.getMetricsPort());
+ }
+
+ @Test
+ public void testMetricsFromJson() throws Exception {
+ Path path = getConfigPath(getClass(),
"tika-config-server-metrics.json");
+ TikaServerConfig config = TikaServerConfig.load(new
DefaultParser().parse(metricsOptions(),
+ new String[]{"-c",
ProcessUtils.escapeCommandLine(path.toAbsolutePath().toString())}));
+ assertEquals(9404, config.getMetricsPort());
+ }
+
+ @Test
+ public void testMetricsPortFromCommandLine() throws Exception {
+ TikaServerConfig config = TikaServerConfig.load(new
DefaultParser().parse(metricsOptions(),
+ new String[]{"-p", "9998", "--metricsPort", "9500"}));
+ assertEquals(9500, config.getMetricsPort());
+ }
+
+ @ParameterizedTest
+ @CsvSource({"abc, --metricsPort", "65536, 0-65535", "-1, 0-65535"})
+ public void testMetricsPortRejected(String value, String expectedMessage)
throws Exception {
+ CommandLine commandLine = new DefaultParser().parse(metricsOptions(),
+ new String[]{"--metricsPort", value});
+ TikaConfigException ex = assertThrows(TikaConfigException.class,
+ () -> TikaServerConfig.load(commandLine));
+ assertContains(expectedMessage, ex.getMessage());
+ }
+
+ @Test
+ public void testMetricsPortZeroIsEphemeral() throws Exception {
+ assertEquals(0, TikaServerConfig.load(new
DefaultParser().parse(metricsOptions(),
+ new String[]{"--metricsPort", "0"})).getMetricsPort());
+ }
+
+ @Test
+ public void testMetricsPortMustDifferFromServerPort() throws Exception {
+ CommandLine commandLine = new DefaultParser().parse(metricsOptions(),
+ new String[]{"-p", "9998", "--metricsPort", "9998"});
+ TikaConfigException ex = assertThrows(TikaConfigException.class,
+ () -> TikaServerConfig.load(commandLine));
+ assertContains("metricsPort", ex.getMessage());
+ }
}
diff --git
a/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerMetricsIntegrationTest.java
b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerMetricsIntegrationTest.java
new file mode 100644
index 0000000000..6f71ca4a01
--- /dev/null
+++
b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerMetricsIntegrationTest.java
@@ -0,0 +1,247 @@
+/*
+ * 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.server.core;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.file.Paths;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import jakarta.ws.rs.core.Response;
+import org.apache.cxf.jaxrs.client.WebClient;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import org.apache.tika.server.core.metrics.MetricsServer;
+import org.apache.tika.server.core.metrics.TikaServerMetrics;
+import org.apache.tika.utils.ProcessUtils;
+
+/**
+ * Runs the real server process with {@code --metricsPort} and checks the
scrape output
+ * and the port isolation both ways.
+ */
+public class TikaServerMetricsIntegrationTest extends IntegrationTestBase {
+
+ private final int metricsPort = TestPortAllocator.findFreePort();
+ private final String metricsEndPoint = "http://localhost:"; + metricsPort;
+
+ @Test
+ @Timeout(120)
+ public void testScrapeAfterParsesAndWorkerRestart() throws Exception {
+ startProcess(new String[]{"-config",
getConfig("tika-config-server-basic.json"),
+ "--metricsPort", String.valueOf(metricsPort)});
+ awaitServerStartup();
+
+ assertEquals(200, rmeta(TEST_HELLO_WORLD).getStatus());
+ assertEquals(503, rmeta(TEST_OOM).getStatus());
+ // The OOM'd worker is restarted on its next use.
+ assertEquals(200, rmeta(TEST_HELLO_WORLD).getStatus());
+ assertEquals(404, WebClient.create(endPoint +
"/no-such-path").get().getStatus());
+
+ HttpResponse<String> scrape = get(metricsEndPoint +
MetricsServer.PATH);
+ assertEquals(200, scrape.statusCode());
+ assertTrue(scrape.headers().firstValue("Content-Type").orElse("")
+ .startsWith("text/plain; version=0.0.4"));
+ String body = scrape.body();
+
+ assertSample(body, "tika_server_requests_seconds_count",
+ "endpoint=\"rmeta\",method=\"PUT\",status=\"2xx\"", 2.0);
+ assertSample(body, "tika_server_requests_seconds_count",
+ "endpoint=\"rmeta\",method=\"PUT\",status=\"5xx\"", 1.0);
+ assertSample(body, "tika_server_requests_seconds_count",
+ "endpoint=\"unmatched\",method=\"GET\",status=\"4xx\"", 1.0);
+ assertSample(body, "tika_server_rejected_total",
"reason=\"crash_503\"", 1.0);
+ assertSample(body, "tika_pipes_worker_restarts_total",
+ "pool=\"sync\",reason=\"oom\"", 1.0);
+ assertSample(body, "tika_pipes_workers",
"pool=\"sync\",state=\"idle\"", 2.0);
+ assertSample(body, "tika_pipes_workers",
"pool=\"sync\",state=\"busy\"", 0.0);
+ assertSample(body, "tika_server_tasks_active", "", 0.0);
+ assertTrue(body.contains("jvm_memory_used_bytes"), body);
+
assertTrue(body.contains("tika_server_request_size_bytes_count{endpoint=\"rmeta\"}"),
body);
+
+ // Isolation both ways.
+ assertEquals(404, WebClient.create(endPoint +
MetricsServer.PATH).get().getStatus());
+ assertEquals(404, WebClient.create(metricsEndPoint + RMETA_PATH)
+ .accept("application/json")
+
.put(ClassLoader.getSystemResourceAsStream(TEST_HELLO_WORLD)).getStatus());
+ }
+
+ @Test
+ @Timeout(120)
+ public void testOffByDefault() throws Exception {
+ startProcess(new String[]{"-config",
getConfig("tika-config-server-basic.json")});
+ awaitServerStartup();
+ assertEquals(200, rmeta(TEST_HELLO_WORLD).getStatus());
+ assertEquals(404, WebClient.create(endPoint +
MetricsServer.PATH).get().getStatus());
+ // The gate is the port: nothing may be listening on the one the
metrics config names.
+ assertThrows(IOException.class, () -> get(metricsEndPoint +
MetricsServer.PATH),
+ "a scrape listener came up with no metrics port configured");
+ }
+
+ /**
+ * /async forks its own workers, separate from the sync pool's. Without a
pool label and
+ * a second binding, a crash in an async worker is counted nowhere.
+ */
+ @Test
+ @Timeout(240)
+ public void testBothWorkerPoolsAreCounted() throws Exception {
+ startProcess(new String[]{"-config",
getConfig("tika-config-server-async-metrics.json"),
+ "--metricsPort", String.valueOf(metricsPort)});
+ awaitServerStartup();
+ assertEquals(200, rmeta(TEST_HELLO_WORLD).getStatus());
+
+ String body = get(metricsEndPoint + MetricsServer.PATH).body();
+ assertSample(body, "tika_pipes_worker_restarts_total",
"pool=\"sync\",reason=\"oom\"", 0.0);
+ assertSample(body, "tika_pipes_worker_restarts_total",
"pool=\"async\",reason=\"oom\"", 0.0);
+ assertSample(body, "tika_pipes_queue_depth", "pool=\"async\"", 0.0);
+ }
+
+ /** Routine restarts (max files, idle exit 24) must not be counted as
crashes. */
+ @Test
+ @Timeout(240)
+ public void testRoutineRestartReasons() throws Exception {
+ startProcess(new String[]{"-config",
getConfig("tika-config-server-metrics-restarts.json"),
+ "--metricsPort", String.valueOf(metricsPort)});
+ awaitServerStartup();
+ for (int i = 0; i < 3; i++) {
+ assertEquals(200, rmeta(TEST_HELLO_WORLD).getStatus());
+ }
+ // Past the idle socket timeout the fork exits 24 and is restarted by
the next request.
+ String body = awaitSample("tika_pipes_worker_restarts_total",
"pool=\"sync\",reason=\"idle\"", 1.0);
+ assertSample(body, "tika_pipes_worker_restarts_total",
"pool=\"sync\",reason=\"max_files\"", 1.0);
+ assertSample(body, "tika_pipes_worker_restarts_total",
"pool=\"sync\",reason=\"crash\"", 0.0);
+ }
+
+ /** Timeout and crash are attributed per client; a 503 for either is a
crash_503 rejection. */
+ @Test
+ @Timeout(240)
+ public void testTimeoutAndCrashReasons() throws Exception {
+ startProcess(new String[]{"-config",
getConfig("tika-config-server-metrics-timeout.json"),
+ "--metricsPort", String.valueOf(metricsPort)});
+ awaitServerStartup();
+ assertEquals(503, rmeta(TEST_HEAVY_HANG).getStatus());
+ assertEquals(200, rmeta(TEST_HELLO_WORLD).getStatus());
+ assertEquals(503, rmeta(TEST_SYSTEM_EXIT).getStatus());
+ assertEquals(200, rmeta(TEST_HELLO_WORLD).getStatus());
+
+ String body = get(metricsEndPoint + MetricsServer.PATH).body();
+ assertSample(body, "tika_pipes_worker_restarts_total",
"pool=\"sync\",reason=\"timeout\"", 1.0);
+ assertSample(body, "tika_pipes_worker_restarts_total",
"pool=\"sync\",reason=\"crash\"", 1.0);
+ assertSample(body, "tika_pipes_worker_restarts_total",
"pool=\"sync\",reason=\"oom\"", 0.0);
+ assertSample(body, "tika_server_rejected_total",
"reason=\"crash_503\"", 2.0);
+ }
+
+ /** Shared server: the client that saw the OOM marks it; the restarter
must not overwrite it with crash. */
+ @Test
+ @Timeout(240)
+ public void testSharedServerOomReason() throws Exception {
+ startProcess(new String[]{"-config",
getConfig("tika-config-server-metrics-shared.json"),
+ "--metricsPort", String.valueOf(metricsPort)});
+ awaitServerStartup();
+ assertEquals(503, rmeta(TEST_OOM).getStatus());
+ assertEquals(200, rmeta(TEST_HELLO_WORLD).getStatus());
+
+ String body = get(metricsEndPoint + MetricsServer.PATH).body();
+ assertSample(body, "tika_pipes_worker_restarts_total",
"pool=\"sync\",reason=\"oom\"", 1.0);
+ assertSample(body, "tika_pipes_worker_restarts_total",
"pool=\"sync\",reason=\"crash\"", 0.0);
+ }
+
+ /** Polls (parse + scrape) until the sample reaches {@code expected};
returns the last body. */
+ private String awaitSample(String name, String labels, double expected)
throws Exception {
+ long deadline = System.currentTimeMillis() + 30_000;
+ String body;
+ do {
+ // Each parse resets the fork's idle clock, so sleep past the
config's socketTimeoutMillis.
+ Thread.sleep(3000);
+ assertEquals(200, rmeta(TEST_HELLO_WORLD).getStatus());
+ body = get(metricsEndPoint + MetricsServer.PATH).body();
+ if (sample(body, name, labels) == expected) {
+ return body;
+ }
+ } while (System.currentTimeMillis() < deadline);
+ throw new AssertionError("timed out waiting for " + name + "{" +
labels + "}=" + expected
+ + " in:\n" + body);
+ }
+
+ /**
+ * The explicit SLO boundaries, not micrometer's ~70-bucket percentile
histogram.
+ * Guards the cardinality decision: a stray publishPercentileHistogram()
fails here.
+ */
+ @Test
+ @Timeout(240)
+ public void testDurationBucketsAreBounded() throws Exception {
+ startProcess(new String[]{"-config",
getConfig("tika-config-server-basic.json"),
+ "--metricsPort", String.valueOf(metricsPort)});
+ awaitServerStartup();
+ assertEquals(200, rmeta(TEST_HELLO_WORLD).getStatus());
+
+ String body = get(metricsEndPoint + MetricsServer.PATH).body();
+ long buckets = body
+ .lines()
+ .filter(l ->
l.startsWith("tika_server_requests_seconds_bucket{")
+ && l.contains("endpoint=\"rmeta\""))
+ .count();
+ int expected = TikaServerMetrics.DURATION_SLOS.length + 1;
+ assertEquals(expected, buckets, "expected SLO buckets + Inf, got " +
buckets + ":\n" + body);
+ }
+
+ private Response rmeta(String resource) {
+ return WebClient.create(endPoint + RMETA_PATH)
+ .accept("application/json")
+ .put(ClassLoader.getSystemResourceAsStream(resource));
+ }
+
+ private static void assertSample(String body, String name, String labels,
double expected) {
+ assertEquals(expected, sample(body, name, labels), 0.0, name + "{" +
labels + "}");
+ }
+
+ private static double sample(String body, String name, String labels) {
+ String labelled = labels.isEmpty() ? "" : "\\{" +
Pattern.quote(labels) + ",?\\}";
+ Matcher m = Pattern.compile("^" + Pattern.quote(name) + labelled + "
(\\S+)$",
+ Pattern.MULTILINE).matcher(body);
+ assertTrue(m.find(), "missing " + name + "{" + labels + "} in:\n" +
body);
+ return Double.parseDouble(m.group(1));
+ }
+
+ private static HttpResponse<String> get(String url) throws Exception {
+ return HttpClient.newHttpClient().send(
+ HttpRequest.newBuilder(URI.create(url)).GET().build(),
+ HttpResponse.BodyHandlers.ofString());
+ }
+
+ private String getConfig(String configName) {
+ try {
+ return ProcessUtils.escapeCommandLine(Paths
+ .get(TikaServerMetricsIntegrationTest.class
+ .getResource("/configs/" + configName)
+ .toURI())
+ .toAbsolutePath()
+ .toString());
+ } catch (URISyntaxException e) {
+ throw new RuntimeException(e);
+ }
+ }
+}
diff --git
a/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/metrics/MetricsServerTest.java
b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/metrics/MetricsServerTest.java
new file mode 100644
index 0000000000..5cad62f758
--- /dev/null
+++
b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/metrics/MetricsServerTest.java
@@ -0,0 +1,96 @@
+/*
+ * 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.server.core.metrics;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+
+import io.micrometer.core.instrument.Counter;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+public class MetricsServerTest {
+
+ private TikaServerMetrics metrics;
+ private MetricsServer server;
+ private String base;
+
+ @BeforeEach
+ public void setUp() throws Exception {
+ metrics = new TikaServerMetrics();
+
Counter.builder("tika_test_total").register(metrics.getRegistry()).increment(3);
+ server = new MetricsServer("localhost", 0, metrics);
+ server.start();
+ base = "http://localhost:"; + server.getPort();
+ }
+
+ @AfterEach
+ public void tearDown() throws Exception {
+ server.close();
+ metrics.close();
+ }
+
+ @Test
+ public void testScrape() throws Exception {
+ HttpResponse<String> response = get(base + MetricsServer.PATH);
+ assertEquals(200, response.statusCode());
+ assertTrue(response.headers().firstValue("Content-Type").orElse("")
+ .startsWith("text/plain; version=0.0.4"),
response.headers().toString());
+ String body = response.body();
+ assertTrue(body.contains("# TYPE tika_test_total counter"), body);
+ assertTrue(body.contains("tika_test_total 3.0"), body);
+ assertEquals(200, get(base + MetricsServer.PATH + "/").statusCode());
+ HttpResponse<String> head = HttpClient.newHttpClient().send(
+ HttpRequest.newBuilder(URI.create(base + MetricsServer.PATH))
+ .method("HEAD",
HttpRequest.BodyPublishers.noBody()).build(),
+ HttpResponse.BodyHandlers.ofString());
+ assertEquals(200, head.statusCode());
+ assertEquals("", head.body());
+ }
+
+ @Test
+ public void testOnlyMetricsPathIsServed() throws Exception {
+ assertEquals(404, get(base + "/tika").statusCode());
+ assertEquals(404, get(base + "/").statusCode());
+ HttpResponse<String> post = HttpClient.newHttpClient().send(
+ HttpRequest.newBuilder(URI.create(base + MetricsServer.PATH))
+ .POST(HttpRequest.BodyPublishers.noBody()).build(),
+ HttpResponse.BodyHandlers.ofString());
+ assertEquals(405, post.statusCode());
+ }
+
+ @Test
+ public void testClosedServerRefusesConnections() throws Exception {
+ assertEquals(200, get(base + MetricsServer.PATH).statusCode());
+ server.close();
+ assertThrows(IOException.class, () -> get(base + MetricsServer.PATH));
+ }
+
+ private static HttpResponse<String> get(String url) throws Exception {
+ return HttpClient.newHttpClient().send(
+ HttpRequest.newBuilder(URI.create(url)).GET().build(),
+ HttpResponse.BodyHandlers.ofString());
+ }
+}
diff --git
a/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/metrics/RejectedReasonTest.java
b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/metrics/RejectedReasonTest.java
new file mode 100644
index 0000000000..556bce87e3
--- /dev/null
+++
b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/metrics/RejectedReasonTest.java
@@ -0,0 +1,65 @@
+/*
+ * 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.server.core.metrics;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.junit.jupiter.api.Test;
+
+import org.apache.tika.pipes.api.PipesResult;
+import org.apache.tika.server.core.resource.PipesParsingHelper;
+
+/**
+ * Ties {@code rejected_total} to the statuses tika-server actually returns,
so a new
+ * {@link PipesResult.RESULT_STATUS} that maps to 429/503/413 fails here until
someone
+ * decides which rejection reason it is.
+ */
+public class RejectedReasonTest {
+
+ private static final Map<PipesResult.RESULT_STATUS, String> EXPECTED = new
HashMap<>();
+
+ static {
+ EXPECTED.put(PipesResult.RESULT_STATUS.CLIENT_UNAVAILABLE_WITHIN_MS,
"busy_429");
+ EXPECTED.put(PipesResult.RESULT_STATUS.TIMEOUT, "crash_503");
+ EXPECTED.put(PipesResult.RESULT_STATUS.OOM, "crash_503");
+ EXPECTED.put(PipesResult.RESULT_STATUS.UNSPECIFIED_CRASH, "crash_503");
+ EXPECTED.put(PipesResult.RESULT_STATUS.PAYLOAD_LIMIT_EXCEEDED,
"payload_413");
+ }
+
+ @Test
+ public void testEveryStatusIsClassified() {
+ for (PipesResult.RESULT_STATUS status :
PipesResult.RESULT_STATUS.values()) {
+ int httpStatus = PipesParsingHelper
+ .responseBuilder(status, 1000)
+ .build()
+ .getStatus();
+ assertEquals(EXPECTED.get(status),
TikaServerMetrics.rejectedReason(httpStatus),
+ status + " maps to HTTP " + httpStatus);
+ }
+ }
+
+ @Test
+ public void testNonRejectionStatusesAreNotCounted() {
+ assertNull(TikaServerMetrics.rejectedReason(200));
+ assertNull(TikaServerMetrics.rejectedReason(400));
+ assertNull(TikaServerMetrics.rejectedReason(500));
+ }
+}
diff --git
a/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/metrics/TikaMetricsFilterTest.java
b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/metrics/TikaMetricsFilterTest.java
new file mode 100644
index 0000000000..911e840143
--- /dev/null
+++
b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/metrics/TikaMetricsFilterTest.java
@@ -0,0 +1,269 @@
+/*
+ * 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.server.core.metrics;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+import java.net.Socket;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+
+import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.DistributionSummary;
+import io.micrometer.core.instrument.Timer;
+import jakarta.ws.rs.GET;
+import jakarta.ws.rs.InternalServerErrorException;
+import jakarta.ws.rs.Path;
+import jakarta.ws.rs.core.Response;
+import jakarta.ws.rs.core.StreamingOutput;
+import org.apache.cxf.jaxrs.JAXRSServerFactoryBean;
+import org.apache.cxf.jaxrs.client.WebClient;
+import org.apache.cxf.jaxrs.lifecycle.SingletonResourceProvider;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+
+import org.apache.tika.server.core.CXFTestBase;
+import org.apache.tika.server.core.MaxRequestSizeFilter;
+import org.apache.tika.server.core.TikaServerParseExceptionMapper;
+import org.apache.tika.server.core.resource.TikaResource;
+import org.apache.tika.server.core.writer.JSONMessageBodyWriter;
+
+public class TikaMetricsFilterTest extends CXFTestBase {
+
+ private static final long MAX_BYTES = 64 * 1024;
+ private static final String HELLO_WORLD =
"test-documents/mock/hello_world.xml";
+
+ private TikaServerMetrics metrics;
+
+ @Path("/boom")
+ public static class BoomResource {
+ @GET
+ public String get() {
+ // Mapped by JAX-RS itself; an unmapped exception never reaches
response filters.
+ throw new InternalServerErrorException("boom");
+ }
+
+ @GET
+ @Path("mid-write")
+ public StreamingOutput midWrite() {
+ return out -> {
+ out.write("partial".getBytes(StandardCharsets.UTF_8));
+ out.flush();
+ throw new InternalServerErrorException("mid-write");
+ };
+ }
+ }
+
+ @Override
+ protected void setUpResources(JAXRSServerFactoryBean sf) {
+ sf.setResourceClasses(TikaResource.class, BoomResource.class);
+ sf.setResourceProvider(TikaResource.class, new
SingletonResourceProvider(tikaResource));
+ sf.setResourceProvider(BoomResource.class, new
SingletonResourceProvider(new BoomResource()));
+ }
+
+ @Override
+ protected void setUpProviders(JAXRSServerFactoryBean sf) {
+ metrics = new TikaServerMetrics();
+ List<Object> providers = new ArrayList<>();
+ providers.add(new TikaServerParseExceptionMapper());
+ providers.add(new JSONMessageBodyWriter());
+ providers.add(new MaxRequestSizeFilter(MAX_BYTES));
+ providers.add(new
MaxRequestSizeFilter.RequestTooLargeExceptionMapper());
+ providers.add(new TikaMetricsFilter(metrics, Set.of("tika", "rmeta")));
+ sf.setProviders(providers);
+ }
+
+ @Test
+ public void testSuccessRecordsTimerAndSize() throws Exception {
+ long helloWorldBytes;
+ try (InputStream is =
ClassLoader.getSystemResourceAsStream(HELLO_WORLD)) {
+ helloWorldBytes = is.readAllBytes().length;
+ }
+ long before = count("tika", "PUT", "2xx");
+ long sizeBefore = sizeCount("tika");
+ double sizeTotalBefore = sizeTotal("tika");
+ Response response = WebClient
+ .create(endPoint + "/tika/text")
+ .accept("text/plain")
+ .put(ClassLoader.getSystemResourceAsStream(HELLO_WORLD));
+ assertEquals(200, response.getStatus());
+ String body = getStringFromInputStream((InputStream)
response.getEntity());
+ assertTrue(body.contains("hello world"), body);
+
+ assertEquals(before + 1, count("tika", "PUT", "2xx"));
+ assertEquals(sizeBefore + 1, sizeCount("tika"));
+ assertEquals(sizeTotalBefore + helloWorldBytes, sizeTotal("tika"),
0.0);
+ }
+
+ /** WebClient streams an InputStream chunked, so this is the
BoundedInputStream + mapper path. */
+ @Test
+ public void testChunkedPayloadTooLargeCountsAsRejected() {
+ long before = count("tika", "PUT", "4xx");
+ double rejectedBefore = rejected("payload_413");
+ Response response = WebClient
+ .create(endPoint + "/tika/text")
+ .put(new ByteArrayInputStream(new byte[(int) MAX_BYTES * 4]));
+ assertEquals(413, response.getStatus());
+ assertEquals(before + 1, count("tika", "PUT", "4xx"));
+ assertEquals(rejectedBefore + 1, rejected("payload_413"), 0.0);
+ }
+
+ /** A declared Content-Length over the limit is refused via abortWith
before the body is read. */
+ @Test
+ public void testDeclaredLengthTooLargeCountsAsRejected() throws Exception {
+ long before = count("tika", "PUT", "4xx");
+ double rejectedBefore = rejected("payload_413");
+ HttpResponse<String> response = put(
+ HttpRequest.BodyPublishers.ofByteArray(new byte[(int)
MAX_BYTES * 4]));
+ assertEquals(413, response.statusCode());
+ assertEquals(before + 1, count("tika", "PUT", "4xx"));
+ assertEquals(rejectedBefore + 1, rejected("payload_413"), 0.0);
+ }
+
+ @Test
+ public void testChunkedRequestHasNoSizeSample() throws Exception {
+ long before = count("tika", "PUT", "2xx");
+ long sizeBefore = sizeCount("tika");
+ HttpResponse<String> response =
put(HttpRequest.BodyPublishers.ofInputStream(
+ () -> ClassLoader.getSystemResourceAsStream(HELLO_WORLD)));
+ assertEquals(200, response.statusCode());
+ assertEquals(before + 1, count("tika", "PUT", "2xx"));
+ assertEquals(sizeBefore, sizeCount("tika"), "no Content-Length must
mean no size sample");
+ }
+
+ @Test
+ public void testThrownExceptionIs5xx() {
+ long before = count(TikaMetricsFilter.OTHER, "GET", "5xx");
+ Response response = WebClient.create(endPoint + "/boom").get();
+ assertEquals(500, response.getStatus());
+ // /boom is not a tika-server endpoint name, so it folds into "other"
+ assertEquals(before + 1, count(TikaMetricsFilter.OTHER, "GET", "5xx"));
+ }
+
+ /** CXF re-runs the response filters for the mapped exception; the request
must be timed once. */
+ @Test
+ public void testMidWriteExceptionRecordedOnce() {
+ long before = count(TikaMetricsFilter.OTHER, "GET");
+ WebClient.create(endPoint + "/boom/mid-write").get();
+ assertEquals(before + 1, count(TikaMetricsFilter.OTHER, "GET"));
+ }
+
+ private HttpResponse<String> put(HttpRequest.BodyPublisher body) throws
Exception {
+ return HttpClient.newHttpClient().send(
+ HttpRequest.newBuilder(URI.create(endPoint + "/tika/text"))
+ .header("Accept", "text/plain").PUT(body).build(),
+ HttpResponse.BodyHandlers.ofString());
+ }
+
+ @Test
+ public void testUnmatchedPathsAreBoundedTags() {
+ long timersBefore = -1;
+ for (int i = 0; i < 25; i++) {
+ Response response = WebClient.create(endPoint + "/no-such-" + i +
"/x").get();
+ assertEquals(404, response.getStatus());
+ long timers =
metrics.getRegistry().find(TikaServerMetrics.REQUESTS).timers().size();
+ if (timersBefore >= 0) {
+ assertEquals(timersBefore, timers, "a new request timer
appeared for request " + i);
+ }
+ timersBefore = timers;
+ }
+ assertEquals(25, count(TikaMetricsFilter.UNMATCHED, "GET", "4xx"));
+ }
+
+ @Test
+ public void testUnknownMethodsAreBoundedTags() throws Exception {
+ long timersBefore = -1;
+ for (int i = 0; i < 25; i++) {
+ assertTrue(rawRequest("BOGUS" + i + " /tika
HTTP/1.1").startsWith("HTTP/1.1"));
+ long timers =
metrics.getRegistry().find(TikaServerMetrics.REQUESTS).timers().size();
+ if (timersBefore >= 0) {
+ assertEquals(timersBefore, timers, "a new request timer
appeared for method BOGUS" + i);
+ }
+ timersBefore = timers;
+ }
+ assertNull(timer("tika", "BOGUS0", "4xx"), "the raw method must not
become a tag value");
+ assertEquals(25, count("tika", "other"), "requests with unknown
methods must still be counted");
+ }
+
+ @ParameterizedTest
+ @CsvSource(nullValues = "null", value = {
+ "GET, GET", "PUT, PUT", "BOGUS, other", "get, other", "null,
other"})
+ public void testMethodTagIsBounded(String method, String expected) {
+ assertEquals(expected, TikaServerMetrics.methodTag(method));
+ }
+
+ /** Arbitrary method tokens: HttpURLConnection and java.net.http both
refuse to send them. */
+ private String rawRequest(String requestLine) throws Exception {
+ URI uri = URI.create(endPoint);
+ try (Socket socket = new Socket(uri.getHost(), uri.getPort())) {
+ socket.setSoTimeout(30000);
+ socket.getOutputStream().write(
+ (requestLine + "\r\nHost: " + uri.getHost() +
"\r\nConnection: close\r\n\r\n")
+ .getBytes(StandardCharsets.US_ASCII));
+ socket.getOutputStream().flush();
+ return new String(socket.getInputStream().readAllBytes(),
StandardCharsets.US_ASCII);
+ }
+ }
+
+ private Timer timer(String endpoint, String method, String status) {
+ return metrics.getRegistry().find(TikaServerMetrics.REQUESTS)
+ .tag("endpoint", endpoint).tag("method", method).tag("status",
status).timer();
+ }
+
+ /** Registry is shared across the class (PER_CLASS), so callers assert
deltas; absent meters read 0. */
+ private long count(String endpoint, String method, String status) {
+ Timer t = timer(endpoint, method, status);
+ return t == null ? 0 : t.count();
+ }
+
+ private long count(String endpoint, String method) {
+ return metrics.getRegistry().find(TikaServerMetrics.REQUESTS)
+ .tag("endpoint", endpoint).tag("method",
method).timers().stream()
+ .mapToLong(Timer::count).sum();
+ }
+
+ private DistributionSummary size(String endpoint) {
+ return metrics.getRegistry().find(TikaServerMetrics.REQUEST_SIZE)
+ .tag("endpoint", endpoint).summary();
+ }
+
+ private long sizeCount(String endpoint) {
+ DistributionSummary s = size(endpoint);
+ return s == null ? 0 : s.count();
+ }
+
+ private double sizeTotal(String endpoint) {
+ DistributionSummary s = size(endpoint);
+ return s == null ? 0 : s.totalAmount();
+ }
+
+ private double rejected(String reason) {
+ Counter c =
metrics.getRegistry().find(TikaServerMetrics.REJECTED).tag("reason",
reason).counter();
+ return c == null ? 0 : c.count();
+ }
+}
diff --git
a/tika-server/tika-server-core/src/test/resources/configs/tika-config-server-async-metrics.json
b/tika-server/tika-server-core/src/test/resources/configs/tika-config-server-async-metrics.json
new file mode 100644
index 0000000000..40c3ab50fc
--- /dev/null
+++
b/tika-server/tika-server-core/src/test/resources/configs/tika-config-server-async-metrics.json
@@ -0,0 +1,24 @@
+{
+ "server": {
+ "port": 9999,
+ "allowPipes": true,
+ "allowPerRequestConfig": true,
+ "endpoints": [
+ "rmeta",
+ "async",
+ "status"
+ ]
+ },
+ "parse-context": {
+ "timeout-limits": {
+ "progressTimeoutMillis": 120000
+ }
+ },
+ "pipes": {
+ "numClients": 2,
+ "forkedJvmArgs": [
+ "-Xmx256m"
+ ]
+ },
+ "plugin-roots": "target/plugins"
+}
diff --git
a/tika-server/tika-server-core/src/test/resources/configs/tika-config-server-metrics-restarts.json
b/tika-server/tika-server-core/src/test/resources/configs/tika-config-server-metrics-restarts.json
new file mode 100644
index 0000000000..665245531c
--- /dev/null
+++
b/tika-server/tika-server-core/src/test/resources/configs/tika-config-server-metrics-restarts.json
@@ -0,0 +1,18 @@
+{
+ "server": {
+ "port": 9999,
+ "allowPipes": true,
+ "endpoints": [
+ "rmeta"
+ ]
+ },
+ "pipes": {
+ "numClients": 1,
+ "maxFilesProcessedPerProcess": 2,
+ "socketTimeoutMillis": 2000,
+ "forkedJvmArgs": [
+ "-Xmx256m"
+ ]
+ },
+ "plugin-roots": "target/plugins"
+}
diff --git
a/tika-server/tika-server-core/src/test/resources/configs/tika-config-server-metrics-shared.json
b/tika-server/tika-server-core/src/test/resources/configs/tika-config-server-metrics-shared.json
new file mode 100644
index 0000000000..5a558f52bc
--- /dev/null
+++
b/tika-server/tika-server-core/src/test/resources/configs/tika-config-server-metrics-shared.json
@@ -0,0 +1,17 @@
+{
+ "server": {
+ "port": 9999,
+ "allowPipes": true,
+ "endpoints": [
+ "rmeta"
+ ]
+ },
+ "pipes": {
+ "useSharedServer": true,
+ "numClients": 2,
+ "forkedJvmArgs": [
+ "-Xmx256m"
+ ]
+ },
+ "plugin-roots": "target/plugins"
+}
diff --git
a/tika-server/tika-server-core/src/test/resources/configs/tika-config-server-metrics-timeout.json
b/tika-server/tika-server-core/src/test/resources/configs/tika-config-server-metrics-timeout.json
new file mode 100644
index 0000000000..53be01e315
--- /dev/null
+++
b/tika-server/tika-server-core/src/test/resources/configs/tika-config-server-metrics-timeout.json
@@ -0,0 +1,21 @@
+{
+ "server": {
+ "port": 9999,
+ "allowPipes": true,
+ "endpoints": [
+ "rmeta"
+ ]
+ },
+ "parse-context": {
+ "timeout-limits": {
+ "progressTimeoutMillis": 5000
+ }
+ },
+ "pipes": {
+ "numClients": 1,
+ "forkedJvmArgs": [
+ "-Xmx256m"
+ ]
+ },
+ "plugin-roots": "target/plugins"
+}
diff --git
a/tika-server/tika-server-core/src/test/resources/configs/tika-config-server-metrics.json
b/tika-server/tika-server-core/src/test/resources/configs/tika-config-server-metrics.json
new file mode 100644
index 0000000000..16dc5c799b
--- /dev/null
+++
b/tika-server/tika-server-core/src/test/resources/configs/tika-config-server-metrics.json
@@ -0,0 +1,10 @@
+{
+ "server": {
+ "port": 9999,
+ "endpoints": [
+ "tika"
+ ],
+ "metricsPort": 9404
+ },
+ "plugin-roots": "target/plugins"
+}
diff --git
a/tika-server/tika-server-standard/src/main/appended-resources/META-INF/LICENSE
b/tika-server/tika-server-standard/src/main/appended-resources/META-INF/LICENSE
index f4f9fb4839..05c659a0d3 100644
---
a/tika-server/tika-server-standard/src/main/appended-resources/META-INF/LICENSE
+++
b/tika-server/tika-server-standard/src/main/appended-resources/META-INF/LICENSE
@@ -1872,3 +1872,87 @@ jetty-session, jetty-util)
Apache Tika elects the Apache License, Version 2.0, whose terms appear
above. The Jetty jars do not carry a copy of either license; both texts
are available at the URLs above.
+
+Prometheus Java client (prometheus-metrics-core, prometheus-metrics-config,
+prometheus-metrics-model, prometheus-metrics-exposition-formats,
+prometheus-metrics-exposition-textformats, prometheus-metrics-tracer-common)
+
+ Copyright The Prometheus Authors
+
+ Licensed under the Apache License, Version 2.0, whose terms appear above.
+ The jars do not carry a copy of the license; the upstream project is
+ https://github.com/prometheus/client_java
+
+ prometheus-metrics-exposition-formats bundles a shaded copy of Protocol
+ Buffers (protobuf-java), Copyright 2008 Google Inc., under the BSD 3-clause
+ license:
+
+ Redistribution and use in source and binary forms, with or without
+ modification, are permitted provided that the following conditions are
+ met:
+
+ * Redistributions of source code must retain the above copyright
+ notice, this list of conditions and the following disclaimer.
+ * Redistributions in binary form must reproduce the above
+ copyright notice, this list of conditions and the following disclaimer
+ in the documentation and/or other materials provided with the
+ distribution.
+ * Neither the name of Google Inc. nor the names of its
+ contributors may be used to endorse or promote products derived from
+ this software without specific prior written permission.
+
+ THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+HdrHistogram library (HdrHistogram)
+
+ Written by Gil Tene, Michael Barker and Matt Warren, and released to the
+ public domain, as explained at
+ https://creativecommons.org/publicdomain/zero/1.0/
+
+ HdrHistogram is dual-licensed: CC0 1.0 Universal (public domain
+ dedication) or the 2-clause BSD license below. Apache Tika elects the
+ 2-clause BSD license. The jar carries both texts in META-INF/LICENSE.txt;
+ the upstream project is https://github.com/HdrHistogram/HdrHistogram
+
+ Redistribution and use in source and binary forms, with or without
+ modification, are permitted provided that the following conditions are met:
+
+ 1. Redistributions of source code must retain the above copyright notice,
+ this list of conditions and the following disclaimer.
+
+ 2. Redistributions in binary form must reproduce the above copyright
+ notice, this list of conditions and the following disclaimer in the
+ documentation and/or other materials provided with the distribution.
+
+ THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+ LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ POSSIBILITY OF SUCH DAMAGE.
+
+Micrometer (micrometer-core, micrometer-commons, micrometer-observation,
+micrometer-registry-prometheus)
+
+ Copyright (c) 2017-Present VMware, Inc. All Rights Reserved.
+
+ Licensed under the Apache License, Version 2.0, whose terms appear above.
+
+JSpecify (jspecify)
+
+ Licensed under the Apache License, Version 2.0, whose terms appear above.
