Introduce index based ingest pipeline

Signed-off-by: Bo Zhang <[email protected]>

Author: bzhangam

CHANGELOG.md

@@ -24,6 +24,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Implement fixed interval refresh task scheduling ([#17777](https://github.com/opensearch-project/OpenSearch/pull/17777))
 - Add GRPC DocumentService and Bulk endpoint ([#17727](https://github.com/opensearch-project/OpenSearch/pull/17727))
 - Added scale to zero (`search_only` mode) support for OpenSearch reader writer separation ([#17299](https://github.com/opensearch-project/OpenSearch/pull/17299)
+- Introduce index based ingest pipeline ([#17817](https://github.com/opensearch-project/OpenSearch/pull/17817)))
 
 ### Changed
 - Migrate BC libs to their FIPS counterparts ([#14912](https://github.com/opensearch-project/OpenSearch/pull/14912))

server/src/main/java/org/opensearch/action/bulk/BulkRequest.java

@@ -46,8 +46,10 @@
 import org.opensearch.action.support.replication.ReplicationRequest;
 import org.opensearch.action.update.UpdateRequest;
 import org.opensearch.common.Nullable;
+import org.opensearch.common.UUIDs;
 import org.opensearch.common.annotation.PublicApi;
 import org.opensearch.common.unit.TimeValue;
+import org.opensearch.common.util.FeatureFlags;
 import org.opensearch.core.common.Strings;
 import org.opensearch.core.common.bytes.BytesArray;
 import org.opensearch.core.common.bytes.BytesReference;
@@ -62,6 +64,7 @@
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Locale;
 import java.util.Objects;
 import java.util.Set;
 
@@ -101,7 +104,13 @@ public class BulkRequest extends ActionRequest implements CompositeIndicesReques
 
     private long sizeInBytes = 0;
 
-    public BulkRequest() {}
+    private String uuid;
+
+    public BulkRequest() {
+        if (FeatureFlags.isEnabled(FeatureFlags.INDEX_BASED_INGEST_PIPELINE)) {
+            uuid = UUIDs.randomBase64UUID().toLowerCase(Locale.ROOT);
+        }
+    }
 
     public BulkRequest(StreamInput in) throws IOException {
         super(in);
@@ -112,6 +121,9 @@ public BulkRequest(StreamInput in) throws IOException {
         if (in.getVersion().onOrAfter(Version.V_2_14_0)) {
             batchSize = in.readInt();
         }
+        if (FeatureFlags.isEnabled(FeatureFlags.INDEX_BASED_INGEST_PIPELINE)) {
+            uuid = in.readOptionalString();
+        }
     }
 
     public BulkRequest(@Nullable String globalIndex) {
@@ -482,6 +494,9 @@ public void writeTo(StreamOutput out) throws IOException {
         if (out.getVersion().onOrAfter(Version.V_2_14_0)) {
             out.writeInt(batchSize);
         }
+        if (FeatureFlags.isEnabled(FeatureFlags.INDEX_BASED_INGEST_PIPELINE)) {
+            out.writeOptionalString(uuid);
+        }
     }
 
     @Override
@@ -515,4 +530,8 @@ public long ramBytesUsed() {
     public Set<String> getIndices() {
         return Collections.unmodifiableSet(indices);
     }
+
+    public String getUuid() {
+        return uuid;
+    }
 }

server/src/main/java/org/opensearch/action/bulk/TransportBulkAction.java

@@ -69,6 +69,7 @@
 import org.opensearch.common.inject.Inject;
 import org.opensearch.common.lease.Releasable;
 import org.opensearch.common.unit.TimeValue;
+import org.opensearch.common.util.FeatureFlags;
 import org.opensearch.common.util.concurrent.AtomicArray;
 import org.opensearch.core.Assertions;
 import org.opensearch.core.action.ActionListener;
@@ -250,8 +251,11 @@ protected void doInternalExecute(Task task, BulkRequest bulkRequest, String exec
         for (DocWriteRequest<?> actionRequest : bulkRequest.requests) {
             IndexRequest indexRequest = getIndexWriteRequest(actionRequest);
             if (indexRequest != null) {
+                if (FeatureFlags.isEnabled(FeatureFlags.INDEX_BASED_INGEST_PIPELINE)) {
+                    indexRequest.setBulkUuid(bulkRequest.getUuid());
+                }
                 // Each index request needs to be evaluated, because this method also modifies the IndexRequest
-                boolean indexRequestHasPipeline = IngestService.resolvePipelines(actionRequest, indexRequest, metadata);
+                boolean indexRequestHasPipeline = ingestService.resolvePipelines(actionRequest, indexRequest, metadata);
                 hasIndexRequestsWithPipelines |= indexRequestHasPipeline;
             }
 

server/src/main/java/org/opensearch/action/index/IndexRequest.java

@@ -49,6 +49,7 @@
 import org.opensearch.common.UUIDs;
 import org.opensearch.common.annotation.PublicApi;
 import org.opensearch.common.lucene.uid.Versions;
+import org.opensearch.common.util.FeatureFlags;
 import org.opensearch.common.xcontent.XContentHelper;
 import org.opensearch.common.xcontent.XContentType;
 import org.opensearch.core.common.Strings;
@@ -123,6 +124,11 @@ public class IndexRequest extends ReplicatedWriteRequest<IndexRequest> implement
 
     private String pipeline;
     private String finalPipeline;
+    private String indexBasedIngestPipeline;
+    /**
+     * The uuid of the bulk request containing this request.
+     */
+    private String bulkUuid;
 
     private boolean isPipelineResolved;
 
@@ -158,6 +164,11 @@ public IndexRequest(@Nullable ShardId shardId, StreamInput in) throws IOExceptio
         versionType = VersionType.fromValue(in.readByte());
         pipeline = in.readOptionalString();
         finalPipeline = in.readOptionalString();
+        // When enable the FF we should add a min version check to ensure the backward compatibility.
+        if (FeatureFlags.isEnabled(FeatureFlags.INDEX_BASED_INGEST_PIPELINE)) {
+            indexBasedIngestPipeline = in.readOptionalString();
+            bulkUuid = in.readOptionalString();
+        }
         isPipelineResolved = in.readBoolean();
         isRetry = in.readBoolean();
         autoGeneratedTimestamp = in.readLong();
@@ -314,6 +325,21 @@ public String getPipeline() {
         return this.pipeline;
     }
 
+    /**
+     * Sets the index based ingest pipeline to be executed before indexing the document
+     */
+    public IndexRequest setIndexBasedIngestPipeline(final String indexBasedIngestPipeline) {
+        this.indexBasedIngestPipeline = indexBasedIngestPipeline;
+        return this;
+    }
+
+    /**
+     * Returns the index based ingest pipeline to be executed before indexing the document
+     */
+    public String getIndexBasedIngestPipeline() {
+        return this.indexBasedIngestPipeline;
+    }
+
     /**
      * Sets the final ingest pipeline to be executed before indexing the document.
      *
@@ -668,6 +694,11 @@ private void writeBody(StreamOutput out) throws IOException {
         out.writeByte(versionType.getValue());
         out.writeOptionalString(pipeline);
         out.writeOptionalString(finalPipeline);
+        // When enable the FF we should add a min version check to ensure the backward compatibility.
+        if (FeatureFlags.isEnabled(FeatureFlags.INDEX_BASED_INGEST_PIPELINE)) {
+            out.writeOptionalString(indexBasedIngestPipeline);
+            out.writeOptionalString(bulkUuid);
+        }
         out.writeBoolean(isPipelineResolved);
         out.writeBoolean(isRetry);
         out.writeLong(autoGeneratedTimestamp);
@@ -743,4 +774,13 @@ public IndexRequest setRequireAlias(boolean requireAlias) {
         this.requireAlias = requireAlias;
         return this;
     }
+
+    public IndexRequest setBulkUuid(final String bulkUuid) {
+        this.bulkUuid = bulkUuid;
+        return this;
+    }
+
+    public String getBulkUuid() {
+        return bulkUuid;
+    }
 }

server/src/main/java/org/opensearch/common/util/FeatureFlags.java

@@ -127,6 +127,13 @@ public class FeatureFlags {
     public static final String ARROW_STREAMS = "opensearch.experimental.feature.arrow.streams.enabled";
     public static final Setting<Boolean> ARROW_STREAMS_SETTING = Setting.boolSetting(ARROW_STREAMS, false, Property.NodeScope);
 
+    public static final String INDEX_BASED_INGEST_PIPELINE = "opensearch.experimental.feature.index_based_ingest_pipeline.enabled";
+    public static final Setting<Boolean> INDEX_BASED_INGEST_PIPELINE_SETTING = Setting.boolSetting(
+        INDEX_BASED_INGEST_PIPELINE,
+        false,
+        Property.NodeScope
+    );
+
     private static final List<Setting<Boolean>> ALL_FEATURE_FLAG_SETTINGS = List.of(
         REMOTE_STORE_MIGRATION_EXPERIMENTAL_SETTING,
         EXTENSIONS_SETTING,
@@ -137,7 +144,8 @@ public class FeatureFlags {
         APPLICATION_BASED_CONFIGURATION_TEMPLATES_SETTING,
         READER_WRITER_SPLIT_EXPERIMENTAL_SETTING,
         TERM_VERSION_PRECOMMIT_ENABLE_SETTING,
-        ARROW_STREAMS_SETTING
+        ARROW_STREAMS_SETTING,
+        INDEX_BASED_INGEST_PIPELINE_SETTING
     );
 
     /**

server/src/main/java/org/opensearch/ingest/AbstractBatchingProcessor.java

@@ -133,4 +133,51 @@ protected abstract AbstractBatchingProcessor newProcessor(
             Map<String, Object> config
         );
     }
+
+    /**
+     * Factory class for creating {@link AbstractBatchingProcessor} instances based on index config.
+     *
+     * Since the processor config is generated based on the index config so the batch size info should also be defined
+     * as part of it. And different processors can have their own logic to decide the batch size so let each
+     * implementation of the newProcessor to handle it.
+     *
+     * @opensearch.internal
+     */
+    public abstract static class IndexBasedFactory implements Processor.Factory {
+        final String processorType;
+
+        protected IndexBasedFactory(String processorType) {
+            this.processorType = processorType;
+        }
+
+        /**
+         * Creates a new processor instance based on index config.
+         *
+         * @param processorFactories The processor factories.
+         * @param tag                 The processor tag.
+         * @param description         The processor description.
+         * @param config              The processor configuration.
+         * @return The new AbstractBatchProcessor instance.
+         * @throws Exception If the processor could not be created.
+         */
+        @Override
+        public AbstractBatchingProcessor create(
+            Map<String, Processor.Factory> processorFactories,
+            String tag,
+            String description,
+            Map<String, Object> config
+        ) throws Exception {
+            return newProcessor(tag, description, config);
+        }
+
+        /**
+         * Returns a new processor instance based on the index config.
+         *
+         * @param tag tag of the processor
+         * @param description description of the processor
+         * @param config configuration of the processor
+         * @return a new batch processor instance
+         */
+        protected abstract AbstractBatchingProcessor newProcessor(String tag, String description, Map<String, Object> config);
+    }
 }

server/src/main/java/org/opensearch/ingest/IndexBasedIngestPipelineCache.java

@@ -0,0 +1,134 @@
+/*
+ * 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.ingest;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+
+import reactor.util.annotation.NonNull;
+
+/**
+ * Cache for index based ingest pipeline
+ */
+public class IndexBasedIngestPipelineCache {
+    private final Map<String, CacheEntry> cache;
+
+    private final int MAX_ENTRIES = 100;
+    private final long EXPIRES_IN_MINUTES = 60;
+    private final int MAX_PROCESSOR_NUMBER = 10;
+
+    public IndexBasedIngestPipelineCache() {
+        this.cache = new ConcurrentHashMap<>();
+    }
+
+    /**
+     * Cache an index based ingest pipeline for an index.
+     * @param index [index_name/index_uuid]
+     * @param indexBasedIngestPipeline A pipeline created based on index configuration.
+     */
+    public void cachePipeline(@NonNull final String index, @NonNull final Pipeline indexBasedIngestPipeline) {
+        if (indexBasedIngestPipeline.getProcessors().size() > MAX_PROCESSOR_NUMBER) {
+            throw new IllegalArgumentException("Too many index based ingest processors for index: " + index);
+        }
+        if (cache.size() >= MAX_ENTRIES) {
+            evictOldestAndExpiredCacheEntry();
+        }
+        cache.put(index, new CacheEntry(indexBasedIngestPipeline));
+    }
+
+    // Evict the oldest and expired cache entry based on time
+    private void evictOldestAndExpiredCacheEntry() {
+        String oldestIndex = null;
+        long oldestTimestamp = Long.MAX_VALUE;
+        final List<String> expiredIndices = new ArrayList<>();
+
+        for (Map.Entry<String, CacheEntry> entry : cache.entrySet()) {
+            final CacheEntry cacheEntry = entry.getValue();
+            if (cacheEntry.getLastAccessTimestamp() < oldestTimestamp) {
+                oldestTimestamp = cacheEntry.getLastAccessTimestamp();
+                oldestIndex = entry.getKey();
+            }
+            if (cacheEntry.isExpired()) {
+                expiredIndices.add(entry.getKey());
+            }
+        }
+
+        if (oldestIndex != null) {
+            // Remove the oldest entry from both the cache and access order
+            cache.remove(oldestIndex);
+        }
+
+        for (final String expiredIndex : expiredIndices) {
+            cache.remove(expiredIndex);
+        }
+    }
+
+    /**
+     * Get the cached index based ingest pipeline for an index.
+     * @param index [index_name/index_uuid]
+     * @return cached index based ingest pipeline
+     */
+    public Pipeline getIndexBasedIngestPipeline(@NonNull final String index) {
+        // Check if the cache contains a valid entry for the index
+        final CacheEntry entry = cache.get(index);
+        if (entry != null) {
+            if (entry.isExpired()) {
+                cache.remove(index);
+                return null;
+            } else {
+                entry.setLastAccessTimestamp(System.currentTimeMillis());
+                return entry.getIndexBasedIngestPipeline();
+            }
+        }
+        return null;
+    }
+
+    public int size() {
+        return cache.size();
+    }
+
+    /**
+     * Invalidate the cache for an index.
+     * @param index [index_name/index_uuid]
+     */
+    public void invalidateCacheForIndex(@NonNull final String index) {
+        cache.remove(index);
+    }
+
+    private class CacheEntry {
+        private final Pipeline indexBasedIngestPipeline;
+        private final long createTimestamp;
+        private long lastAccessTimestamp;
+
+        public CacheEntry(Pipeline indexBasedIngestPipeline) {
+            this.indexBasedIngestPipeline = indexBasedIngestPipeline;
+            this.createTimestamp = System.currentTimeMillis();
+            this.lastAccessTimestamp = createTimestamp;
+        }
+
+        public Pipeline getIndexBasedIngestPipeline() {
+            return indexBasedIngestPipeline;
+        }
+
+        public boolean isExpired() {
+            return System.currentTimeMillis() - createTimestamp > TimeUnit.MINUTES.toMillis(EXPIRES_IN_MINUTES);
+        }
+
+        public void setLastAccessTimestamp(final long lastAccessTimestamp) {
+            this.lastAccessTimestamp = lastAccessTimestamp;
+        }
+
+        public long getLastAccessTimestamp() {
+            return this.lastAccessTimestamp;
+        }
+    }
+}