Merge branch 'main' into index-stats-issue-10766

Signed-off-by: Andrew Ross <[email protected]>

Author: andrross

CHANGELOG.md

@@ -12,6 +12,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Create equivalents of JSM's AccessController in the java agent ([#18346](https://github.com/opensearch-project/OpenSearch/issues/18346))
 - Introduced a new cluster-level API to fetch remote store metadata (segments and translogs) for each shard of an index. ([#18257](https://github.com/opensearch-project/OpenSearch/pull/18257))
 - Add last index request timestamp columns to the `_cat/indices` API. ([10766](https://github.com/opensearch-project/OpenSearch/issues/10766))
+- Introduce a new pull-based ingestion plugin for file-based indexing (for local testing) ([#18591](https://github.com/opensearch-project/OpenSearch/pull/18591))
 - Add support for search pipeline in search and msearch template ([#18564](https://github.com/opensearch-project/OpenSearch/pull/18564))
 
 ### Changed

plugins/ingestion-fs/README.md

@@ -0,0 +1,64 @@
+# ingestion-fs plugin
+
+The ingestion-fs plugin enables pull-based ingestion from the local file system into OpenSearch. It is primarily intended for local testing and development purposes to facilitate testing without setting up Kafka/Kinesis.
+
+## Overview
+
+This plugin implements a custom ingestion source for the [pull-based ingestion framework](https://docs.opensearch.org/docs/latest/api-reference/document-apis/pull-based-ingestion/). It allows OpenSearch to ingest documents from `.ndjson` files on the file system.
+
+Each shard-specific file is expected to follow the path:
+```
+${base_directory}/${stream}/${shard_id}.ndjson
+```
+
+## Usage
+
+### 1. Prepare test data
+
+Create the `ndjson` files with sample data following the format mentioned [here](https://docs.opensearch.org/docs/latest/api-reference/document-apis/pull-based-ingestion/).
+For example, create a file `${base_directory}/test-stream/0.ndjson` with data
+
+```
+{"_id":"1","_version":"1","_op_type":"index","_source":{"name":"name1", "age": 30}}
+{"_id":"2","_version":"1","_op_type":"index","_source":{"name":"name2", "age": 31}}
+```
+
+### 2. Start OpenSearch with the Plugin
+
+```
+./gradlew run -PinstalledPlugins="['ingestion-fs']"
+```
+
+### 3. Create a pull-based index
+
+Create an index by specifying ingestion source settings as follows
+
+<pre>
+PUT /test-index
+{
+  "settings": {
+    "ingestion_source": {
+      "type": "file",
+      "param": {
+        "stream": "test_stream",
+        "base_directory": "path to the base directory"
+      }
+    },
+    "index.number_of_shards": 1,
+    "index.number_of_replicas": 0,
+    "index": {
+      "replication.type": "SEGMENT"
+    }
+  },
+  "mappings": {
+    "properties": {
+      "name": {
+        "type": "text"
+      },
+      "age": {
+        "type": "integer"
+      }
+    }
+  }
+}
+</pre>

plugins/ingestion-fs/build.gradle

@@ -0,0 +1,14 @@
+/*
+ * 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.
+ */
+
+apply plugin: 'opensearch.internal-cluster-test'
+
+opensearchplugin {
+  description = 'Pull-based ingestion plugin to read from local files for testing'
+  classname = 'org.opensearch.plugin.ingestion.fs.FilePlugin'
+}

plugins/ingestion-fs/src/main/java/org/opensearch/plugin/ingestion/fs/FileConsumerFactory.java

@@ -0,0 +1,42 @@
+/*
+ * 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.plugin.ingestion.fs;
+
+import org.opensearch.index.IngestionConsumerFactory;
+
+import java.util.Map;
+
+/**
+ * Factory for creating file-based ingestion consumers.
+ */
+public class FileConsumerFactory implements IngestionConsumerFactory<FilePartitionConsumer, FileOffset> {
+
+    private FileSourceConfig config;
+
+    /**
+     * Initialize a FileConsumerFactory for file-based indexing.
+     */
+    public FileConsumerFactory() {}
+
+    @Override
+    public void initialize(Map<String, Object> params) {
+        this.config = new FileSourceConfig(params);
+    }
+
+    @Override
+    public FilePartitionConsumer createShardConsumer(String clientId, int shardId) {
+        assert config != null;
+        return new FilePartitionConsumer(config, shardId);
+    }
+
+    @Override
+    public FileOffset parsePointerFromString(String pointer) {
+        return new FileOffset(Long.parseLong(pointer));
+    }
+}

plugins/ingestion-fs/src/main/java/org/opensearch/plugin/ingestion/fs/FileMessage.java

@@ -0,0 +1,39 @@
+/*
+ * 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.plugin.ingestion.fs;
+
+import org.opensearch.index.Message;
+
+/**
+ * Message abstraction for file-based ingestion.
+ */
+public class FileMessage implements Message<byte[]> {
+    private final byte[] payload;
+    private final Long timestamp;
+
+    /**
+     * Create a file message.
+     * @param payload Line contents from the file, as a byte array.
+     * @param timestamp Millisecond timestamp for when the line was read.
+     */
+    public FileMessage(byte[] payload, Long timestamp) {
+        this.payload = payload;
+        this.timestamp = timestamp;
+    }
+
+    @Override
+    public byte[] getPayload() {
+        return payload;
+    }
+
+    @Override
+    public Long getTimestamp() {
+        return timestamp;
+    }
+}

plugins/ingestion-fs/src/main/java/org/opensearch/plugin/ingestion/fs/FileOffset.java

@@ -0,0 +1,87 @@
+/*
+ * 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.plugin.ingestion.fs;
+
+import org.apache.lucene.document.Field;
+import org.apache.lucene.document.LongPoint;
+import org.apache.lucene.search.Query;
+import org.opensearch.index.IngestionShardPointer;
+
+import java.nio.ByteBuffer;
+
+/**
+ * Offset for a file-based ingestion source (line number).
+ */
+public class FileOffset implements IngestionShardPointer {
+
+    private final long line;
+
+    /**
+     * Create a new file offset based on line number.
+     * @param line line number (offset)
+     */
+    public FileOffset(long line) {
+        assert line >= 0;
+        this.line = line;
+    }
+
+    /**
+     * Returns the line number (offset).
+     */
+    public long getLine() {
+        return line;
+    }
+
+    @Override
+    public byte[] serialize() {
+        ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES);
+        buffer.putLong(line);
+        return buffer.array();
+    }
+
+    @Override
+    public String asString() {
+        return Long.toString(line);
+    }
+
+    @Override
+    public Field asPointField(String fieldName) {
+        return new LongPoint(fieldName, line);
+    }
+
+    @Override
+    public Query newRangeQueryGreaterThan(String fieldName) {
+        return LongPoint.newRangeQuery(fieldName, line, Long.MAX_VALUE);
+    }
+
+    @Override
+    public int compareTo(IngestionShardPointer o) {
+        if (o == null || !(o instanceof FileOffset)) {
+            throw new IllegalArgumentException("Incompatible pointer type: " + (o == null ? "null" : o.getClass()));
+        }
+        return Long.compare(this.line, ((FileOffset) o).line);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (!(o instanceof FileOffset)) return false;
+        return this.line == ((FileOffset) o).line;
+    }
+
+    @Override
+    public int hashCode() {
+        return Long.hashCode(line);
+    }
+
+    @Override
+    public String toString() {
+        return "FileOffset{" + "line=" + line + '}';
+    }
+}