From 4d2a1bc8f9273321d7d4894ef0b71f614eae716b Mon Sep 17 00:00:00 2001 From: Siqi Ding Date: Thu, 27 Aug 2026 19:26:44 +0000 Subject: [PATCH] feat(s3-source): Add format auto-detection for zero-config S3 ingestion Add a format auto-detection library that identifies data format and compression from S3 object bytes, so customers do not need to specify a codec in their pipeline configuration. - New format-detection module using Apache Tika for MIME/format detection, with lightweight text heuristics to distinguish JSON arrays, ND-JSON, CSV, and TSV. Detects gzip/snappy/zstd compression and decompresses gzip/snappy samples. - S3 source integration via a Function codec provider. When no codec is configured, the source reads a byte-range sample (bytes=0-65535) per object, detects the format, and selects the matching codec. No forked object worker; the existing S3ObjectWorker resolves the codec through the provider. - Codec is now optional in S3SourceConfig (auto-detect when absent). Includes unit tests covering compression detection, binary/text formats, edge cases (BOM, quoted CSV, truncated JSON, binary garbage), and gzip integration. Compressed test inputs are generated at test time. Signed-off-by: Siqi Ding --- .../format-detection/build.gradle | 24 ++ .../plugins/formatdetection/Confidence.java | 19 + .../formatdetection/DetectedCompression.java | 20 + .../formatdetection/DetectedFormat.java | 28 ++ .../FormatDetectionResult.java | 46 +++ .../formatdetection/FormatDetector.java | 360 ++++++++++++++++++ .../FormatDetectorFileTest.java | 101 +++++ .../formatdetection/FormatDetectorTest.java | 264 +++++++++++++ .../src/test/resources/samples/sample.csv | 6 + .../src/test/resources/samples/sample.json | 5 + .../src/test/resources/samples/sample.log | 6 + .../src/test/resources/samples/sample.ndjson | 6 + .../src/test/resources/samples/sample.tsv | 6 + .../src/test/resources/samples/sample.xml | 18 + data-prepper-plugins/s3-source/build.gradle | 1 + .../source/s3/AutoDetectCodecFactory.java | 129 +++++++ .../plugins/source/s3/S3ObjectRequest.java | 13 + .../plugins/source/s3/S3ObjectWorker.java | 15 +- .../plugins/source/s3/S3Source.java | 25 +- .../plugins/source/s3/S3SourceConfig.java | 4 +- .../plugins/source/s3/S3SourceConfigTest.java | 5 +- settings.gradle | 1 + 22 files changed, 1092 insertions(+), 10 deletions(-) create mode 100644 data-prepper-plugins/format-detection/build.gradle create mode 100644 data-prepper-plugins/format-detection/src/main/java/org/opensearch/dataprepper/plugins/formatdetection/Confidence.java create mode 100644 data-prepper-plugins/format-detection/src/main/java/org/opensearch/dataprepper/plugins/formatdetection/DetectedCompression.java create mode 100644 data-prepper-plugins/format-detection/src/main/java/org/opensearch/dataprepper/plugins/formatdetection/DetectedFormat.java create mode 100644 data-prepper-plugins/format-detection/src/main/java/org/opensearch/dataprepper/plugins/formatdetection/FormatDetectionResult.java create mode 100644 data-prepper-plugins/format-detection/src/main/java/org/opensearch/dataprepper/plugins/formatdetection/FormatDetector.java create mode 100644 data-prepper-plugins/format-detection/src/test/java/org/opensearch/dataprepper/plugins/formatdetection/FormatDetectorFileTest.java create mode 100644 data-prepper-plugins/format-detection/src/test/java/org/opensearch/dataprepper/plugins/formatdetection/FormatDetectorTest.java create mode 100644 data-prepper-plugins/format-detection/src/test/resources/samples/sample.csv create mode 100644 data-prepper-plugins/format-detection/src/test/resources/samples/sample.json create mode 100644 data-prepper-plugins/format-detection/src/test/resources/samples/sample.log create mode 100644 data-prepper-plugins/format-detection/src/test/resources/samples/sample.ndjson create mode 100644 data-prepper-plugins/format-detection/src/test/resources/samples/sample.tsv create mode 100644 data-prepper-plugins/format-detection/src/test/resources/samples/sample.xml create mode 100644 data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/AutoDetectCodecFactory.java diff --git a/data-prepper-plugins/format-detection/build.gradle b/data-prepper-plugins/format-detection/build.gradle new file mode 100644 index 0000000000..bfca14e762 --- /dev/null +++ b/data-prepper-plugins/format-detection/build.gradle @@ -0,0 +1,24 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +plugins { + id 'java' +} + +dependencies { + implementation 'org.apache.tika:tika-core:2.9.2' + implementation 'org.xerial.snappy:snappy-java:1.1.10.8' + testImplementation 'org.junit.jupiter:junit-jupiter:5.10.2' + testImplementation 'org.assertj:assertj-core:3.25.3' + testRuntimeOnly 'org.junit.platform:junit-platform-launcher' +} + +test { + useJUnitPlatform() +} diff --git a/data-prepper-plugins/format-detection/src/main/java/org/opensearch/dataprepper/plugins/formatdetection/Confidence.java b/data-prepper-plugins/format-detection/src/main/java/org/opensearch/dataprepper/plugins/formatdetection/Confidence.java new file mode 100644 index 0000000000..75cb688f96 --- /dev/null +++ b/data-prepper-plugins/format-detection/src/main/java/org/opensearch/dataprepper/plugins/formatdetection/Confidence.java @@ -0,0 +1,19 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.dataprepper.plugins.formatdetection; + +/** + * Confidence level of a format detection result. + */ +public enum Confidence { + HIGH, + MEDIUM, + LOW +} diff --git a/data-prepper-plugins/format-detection/src/main/java/org/opensearch/dataprepper/plugins/formatdetection/DetectedCompression.java b/data-prepper-plugins/format-detection/src/main/java/org/opensearch/dataprepper/plugins/formatdetection/DetectedCompression.java new file mode 100644 index 0000000000..5d14b14466 --- /dev/null +++ b/data-prepper-plugins/format-detection/src/main/java/org/opensearch/dataprepper/plugins/formatdetection/DetectedCompression.java @@ -0,0 +1,20 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.dataprepper.plugins.formatdetection; + +/** + * Compression types detectable from magic bytes. + */ +public enum DetectedCompression { + GZIP, + ZSTD, + SNAPPY, + NONE +} diff --git a/data-prepper-plugins/format-detection/src/main/java/org/opensearch/dataprepper/plugins/formatdetection/DetectedFormat.java b/data-prepper-plugins/format-detection/src/main/java/org/opensearch/dataprepper/plugins/formatdetection/DetectedFormat.java new file mode 100644 index 0000000000..019dc0c616 --- /dev/null +++ b/data-prepper-plugins/format-detection/src/main/java/org/opensearch/dataprepper/plugins/formatdetection/DetectedFormat.java @@ -0,0 +1,28 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.dataprepper.plugins.formatdetection; + +/** + * Data formats that can be detected from S3 object bytes. + */ +public enum DetectedFormat { + PARQUET, + AVRO, + ORC, + PDF, + IMAGE, + JSON, + NDJSON, + XML, + CSV, + TSV, + TEXT, + UNKNOWN +} diff --git a/data-prepper-plugins/format-detection/src/main/java/org/opensearch/dataprepper/plugins/formatdetection/FormatDetectionResult.java b/data-prepper-plugins/format-detection/src/main/java/org/opensearch/dataprepper/plugins/formatdetection/FormatDetectionResult.java new file mode 100644 index 0000000000..62b9d97b19 --- /dev/null +++ b/data-prepper-plugins/format-detection/src/main/java/org/opensearch/dataprepper/plugins/formatdetection/FormatDetectionResult.java @@ -0,0 +1,46 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.dataprepper.plugins.formatdetection; + +/** + * Result of detecting the format and compression of a data sample. + */ +public class FormatDetectionResult { + private final DetectedCompression compression; + private final DetectedFormat format; + private final Confidence confidence; + + public FormatDetectionResult(final DetectedCompression compression, + final DetectedFormat format, + final Confidence confidence) { + this.compression = compression; + this.format = format; + this.confidence = confidence; + } + + public DetectedCompression getCompression() { + return compression; + } + + public DetectedFormat getFormat() { + return format; + } + + public Confidence getConfidence() { + return confidence; + } + + @Override + public String toString() { + return "FormatDetectionResult{compression=" + compression + + ", format=" + format + + ", confidence=" + confidence + "}"; + } +} diff --git a/data-prepper-plugins/format-detection/src/main/java/org/opensearch/dataprepper/plugins/formatdetection/FormatDetector.java b/data-prepper-plugins/format-detection/src/main/java/org/opensearch/dataprepper/plugins/formatdetection/FormatDetector.java new file mode 100644 index 0000000000..d25b3e1547 --- /dev/null +++ b/data-prepper-plugins/format-detection/src/main/java/org/opensearch/dataprepper/plugins/formatdetection/FormatDetector.java @@ -0,0 +1,360 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.dataprepper.plugins.formatdetection; + +import org.apache.tika.Tika; +import org.apache.tika.mime.MediaType; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.xerial.snappy.SnappyInputStream; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import java.util.zip.GZIPInputStream; + +/** + * Detects data format and compression from raw bytes using Apache Tika. + * + *

Detection proceeds in two layers: + *

    + *
  1. Compression: Tika identifies compression wrappers (gzip, zstd, snappy) from magic bytes. + * If compressed, the sample is decompressed before format detection.
    2. + *
  2. Format: Tika detects the MIME type of the (decompressed) content, which is mapped to a + * {@link DetectedFormat}. Additional text heuristics disambiguate JSON arrays, ND-JSON, + * CSV, and TSV, which Tika reports generically.
    3. + *
+ */ +public class FormatDetector { + private static final Logger LOG = LoggerFactory.getLogger(FormatDetector.class); + + private static final int DEFAULT_SAMPLE_SIZE = 65536; // 64 KB + private static final double NDJSON_LINE_THRESHOLD = 0.8; + private static final double CSV_CONSISTENCY_THRESHOLD = 0.8; + private static final int CSV_MIN_LINES = 2; + private static final int CSV_MAX_SAMPLE_LINES = 10; + + private final Tika tika; + + public FormatDetector() { + this.tika = new Tika(); + } + + /** + * Detect format and compression from a byte array sample. + * + * @param sample raw bytes from the beginning of the data source + * @return detection result with compression, format, and confidence + */ + public FormatDetectionResult detect(final byte[] sample) { + if (sample == null || sample.length < 4) { + return new FormatDetectionResult(DetectedCompression.NONE, DetectedFormat.UNKNOWN, Confidence.LOW); + } + + final DetectedCompression compression = detectCompression(sample); + final byte[] decompressed = decompress(sample, compression); + + if (decompressed.length < 4) { + return new FormatDetectionResult(compression, DetectedFormat.UNKNOWN, Confidence.LOW); + } + + return detectFormat(decompressed, compression); + } + + /** + * Detect compression by asking Tika for the MIME type of the raw bytes. + */ + DetectedCompression detectCompression(final byte[] data) { + final String mimeType = tikaDetect(data); + if (mimeType == null) { + return DetectedCompression.NONE; + } + if (mimeType.contains("gzip")) { + return DetectedCompression.GZIP; + } + if (mimeType.contains("zstd") || mimeType.contains("zstandard")) { + return DetectedCompression.ZSTD; + } + if (mimeType.contains("snappy") || mimeType.contains("x-snappy")) { + return DetectedCompression.SNAPPY; + } + return DetectedCompression.NONE; + } + + private FormatDetectionResult detectFormat(final byte[] decompressed, final DetectedCompression compression) { + final String mimeType = tikaDetect(decompressed); + + // Binary and document formats — Tika identifies these definitively from magic bytes. + final DetectedFormat binaryFormat = mapBinaryMimeType(mimeType); + if (binaryFormat != null) { + return new FormatDetectionResult(compression, binaryFormat, Confidence.HIGH); + } + + // Text-based content — refine with heuristics since Tika reports generic text/plain, + // application/json, etc. without distinguishing JSON array vs ND-JSON vs CSV. + String text = new String(decompressed, StandardCharsets.UTF_8).trim(); + if (text.startsWith("\uFEFF")) { + text = text.substring(1); + } + if (text.isEmpty()) { + return new FormatDetectionResult(compression, DetectedFormat.UNKNOWN, Confidence.LOW); + } + + return detectTextFormat(text, mimeType, compression); + } + + /** + * Map Tika binary/document MIME types to {@link DetectedFormat}. + * Returns null for text-based types that require further heuristics. + */ + private DetectedFormat mapBinaryMimeType(final String mimeType) { + if (mimeType == null) { + return null; + } + if (mimeType.contains("parquet")) { + return DetectedFormat.PARQUET; + } + if (mimeType.contains("avro")) { + return DetectedFormat.AVRO; + } + if (mimeType.contains("orc")) { + return DetectedFormat.ORC; + } + if (mimeType.equals("application/pdf")) { + return DetectedFormat.PDF; + } + if (mimeType.startsWith("image/")) { + return DetectedFormat.IMAGE; + } + return null; + } + + FormatDetectionResult detectTextFormat(final String text, final String mimeType, + final DetectedCompression compression) { + // XML — trust Tika's MIME detection plus a structural check. + if ((mimeType != null && mimeType.contains("xml")) + || text.startsWith(""))) { + return new FormatDetectionResult(compression, DetectedFormat.XML, Confidence.HIGH); + } + + // JSON array vs ND-JSON vs single object. + if (text.startsWith("{") || text.startsWith("[")) { + if (isCompleteJson(text)) { + if (text.startsWith("[")) { + return new FormatDetectionResult(compression, DetectedFormat.JSON, Confidence.HIGH); + } + // A single JSON object maps to ND-JSON (the json codec only handles arrays). + return new FormatDetectionResult(compression, DetectedFormat.NDJSON, Confidence.HIGH); + } + // Truncated JSON array (sample cut mid-array). + if (text.startsWith("[") && text.length() > 2 && text.substring(1).trim().startsWith("{")) { + return new FormatDetectionResult(compression, DetectedFormat.JSON, Confidence.MEDIUM); + } + // Truncated single JSON object (single line only). + if (text.startsWith("{") && text.length() > 10 && !text.contains("\n") + && (text.contains("\":") || text.contains("\": "))) { + return new FormatDetectionResult(compression, DetectedFormat.NDJSON, Confidence.MEDIUM); + } + } + + // ND-JSON: most lines parse as JSON objects. + final String[] lines = text.split("\\r?\\n"); + final String[] nonEmptyLines = filterNonEmpty(lines); + if (nonEmptyLines.length > 0) { + final long jsonLineCount = countJsonLines(nonEmptyLines); + final double ratio = (double) jsonLineCount / nonEmptyLines.length; + if (ratio >= NDJSON_LINE_THRESHOLD) { + final Confidence confidence = ratio >= 0.95 ? Confidence.HIGH : Confidence.MEDIUM; + return new FormatDetectionResult(compression, DetectedFormat.NDJSON, confidence); + } + } + + // CSV / TSV: consistent delimiter count across lines. + if (nonEmptyLines.length >= CSV_MIN_LINES) { + final DetectedFormat delimiterFormat = detectDelimitedFormat(nonEmptyLines); + if (delimiterFormat != null) { + return new FormatDetectionResult(compression, delimiterFormat, Confidence.MEDIUM); + } + } + + // Verify it is genuinely readable text before falling back. + if (!isReadableText(text)) { + return new FormatDetectionResult(compression, DetectedFormat.UNKNOWN, Confidence.LOW); + } + + return new FormatDetectionResult(compression, DetectedFormat.TEXT, Confidence.LOW); + } + + private String tikaDetect(final byte[] data) { + try { + final MediaType mediaType = MediaType.parse(tika.detect(data)); + return mediaType == null ? null : mediaType.toString(); + } catch (final Exception e) { + LOG.debug("Tika detection failed: {}", e.getMessage()); + return null; + } + } + + private DetectedFormat detectDelimitedFormat(final String[] lines) { + if (isConsistentDelimiter(lines, ',')) { + return DetectedFormat.CSV; + } + if (isConsistentDelimiter(lines, '\t')) { + return DetectedFormat.TSV; + } + if (isConsistentDelimiter(lines, '|')) { + return DetectedFormat.CSV; + } + return null; + } + + private boolean isConsistentDelimiter(final String[] lines, final char delimiter) { + final int sampleSize = Math.min(lines.length, CSV_MAX_SAMPLE_LINES); + final long firstLineCount = countDelimiterOutsideQuotes(lines[0], delimiter); + if (firstLineCount == 0) { + return false; + } + int matches = 0; + for (int i = 1; i < sampleSize; i++) { + if (countDelimiterOutsideQuotes(lines[i], delimiter) == firstLineCount) { + matches++; + } + } + return (double) matches / (sampleSize - 1) >= CSV_CONSISTENCY_THRESHOLD; + } + + private boolean isCompleteJson(final String text) { + if (text.startsWith("{") && text.endsWith("}")) { + return bracketsBalanced(text, '{', '}'); + } + if (text.startsWith("[") && text.endsWith("]")) { + return bracketsBalanced(text, '[', ']'); + } + return false; + } + + private boolean bracketsBalanced(final String text, final char open, final char close) { + int depth = 0; + boolean inString = false; + boolean escaped = false; + for (int i = 0; i < text.length(); i++) { + final char c = text.charAt(i); + if (escaped) { + escaped = false; + continue; + } + if (c == '\\' && inString) { + escaped = true; + continue; + } + if (c == '"') { + inString = !inString; + continue; + } + if (inString) { + continue; + } + if (c == open) { + depth++; + } else if (c == close) { + depth--; + if (depth == 0 && i < text.length() - 1) { + return false; + } + } + } + return depth == 0; + } + + private long countJsonLines(final String[] lines) { + long count = 0; + for (final String line : lines) { + final String trimmed = line.trim(); + if (trimmed.startsWith("{") && trimmed.endsWith("}")) { + count++; + } + } + return count; + } + + private String[] filterNonEmpty(final String[] lines) { + return java.util.Arrays.stream(lines) + .filter(l -> !l.trim().isEmpty()) + .toArray(String[]::new); + } + + private long countDelimiterOutsideQuotes(final String line, final char delimiter) { + long count = 0; + boolean inQuotes = false; + for (int i = 0; i < line.length(); i++) { + final char c = line.charAt(i); + if (c == '"') { + inQuotes = !inQuotes; + } else if (c == delimiter && !inQuotes) { + count++; + } + } + return count; + } + + /** + * Checks if the text content is readable (mostly printable characters). + * Binary data disguised as text will have a high ratio of non-printable chars. + */ + private boolean isReadableText(final String text) { + if (text.isEmpty()) { + return false; + } + final int sampleLength = Math.min(text.length(), 1024); + int nonPrintable = 0; + for (int i = 0; i < sampleLength; i++) { + final char c = text.charAt(i); + if (c < 0x20 && c != '\n' && c != '\r' && c != '\t') { + nonPrintable++; + } else if (c >= 0x7F && c <= 0x9F) { + nonPrintable++; + } + } + return (double) nonPrintable / sampleLength < 0.10; + } + + private byte[] decompress(final byte[] data, final DetectedCompression compression) { + if (compression == DetectedCompression.GZIP) { + return decompressStream(data, true); + } + if (compression == DetectedCompression.SNAPPY) { + return decompressStream(data, false); + } + // Zstd decompression not yet implemented. + return data; + } + + private byte[] decompressStream(final byte[] data, final boolean gzip) { + try (final InputStream is = gzip + ? new GZIPInputStream(new ByteArrayInputStream(data)) + : new SnappyInputStream(new ByteArrayInputStream(data))) { + final ByteArrayOutputStream out = new ByteArrayOutputStream(); + final byte[] buffer = new byte[4096]; + int bytesRead; + int totalRead = 0; + while ((bytesRead = is.read(buffer)) != -1 && totalRead < DEFAULT_SAMPLE_SIZE) { + out.write(buffer, 0, bytesRead); + totalRead += bytesRead; + } + return out.toByteArray(); + } catch (final IOException e) { + // Corrupt compression — return empty so format comes back UNKNOWN. + return new byte[0]; + } + } +} diff --git a/data-prepper-plugins/format-detection/src/test/java/org/opensearch/dataprepper/plugins/formatdetection/FormatDetectorFileTest.java b/data-prepper-plugins/format-detection/src/test/java/org/opensearch/dataprepper/plugins/formatdetection/FormatDetectorFileTest.java new file mode 100644 index 0000000000..0017b4fe8e --- /dev/null +++ b/data-prepper-plugins/format-detection/src/test/java/org/opensearch/dataprepper/plugins/formatdetection/FormatDetectorFileTest.java @@ -0,0 +1,101 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.dataprepper.plugins.formatdetection; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.util.stream.Stream; +import java.util.zip.GZIPOutputStream; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests format detection against plain-text sample files from test resources. + * + * Compressed variants are generated at test time (rather than checked in) to keep + * the repository free of binary artifacts. + */ +class FormatDetectorFileTest { + + private FormatDetector detector; + + @BeforeEach + void setUp() { + detector = new FormatDetector(); + } + + @ParameterizedTest(name = "{0} -> format={1}") + @MethodSource("sampleFiles") + void detectsFormatFromSampleFile(final String fileName, + final DetectedFormat expectedFormat) throws IOException { + final byte[] sample = readResource("samples/" + fileName); + final FormatDetectionResult result = detector.detect(sample); + + assertThat(result.getFormat()) + .as("format for %s", fileName) + .isEqualTo(expectedFormat); + assertThat(result.getCompression()).isEqualTo(DetectedCompression.NONE); + } + + static Stream sampleFiles() { + return Stream.of( + Arguments.of("sample.json", DetectedFormat.JSON), + Arguments.of("sample.ndjson", DetectedFormat.NDJSON), + Arguments.of("sample.csv", DetectedFormat.CSV), + Arguments.of("sample.tsv", DetectedFormat.TSV), + Arguments.of("sample.xml", DetectedFormat.XML), + Arguments.of("sample.log", DetectedFormat.TEXT) + ); + } + + @Test + void detectsGzippedNdjsonGeneratedAtTestTime() throws IOException { + final byte[] plain = readResource("samples/sample.ndjson"); + final byte[] gzipped = gzip(plain); + + final FormatDetectionResult result = detector.detect(gzipped); + assertThat(result.getCompression()).isEqualTo(DetectedCompression.GZIP); + assertThat(result.getFormat()).isEqualTo(DetectedFormat.NDJSON); + } + + @Test + void detectsGzippedCsvGeneratedAtTestTime() throws IOException { + final byte[] plain = readResource("samples/sample.csv"); + final byte[] gzipped = gzip(plain); + + final FormatDetectionResult result = detector.detect(gzipped); + assertThat(result.getCompression()).isEqualTo(DetectedCompression.GZIP); + assertThat(result.getFormat()).isEqualTo(DetectedFormat.CSV); + } + + private byte[] readResource(final String path) throws IOException { + try (final InputStream is = getClass().getClassLoader().getResourceAsStream(path)) { + if (is == null) { + throw new IOException("Resource not found: " + path); + } + return is.readAllBytes(); + } + } + + private byte[] gzip(final byte[] data) throws IOException { + final ByteArrayOutputStream bos = new ByteArrayOutputStream(); + try (final GZIPOutputStream gos = new GZIPOutputStream(bos)) { + gos.write(data); + } + return bos.toByteArray(); + } +} diff --git a/data-prepper-plugins/format-detection/src/test/java/org/opensearch/dataprepper/plugins/formatdetection/FormatDetectorTest.java b/data-prepper-plugins/format-detection/src/test/java/org/opensearch/dataprepper/plugins/formatdetection/FormatDetectorTest.java new file mode 100644 index 0000000000..9f174ce0ac --- /dev/null +++ b/data-prepper-plugins/format-detection/src/test/java/org/opensearch/dataprepper/plugins/formatdetection/FormatDetectorTest.java @@ -0,0 +1,264 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.dataprepper.plugins.formatdetection; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Nested; +import org.junit.jupiter.api.Test; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.zip.GZIPOutputStream; + +import static org.assertj.core.api.Assertions.assertThat; + +class FormatDetectorTest { + + private FormatDetector detector; + + @BeforeEach + void setUp() { + detector = new FormatDetector(); + } + + @Nested + class CompressionDetection { + + @Test + void detectsGzip() throws IOException { + final byte[] gzipData = gzip("{\"a\":1}\n{\"b\":2}\n".getBytes(StandardCharsets.UTF_8)); + assertThat(detector.detectCompression(gzipData)).isEqualTo(DetectedCompression.GZIP); + } + + @Test + void returnsNoneForUncompressed() { + final byte[] plainData = "hello world this is plain text content".getBytes(StandardCharsets.UTF_8); + assertThat(detector.detectCompression(plainData)).isEqualTo(DetectedCompression.NONE); + } + } + + @Nested + class JsonDetection { + + @Test + void detectsSingleJsonObjectAsNdjson() { + final byte[] json = "{\"name\": \"test\", \"value\": 42}".getBytes(StandardCharsets.UTF_8); + final FormatDetectionResult result = detector.detect(json); + assertThat(result.getFormat()).isEqualTo(DetectedFormat.NDJSON); + assertThat(result.getConfidence()).isEqualTo(Confidence.HIGH); + assertThat(result.getCompression()).isEqualTo(DetectedCompression.NONE); + } + + @Test + void detectsJsonArray() { + final byte[] json = "[{\"a\":1},{\"a\":2},{\"a\":3}]".getBytes(StandardCharsets.UTF_8); + final FormatDetectionResult result = detector.detect(json); + assertThat(result.getFormat()).isEqualTo(DetectedFormat.JSON); + assertThat(result.getConfidence()).isEqualTo(Confidence.HIGH); + } + + @Test + void detectsTruncatedJsonArrayAsMediumConfidence() { + final byte[] json = "[{\"a\":1},{\"b\":2},{\"c\":3}".getBytes(StandardCharsets.UTF_8); + final FormatDetectionResult result = detector.detect(json); + assertThat(result.getFormat()).isEqualTo(DetectedFormat.JSON); + assertThat(result.getConfidence()).isEqualTo(Confidence.MEDIUM); + } + + @Test + void handlesJsonWithEscapedBrackets() { + final byte[] json = "{\"msg\": \"value with \\\"nested\\\" and {braces}\"}".getBytes(StandardCharsets.UTF_8); + final FormatDetectionResult result = detector.detect(json); + assertThat(result.getFormat()).isEqualTo(DetectedFormat.NDJSON); + assertThat(result.getConfidence()).isEqualTo(Confidence.HIGH); + } + } + + @Nested + class NdjsonDetection { + + @Test + void detectsNdjson() { + final String ndjson = "{\"a\":1}\n{\"b\":2}\n{\"c\":3}\n{\"d\":4}\n{\"e\":5}\n"; + final FormatDetectionResult result = detector.detect(ndjson.getBytes(StandardCharsets.UTF_8)); + assertThat(result.getFormat()).isEqualTo(DetectedFormat.NDJSON); + assertThat(result.getConfidence()).isEqualTo(Confidence.HIGH); + } + + @Test + void detectsNdjsonMediumWhenSomeLinesInvalid() { + final String ndjson = "{\"a\":1}\n{\"b\":2}\n{\"c\":3}\n{\"d\":4}\nnot json\n"; + final FormatDetectionResult result = detector.detect(ndjson.getBytes(StandardCharsets.UTF_8)); + assertThat(result.getFormat()).isEqualTo(DetectedFormat.NDJSON); + assertThat(result.getConfidence()).isEqualTo(Confidence.MEDIUM); + } + + @Test + void doesNotDetectNdjsonWhenTooFewJsonLines() { + final String text = "{\"a\":1}\nnot json\nalso not\nnope\nstill no\n"; + final FormatDetectionResult result = detector.detect(text.getBytes(StandardCharsets.UTF_8)); + assertThat(result.getFormat()).isNotEqualTo(DetectedFormat.NDJSON); + } + + @Test + void detectsNdjsonWithEmptyLines() { + final String ndjson = "{\"a\":1}\n\n{\"b\":2}\n\n{\"c\":3}\n\n{\"d\":4}\n\n{\"e\":5}\n"; + final FormatDetectionResult result = detector.detect(ndjson.getBytes(StandardCharsets.UTF_8)); + assertThat(result.getFormat()).isEqualTo(DetectedFormat.NDJSON); + } + } + + @Nested + class XmlDetection { + + @Test + void detectsXmlWithDeclaration() { + final byte[] xml = "".getBytes(StandardCharsets.UTF_8); + final FormatDetectionResult result = detector.detect(xml); + assertThat(result.getFormat()).isEqualTo(DetectedFormat.XML); + assertThat(result.getConfidence()).isEqualTo(Confidence.HIGH); + } + + @Test + void detectsXmlWithRootTag() { + final byte[] xml = "data".getBytes(StandardCharsets.UTF_8); + final FormatDetectionResult result = detector.detect(xml); + assertThat(result.getFormat()).isEqualTo(DetectedFormat.XML); + } + } + + @Nested + class CsvDetection { + + @Test + void detectsCsv() { + final String csv = "name,age,city\nAlice,30,NYC\nBob,25,LA\nCharlie,35,Chicago\n"; + final FormatDetectionResult result = detector.detect(csv.getBytes(StandardCharsets.UTF_8)); + assertThat(result.getFormat()).isEqualTo(DetectedFormat.CSV); + assertThat(result.getConfidence()).isEqualTo(Confidence.MEDIUM); + } + + @Test + void detectsTsv() { + final String tsv = "name\tage\tcity\nAlice\t30\tNYC\nBob\t25\tLA\nCharlie\t35\tChicago\n"; + final FormatDetectionResult result = detector.detect(tsv.getBytes(StandardCharsets.UTF_8)); + assertThat(result.getFormat()).isEqualTo(DetectedFormat.TSV); + } + + @Test + void detectsCsvWithQuotedCommas() { + final String csv = "name,address,city\n\"Smith, John\",\"123 Main St\",NYC\n" + + "\"Doe, Jane\",\"456 Oak Ave\",LA\n\"Lee, Amy\",\"789 Pine Rd\",SF\n"; + final FormatDetectionResult result = detector.detect(csv.getBytes(StandardCharsets.UTF_8)); + assertThat(result.getFormat()).isIn(DetectedFormat.CSV, DetectedFormat.TEXT); + } + + @Test + void doesNotDetectCsvForInconsistentDelimiters() { + final String text = "one,two,three\nfour\nfive,six\n"; + final FormatDetectionResult result = detector.detect(text.getBytes(StandardCharsets.UTF_8)); + assertThat(result.getFormat()).isNotEqualTo(DetectedFormat.CSV); + } + } + + @Nested + class GzipIntegration { + + @Test + void detectsGzippedNdjson() throws IOException { + final String ndjson = "{\"a\":1}\n{\"b\":2}\n{\"c\":3}\n{\"d\":4}\n{\"e\":5}\n"; + final byte[] gzipped = gzip(ndjson.getBytes(StandardCharsets.UTF_8)); + final FormatDetectionResult result = detector.detect(gzipped); + assertThat(result.getCompression()).isEqualTo(DetectedCompression.GZIP); + assertThat(result.getFormat()).isEqualTo(DetectedFormat.NDJSON); + } + + @Test + void detectsGzippedCsv() throws IOException { + final String csv = "a,b,c\n1,2,3\n4,5,6\n7,8,9\n"; + final byte[] gzipped = gzip(csv.getBytes(StandardCharsets.UTF_8)); + final FormatDetectionResult result = detector.detect(gzipped); + assertThat(result.getCompression()).isEqualTo(DetectedCompression.GZIP); + assertThat(result.getFormat()).isEqualTo(DetectedFormat.CSV); + } + + @Test + void handlesCorruptGzip() { + final byte[] corrupt = new byte[]{0x1F, (byte) 0x8B, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x7F, 0x7F}; + final FormatDetectionResult result = detector.detect(corrupt); + assertThat(result.getCompression()).isEqualTo(DetectedCompression.GZIP); + assertThat(result.getFormat()).isEqualTo(DetectedFormat.UNKNOWN); + } + } + + @Nested + class EdgeCases { + + @Test + void returnsUnknownForNull() { + final FormatDetectionResult result = detector.detect(null); + assertThat(result.getFormat()).isEqualTo(DetectedFormat.UNKNOWN); + assertThat(result.getConfidence()).isEqualTo(Confidence.LOW); + } + + @Test + void returnsUnknownForEmptyArray() { + final FormatDetectionResult result = detector.detect(new byte[0]); + assertThat(result.getFormat()).isEqualTo(DetectedFormat.UNKNOWN); + } + + @Test + void returnsUnknownForTinyInput() { + final FormatDetectionResult result = detector.detect(new byte[]{0x00, 0x01}); + assertThat(result.getFormat()).isEqualTo(DetectedFormat.UNKNOWN); + } + + @Test + void fallsBackToTextForPlainContent() { + final String plain = "This is just a plain text log line.\nAnother line here.\nAnd another.\n"; + final FormatDetectionResult result = detector.detect(plain.getBytes(StandardCharsets.UTF_8)); + assertThat(result.getFormat()).isEqualTo(DetectedFormat.TEXT); + assertThat(result.getConfidence()).isEqualTo(Confidence.LOW); + } + + @Test + void detectsBinaryGarbageAsUnknown() { + final byte[] garbage = new byte[512]; + for (int i = 0; i < garbage.length; i++) { + garbage[i] = (byte) ((i * 37 + 13) % 256); + } + // Ensure it doesn't start with a known magic byte + garbage[0] = 0x03; + garbage[1] = 0x04; + garbage[2] = 0x05; + garbage[3] = 0x06; + final FormatDetectionResult result = detector.detect(garbage); + assertThat(result.getFormat()).isIn(DetectedFormat.UNKNOWN, DetectedFormat.IMAGE); + } + + @Test + void stripsBomBeforeDetection() { + final byte[] bomJson = ("\uFEFF{\"event\":\"login\"}\n{\"event\":\"logout\"}\n") + .getBytes(StandardCharsets.UTF_8); + final FormatDetectionResult result = detector.detect(bomJson); + assertThat(result.getFormat()).isEqualTo(DetectedFormat.NDJSON); + } + } + + // Helper + private byte[] gzip(final byte[] data) throws IOException { + final ByteArrayOutputStream bos = new ByteArrayOutputStream(); + try (final GZIPOutputStream gos = new GZIPOutputStream(bos)) { + gos.write(data); + } + return bos.toByteArray(); + } +} diff --git a/data-prepper-plugins/format-detection/src/test/resources/samples/sample.csv b/data-prepper-plugins/format-detection/src/test/resources/samples/sample.csv new file mode 100644 index 0000000000..1cd8773601 --- /dev/null +++ b/data-prepper-plugins/format-detection/src/test/resources/samples/sample.csv @@ -0,0 +1,6 @@ +id,name,email,age,city,active +1,Alice,alice@example.com,30,New York,true +2,Bob,bob@example.com,25,Los Angeles,false +3,Charlie,charlie@example.com,35,Chicago,true +4,Diana,diana@example.com,28,Houston,true +5,Eve,eve@example.com,32,Phoenix,false diff --git a/data-prepper-plugins/format-detection/src/test/resources/samples/sample.json b/data-prepper-plugins/format-detection/src/test/resources/samples/sample.json new file mode 100644 index 0000000000..4b8f7b36c3 --- /dev/null +++ b/data-prepper-plugins/format-detection/src/test/resources/samples/sample.json @@ -0,0 +1,5 @@ +[ + {"id": 1, "name": "Alice", "email": "alice@example.com", "active": true}, + {"id": 2, "name": "Bob", "email": "bob@example.com", "active": false}, + {"id": 3, "name": "Charlie", "email": "charlie@example.com", "active": true} +] diff --git a/data-prepper-plugins/format-detection/src/test/resources/samples/sample.log b/data-prepper-plugins/format-detection/src/test/resources/samples/sample.log new file mode 100644 index 0000000000..04bae0db94 --- /dev/null +++ b/data-prepper-plugins/format-detection/src/test/resources/samples/sample.log @@ -0,0 +1,6 @@ +2026-06-17 10:00:00 INFO [main] Application starting up +2026-06-17 10:00:01 DEBUG [main] Loading configuration from /etc/app/config.yml +2026-06-17 10:00:02 INFO [main] Connected to database at localhost:5432 +2026-06-17 10:00:03 WARN [pool-1] Connection pool nearing capacity: 45/50 +2026-06-17 10:00:04 ERROR [handler-3] Failed to process request: timeout after 30s +2026-06-17 10:00:05 INFO [main] Health check endpoint ready on :8080/health diff --git a/data-prepper-plugins/format-detection/src/test/resources/samples/sample.ndjson b/data-prepper-plugins/format-detection/src/test/resources/samples/sample.ndjson new file mode 100644 index 0000000000..f31efeab63 --- /dev/null +++ b/data-prepper-plugins/format-detection/src/test/resources/samples/sample.ndjson @@ -0,0 +1,6 @@ +{"timestamp":"2026-06-17T10:00:00Z","level":"INFO","service":"api","message":"Request received","duration_ms":12} +{"timestamp":"2026-06-17T10:00:01Z","level":"DEBUG","service":"api","message":"Processing payload","duration_ms":45} +{"timestamp":"2026-06-17T10:00:02Z","level":"INFO","service":"worker","message":"Task completed","duration_ms":230} +{"timestamp":"2026-06-17T10:00:03Z","level":"WARN","service":"api","message":"Slow response","duration_ms":1500} +{"timestamp":"2026-06-17T10:00:04Z","level":"ERROR","service":"worker","message":"Connection timeout","duration_ms":5000} +{"timestamp":"2026-06-17T10:00:05Z","level":"INFO","service":"api","message":"Health check OK","duration_ms":2} diff --git a/data-prepper-plugins/format-detection/src/test/resources/samples/sample.tsv b/data-prepper-plugins/format-detection/src/test/resources/samples/sample.tsv new file mode 100644 index 0000000000..58dd3448ed --- /dev/null +++ b/data-prepper-plugins/format-detection/src/test/resources/samples/sample.tsv @@ -0,0 +1,6 @@ +id name email age city active +1 Alice alice@example.com 30 New York true +2 Bob bob@example.com 25 Los Angeles false +3 Charlie charlie@example.com 35 Chicago true +4 Diana diana@example.com 28 Houston true +5 Eve eve@example.com 32 Phoenix false diff --git a/data-prepper-plugins/format-detection/src/test/resources/samples/sample.xml b/data-prepper-plugins/format-detection/src/test/resources/samples/sample.xml new file mode 100644 index 0000000000..06d5e3c111 --- /dev/null +++ b/data-prepper-plugins/format-detection/src/test/resources/samples/sample.xml @@ -0,0 +1,18 @@ + + + + Widget + 9.99 + Tools + + + Gadget + 24.99 + Electronics + + + Doohickey + 4.50 + Misc + + diff --git a/data-prepper-plugins/s3-source/build.gradle b/data-prepper-plugins/s3-source/build.gradle index 59dd1e979c..f1f3f2614c 100644 --- a/data-prepper-plugins/s3-source/build.gradle +++ b/data-prepper-plugins/s3-source/build.gradle @@ -13,6 +13,7 @@ dependencies { implementation project(':data-prepper-plugins:common') implementation project(':data-prepper-plugins:sqs-common') implementation project(':data-prepper-plugins:s3-common') + implementation project(':data-prepper-plugins:format-detection') implementation libs.armeria.core implementation 'io.micrometer:micrometer-core' diff --git a/data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/AutoDetectCodecFactory.java b/data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/AutoDetectCodecFactory.java new file mode 100644 index 0000000000..25829cad2d --- /dev/null +++ b/data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/AutoDetectCodecFactory.java @@ -0,0 +1,129 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.dataprepper.plugins.source.s3; + +import org.opensearch.dataprepper.model.codec.InputCodec; +import org.opensearch.dataprepper.model.configuration.PluginSetting; +import org.opensearch.dataprepper.model.plugin.PluginFactory; +import org.opensearch.dataprepper.plugins.formatdetection.DetectedFormat; +import org.opensearch.dataprepper.plugins.formatdetection.FormatDetectionResult; +import org.opensearch.dataprepper.plugins.formatdetection.FormatDetector; +import org.opensearch.dataprepper.plugins.s3.common.ownership.BucketOwnerProvider; +import org.opensearch.dataprepper.plugins.s3.common.source.S3ObjectReference; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import software.amazon.awssdk.core.ResponseInputStream; +import software.amazon.awssdk.services.s3.S3Client; +import software.amazon.awssdk.services.s3.model.GetObjectRequest; +import software.amazon.awssdk.services.s3.model.GetObjectResponse; + +import java.io.IOException; +import java.util.Collections; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; +import java.util.function.Function; + +/** + * A {@link Function} that, given an {@link S3ObjectReference}, reads a small byte-range sample + * from the object, auto-detects its format, and returns the appropriate {@link InputCodec}. + * + *

Used when no codec is explicitly configured in the S3 source. It reads only the first + * {@code SAMPLE_SIZE} bytes via a ranged GET so it does not consume the object's full content + * or advance the stream used for processing. + */ +class AutoDetectCodecFactory implements Function { + private static final Logger LOG = LoggerFactory.getLogger(AutoDetectCodecFactory.class); + static final int SAMPLE_SIZE = 65536; // 64 KB + + private final PluginFactory pluginFactory; + private final S3Client s3Client; + private final BucketOwnerProvider bucketOwnerProvider; + private final FormatDetector formatDetector; + private final Map codecCache; + + AutoDetectCodecFactory(final PluginFactory pluginFactory, + final S3Client s3Client, + final BucketOwnerProvider bucketOwnerProvider) { + this.pluginFactory = pluginFactory; + this.s3Client = s3Client; + this.bucketOwnerProvider = bucketOwnerProvider; + this.formatDetector = new FormatDetector(); + this.codecCache = new ConcurrentHashMap<>(); + } + + @Override + public InputCodec apply(final S3ObjectReference s3ObjectReference) { + final byte[] sample = readSample(s3ObjectReference); + final FormatDetectionResult detection = formatDetector.detect(sample); + + LOG.debug("Auto-detected format for s3://{}/{}: format={}, compression={}, confidence={}", + s3ObjectReference.getBucketName(), s3ObjectReference.getKey(), + detection.getFormat(), detection.getCompression(), detection.getConfidence()); + + return getCodecForFormat(detection.getFormat()); + } + + private byte[] readSample(final S3ObjectReference s3ObjectReference) { + final GetObjectRequest.Builder requestBuilder = GetObjectRequest.builder() + .bucket(s3ObjectReference.getBucketName()) + .key(s3ObjectReference.getKey()) + .range(String.format("bytes=0-%d", SAMPLE_SIZE - 1)); + + final Optional bucketOwner = bucketOwnerProvider.getBucketOwner(s3ObjectReference.getBucketName()); + bucketOwner.ifPresent(requestBuilder::expectedBucketOwner); + + try (final ResponseInputStream stream = s3Client.getObject(requestBuilder.build())) { + return stream.readAllBytes(); + } catch (final IOException | RuntimeException e) { + LOG.warn("Failed to read sample for format detection of s3://{}/{}: {}", + s3ObjectReference.getBucketName(), s3ObjectReference.getKey(), e.getMessage()); + return new byte[0]; + } + } + + private InputCodec getCodecForFormat(final DetectedFormat format) { + return codecCache.computeIfAbsent(format, this::createCodec); + } + + private InputCodec createCodec(final DetectedFormat format) { + final String pluginName = mapFormatToCodecPlugin(format); + if (pluginName == null) { + LOG.warn("No codec available for detected format: {}", format); + return null; + } + + LOG.debug("Auto-detecting codec: {} -> {}", format, pluginName); + final PluginSetting pluginSetting = new PluginSetting(pluginName, Collections.emptyMap()); + return pluginFactory.loadPlugin(InputCodec.class, pluginSetting); + } + + private String mapFormatToCodecPlugin(final DetectedFormat format) { + if (format == DetectedFormat.NDJSON) { + return "ndjson"; + } + if (format == DetectedFormat.JSON) { + return "json"; + } + if (format == DetectedFormat.CSV || format == DetectedFormat.TSV) { + return "csv"; + } + if (format == DetectedFormat.PARQUET) { + return "parquet"; + } + if (format == DetectedFormat.AVRO) { + return "avro"; + } + if (format == DetectedFormat.TEXT) { + return "newline"; + } + return null; + } +} diff --git a/data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/S3ObjectRequest.java b/data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/S3ObjectRequest.java index 10a679c036..62762e3134 100644 --- a/data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/S3ObjectRequest.java +++ b/data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/S3ObjectRequest.java @@ -21,6 +21,7 @@ import java.time.Duration; import java.util.function.BiConsumer; +import java.util.function.Function; public class S3ObjectRequest { private final Buffer> buffer; @@ -34,6 +35,7 @@ public class S3ObjectRequest { private final CompressionOption compressionOption; private final BucketOwnerProvider bucketOwnerProvider; private final InputCodec codec; + private final Function codecProvider; private final BiConsumer eventConsumer; private final S3Client s3Client; private final CompressionType compressionType; @@ -55,6 +57,7 @@ private S3ObjectRequest(Builder builder) { this.compressionOption = builder.compressionOption; this.bucketOwnerProvider = builder.bucketOwnerProvider; this.codec = builder.codec; + this.codecProvider = builder.codecProvider; this.eventConsumer = builder.eventConsumer; this.s3Client = builder.s3Client; this.compressionType = builder.compressionType; @@ -107,6 +110,10 @@ public InputCodec getCodec() { return codec; } + public Function getCodecProvider() { + return codecProvider; + } + public BiConsumer getEventConsumer() { return eventConsumer; } @@ -143,6 +150,7 @@ public static class Builder { private S3SelectResponseHandlerFactory s3SelectResponseHandlerFactory; private CompressionOption compressionOption; private InputCodec codec; + private Function codecProvider; private BiConsumer eventConsumer; private S3Client s3Client; private CompressionType compressionType; @@ -195,6 +203,11 @@ public Builder codec(InputCodec codec) { return this; } + public Builder codecProvider(Function codecProvider) { + this.codecProvider = codecProvider; + return this; + } + public Builder eventConsumer(BiConsumer eventConsumer) { this.eventConsumer = eventConsumer; return this; diff --git a/data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/S3ObjectWorker.java b/data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/S3ObjectWorker.java index 2c96df07ee..8b6c0c88eb 100644 --- a/data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/S3ObjectWorker.java +++ b/data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/S3ObjectWorker.java @@ -35,6 +35,7 @@ import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; import java.util.function.BiConsumer; +import java.util.function.Function; /** * Class responsible for taking an {@link S3ObjectReference} and creating all the necessary {@link Event} @@ -56,6 +57,7 @@ class S3ObjectWorker implements S3ObjectHandler { private final CompressionOption compressionOption; private final InputCodec codec; + private final Function codecProvider; private final BucketOwnerProvider bucketOwnerProvider; private final Duration bufferTimeout; private final int numberOfRecordsToAccumulate; @@ -67,6 +69,7 @@ public S3ObjectWorker(final S3ObjectRequest s3ObjectRequest) { this.buffer = s3ObjectRequest.getBuffer(); this.compressionOption = s3ObjectRequest.getCompressionOption(); this.codec = s3ObjectRequest.getCodec(); + this.codecProvider = s3ObjectRequest.getCodecProvider(); this.bucketOwnerProvider = s3ObjectRequest.getBucketOwnerProvider(); this.bufferTimeout = s3ObjectRequest.getBufferTimeout(); this.numberOfRecordsToAccumulate = s3ObjectRequest.getNumberOfRecordsToAccumulate(); @@ -158,8 +161,16 @@ public long consumeS3Object(final S3InputFile inputFile, final S3DataSelection d final CompressionOption fileCompressionOption = compressionOption != CompressionOption.AUTOMATIC ? compressionOption : CompressionOption.fromFileName(s3ObjectReference.getKey()); + // Resolve the codec: use the fixed codec if configured, otherwise auto-detect via the provider. + final InputCodec resolvedCodec = codec != null ? codec : codecProvider.apply(s3ObjectReference); + if (resolvedCodec == null) { + s3ObjectPluginMetrics.getS3ObjectReadFailedCounter().increment(); + throw new S3ReadFailedException( + new IllegalStateException("Unable to determine codec for object " + s3ObjectReference.getKey())); + } + try { - codec.parse(inputFile, fileCompressionOption.getDecompressionEngine(), record -> { + resolvedCodec.parse(inputFile, fileCompressionOption.getDecompressionEngine(), record -> { consumer.accept(record, dataSelection); }); return inputFile.getLength(); @@ -214,7 +225,7 @@ private void doProcessObject(final AcknowledgementSet acknowledgementSet, saveStateCounter.getAndIncrement(); } } catch (final Exception e) { - LOG.error("Failed writing S3 objects to buffer.", e); + LOG.error("Failed writing S3 objects to buffer due to: {}", e.getMessage()); } }); diff --git a/data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/S3Source.java b/data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/S3Source.java index ed4c4e03fc..3df7a2f609 100644 --- a/data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/S3Source.java +++ b/data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/S3Source.java @@ -29,15 +29,20 @@ import org.opensearch.dataprepper.plugins.source.s3.configuration.S3SelectOptions; import org.opensearch.dataprepper.plugins.source.s3.ownership.ConfigBucketOwnerProviderFactory; import org.opensearch.dataprepper.plugins.s3.common.ownership.BucketOwnerProvider; +import org.opensearch.dataprepper.plugins.codec.CompressionOption; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider; import software.amazon.awssdk.services.s3.model.CompressionType; import java.util.Objects; import java.util.Optional; import java.util.function.BiConsumer; +import java.util.function.Function; @DataPrepperPlugin(name = "s3", pluginType = Source.class, pluginConfigurationType = S3SourceConfig.class) public class S3Source implements Source>, UsesSourceCoordination { + private static final Logger LOG = LoggerFactory.getLogger(S3Source.class); private final PluginMetrics pluginMetrics; private final S3SourceConfig s3SourceConfig; @@ -113,14 +118,26 @@ public void start(Buffer> buffer) { s3Handler = new S3SelectObjectWorker(s3ObjectRequest); } else { final PluginModel codecConfiguration = s3SourceConfig.getCodec(); - final PluginSetting codecPluginSettings = new PluginSetting(codecConfiguration.getPluginName(), codecConfiguration.getPluginSettings()); - final InputCodec codec = pluginFactory.loadPlugin(InputCodec.class, codecPluginSettings); + final Function codecProvider; + final CompressionOption compressionOption; + if (codecConfiguration != null) { + final PluginSetting codecPluginSettings = new PluginSetting(codecConfiguration.getPluginName(), codecConfiguration.getPluginSettings()); + final InputCodec codec = pluginFactory.loadPlugin(InputCodec.class, codecPluginSettings); + codecProvider = s3ObjectReference -> codec; + compressionOption = s3SourceConfig.getCompression(); + } else { + // Auto-detect mode: no codec specified. Detect format per-object from a byte-range sample. + LOG.info("No codec configured - using auto-detect mode"); + codecProvider = new AutoDetectCodecFactory(pluginFactory, s3ClientBuilderFactory.getS3Client(), bucketOwnerProvider); + compressionOption = CompressionOption.AUTOMATIC; + } + final S3ObjectRequest s3ObjectRequest = s3ObjectRequestBuilder .bucketOwnerProvider(bucketOwnerProvider) - .codec(codec) + .codecProvider(codecProvider) .eventConsumer(eventMetadataModifier) .s3Client(s3ClientBuilderFactory.getS3Client()) - .compressionOption(s3SourceConfig.getCompression()) + .compressionOption(compressionOption) .build(); s3Handler = new S3ObjectWorker(s3ObjectRequest); } diff --git a/data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/S3SourceConfig.java b/data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/S3SourceConfig.java index 0827bbe1ef..5c63b05579 100644 --- a/data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/S3SourceConfig.java +++ b/data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/S3SourceConfig.java @@ -110,10 +110,10 @@ public class S3SourceConfig { @Valid private Map filters; - @AssertTrue(message = "A codec is required for reading objects.") + @AssertTrue(message = "A codec is required for reading objects unless using auto-detect mode.") boolean isCodecProvidedWhenNeeded() { if(s3SelectOptions == null) - return codec != null; + return true; // codec is now optional — auto-detect when null return true; } diff --git a/data-prepper-plugins/s3-source/src/test/java/org/opensearch/dataprepper/plugins/source/s3/S3SourceConfigTest.java b/data-prepper-plugins/s3-source/src/test/java/org/opensearch/dataprepper/plugins/source/s3/S3SourceConfigTest.java index 5dce6638ff..1e46c9834a 100644 --- a/data-prepper-plugins/s3-source/src/test/java/org/opensearch/dataprepper/plugins/source/s3/S3SourceConfigTest.java +++ b/data-prepper-plugins/s3-source/src/test/java/org/opensearch/dataprepper/plugins/source/s3/S3SourceConfigTest.java @@ -95,9 +95,10 @@ void isCodecProvidedWhenNeeded_returns_true_when_s3SelectOptions_is_null_and_cod } @Test - void isCodecProvidedWhenNeeded_returns_false_when_s3SelectOptions_is_null_and_codec_is_null(){ + void isCodecProvidedWhenNeeded_returns_true_when_s3SelectOptions_is_null_and_codec_is_null(){ + // codec is optional — when null, the source uses auto-detect mode final S3SourceConfig s3SourceConfig = new S3SourceConfig(); - assertFalse(s3SourceConfig.isCodecProvidedWhenNeeded()); + assertTrue(s3SourceConfig.isCodecProvidedWhenNeeded()); } @Test diff --git a/settings.gradle b/settings.gradle index 7f330e125a..08503f1e01 100644 --- a/settings.gradle +++ b/settings.gradle @@ -121,6 +121,7 @@ include 'data-prepper-plugins:common' include 'data-prepper-plugins:armeria-common' include 'data-prepper-plugins:anomaly-detector-processor' include 'data-prepper-plugins:detect-format-processor' +include 'data-prepper-plugins:format-detection' include 'data-prepper-plugins:opensearch' include 'data-prepper-plugins:opensearch-pull-kafka' include 'data-prepper-plugins:ocsf'