Support to emit multiple streams for a file content each responsible …

…for processing a specific part of the file

Signed-off-by: Raghuvansh Raj <[email protected]>

buildSrc/version.properties

@@ -41,6 +41,7 @@ commonscodec      = 1.15
 
 # plugin dependencies
 aws               = 1.12.270
+aws_v2            = 2.20.26
 
 # when updating this version, you need to ensure compatibility with:
 #  - plugins/ingest-attachment (transitive dependency, check the upstream POM)
@@ -56,3 +57,7 @@ bytebuddy         = 1.14.3
 
 # benchmark dependencies
 jmh               = 1.35
+
+jzlib = 1.1.3
+
+reactivestreams = 1.0.4

modules/transport-netty4/build.gradle

@@ -188,10 +188,6 @@ thirdPartyAudit {
     'com.github.luben.zstd.Zstd',
     'com.google.protobuf.nano.CodedOutputByteBufferNano',
     'com.google.protobuf.nano.MessageNano',
-    'com.jcraft.jzlib.Deflater',
-    'com.jcraft.jzlib.Inflater',
-    'com.jcraft.jzlib.JZlib$WrapperType',
-    'com.jcraft.jzlib.JZlib',
     'com.ning.compress.BufferRecycler',
     'com.ning.compress.lzf.ChunkDecoder',
     'com.ning.compress.lzf.ChunkEncoder',

server/build.gradle

@@ -146,9 +146,15 @@ dependencies {
   // jna
   api "net.java.dev.jna:jna:${versions.jna}"
 
+  // jcraft
+  api "com.jcraft:jzlib:${versions.jzlib}"
+
   // protobuf
   api "com.google.protobuf:protobuf-java:${versions.protobuf}"
 
+  // jcraft
+  api "com.jcraft:jzlib:${versions.jzlib}"
+
   testImplementation(project(":test:framework")) {
     // tests use the locally compiled version of server
     exclude group: 'org.opensearch', module: 'server'

server/licenses/jzlib-1.1.3.jar.sha1

@@ -0,0 +1 @@
+c01428efa717624f7aabf4df319939dda9646b2d

server/licenses/jzlib-LICENSE.txt

@@ -0,0 +1,29 @@
+JZlib 0.0.* were released under the GNU LGPL license.  Later, we have switched
+over to a BSD-style license.
+
+------------------------------------------------------------------------------
+Copyright (c) 2000-2011 ymnk, JCraft,Inc. All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are met:
+
+  Redistributions of source code must retain the above copyright notice, this
+  list of conditions and the following disclaimer.
+
+  Redistributions in binary form must reproduce the above copyright notice,
+  this list of conditions and the following disclaimer in the documentation
+  and/or other materials provided with the distribution.
+
+  The name of the authors may not be used to endorse or promote products
+  derived from this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+DISCLAIMED. IN NO EVENT SHALL JCRAFT, INC. OR ANY CONTRIBUTORS TO THIS SOFTWARE
+BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
+ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

server/src/main/java/org/opensearch/common/CheckedTriFunction.java

@@ -0,0 +1,19 @@
+/*
+ * 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.common;
+
+/**
+ * A {@link TriFunction}-like interface which allows throwing checked exceptions.
+ *
+ * @opensearch.internal
+ */
+@FunctionalInterface
+public interface CheckedTriFunction<S, T, U, R, E extends Exception> {
+    R apply(S s, T t, U u) throws E;
+}

server/src/main/java/org/opensearch/common/StreamContext.java

@@ -0,0 +1,66 @@
+/*
+ * 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.common;
+
+import org.opensearch.common.io.InputStreamContainer;
+
+import java.io.IOException;
+
+/**
+ * StreamContext is used to supply streams to vendor plugins using {@link StreamContext#provideStream}
+ *
+ * @opensearch.internal
+ */
+public class StreamContext {
+
+    private final CheckedTriFunction<Integer, Long, Long, InputStreamContainer, IOException> streamSupplier;
+    private final long partSize;
+    private final long lastPartSize;
+    private final int numberOfParts;
+
+    /**
+     * Construct a new StreamProvider object
+     *
+     * @param streamSupplier An implementation of TransferPartStreamSupplier which will be called when provideStreams is called
+     * @param partSize Size of all parts apart from the last one
+     * @param lastPartSize Size of the last part
+     * @param numberOfParts Total number of parts
+     */
+    public StreamContext(
+        CheckedTriFunction<Integer, Long, Long, InputStreamContainer, IOException> streamSupplier,
+        long partSize,
+        long lastPartSize,
+        int numberOfParts
+    ) {
+        this.streamSupplier = streamSupplier;
+        this.partSize = partSize;
+        this.lastPartSize = lastPartSize;
+        this.numberOfParts = numberOfParts;
+    }
+
+    /**
+     * Vendor plugins can use this method to create new streams only when they are required for processing
+     * New streams won't be created till this method is called with the specific <code>partNumber</code>
+     *
+     * @param partNumber The index of the part
+     * @return A stream reference to the part requested
+     */
+    public InputStreamContainer provideStream(int partNumber) throws IOException {
+        long position = partSize * partNumber;
+        long size = (partNumber == numberOfParts - 1) ? lastPartSize : partSize;
+        return streamSupplier.apply(partNumber, size, position);
+    }
+
+    /**
+     * @return The number of parts in which this file is supposed to be uploaded
+     */
+    public int getNumberOfParts() {
+        return numberOfParts;
+    }
+}

server/src/main/java/org/opensearch/common/blobstore/stream/package-info.java

@@ -0,0 +1,10 @@
+/*
+ * 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.
+ */
+
+/** Abstractions for stream based file transfers */
+package org.opensearch.common.blobstore.stream;

server/src/main/java/org/opensearch/common/blobstore/stream/write/StreamContextSupplier.java

@@ -0,0 +1,25 @@
+/*
+ * 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.common.blobstore.stream.write;
+
+import org.opensearch.common.StreamContext;
+
+/**
+ * Will return the <code>StreamContext</code> to the caller given the part size
+ *
+ * @opensearch.internal
+ */
+public interface StreamContextSupplier {
+
+    /**
+     * @param partSize The size of a single part to be uploaded
+     * @return The <code>StreamContext</code> based on the part size provided
+     */
+    StreamContext supplyStreamContext(long partSize);
+}

server/src/main/java/org/opensearch/common/blobstore/stream/write/WriteContext.java

@@ -0,0 +1,118 @@
+/*
+ * 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.common.blobstore.stream.write;
+
+import org.opensearch.common.Nullable;
+import org.opensearch.common.StreamContext;
+import org.opensearch.common.blobstore.transfer.UploadFinalizer;
+
+/**
+ * WriteContext is used to encapsulate all data needed by <code>BlobContainer#writeStreams</code>
+ *
+ * @opensearch.internal
+ */
+public class WriteContext {
+
+    private final String fileName;
+    private final StreamContextSupplier streamContextSupplier;
+    private final long fileSize;
+    private final boolean failIfAlreadyExists;
+    private final WritePriority writePriority;
+    private final UploadFinalizer uploadFinalizer;
+    private final boolean doRemoteDataIntegrityCheck;
+    private final Long expectedChecksum;
+
+    /**
+     * Construct a new WriteContext object
+     *
+     * @param fileName                   The name of the file being uploaded
+     * @param streamContextSupplier      A supplier that will provide StreamContext to the plugin
+     * @param fileSize                   The total size of the file being uploaded
+     * @param failIfAlreadyExists        A boolean to fail the upload is the file exists
+     * @param writePriority              The <code>WritePriority</code> of this upload
+     * @param doRemoteDataIntegrityCheck A boolean to inform vendor plugins whether remote data integrity checks need to be done
+     * @param expectedChecksum           This parameter expected only when the vendor plugin is expected to do server side data integrity verification
+     */
+    public WriteContext(
+        String fileName,
+        StreamContextSupplier streamContextSupplier,
+        long fileSize,
+        boolean failIfAlreadyExists,
+        WritePriority writePriority,
+        UploadFinalizer uploadFinalizer,
+        boolean doRemoteDataIntegrityCheck,
+        @Nullable Long expectedChecksum
+    ) {
+        this.fileName = fileName;
+        this.streamContextSupplier = streamContextSupplier;
+        this.fileSize = fileSize;
+        this.failIfAlreadyExists = failIfAlreadyExists;
+        this.writePriority = writePriority;
+        this.uploadFinalizer = uploadFinalizer;
+        this.doRemoteDataIntegrityCheck = doRemoteDataIntegrityCheck;
+        this.expectedChecksum = expectedChecksum;
+    }
+
+    /**
+     * @return The file name
+     */
+    public String getFileName() {
+        return fileName;
+    }
+
+    /**
+     * @return The boolean representing whether to fail the file upload if it exists
+     */
+    public boolean isFailIfAlreadyExists() {
+        return failIfAlreadyExists;
+    }
+
+    /**
+     * @param partSize The size of a single part to be uploaded
+     * @return The stream context which will be used by the plugin to initialize streams from the file
+     */
+    public StreamContext getStreamProvider(long partSize) {
+        return streamContextSupplier.supplyStreamContext(partSize);
+    }
+
+    /**
+     * @return The total size of the file
+     */
+    public long getFileSize() {
+        return fileSize;
+    }
+
+    /**
+     * @return The <code>WritePriority</code> of the upload
+     */
+    public WritePriority getWritePriority() {
+        return writePriority;
+    }
+
+    /**
+     * @return The <code>UploadFinalizer</code> for this upload
+     */
+    public UploadFinalizer getUploadFinalizer() {
+        return uploadFinalizer;
+    }
+
+    /**
+     * @return A boolean for whether remote data integrity check has to be done for this upload or not
+     */
+    public boolean doRemoteDataIntegrityCheck() {
+        return doRemoteDataIntegrityCheck;
+    }
+
+    /**
+     * @return The CRC32 checksum associated with this file
+     */
+    public Long getExpectedChecksum() {
+        return expectedChecksum;
+    }
+}

server/src/main/java/org/opensearch/common/blobstore/stream/write/WritePriority.java

@@ -0,0 +1,19 @@
+/*
+ * 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.common.blobstore.stream.write;
+
+/**
+ * WritePriority for upload
+ *
+ * @opensearch.internal
+ */
+public enum WritePriority {
+    NORMAL,
+    HIGH
+}

server/src/main/java/org/opensearch/common/blobstore/stream/write/package-info.java

@@ -0,0 +1,10 @@
+/*
+ * 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.
+ */
+
+/** Abstractions for stream based file writes */
+package org.opensearch.common.blobstore.stream.write;