This is an automated email from the ASF dual-hosted git repository. tballison pushed a commit to branch TIKA-4846 in repository https://gitbox.apache.org/repos/asf/tika.git
commit a1406515ede22a0a5c893e1aa51a474ae5d9393b Author: tallison <[email protected]> AuthorDate: Thu Aug 27 08:37:13 2026 -0400 TIKA-4846 -- add jsonl reporter --- CHANGES.txt | 7 + .../ROOT/examples/pipes-fs-jsonl-reporter.json | 1 + .../ROOT/pages/pipes/plugins/filesystem.adoc | 56 ++++++ docs/modules/ROOT/pages/pipes/reporters.adoc | 4 + .../tika-pipes-file-system/pom.xml | 5 + .../pipes/reporter/fs/FileSystemJsonlReporter.java | 217 +++++++++++++++++++++ .../reporter/fs/FileSystemJsonlReporterConfig.java | 45 +++++ .../fs/FileSystemJsonlReporterFactory.java | 41 ++++ .../apache/tika/pipes/fs/ConfigExamplesTest.java | 5 + .../reporter/fs/FileSystemJsonlReporterTest.java | 159 +++++++++++++++ .../file-system-jsonl-reporter.json | 10 + 11 files changed, 550 insertions(+) diff --git a/CHANGES.txt b/CHANGES.txt index d517042ccb..662496cedf 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,5 +1,12 @@ Release 4.1.0 - unreleased + * New file-system-jsonl-reporter pipes reporter: an append-only, one-JSON- + line-per-document audit log written by the driver process. Documents whose + forked worker crashed (OOM, TIMEOUT, UNSPECIFIED_CRASH) produce nothing in + the emitter's output; this file is the record that they were attempted. + Refuses to start over an existing file unless onExists is APPEND or + REPLACE (TIKA-4846). + * Add Micrometer reporting and opt-in endpoint for tika-server (TIKA-4839). * Improve spooling/decrease number of spills to disk (TIKA-4835). diff --git a/docs/modules/ROOT/examples/pipes-fs-jsonl-reporter.json b/docs/modules/ROOT/examples/pipes-fs-jsonl-reporter.json new file mode 120000 index 0000000000..e83e6c241b --- /dev/null +++ b/docs/modules/ROOT/examples/pipes-fs-jsonl-reporter.json @@ -0,0 +1 @@ +../../../../tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/test/resources/config-examples/file-system-jsonl-reporter.json \ No newline at end of file diff --git a/docs/modules/ROOT/pages/pipes/plugins/filesystem.adoc b/docs/modules/ROOT/pages/pipes/plugins/filesystem.adoc index 6fc9333f20..600ec03eba 100644 --- a/docs/modules/ROOT/pages/pipes/plugins/filesystem.adoc +++ b/docs/modules/ROOT/pages/pipes/plugins/filesystem.adoc @@ -40,6 +40,10 @@ The File System plugin (`tika-pipes-file-system`) is the most common starting po |Reporter |`file-system-reporter` |`FileSystemStatusReporter` + +|Reporter +|`file-system-jsonl-reporter` +|`FileSystemJsonlReporter` |=== == Complete Pipeline Example @@ -253,6 +257,58 @@ Tradeoffs: * Per-record `report()` calls are cheap (counter increment only). The cost of "watching" is bounded by the periodic write, not by document throughput. [#security-notes] +[#file-system-jsonl-reporter] +== File System JSONL Reporter (`file-system-jsonl-reporter`) + +Append-only per-document audit log: one JSON object per line for every result that passes the `includes`/`excludes` filter. Where the status reporter above summarizes counts, this one records *which* documents ended in which state, so a downstream process (a dead-letter queue, `tika-eval`) can act on them. + +A crash result (`OOM`, `TIMEOUT`, `UNSPECIFIED_CRASH`) leaves nothing in the emitter's output; this file is the only record that the document was attempted. The driver process writes it, so it survives the forked worker's death. + +[source,json] +---- +include::example$pipes-fs-jsonl-reporter.json[] +---- + +=== Line format + +[source,json] +---- +{"id":"reports/q3.pdf","status":"OOM","message":"...","elapsedMs":4120,"ts":"2026-08-27T14:02:11.482Z"} +---- + +* `id` — `FetchEmitTuple.getId()` verbatim. For the file system iterator that is the path relative to `basePath`, so it matches the emitted file name minus the extract suffix. +* `status` — `PipesResult.RESULT_STATUS` name. +* `message` — the result's message (a stack trace for crashes), capped at `maxMessageLength`. +* `elapsedMs` — wall-clock time the driver spent on this document. +* `ts` — ISO-8601 UTC timestamp of the report. + +If the pipeline dies, a final line of the form `{"error":"<stack trace>","ts":"..."}` is written and the file is flushed. + +=== Configuration + +[cols="1,1,3"] +|=== +|Field |Default |Description + +|`path` +|_required_ +|Path of the JSONL file. Missing parent directories are created at startup. + +|`onExists` +|`EXCEPTION` +|What to do when `path` already exists at startup: `EXCEPTION` refuses to start, `APPEND` continues the existing file, `REPLACE` truncates it. The default is deliberately strict — appending a new run onto an old ledger is the mistake this reporter exists to prevent. + +|`includes` / `excludes` +|_all statuses_ +|Mutually exclusive sets of `RESULT_STATUS` names. For a crash ledger, `includes` the crash and exception statuses; leave both unset for a full per-document audit trail. + +|`maxMessageLength` +|`10000` +|Characters of `message` to keep; longer messages are truncated with a suffix stating how many characters were dropped. +|=== + +Reports are queued and written by a single background thread that flushes whenever the queue is empty, so the ordering across documents is the order the driver finished them, not the iterator's order. If the writer fails (disk full), the next `report()` throws rather than dropping lines silently. + == Security Notes * **`basePath` is the sandbox boundary, and it is the only one.** With `basePath` set, the fetcher and emitter reject any key that resolves outside it, including absolute paths and `../` traversal. `allowAbsolutePaths` has no effect in this state. diff --git a/docs/modules/ROOT/pages/pipes/reporters.adoc b/docs/modules/ROOT/pages/pipes/reporters.adoc index 01bc05e604..bd59ddbf4e 100644 --- a/docs/modules/ROOT/pages/pipes/reporters.adoc +++ b/docs/modules/ROOT/pages/pipes/reporters.adoc @@ -58,6 +58,10 @@ Each entry's outer key is the reporter's component name — there is no separate |`file-system-reporter` |Writes a JSON status file periodically. Pair with an external watcher — see xref:pipes/plugins/filesystem.adoc#watching[Live status for watching applications]. +|xref:pipes/plugins/filesystem.adoc#file-system-jsonl-reporter[File System] +|`file-system-jsonl-reporter` +|Appends one JSON line per document to a file. The record of documents whose worker crashed (nothing reaches the emitter for those). + |xref:pipes/plugins/jdbc.adoc[JDBC] |`jdbc-reporter` |Writes per-doc status rows to a SQL table. diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/pom.xml b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/pom.xml index 06d1602e00..d6ce406eeb 100644 --- a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/pom.xml +++ b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/pom.xml @@ -41,6 +41,11 @@ <version>${project.version}</version> <scope>provided</scope> </dependency> + <dependency> + <groupId>${project.groupId}</groupId> + <artifactId>tika-pipes-reporter-commons</artifactId> + <version>${project.version}</version> + </dependency> <dependency> <groupId>com.fasterxml.jackson.datatype</groupId> <artifactId>jackson-datatype-jsr310</artifactId> diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/reporter/fs/FileSystemJsonlReporter.java b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/reporter/fs/FileSystemJsonlReporter.java new file mode 100644 index 0000000000..5fdf3dcf80 --- /dev/null +++ b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/reporter/fs/FileSystemJsonlReporter.java @@ -0,0 +1,217 @@ +/* + * 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.reporter.fs; + +import java.io.BufferedWriter; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.FileAlreadyExistsException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; +import java.time.Instant; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.TimeUnit; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.tika.exception.TikaConfigException; +import org.apache.tika.pipes.api.FetchEmitTuple; +import org.apache.tika.pipes.api.PipesResult; +import org.apache.tika.pipes.api.pipesiterator.TotalCountResult; +import org.apache.tika.pipes.reporters.PipesReporterBase; +import org.apache.tika.plugins.ExtensionConfig; +import org.apache.tika.utils.ExceptionUtils; + +/** + * Append-only per-document audit log: one JSON object per line for every result + * accepted by the includes/excludes filter. The line's {@code id} is the + * {@link FetchEmitTuple#getId()} verbatim, so consumers join on it. + * <p> + * A single writer thread drains a bounded queue and flushes whenever the queue + * runs dry, so a crash line is on disk within moments of being reported. A + * writer failure (disk full, etc.) fails the next {@link #report} rather than + * silently dropping lines. + */ +public class FileSystemJsonlReporter extends PipesReporterBase { + + private static final Logger LOG = LoggerFactory.getLogger(FileSystemJsonlReporter.class); + + private static final int QUEUE_SIZE = 10_000; + private static final long MAX_OFFER_WAIT_MS = 60_000; + private static final long CLOSE_WAIT_MS = 60_000; + private static final Line END = new Line(null, null, null, -1, null); + + public record Line(String id, String status, String message, long elapsedMs, String ts) { + } + + private record ErrorLine(String error, String ts) { + } + + public static FileSystemJsonlReporter build(ExtensionConfig pluginConfig) throws TikaConfigException, IOException { + FileSystemJsonlReporterConfig config = FileSystemJsonlReporterConfig.load(pluginConfig.json()); + return new FileSystemJsonlReporter(pluginConfig, config); + } + + private final FileSystemJsonlReporterConfig config; + private final ObjectMapper mapper = new ObjectMapper(); + private final ArrayBlockingQueue<Object> queue = new ArrayBlockingQueue<>(QUEUE_SIZE); + private final Thread writerThread; + private final BufferedWriter writer; + private volatile IOException writerFailure; + private volatile boolean closed; + + public FileSystemJsonlReporter(ExtensionConfig pluginConfig, FileSystemJsonlReporterConfig config) throws TikaConfigException, IOException { + super(pluginConfig, config.includes(), config.excludes()); + this.config = config; + if (config.path() == null) { + throw new TikaConfigException("must initialize 'path'"); + } + this.writer = open(config); + this.writerThread = new Thread(this::drain, "tika-jsonl-reporter"); + writerThread.setDaemon(true); + writerThread.start(); + } + + private static BufferedWriter open(FileSystemJsonlReporterConfig config) throws TikaConfigException, IOException { + Path path = config.path(); + if (path.getParent() != null) { + Files.createDirectories(path.getParent()); + } + StandardOpenOption[] options = switch (config.onExists()) { + case EXCEPTION -> new StandardOpenOption[]{StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE}; + case APPEND -> new StandardOpenOption[]{StandardOpenOption.CREATE, StandardOpenOption.WRITE, StandardOpenOption.APPEND}; + case REPLACE -> new StandardOpenOption[]{StandardOpenOption.CREATE, StandardOpenOption.WRITE, StandardOpenOption.TRUNCATE_EXISTING}; + }; + try { + return Files.newBufferedWriter(path, StandardCharsets.UTF_8, options); + } catch (FileAlreadyExistsException e) { + throw new TikaConfigException("'" + path + "' already exists; set onExists to APPEND or REPLACE to reuse it", e); + } + } + + @Override + public void report(FetchEmitTuple t, PipesResult result, long elapsed) { + if (!accept(result.status())) { + return; + } + enqueue(new Line(t.getId(), result.status().name(), truncate(result.message()), elapsed, Instant.now().toString())); + } + + private String truncate(String msg) { + if (msg == null || msg.length() <= config.maxMessageLength()) { + return msg; + } + return msg.substring(0, config.maxMessageLength()) + "...[truncated " + (msg.length() - config.maxMessageLength()) + " chars]"; + } + + private void enqueue(Object line) { + if (writerFailure != null) { + throw new IllegalStateException("jsonl reporter writer failed; refusing to drop lines silently", writerFailure); + } + if (closed) { + throw new IllegalStateException("jsonl reporter is closed"); + } + try { + if (!queue.offer(line, MAX_OFFER_WAIT_MS, TimeUnit.MILLISECONDS)) { + throw new IllegalStateException("jsonl reporter queue full for " + MAX_OFFER_WAIT_MS + " ms"); + } + } catch (InterruptedException e) { + LOG.warn("interrupted before queuing report for {}; line dropped", line); + Thread.currentThread().interrupt(); + } + } + + private void drain() { + try { + while (true) { + Object line = queue.take(); + if (line == END) { + return; + } + writer.write(mapper.writeValueAsString(line)); + writer.newLine(); + if (queue.isEmpty()) { + writer.flush(); + } + } + } catch (IOException e) { + LOG.error("jsonl reporter failed writing {}", config.path(), e); + writerFailure = e; + } catch (InterruptedException e) { + //fall through to close + } finally { + try { + writer.close(); + } catch (IOException e) { + LOG.warn("problem closing {}", config.path(), e); + } + } + } + + @Override + public void report(TotalCountResult totalCountResult) { + //no-op + } + + @Override + public boolean supportsTotalCount() { + return false; + } + + @Override + public void error(Throwable t) { + error(ExceptionUtils.getStackTrace(t)); + } + + @Override + public void error(String msg) { + // close() may never be called after this; get the line on disk now + try { + enqueue(new ErrorLine(truncate(msg), Instant.now().toString())); + } catch (IllegalStateException e) { + LOG.warn("couldn't record error in jsonl reporter", e); + } + finish(); + } + + @Override + public void close() throws IOException { + finish(); + if (writerFailure != null) { + throw writerFailure; + } + } + + private void finish() { + if (closed) { + return; + } + closed = true; + try { + if (!queue.offer(END, CLOSE_WAIT_MS, TimeUnit.MILLISECONDS)) { + LOG.warn("jsonl reporter queue never drained; interrupting writer"); + writerThread.interrupt(); + } + writerThread.join(CLOSE_WAIT_MS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } +} diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/reporter/fs/FileSystemJsonlReporterConfig.java b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/reporter/fs/FileSystemJsonlReporterConfig.java new file mode 100644 index 0000000000..bbc1ec09e0 --- /dev/null +++ b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/reporter/fs/FileSystemJsonlReporterConfig.java @@ -0,0 +1,45 @@ +/* + * 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.reporter.fs; + +import java.nio.file.Path; +import java.util.Set; + +import org.apache.tika.exception.TikaConfigException; +import org.apache.tika.plugins.PluginJson; + +public record FileSystemJsonlReporterConfig(Path path, Set<String> includes, Set<String> excludes, ON_EXISTS onExists, int maxMessageLength) { + + public enum ON_EXISTS { + EXCEPTION, APPEND, REPLACE + } + + public static final int DEFAULT_MAX_MESSAGE_LENGTH = 10_000; + + public FileSystemJsonlReporterConfig { + if (onExists == null) { + onExists = ON_EXISTS.EXCEPTION; + } + if (maxMessageLength <= 0) { + maxMessageLength = DEFAULT_MAX_MESSAGE_LENGTH; + } + } + + public static FileSystemJsonlReporterConfig load(final String json) throws TikaConfigException { + return PluginJson.read(json, FileSystemJsonlReporterConfig.class); + } +} diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/reporter/fs/FileSystemJsonlReporterFactory.java b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/reporter/fs/FileSystemJsonlReporterFactory.java new file mode 100644 index 0000000000..8d80a67166 --- /dev/null +++ b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/reporter/fs/FileSystemJsonlReporterFactory.java @@ -0,0 +1,41 @@ +/* + * 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.reporter.fs; + +import java.io.IOException; + +import org.pf4j.Extension; + +import org.apache.tika.exception.TikaConfigException; +import org.apache.tika.pipes.api.reporter.PipesReporterFactory; +import org.apache.tika.plugins.ExtensionConfig; + +@Extension +public class FileSystemJsonlReporterFactory implements PipesReporterFactory { + + public static final String NAME = "file-system-jsonl-reporter"; + + @Override + public String getName() { + return NAME; + } + + @Override + public FileSystemJsonlReporter buildExtension(ExtensionConfig extensionConfig) throws IOException, TikaConfigException { + return FileSystemJsonlReporter.build(extensionConfig); + } +} diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/test/java/org/apache/tika/pipes/fs/ConfigExamplesTest.java b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/test/java/org/apache/tika/pipes/fs/ConfigExamplesTest.java index 041e079e15..be1aaf91b0 100644 --- a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/test/java/org/apache/tika/pipes/fs/ConfigExamplesTest.java +++ b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/test/java/org/apache/tika/pipes/fs/ConfigExamplesTest.java @@ -42,4 +42,9 @@ public class ConfigExamplesTest extends AbstractConfigExamplesTest { public void testFileSystemPipelineConfig() throws Exception { loadAndValidate("file-system-pipeline.json"); } + + @Test + public void testFileSystemJsonlReporterConfig() throws Exception { + loadAndValidate("file-system-jsonl-reporter.json"); + } } diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/test/java/org/apache/tika/pipes/reporter/fs/FileSystemJsonlReporterTest.java b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/test/java/org/apache/tika/pipes/reporter/fs/FileSystemJsonlReporterTest.java new file mode 100644 index 0000000000..db9c9a082a --- /dev/null +++ b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/test/java/org/apache/tika/pipes/reporter/fs/FileSystemJsonlReporterTest.java @@ -0,0 +1,159 @@ +/* + * 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.reporter.fs; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import org.apache.tika.exception.TikaConfigException; +import org.apache.tika.pipes.api.FetchEmitTuple; +import org.apache.tika.pipes.api.PipesResult; +import org.apache.tika.pipes.api.emitter.EmitKey; +import org.apache.tika.pipes.api.fetcher.FetchKey; +import org.apache.tika.plugins.ExtensionConfig; + +public class FileSystemJsonlReporterTest { + + private static final ObjectMapper MAPPER = new ObjectMapper(); + + private static FileSystemJsonlReporter build(Path path, String extraJson) throws Exception { + String json = "{\"path\":\"" + path.toAbsolutePath().toString().replace("\\", "/") + "\"" + extraJson + "}"; + return new FileSystemJsonlReporterFactory().buildExtension(new ExtensionConfig("test", FileSystemJsonlReporterFactory.NAME, json)); + } + + private static void report(FileSystemJsonlReporter r, String id, PipesResult.RESULT_STATUS status, String msg) { + r.report(new FetchEmitTuple(id, new FetchKey("f", id), new EmitKey("e", id)), new PipesResult(status, msg), 7); + } + + private static List<Map<String, Object>> lines(Path path) throws IOException { + List<Map<String, Object>> ret = new ArrayList<>(); + for (String line : Files.readAllLines(path, StandardCharsets.UTF_8)) { + ret.add(MAPPER.readValue(line, Map.class)); + } + return ret; + } + + @Test + public void testIncludesAndFields(@TempDir Path tmp) throws Exception { + Path path = tmp.resolve("audit.jsonl"); + try (FileSystemJsonlReporter r = build(path, ",\"includes\":[\"OOM\",\"TIMEOUT\"]")) { + report(r, "a/b.pdf", PipesResult.RESULT_STATUS.OOM, "boom"); + report(r, "c.doc", PipesResult.RESULT_STATUS.PARSE_SUCCESS, null); + report(r, "d.doc", PipesResult.RESULT_STATUS.TIMEOUT, null); + } + List<Map<String, Object>> lines = lines(path); + assertEquals(2, lines.size()); + assertEquals("a/b.pdf", lines.get(0).get("id")); + assertEquals("OOM", lines.get(0).get("status")); + assertEquals("boom", lines.get(0).get("message")); + assertEquals(7, lines.get(0).get("elapsedMs")); + assertTrue(lines.get(0).get("ts").toString().endsWith("Z")); + assertEquals("d.doc", lines.get(1).get("id")); + } + + @Test + public void testConcurrentReportsAllLand(@TempDir Path tmp) throws Exception { + Path path = tmp.resolve("audit.jsonl"); + int threads = 8; + int perThread = 500; + try (FileSystemJsonlReporter r = build(path, "")) { + ExecutorService ex = Executors.newFixedThreadPool(threads); + List<Future<?>> futures = new ArrayList<>(); + for (int t = 0; t < threads; t++) { + final int tid = t; + futures.add(ex.submit(() -> { + for (int i = 0; i < perThread; i++) { + report(r, tid + "/" + i, PipesResult.RESULT_STATUS.PARSE_SUCCESS, null); + } + })); + } + for (Future<?> f : futures) { + f.get(); + } + ex.shutdown(); + } + assertEquals(threads * perThread, lines(path).size()); + } + + @Test + public void testOnExists(@TempDir Path tmp) throws Exception { + Path path = tmp.resolve("audit.jsonl"); + try (FileSystemJsonlReporter r = build(path, "")) { + report(r, "first", PipesResult.RESULT_STATUS.OOM, null); + } + assertThrows(TikaConfigException.class, () -> build(path, "")); + assertThrows(TikaConfigException.class, () -> build(path, ",\"onExists\":\"EXCEPTION\"")); + assertEquals(1, lines(path).size()); + + try (FileSystemJsonlReporter r = build(path, ",\"onExists\":\"APPEND\"")) { + report(r, "second", PipesResult.RESULT_STATUS.OOM, null); + } + assertEquals(List.of("first", "second"), lines(path).stream().map(m -> m.get("id")).toList()); + + try (FileSystemJsonlReporter r = build(path, ",\"onExists\":\"REPLACE\"")) { + report(r, "third", PipesResult.RESULT_STATUS.OOM, null); + } + assertEquals(List.of("third"), lines(path).stream().map(m -> m.get("id")).toList()); + } + + @Test + public void testMessageCap(@TempDir Path tmp) throws Exception { + Path path = tmp.resolve("audit.jsonl"); + try (FileSystemJsonlReporter r = build(path, ",\"maxMessageLength\":10")) { + report(r, "x", PipesResult.RESULT_STATUS.OOM, "0123456789abcdef"); + } + String msg = (String) lines(path).get(0).get("message"); + assertTrue(msg.startsWith("0123456789...[truncated 6 chars]"), msg); + } + + @Test + public void testErrorFlushesWithoutClose(@TempDir Path tmp) throws Exception { + Path path = tmp.resolve("audit.jsonl"); + FileSystemJsonlReporter r = build(path, ""); + report(r, "x", PipesResult.RESULT_STATUS.OOM, null); + r.error(new RuntimeException("fatal")); + List<Map<String, Object>> lines = lines(path); + assertEquals(2, lines.size()); + assertTrue(lines.get(1).get("error").toString().contains("fatal")); + assertThrows(IllegalStateException.class, () -> report(r, "y", PipesResult.RESULT_STATUS.OOM, null)); + } + + @Test + public void testCreatesParentDirs(@TempDir Path tmp) throws Exception { + Path path = tmp.resolve("a/b/audit.jsonl"); + try (FileSystemJsonlReporter r = build(path, "")) { + report(r, "x", PipesResult.RESULT_STATUS.OOM, null); + } + assertEquals(1, lines(path).size()); + } +} diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/test/resources/config-examples/file-system-jsonl-reporter.json b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/test/resources/config-examples/file-system-jsonl-reporter.json new file mode 100644 index 0000000000..b720101362 --- /dev/null +++ b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/test/resources/config-examples/file-system-jsonl-reporter.json @@ -0,0 +1,10 @@ +{ + "pipes-reporters": { + "file-system-jsonl-reporter": { + "path": "/var/log/tika/pipes-audit.jsonl", + "includes": ["OOM", "TIMEOUT", "UNSPECIFIED_CRASH", "PAYLOAD_LIMIT_EXCEEDED", "EMIT_EXCEPTION", "FETCH_EXCEPTION"], + "onExists": "EXCEPTION", + "maxMessageLength": 10000 + } + } +}
