This is an automated email from the ASF dual-hosted git repository. tballison pushed a commit to branch TIKA-4839-metrics in repository https://gitbox.apache.org/repos/asf/tika.git
commit aaaa67795934f8ca23df0439d3c7e6f9d949c67d Author: tallison <[email protected]> AuthorDate: Mon Aug 24 13:52:50 2026 -0400 TIKA-4839 - add micrometer to tika-server --- CHANGES.txt | 18 ++ docs/modules/ROOT/nav.adoc | 1 + .../ROOT/pages/using-tika/server/index.adoc | 11 +- .../ROOT/pages/using-tika/server/monitoring.adoc | 208 +++++++++++++++++++ tika-parent/pom.xml | 11 + .../tika/pipes/core/PerClientServerManager.java | 28 ++- .../org/apache/tika/pipes/core/PipesClient.java | 12 +- .../org/apache/tika/pipes/core/PipesParser.java | 18 ++ .../org/apache/tika/pipes/core/RestartCounter.java | 63 ++++++ .../org/apache/tika/pipes/core/RestartReason.java | 31 +++ .../org/apache/tika/pipes/core/ServerManager.java | 10 + .../tika/pipes/core/SharedServerManager.java | 19 +- .../tika/pipes/core/async/AsyncProcessor.java | 5 + .../apache/tika/pipes/core/server/PipesServer.java | 7 +- .../apache/tika/pipes/core/RestartCounterTest.java | 67 ++++++ tika-server/tika-server-core/pom.xml | 8 + .../org/apache/tika/server/core/TikaServerCli.java | 5 + .../apache/tika/server/core/TikaServerConfig.java | 33 +++ .../apache/tika/server/core/TikaServerProcess.java | 91 ++++++++- .../tika/server/core/metrics/MetricsConfig.java | 62 ++++++ .../tika/server/core/metrics/MetricsServer.java | 99 +++++++++ .../server/core/metrics/TikaMetricsFilter.java | 117 +++++++++++ .../server/core/metrics/TikaServerMetrics.java | 225 +++++++++++++++++++++ .../tika/server/core/resource/AsyncResource.java | 4 + .../tika/server/core/TikaServerConfigTest.java | 44 ++++ .../core/TikaServerMetricsIntegrationTest.java | 129 ++++++++++++ .../server/core/metrics/MetricsServerTest.java | 84 ++++++++ .../server/core/metrics/TikaMetricsFilterTest.java | 156 ++++++++++++++ .../configs/tika-config-server-metrics.json | 16 ++ .../src/main/appended-resources/META-INF/LICENSE | 10 + 30 files changed, 1581 insertions(+), 11 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index eea5ad329b..ff6e11239d 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,5 +1,23 @@ Release 4.1.0 - unreleased + * tika-server can publish Prometheus metrics via Micrometer. Set + --metricsPort <port> (or server.metrics.port in the JSON config) and a + separate scrape listener serves GET /metrics on that port; unset, nothing + is created. Meters: tika_server_requests_seconds (endpoint/method/status + class, fixed 12-bucket histogram), request/response size summaries, + tika_server_rejected_total (busy_429, crash_503, payload_413), + tika_pipes_workers (busy/idle), tika_pipes_worker_restarts_total by + reason (oom, timeout, crash, max_files, idle, connection_abandoned), + tika_pipes_queue_depth, the parse port's Jetty thread pool, and the JVM + binders. The scrape listener binds to the server host by default + (server.metrics.host overrides). New public API in tika-pipes-core: + RestartReason, ServerManager.markServerForRestart(RestartReason), + PipesParser.getNumClients/getIdleClientCount/getRestartCount and + AsyncProcessor.getQueueDepth. A forked PipesServer that shuts down after + its idle socket timeout now exits with code 24 (PipesServer.IDLE_EXIT_CODE) + instead of 0, so the parent can tell an idle exit from a requested + shutdown; 0 remains the SHUT_DOWN reply (TIKA-4839). + * The Kafka pipes iterator no longer stops at the first empty poll. A newly subscribed consumer spends its first poll(s) joining the group and returns empty even when the topic has a backlog, so the iterator could enqueue zero 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..0a288dcbb5 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 @@ -574,5 +581,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..dfb1ae770a --- /dev/null +++ b/docs/modules/ROOT/pages/using-tika/server/monitoring.adoc @@ -0,0 +1,208 @@ +// +// 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, + "metrics": { + "port": 9404, + "host": "127.0.0.1", + "commonTags": { "cluster": "blue" } + } + } +} +---- + +[cols="1,3"] +|=== +|Key |Meaning + +|`port` +|The scrape port. Unset means metrics are entirely off: no registry, no meters, no +listener. `--metricsPort` on the command line overrides the JSON value. Must differ from +the server port. + +|`host` +|Bind address for the scrape listener. Defaults 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. Set this to `127.0.0.1` to restrict scraping to a sidecar. + +|`commonTags` +|Labels added to every series, for example a cluster or tier name. +|=== + +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 answers only `GET /metrics` -- every parse endpoint is `404` +there, and `/metrics` is `404` on the parse port. It also has its own small thread pool, +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 to the response. `endpoint` +is the first path segment when it is one of the server's endpoints (`tika`, `rmeta`, +`meta`, `unpack`, `detect`, `language`, `mime-types`, `detectors`, `parsers`, `version`, +`status`, `pipes`, `async`), `other` for any other resource, `unmatched` for a `404` +that matched nothing. `status` is the class: `2xx`, `4xx`, `5xx`. + +|`tika_server_request_size_bytes` +|summary +|`endpoint` +|Request body size from `Content-Length`; chunked uploads have none and are not recorded. + +|`tika_server_response_size_bytes` +|summary +|`endpoint` +|Bytes of the response entity, before any compression. + +|`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 +| +|Parse/detect tasks in flight right now. + +|`tika_server_tasks_started_total` +|counter +| +|Tasks started since the server came up. + +|`jetty_threads_*` +|gauges +| +|The parse port's request thread pool. `jetty_threads_jobs` (queued jobs) is HTTP-level +backpressure: requests waiting for a thread. +|=== + +=== Forked workers + +[cols="2,1,2,3"] +|=== +|Meter |Type |Labels |Meaning + +|`tika_pipes_workers` +|gauge +|`state` +|Forked pipes workers `busy` and `idle`. `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 +|`reason` +|Forked workers restarted, by why: `oom`, `timeout`, `crash` (an unexplained exit), +`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), `connection_abandoned` (a request was interrupted mid-flight). Alert on +`oom`, `timeout` and `crash`; `max_files` and `idle` are expected. + +|`tika_pipes_queue_depth` +|gauge +| +|Tuples accepted by `/async` and not yet picked up by a worker. Only present when the +`async` endpoint is enabled. +|=== + +=== JVM and process + +The standard Micrometer binders are enabled: `jvm_memory_*`, `jvm_gc_*`, `jvm_threads_*`, +`process_cpu_usage`, `system_cpu_usage`, `process_files_*`, `process_uptime_seconds`. +These describe the server JVM only, not the forked workers. + +== What to scale and alert on + +* Saturation, for an autoscaler: `tika_pipes_workers{state="busy"}` as a ratio of the + total, the rate of `tika_server_rejected_total{reason="busy_429"}`, and + `jetty_threads_jobs`. These move before latency does, which CPU does not. +* Failure, for alerting: `rate(tika_pipes_worker_restarts_total{reason=~"oom|timeout|crash"}[5m])` + 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 c19d259e3c..c01fca2500 100644 --- a/tika-parent/pom.xml +++ b/tika-parent/pom.xml @@ -402,6 +402,7 @@ <log4j2.version>2.26.1</log4j2.version> <!-- can't update to 10, needs jdk21 --> <lucene.version>9.12.3</lucene.version> + <micrometer.version>1.17.1</micrometer.version> <maven.plugin.annotations.version>3.15.2</maven.plugin.annotations.version> <metadata.extractor.version>2.21.0</metadata.extractor.version> <microsoft.translator.version>0.6.2</microsoft.translator.version> @@ -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> 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..56900dceef 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,6 +282,7 @@ public class PerClientServerManager implements ServerManager { if (filesProcessed >= maxFilesPerProcess) { LOG.info("clientId={}: reached max files limit ({}/{}), marking for restart", clientId, filesProcessed, maxFilesPerProcess); + restarts.mark(RestartReason.MAX_FILES); pendingRestart = true; } } @@ -292,18 +294,31 @@ 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); + 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); + restarts.mark(RestartReason.CONNECTION_ABANDONED); pendingRestart = true; } @Override public int handleCrashAndGetExitCode() { + restarts.mark(RestartReason.CRASH); pendingRestart = true; if (process != null) { try { @@ -312,6 +327,8 @@ public class PerClientServerManager implements ServerManager { int exitValue = process.exitValue(); if (exitValue == 0) { LOG.info("clientId={}: process exited cleanly", 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 +352,11 @@ public class PerClientServerManager implements ServerManager { if (isRunning() && !pendingRestart) { return; } + Process previous = process; startServer(); + if (previous != null) { + restarts.restarted(exitCodeOrMinusOne(previous)); + } } @Override @@ -374,6 +395,7 @@ public class PerClientServerManager implements ServerManager { // 1. pb.start() fails (can't launch process) - handled in startServer() // 2. Server explicitly reports bad config via protocol - handled in waitForStartup() // 3. Exhausted all retry attempts - handled in maybeInit() + restarts.mark(RestartReason.CRASH); pendingRestart = true; throw new IOException( "Server process died before connecting (exit code " + exitValue + ") - will retry"); @@ -391,6 +413,10 @@ public class PerClientServerManager implements ServerManager { } } + static int exitCodeOrMinusOne(Process p) { + return p.isAlive() ? -1 : p.exitValue(); + } + private synchronized void startServer() throws IOException, InterruptedException, TimeoutException, ServerInitializationException { if (closed) { throw new IllegalStateException("PerClientServerManager is closed"); 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 7b2b34a974..34cea6cf44 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 @@ -404,7 +404,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(); + serverManager.markServerForRestart(RestartReason.TIMEOUT); closeConnection(); return buildFatalResult(t.getId(), t.getEmitKey(), TIMEOUT, intermediateResult.get()); } @@ -420,19 +420,19 @@ public class PipesClient implements Closeable { switch (msg.type()) { case OOM: String oomMsg = JsonPipesIpc.fromBytes(msg.payload(), String.class); - serverManager.markServerForRestart(); + serverManager.markServerForRestart(RestartReason.OOM); 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(); + serverManager.markServerForRestart(RestartReason.TIMEOUT); closeConnection(); return buildFatalResult(t.getId(), t.getEmitKey(), TIMEOUT, intermediateResult.get(), timeoutMsg); case UNSPECIFIED_CRASH: String crashMsg = JsonPipesIpc.fromBytes(msg.payload(), String.class); - serverManager.markServerForRestart(); + serverManager.markServerForRestart(RestartReason.CRASH); closeConnection(); return buildFatalResult(t.getId(), t.getEmitKey(), UNSPECIFIED_CRASH, intermediateResult.get(), crashMsg); @@ -466,7 +466,7 @@ 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(); + serverManager.markServerForRestart(RestartReason.TIMEOUT); closeConnection(); return buildFatalResult(t.getId(), t.getEmitKey(), TIMEOUT, intermediateResult.get(), ExceptionUtils.getStackTrace(e)); @@ -486,8 +486,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); } else if (exitCode == PipesMessageType.TIMEOUT.getExitCode().orElse(-1)) { status = PipesResult.RESULT_STATUS.TIMEOUT; + serverManager.markServerForRestart(RestartReason.TIMEOUT); } 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 dd68dca002..89b527b243 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,24 @@ 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) { + long total = 0; + for (ServerManager m : serverManagers) { + total += m.getRestartCount(reason); + } + return total; + } + /** * 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..141f84a97e --- /dev/null +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/RestartCounter.java @@ -0,0 +1,63 @@ +/* + * 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.server.PipesServer; + +/** + * Per-manager restart bookkeeping: the reason recorded while a restart is pending, and + * monotonic counts of restarts actually performed, by reason. + * <p> + * Last recorded reason wins: a crash is first seen generically and then refined + * (OOM/TIMEOUT) from the exit code, so the refinement must overwrite. + */ +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; + } + + /** + * Records that a restart happened. With no reason marked, the worker died on its own: + * the idle exit code is the idle shutdown, anything else -- including an unexplained + * 0 -- is a crash. + */ + void restarted(int exitCode) { + RestartReason reason = pending; + pending = null; + if (reason == null) { + reason = exitCode == PipesServer.IDLE_EXIT_CODE ? RestartReason.IDLE : RestartReason.CRASH; + } + counts.get(reason).increment(); + } + + 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..994667a5f3 --- /dev/null +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/RestartReason.java @@ -0,0 +1,31 @@ +/* + * 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 cleanly on its own after sitting idle (socket timeout). */ + IDLE +} 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 3a663193e6..e890e03d15 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 @@ -110,6 +110,16 @@ public interface ServerManager extends Closeable { // Default no-op for backward compatibility } + /** As {@link #markServerForRestart()}, attributing the restart to {@code reason}. */ + default void markServerForRestart(RestartReason reason) { + markServerForRestart(); + } + + /** Restarts performed so far for {@code reason}; monotonic, never reset. */ + default long getRestartCount(RestartReason reason) { + return 0; + } + /** * Signals that the client walked away from its connection, whether * mid-handshake or with a request still in flight. 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 ea6db80039..a62e7018c9 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 @@ -77,6 +77,7 @@ public class SharedServerManager implements ServerManager { private final AtomicLong filesProcessed = new AtomicLong(0); private volatile boolean restarting = false; 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 Process process; private Path tmpDir; @@ -142,10 +143,14 @@ public class SharedServerManager implements ServerManager { return; } restarting = true; + Process previous = process; try { startServer(); pendingRestart = false; // Clear the flag after successful restart filesProcessed.set(0); // Reset file counter on restart + if (previous != null) { + restarts.restarted(PerClientServerManager.exitCodeOrMinusOne(previous)); + } } finally { restarting = false; lock.notifyAll(); // Wake up any threads waiting in getPort() @@ -165,12 +170,23 @@ public class SharedServerManager implements ServerManager { */ @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()"); + LOG.debug("Server marked for restart ({}) - will restart on next ensureRunning()", reason); + restarts.mark(reason); pendingRestart = true; } } + @Override + public long getRestartCount(RestartReason reason) { + return restarts.count(reason); + } + /** * Increments the count of files processed and marks for restart if limit reached. */ @@ -184,6 +200,7 @@ public class SharedServerManager implements ServerManager { synchronized (lock) { LOG.info("Shared server reached max files limit ({}/{}), marking for restart", count, maxFilesPerProcess); + restarts.mark(RestartReason.MAX_FILES); pendingRestart = true; } } 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..5c02398fcb 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 @@ -247,6 +247,11 @@ public class AsyncProcessor implements Closeable { return fetchEmitTuples.remainingCapacity(); } + /** Tuples accepted but not yet picked up by a worker. */ + public int getQueueDepth() { + return fetchEmitTuples.size(); + } + /** * 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 ee14e103f5..0c3e247641 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 @@ -148,6 +148,11 @@ 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; + /** Exit code for the idle shutdown (no request within socketTimeoutMillis). Not 0: + * 0 is reserved for a parent-requested SHUT_DOWN, so the parent can tell a worker + * that timed out from one that stopped for any other reason. */ + public static final int IDLE_EXIT_CODE = 24; + private final long heartbeatIntervalMillis; private final String pipesClientId; @@ -402,7 +407,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..d19e47bc53 --- /dev/null +++ b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/RestartCounterTest.java @@ -0,0 +1,67 @@ +/* + * 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 org.junit.jupiter.api.Test; + +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() { + // generic crash first, then refined from the exit code + 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 testUnmarkedAttributedByExitCode() { + RestartCounter c = new RestartCounter(); + c.restarted(PipesServer.IDLE_EXIT_CODE); + c.restarted(0); // unexplained clean exit: the parent did not ask for it + c.restarted(1); + c.restarted(-1); + assertEquals(1, c.count(RestartReason.IDLE)); + assertEquals(3, c.count(RestartReason.CRASH)); + } + + @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)); + } +} diff --git a/tika-server/tika-server-core/pom.xml b/tika-server/tika-server-core/pom.xml index 90de66d67d..dee2f69761 100644 --- a/tika-server/tika-server-core/pom.xml +++ b/tika-server/tika-server-core/pom.xml @@ -88,6 +88,14 @@ <groupId>jakarta.servlet</groupId> <artifactId>jakarta.servlet-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/TikaServerCli.java b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TikaServerCli.java index 6e5da93fd5..57590d41b8 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 @@ -28,6 +28,8 @@ import org.apache.commons.cli.help.HelpFormatter; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.apache.tika.server.core.metrics.MetricsConfig; + public class TikaServerCli { private static final Logger LOG = LoggerFactory.getLogger(TikaServerCli.class); @@ -39,6 +41,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, + "enable Prometheus metrics on this port (default when enabled in config = " + + MetricsConfig.DEFAULT_PORT + ")"); 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..033285707e 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 @@ -33,6 +33,7 @@ import org.slf4j.LoggerFactory; import org.apache.tika.config.loader.TikaLoader; import org.apache.tika.exception.TikaConfigException; import org.apache.tika.exception.TikaException; +import org.apache.tika.server.core.metrics.MetricsConfig; public class TikaServerConfig { @@ -74,6 +75,7 @@ public class TikaServerConfig { private ArrayList<String> endpoints = new ArrayList<>(); private TlsConfig tlsConfig = new TlsConfig(); + private MetricsConfig metrics = new MetricsConfig(); /** * Config with only the defaults @@ -109,6 +111,11 @@ public class TikaServerConfig { settings.add("id"); } + if (commandLine.hasOption("metricsPort")) { + config.getMetrics().setPort(Integer.parseInt(commandLine.getOptionValue("metricsPort"))); + settings.add("metricsPort"); + } + config.validateConsistency(settings); return config; } @@ -180,6 +187,16 @@ public class TikaServerConfig { } } tlsConfig.checkInitialization(); + if (metrics.isEnabled()) { + int metricsPort = metrics.getPort(); + if (metricsPort < 0 || metricsPort > 65535) { + throw new TikaConfigException("metrics.port must be 0-65535, got " + metricsPort); + } + if (metricsPort != 0 && metricsPort == port) { + throw new TikaConfigException("metrics.port (" + metricsPort + + ") must differ from the server port"); + } + } } public String getHost() { @@ -269,6 +286,22 @@ public class TikaServerConfig { this.tlsConfig = tlsConfig; } + /** + * Prometheus metrics; off unless {@code metrics.port} (or {@code --metricsPort}) is set. + */ + public MetricsConfig getMetrics() { + return metrics; + } + + public void setMetrics(MetricsConfig metrics) { + this.metrics = metrics == null ? new MetricsConfig() : metrics; + } + + /** Bind address for the scrape listener: {@code metrics.host} if set, else the server host. */ + public String getMetricsHost() { + return metrics.getHost() == null ? host : metrics.getHost(); + } + 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..65126c1a50 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 @@ -52,6 +52,7 @@ import org.apache.cxf.rs.security.cors.CrossOriginResourceSharingFilter; import org.apache.cxf.service.factory.ServiceConstructionException; import org.apache.cxf.transport.common.gzip.GZIPInInterceptor; import org.apache.cxf.transport.common.gzip.GZIPOutInterceptor; +import org.apache.cxf.transport.http_jetty.JettyHTTPServerEngine; import org.apache.cxf.transport.http_jetty.JettyHTTPServerEngineFactory; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -70,6 +71,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; @@ -103,7 +107,7 @@ public class TikaServerProcess { private static final Logger LOG = LoggerFactory.getLogger(TikaServerProcess.class); public static int DO_NOT_RESTART_EXIT_VALUE = -100; - private static final List<String> VALID_ENDPOINTS = List.of("tika", "rmeta", "meta", + public static final List<String> VALID_ENDPOINTS = List.of("tika", "rmeta", "meta", "unpack", "detect", "language", "mime", "mime-types", "detectors", "parsers", "version", "status", "pipes", "async"); @@ -118,6 +122,7 @@ 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, "enable Prometheus metrics on this port"); options.addOption("?", "help", false, "this help message"); return options; } @@ -168,6 +173,42 @@ public class TikaServerProcess { System.exit(DO_NOT_RESTART_EXIT_VALUE); } LOG.info("Started Apache Tika server {} at {}", serverDetails.serverId, serverDetails.url); + if (serverDetails.metrics != null) { + bindJettyThreadPool(serverDetails); + startMetricsServer(serverDetails); + } + } + + // Read-only: the CXF-owned Jetty engine is only consulted for its request pool. + private static void bindJettyThreadPool(ServerDetails serverDetails) { + try { + JettyHTTPServerEngineFactory factory = serverDetails.sf.getBus() + .getExtension(JettyHTTPServerEngineFactory.class); + JettyHTTPServerEngine engine = factory == null ? null : + factory.retrieveJettyHTTPServerEngine(serverDetails.port); + if (engine == null || engine.getServer() == null) { + LOG.warn("metrics: CXF Jetty engine not found for port {}; skipping request " + + "thread pool metrics", serverDetails.port); + return; + } + serverDetails.metrics.bindJettyThreadPool(engine.getServer().getThreadPool()); + } catch (Exception e) { + LOG.warn("metrics: could not bind request thread pool metrics", e); + } + } + + 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 on {}:{}{}", serverDetails.metricsHost, + serverDetails.metricsServer.getPort(), MetricsServer.PATH); } /** @@ -185,9 +226,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 +300,24 @@ public class TikaServerProcess { List<Object> providers = new ArrayList<>(); loadAllProviders(tikaServerConfig, serverStatus, tikaResource, resourceProviders, providers); + TikaServerMetrics metrics = null; + MetricsServer metricsServer = null; + if (tikaServerConfig.getMetrics().isEnabled()) { + metrics = new TikaServerMetrics(tikaServerConfig.getMetrics()); + metrics.bindJvm(); + metrics.bindServerStatus(serverStatus); + if (pipesParsingHelper != null) { + metrics.bindPipes(pipesParsingHelper.getPipesParser()); + } + AsyncResource asyncResource = findAsyncResource(resourceProviders); + if (asyncResource != null) { + metrics.bindAsyncQueue(asyncResource); + } + providers.add(new TikaMetricsFilter(metrics)); + metricsServer = new MetricsServer(tikaServerConfig.getMetricsHost(), + tikaServerConfig.getMetrics().getPort(), metrics); + } + sf.setResourceProviders(resourceProviders); sf.setProviders(providers); @@ -290,9 +359,23 @@ public class TikaServerProcess { details.serverId = tikaServerConfig.getId(); details.serverStatus = serverStatus; details.pipesParsingHelper = pipesParsingHelper; + details.port = port; + details.metrics = metrics; + details.metricsServer = metricsServer; + details.metricsHost = tikaServerConfig.getMetricsHost(); 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 @@ -551,7 +634,7 @@ public class TikaServerProcess { * Root segment of the class-level {@code @Path}, walking up the hierarchy: * {@code @Path} is not {@code @Inherited}, but JAX-RS resolves it from superclasses. */ - static String resourcePathRoot(Class<?> resourceClass) { + public static String resourcePathRoot(Class<?> resourceClass) { for (Class<?> c = resourceClass; c != null && c != Object.class; c = c.getSuperclass()) { jakarta.ws.rs.Path path = c.getAnnotation(jakarta.ws.rs.Path.class); if (path != null) { @@ -753,5 +836,9 @@ public class TikaServerProcess { String url; ServerStatus serverStatus; PipesParsingHelper pipesParsingHelper; + int port; + TikaServerMetrics metrics; + MetricsServer metricsServer; + String metricsHost; } } diff --git a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/metrics/MetricsConfig.java b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/metrics/MetricsConfig.java new file mode 100644 index 0000000000..25b773760c --- /dev/null +++ b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/metrics/MetricsConfig.java @@ -0,0 +1,62 @@ +/* + * 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.LinkedHashMap; +import java.util.Map; + +/** + * The {@code metrics} block of the tika-server config. The port is the only switch: + * unset means no registry, no meters, no scrape listener. + */ +public class MetricsConfig { + + public static final int DEFAULT_PORT = 9404; + + private Integer port; + private String host; + private Map<String, String> commonTags = new LinkedHashMap<>(); + + public boolean isEnabled() { + return port != null; + } + + public Integer getPort() { + return port; + } + + public void setPort(Integer port) { + this.port = port; + } + + /** Bind address for the scrape listener; null follows the server's own host. */ + public String getHost() { + return host; + } + + public void setHost(String host) { + this.host = host; + } + + public Map<String, String> getCommonTags() { + return commonTags; + } + + public void setCommonTags(Map<String, String> commonTags) { + this.commonTags = commonTags == null ? new LinkedHashMap<>() : commonTags; + } +} 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..2436592437 --- /dev/null +++ b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/metrics/MetricsServer.java @@ -0,0 +1,99 @@ +/* + * 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.Handler; +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; + +/** + * The scrape listener: a separate, minimal Jetty server that answers {@code GET /metrics} + * and nothing else. Deliberately not a connector on the CXF-owned server, so the parse + * endpoints are unreachable on this port by construction and a slow parse can never + * queue a scrape -- it has its own (tiny) thread pool. + */ +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; + + 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); + connector = new ServerConnector(server); + connector.setHost(host); + connector.setPort(port); + server.addConnector(connector); + server.setHandler(new ScrapeHandler(metrics)); + } + + public void start() throws Exception { + server.start(); + } + + /** 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) { + if (!PATH.equals(Request.getPathInContext(request))) { + Response.writeError(request, response, callback, HttpStatus.NOT_FOUND_404); + return true; + } + if (!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, 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..e04ce7c673 --- /dev/null +++ b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/metrics/TikaMetricsFilter.java @@ -0,0 +1,117 @@ +/* + * 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.io.IOException; +import java.io.OutputStream; +import java.util.List; +import java.util.Set; + +import jakarta.ws.rs.WebApplicationException; +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; +import jakarta.ws.rs.ext.WriterInterceptor; +import jakarta.ws.rs.ext.WriterInterceptorContext; +import org.apache.commons.io.output.CountingOutputStream; + +import org.apache.tika.server.core.TikaServerProcess; + +/** + * Records {@code tika_server_requests} and the size summaries for every request on the + * parse-side listener. + * <p> + * Timing starts pre-matching so unmatched (404) and aborted (413) requests are counted + * too. The endpoint tag is the first path segment when it names one of the server's + * endpoints; any other matched resource is {@code other} and an unmatched request is + * {@code unmatched}. Response bytes are counted by wrapping the entity stream, since a + * response filter runs before a streamed body is written. + * <p> + * Not counted: an exception no {@code ExceptionMapper} handles. CXF answers those from + * its fault chain, which does not run response filters. + */ +@Provider +@PreMatching +public class TikaMetricsFilter implements ContainerRequestFilter, ContainerResponseFilter, + WriterInterceptor { + + static final String START_NANOS = TikaMetricsFilter.class.getName() + ".start"; + static final String ENDPOINT = TikaMetricsFilter.class.getName() + ".endpoint"; + static final String UNMATCHED = "unmatched"; + static final String OTHER = "other"; + + private final TikaServerMetrics metrics; + private final Set<String> knownEndpoints; + + public TikaMetricsFilter(TikaServerMetrics metrics) { + this(metrics, Set.copyOf(TikaServerProcess.VALID_ENDPOINTS)); + } + + 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) { + String endpoint = endpoint(requestContext); + requestContext.setProperty(ENDPOINT, endpoint); + 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); + } + } + + @Override + public void aroundWriteTo(WriterInterceptorContext context) + throws IOException, WebApplicationException { + OutputStream original = context.getOutputStream(); + CountingOutputStream counting = new CountingOutputStream(original); + context.setOutputStream(counting); + try { + context.proceed(); + } finally { + Object endpoint = context.getProperty(ENDPOINT); + metrics.recordResponseSize(endpoint instanceof String e ? e : OTHER, + counting.getByteCount()); + } + } + + private String endpoint(ContainerRequestContext requestContext) { + List<Object> matched = requestContext.getUriInfo().getMatchedResources(); + if (matched == null || matched.isEmpty()) { + return UNMATCHED; + } + String path = requestContext.getUriInfo().getPath(); + int slash = path.indexOf('/'); + String root = slash < 0 ? path : path.substring(0, slash); + return knownEndpoints.contains(root) ? root : 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..3316a58e5f --- /dev/null +++ b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/metrics/TikaServerMetrics.java @@ -0,0 +1,225 @@ +/* + * 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.ArrayList; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.concurrent.TimeUnit; + +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.JvmGcMetrics; +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.ProcessorMetrics; +import io.micrometer.core.instrument.binder.system.UptimeMetrics; +import io.micrometer.prometheusmetrics.PrometheusConfig; +import io.micrometer.prometheusmetrics.PrometheusMeterRegistry; +import org.eclipse.jetty.util.thread.ThreadPool; + +import org.apache.tika.pipes.core.PipesParser; +import org.apache.tika.pipes.core.RestartReason; +import org.apache.tika.server.core.ServerStatus; +import org.apache.tika.server.core.resource.AsyncResource; + +/** + * Owns the one {@link PrometheusMeterRegistry} for a tika-server process and every meter + * recorded against it. Constructed only when metrics are enabled; callers hold null + * otherwise, so no meter exists when the feature is off. + * <p> + * Every tag value is drawn from a bounded set (resource root, status class, enum names): + * nothing request-derived ever 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 + }; + + public static final String REQUESTS = "tika_server_requests"; + public static final String REQUEST_SIZE = "tika_server_request_size_bytes"; + public static final String RESPONSE_SIZE = "tika_server_response_size_bytes"; + public static final String REJECTED = "tika_server_rejected_total"; + public static final String TASKS_ACTIVE = "tika_server_tasks_active"; + public static final String TASKS_STARTED = "tika_server_tasks_started_total"; + public static final String PIPES_WORKERS = "tika_pipes_workers"; + public static final String PIPES_RESTARTS = "tika_pipes_worker_restarts_total"; + public static final String PIPES_QUEUE_DEPTH = "tika_pipes_queue_depth"; + + private final PrometheusMeterRegistry registry; + private final List<AutoCloseable> closeables = new ArrayList<>(); + + public TikaServerMetrics(MetricsConfig config) { + this.registry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT); + for (Map.Entry<String, String> e : config.getCommonTags().entrySet()) { + registry.config().commonTags(e.getKey(), e.getValue()); + } + } + + public MeterRegistry getRegistry() { + return registry; + } + + /** Prometheus text exposition (format 0.0.4). */ + public String scrape() { + return registry.scrape(); + } + + public void bindJvm() { + JvmGcMetrics gc = new JvmGcMetrics(); + gc.bindTo(registry); + closeables.add(gc); + new JvmMemoryMetrics().bindTo(registry); + new JvmThreadMetrics().bindTo(registry); + new ProcessorMetrics().bindTo(registry); + new UptimeMetrics().bindTo(registry); + new FileDescriptorMetrics().bindTo(registry); + } + + public void bindServerStatus(ServerStatus serverStatus) { + Gauge.builder(TASKS_ACTIVE, serverStatus, s -> s.getTasks().size()) + .description("Parse/detect tasks currently running in this server") + .register(registry); + FunctionCounter.builder(TASKS_STARTED, serverStatus, ServerStatus::getFilesProcessed) + .description("Parse/detect tasks started since server start") + .register(registry); + } + + public void bindPipes(PipesParser pipesParser) { + Gauge.builder(PIPES_WORKERS, pipesParser, p -> p.getNumClients() - p.getIdleClientCount()) + .tag("state", "busy") + .description("Forked pipes workers by state") + .register(registry); + Gauge.builder(PIPES_WORKERS, pipesParser, PipesParser::getIdleClientCount) + .tag("state", "idle") + .description("Forked pipes workers by state") + .register(registry); + for (RestartReason reason : RestartReason.values()) { + FunctionCounter.builder(PIPES_RESTARTS, pipesParser, p -> p.getRestartCount(reason)) + .tag("reason", tagValue(reason.name())) + .description("Forked pipes worker restarts by reason") + .register(registry); + } + } + + public void bindAsyncQueue(AsyncResource asyncResource) { + Gauge.builder(PIPES_QUEUE_DEPTH, asyncResource, AsyncResource::getQueueDepth) + .description("Tuples waiting in the /async queue") + .register(registry); + } + + /** The CXF/Jetty request thread pool: queued jobs are HTTP-level backpressure. */ + public void bindJettyThreadPool(ThreadPool threadPool) { + new io.micrometer.core.instrument.binder.jetty.JettyServerThreadPoolMetrics( + threadPool, List.of()).bindTo(registry); + } + + void recordRequest(String endpoint, String method, int status, long nanos) { + Timer.builder(REQUESTS) + .tag("endpoint", endpoint) + .tag("method", method) + .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("reason", rejected) + .description("Requests refused for capacity reasons") + .register(registry) + .increment(); + } + } + + void recordRequestSize(String endpoint, long bytes) { + sizeSummary(REQUEST_SIZE, endpoint, "Request body bytes (Content-Length)") + .record(bytes); + } + + void recordResponseSize(String endpoint, long bytes) { + sizeSummary(RESPONSE_SIZE, endpoint, "Response entity bytes before compression") + .record(bytes); + } + + private DistributionSummary sizeSummary(String name, String endpoint, String description) { + return DistributionSummary.builder(name) + .tag("endpoint", endpoint) + .baseUnit("bytes") + .description(description) + .serviceLevelObjectives(SIZE_SLOS) + .register(registry); + } + + static String statusClass(int status) { + if (status < 200 || status >= 600) { + return "other"; + } + return (status / 100) + "xx"; + } + + /** + * By status only, which is what tika-server's mapping already encodes: 429 is + * capacity (the /async queue full, or no free fork within maxWaitForClientMillis), + * 503 is a fork that OOM'd, timed out or crashed, 413 is an over-limit body. + */ + static String rejectedReason(int status) { + return switch (status) { + case 429 -> "busy_429"; + case 413 -> "payload_413"; + case 503 -> "crash_503"; + default -> null; + }; + } + + static String tagValue(String enumName) { + return enumName.toLowerCase(Locale.ROOT); + } + + @Override + public void close() { + for (AutoCloseable c : closeables) { + try { + c.close(); + } catch (Exception ignore) { + // best-effort teardown of JVM binders + } + } + 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..9c646da885 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 int getQueueDepth() { + return asyncProcessor.getQueueDepth(); + } + 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..64e5d3ca8e 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 @@ -238,4 +238,48 @@ 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[]{})); + assertFalse(config.getMetrics().isEnabled()); + assertEquals(config.getHost(), config.getMetricsHost()); + } + + @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())})); + assertTrue(config.getMetrics().isEnabled()); + assertEquals(9404, config.getMetrics().getPort()); + assertEquals("127.0.0.1", config.getMetricsHost()); + assertEquals("blue", config.getMetrics().getCommonTags().get("cluster")); + } + + @Test + public void testMetricsPortFromCommandLine() throws Exception { + TikaServerConfig config = TikaServerConfig.load(new DefaultParser().parse(metricsOptions(), + new String[]{"-p", "9998", "--metricsPort", "9500"})); + assertTrue(config.getMetrics().isEnabled()); + assertEquals(9500, config.getMetrics().getPort()); + // no metrics.host: follows the server host + assertEquals(TikaServerConfig.DEFAULT_HOST, config.getMetricsHost()); + } + + @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("metrics.port", 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..9f1ce0e973 --- /dev/null +++ b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerMetricsIntegrationTest.java @@ -0,0 +1,129 @@ +/* + * 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.assertTrue; + +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.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", "reason=\"oom\"", 1.0); + assertSample(body, "tika_pipes_workers", "state=\"idle\"", 2.0); + assertSample(body, "tika_pipes_workers", "state=\"busy\"", 0.0); + assertTrue(body.contains("tika_server_tasks_started_total"), body); + assertTrue(body.contains("jvm_memory_used_bytes"), body); + assertTrue(body.contains("tika_server_request_size_bytes_count{endpoint=\"rmeta\"}"), body); + assertTrue(body.contains("jetty_threads_"), "CXF request pool metrics missing:\n" + 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()); + } + + 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) { + Matcher m = Pattern.compile("^" + Pattern.quote(name) + "\\{" + Pattern.quote(labels) + + ",?\\} (\\S+)$", Pattern.MULTILINE).matcher(body); + assertTrue(m.find(), "missing " + name + "{" + labels + "} in:\n" + body); + assertEquals(expected, Double.parseDouble(m.group(1)), 0.0, name + "{" + labels + "}"); + } + + 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..17cc2e0991 --- /dev/null +++ b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/metrics/MetricsServerTest.java @@ -0,0 +1,84 @@ +/* + * 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.assertTrue; + +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.util.Map; + +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 { + MetricsConfig config = new MetricsConfig(); + config.setCommonTags(Map.of("cluster", "test")); + metrics = new TikaServerMetrics(config); + 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{cluster=\"test\"} 3.0"), body); + } + + @Test + public void testOnlyMetricsPathIsServed() throws Exception { + assertEquals(404, get(base + "/tika").statusCode()); + assertEquals(404, get(base + "/").statusCode()); + assertEquals(404, get(base + "/metrics/").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()); + } + + 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/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..b6f74a1542 --- /dev/null +++ b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/metrics/TikaMetricsFilterTest.java @@ -0,0 +1,156 @@ +/* + * 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.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.ByteArrayInputStream; +import java.io.InputStream; +import java.util.ArrayList; +import java.util.List; + +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 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.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"); + } + } + + @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(new MetricsConfig()); + 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)); + sf.setProviders(providers); + } + + @Test + public void testSuccessRecordsTimerAndSizes() throws Exception { + long helloWorldBytes; + try (InputStream is = ClassLoader.getSystemResourceAsStream(HELLO_WORLD)) { + helloWorldBytes = is.readAllBytes().length; + } + 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); + + Timer timer = timer("tika", "PUT", "2xx"); + assertNotNull(timer); + assertEquals(1, timer.count()); + DistributionSummary requestSize = metrics.getRegistry() + .find(TikaServerMetrics.REQUEST_SIZE).tag("endpoint", "tika").summary(); + assertNotNull(requestSize); + assertEquals(helloWorldBytes, requestSize.totalAmount(), 0.0); + // Recorded after the last byte is flushed, so possibly a beat after the client returns. + DistributionSummary responseSize = null; + for (int i = 0; i < 50 && responseSize == null; i++) { + responseSize = metrics.getRegistry() + .find(TikaServerMetrics.RESPONSE_SIZE).tag("endpoint", "tika").summary(); + if (responseSize == null) { + Thread.sleep(50); + } + } + assertNotNull(responseSize); + assertTrue(responseSize.totalAmount() >= 11.0, "response bytes: " + responseSize.totalAmount()); + } + + @Test + public void testPayloadTooLargeCountsAsRejected() { + Response response = WebClient + .create(endPoint + "/tika/text") + .put(new ByteArrayInputStream(new byte[(int) MAX_BYTES * 4])); + assertEquals(413, response.getStatus()); + assertEquals(1, timer("tika", "PUT", "4xx").count()); + assertEquals(1.0, rejected("payload_413").count(), 0.0); + } + + @Test + public void testThrownExceptionIs5xx() { + 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(1, timer(TikaMetricsFilter.OTHER, "GET", "5xx").count()); + } + + @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, timer(TikaMetricsFilter.UNMATCHED, "GET", "4xx").count()); + } + + 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(); + } + + private Counter rejected(String reason) { + return metrics.getRegistry().find(TikaServerMetrics.REJECTED).tag("reason", reason).counter(); + } +} 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..9ce5c95ff2 --- /dev/null +++ b/tika-server/tika-server-core/src/test/resources/configs/tika-config-server-metrics.json @@ -0,0 +1,16 @@ +{ + "server": { + "port": 9999, + "endpoints": [ + "tika" + ], + "metrics": { + "port": 9404, + "host": "127.0.0.1", + "commonTags": { + "cluster": "blue" + } + } + }, + "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..a48fe3c4f9 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,13 @@ 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
