TIKA-4839 - add micrometer to tika-server

Author: tballison

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

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]

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

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.

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>

tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PerClientServerManager.java

@@ -180,6 +180,7 @@ private static long totalMemorySize() {
     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 void incrementFilesProcessed(long maxFilesPerProcess) {
         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 boolean needsRestart() {
 
     @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 int handleCrashAndGetExitCode() {
                     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 synchronized void ensureRunning() throws IOException, InterruptedExceptio
         if (isRunning() && !pendingRestart) {
             return;
         }
+        Process previous = process;
         startServer();
+        if (previous != null) {
+            restarts.restarted(exitCodeOrMinusOne(previous));
+        }
     }
 
     @Override
@@ -374,6 +395,7 @@ public Socket connect(int socketTimeoutMillis) throws IOException, ServerInitial
                     // 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 Socket connect(int socketTimeoutMillis) throws IOException, ServerInitial
         }
     }
 
+    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");