Refactor Compressors from CompressorFactory to CompressorRegistry for extensibility

gradle/missing-javadoc.gradle

@@ -166,6 +166,7 @@ configure([
 configure([
  project(":libs:opensearch-common"),
  project(":libs:opensearch-core"),
+ project(":libs:opensearch-compress"),
  project(":plugins:events-correlation-engine"),
  project(":server")
 ]) {

libs/compress/build.gradle

@@ -0,0 +1,38 @@
+/*
+ * 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.
+ *
+ * Modifications Copyright OpenSearch Contributors. See
+ * GitHub history for details.
+ */
+
+apply plugin: 'opensearch.build'
+apply plugin: 'opensearch.publish'
+
+base {
+ archivesName = 'opensearch-compress'
+}
+
+dependencies {
+ api project(':libs:opensearch-common')
+ api project(':libs:opensearch-core')
+
+ //zstd
+ api "com.github.luben:zstd-jni:${versions.zstd}"
+
+ testImplementation(project(":test:framework")) {
+ // tests use the locally compiled version of server
+ exclude group: 'org.opensearch', module: 'opensearch-compress'
+ }
+}
+
+tasks.named('forbiddenApisMain').configure {
+ // :libs:opensearch-compress does not depend on server
+ // TODO: Need to decide how we want to handle for forbidden signatures with the changes to server
+ replaceSignatureFiles 'jdk-signatures'
+}
+
+jarHell.enabled = false

server/src/main/java/org/opensearch/common/compress/ZstdCompressor.java → libs/compress/src/main/java/org/opensearch/compress/ZstdCompressor.java

@@ -6,13 +6,14 @@
  * compatible open source license.
  */
 
-package org.opensearch.common.compress;
+package org.opensearch.compress;
 
 import com.github.luben.zstd.RecyclingBufferPool;
 import com.github.luben.zstd.ZstdInputStreamNoFinalizer;
 import com.github.luben.zstd.ZstdOutputStreamNoFinalizer;
+import org.opensearch.common.annotation.PublicApi;
 import org.opensearch.core.common.bytes.BytesReference;
-import org.opensearch.core.common.compress.Compressor;
+import org.opensearch.core.compress.Compressor;
 
 import java.io.BufferedInputStream;
 import java.io.BufferedOutputStream;
@@ -24,7 +25,8 @@
 /**
  * {@link Compressor} implementation based on the ZSTD compression algorithm.
  *
- * @opensearch.internal
+ * @opensearch.api - registered name requires BWC support
+ * @opensearch.experimental - class methods might change
  */
 public class ZstdCompressor implements Compressor {
  // An arbitrary header that we use to identify compressed streams
@@ -33,6 +35,14 @@ public class ZstdCompressor implements Compressor {
  // a XContent
  private static final byte[] HEADER = new byte[] { 'Z', 'S', 'T', 'D', '\0' };
 
+ /**
+ * The name to register the compressor by
+ *
+ * @opensearch.api - requires BWC support
+ */
+ @PublicApi(since = "2.10.0")
+ public static final String NAME = "ZSTD";
+
  private static final int LEVEL = 3;
 
  private static final int BUFFER_SIZE = 4096;

libs/core/src/main/java/org/opensearch/core/common/compress/package-info.java → libs/compress/src/main/java/org/opensearch/compress/package-info.java

@@ -6,5 +6,7 @@
  * compatible open source license.
  */
 
-/** Classes for core compress module */
-package org.opensearch.core.common.compress;
+/**
+ * Concrete {@link org.opensearch.core.compress.Compressor} implementations
+ */
+package org.opensearch.compress;

libs/compress/src/main/java/org/opensearch/compress/spi/CompressionProvider.java

@@ -0,0 +1,32 @@
+/*
+ * 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.compress.spi;
+
+import org.opensearch.compress.ZstdCompressor;
+import org.opensearch.core.compress.Compressor;
+import org.opensearch.core.compress.spi.CompressorProvider;
+
+import java.util.AbstractMap.SimpleEntry;
+import java.util.Map.Entry;
+import java.util.List;
+
+/**
+ * Additional "optional" compressor implementations provided by the opensearch compress library
+ *
+ * @opensearch.internal
+ */
+public class CompressionProvider implements CompressorProvider {
+
+ /** Returns the concrete {@link Compressor}s provided by the compress library */
+ @SuppressWarnings({ "unchecked", "rawtypes" })
+ @Override
+ public List<Entry<String, Compressor>> getCompressors() {
+ return List.of(new SimpleEntry<>(ZstdCompressor.NAME, new ZstdCompressor()));
+ }
+}

libs/compress/src/main/java/org/opensearch/compress/spi/package-info.java

@@ -0,0 +1,15 @@
+/*
+ * 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.
+ */
+
+/**
+ * Service Provider Interface for registering concrete {@link org.opensearch.core.compress.Compressor}
+ * implementations.
+ *
+ * See {@link org.opensearch.compress.ZstdCompressor}
+ */
+package org.opensearch.compress.spi;

libs/compress/src/main/java/org/opensearch/package-info.java

@@ -0,0 +1,13 @@
+/*
+ * 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.
+ */
+
+/**
+ * This is the compress library for registering optional
+ * {@link org.opensearch.core.compress.Compressor} implementations
+ */
+package org.opensearch;

libs/compress/src/main/resources/META-INF/services/org.opensearch.core.compress.spi.CompressorProvider

@@ -0,0 +1,9 @@
+#
+# 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.
+#
+
+org.opensearch.compress.spi.CompressionProvider

server/src/test/java/org/opensearch/common/compress/ZstdCompressTests.java → libs/compress/src/test/java/org/opensearch/compress/ZstdCompressTests.java

@@ -6,19 +6,20 @@
  * compatible open source license.
  */
 
-package org.opensearch.common.compress;
+package org.opensearch.compress;
 
-import org.opensearch.core.common.compress.Compressor;
+import org.opensearch.core.compress.Compressor;
+import org.opensearch.test.core.compress.AbstractCompressorTestCase;
 
 /**
  * Test streaming compression
  */
-public class ZstdCompressTests extends AbstractCompressorTests {
+public class ZstdCompressTests extends AbstractCompressorTestCase {
 
  private final Compressor compressor = new ZstdCompressor();
 
  @Override
- Compressor compressor() {
+ protected Compressor compressor() {
  return compressor;
  }
 }

libs/core/src/main/java/org/opensearch/ExceptionsHelper.java

@@ -43,7 +43,7 @@
 import org.opensearch.common.CheckedSupplier;
 import org.opensearch.common.Nullable;
 import org.opensearch.core.action.ShardOperationFailedException;
-import org.opensearch.core.common.compress.NotXContentException;
+import org.opensearch.core.compress.NotXContentException;
 import org.opensearch.core.concurrency.OpenSearchRejectedExecutionException;
 import org.opensearch.core.index.Index;
 import org.opensearch.core.rest.RestStatus;

libs/core/src/main/java/org/opensearch/core/common/compress/Compressor.java → libs/core/src/main/java/org/opensearch/core/compress/Compressor.java

@@ -30,19 +30,29 @@
  * GitHub history for details.
  */
 
-package org.opensearch.core.common.compress;
+package org.opensearch.core.compress;
 
+import org.opensearch.common.annotation.ExperimentalApi;
+import org.opensearch.common.annotation.PublicApi;
 import org.opensearch.core.common.bytes.BytesReference;
 
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
 
 /**
- * Compressor interface
+ * Compressor interface used for compressing {@link org.opensearch.core.xcontent.MediaType} and
+ * {@code org.opensearch.repositories.blobstore.BlobStoreRepository} implementations.
  *
- * @opensearch.internal
+ * This is not to be confused with {@link org.apache.lucene.codecs.compressing.Compressor} which is used
+ * for codec implementations such as {@code org.opensearch.index.codec.customcodecs.Lucene95CustomCodec}
+ * for compressing {@link org.apache.lucene.document.StoredField}s
+ *
+ * @opensearch.api - intended to be extended
+ * @opensearch.experimental - however, bwc is not guaranteed at this time
  */
+@ExperimentalApi
+@PublicApi(since = "2.10.0")
 public interface Compressor {
 
  boolean isCompressed(BytesReference bytes);

libs/core/src/main/java/org/opensearch/core/compress/CompressorRegistry.java

@@ -0,0 +1,115 @@
+/*
+ * 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.core.compress;
+
+import org.opensearch.common.Nullable;
+import org.opensearch.common.annotation.InternalApi;
+import org.opensearch.core.common.bytes.BytesReference;
+import org.opensearch.core.compress.spi.CompressorProvider;
+import org.opensearch.core.xcontent.MediaTypeRegistry;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Objects;
+import java.util.ServiceLoader;
+import java.util.stream.Collectors;
+
+/**
+ * A registry that wraps a static Map singleton which holds a mapping of unique String names (typically the
+ * compressor header as a string) to registerd {@link Compressor} implementations.
+ *
+ * This enables plugins, modules, extensions to register their own compression implementations through SPI
+ *
+ * @opensearch.experimental
+ * @opensearch.internal
+ */
+@InternalApi
+public final class CompressorRegistry {
+
+ // the backing registry map
+ private static final Map<String, Compressor> registeredCompressors = ServiceLoader.load(
+ CompressorProvider.class,
+ CompressorProvider.class.getClassLoader()
+ )
+ .stream()
+ .flatMap(p -> p.get().getCompressors().stream())
+ .collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, Map.Entry::getValue));
+
+ // no instance:
+ private CompressorRegistry() {}
+
+ /**
+ * Returns the default compressor
+ */
+ public static Compressor defaultCompressor() {
+ return registeredCompressors.get("DEFLATE");
+ }
+
+ public static Compressor none() {
+ return registeredCompressors.get(NoneCompressor.NAME);
+ }
+
+ public static boolean isCompressed(BytesReference bytes) {
+ return compressor(bytes) != null;
+ }
+
+ @Nullable
+ public static Compressor compressor(final BytesReference bytes) {
+ for (Compressor compressor : registeredCompressors.values()) {
+ if (compressor.isCompressed(bytes) == true) {
+ // bytes should be either detected as compressed or as xcontent,
+ // if we have bytes that can be either detected as compressed or
+ // as a xcontent, we have a problem
+ assert MediaTypeRegistry.xContentType(bytes) == null;
+ return compressor;
+ }
+ }
+
+ if (MediaTypeRegistry.xContentType(bytes) == null) {
+ throw new NotXContentException("Compressor detection can only be called on some xcontent bytes or compressed xcontent bytes");
+ }
+
+ return null;
+ }
+
+ /** Decompress the provided {@link BytesReference}. */
+ public static BytesReference uncompress(BytesReference bytes) throws IOException {
+ Compressor compressor = compressor(bytes);
+ if (compressor == null) {
+ throw new NotCompressedException();
+ }
+ return compressor.uncompress(bytes);
+ }
+
+ /**
+ * Uncompress the provided data, data can be detected as compressed using {@link #isCompressed(BytesReference)}.
+ */
+ public static BytesReference uncompressIfNeeded(BytesReference bytes) throws IOException {
+ Compressor compressor = compressor(Objects.requireNonNull(bytes, "the BytesReference must not be null"));
+ return compressor == null ? bytes : compressor.uncompress(bytes);
+ }
+
+ /** Returns a registered compressor by its registered name */
+ public static Compressor getCompressor(final String name) {
+ if (registeredCompressors.containsKey(name)) {
+ return registeredCompressors.get(name);
+ }
+ throw new IllegalArgumentException("No registered compressor found by name [" + name + "]");
+ }
+
+ /**
+ * Returns the registered compressors as an Immutable collection
+ *
+ * note: used for testing
+ */
+ public static Map<String, Compressor> registeredCompressors() {
+ // no destructive danger as backing map is immutable
+ return registeredCompressors;
+ }
+}

server/src/main/java/org/opensearch/common/compress/NoneCompressor.java → libs/core/src/main/java/org/opensearch/core/compress/NoneCompressor.java

@@ -6,10 +6,10 @@
  * compatible open source license.
  */
 
-package org.opensearch.common.compress;
+package org.opensearch.core.compress;
 
+import org.opensearch.common.annotation.PublicApi;
 import org.opensearch.core.common.bytes.BytesReference;
-import org.opensearch.core.common.compress.Compressor;
 
 import java.io.IOException;
 import java.io.InputStream;
@@ -18,9 +18,18 @@
 /**
  * {@link Compressor} no compressor implementation.
  *
- * @opensearch.internal
+ * @opensearch.api - registered name requires BWC support
+ * @opensearch.experimental - class methods might change
  */
 public class NoneCompressor implements Compressor {
+ /**
+ * The name to register the compressor by
+ *
+ * @opensearch.api - requires BWC support
+ */
+ @PublicApi(since = "2.10.0")
+ public static final String NAME = "NONE";
+
  @Override
  public boolean isCompressed(BytesReference bytes) {
  return false;

server/src/main/java/org/opensearch/common/compress/NotCompressedException.java → libs/core/src/main/java/org/opensearch/core/compress/NotCompressedException.java

@@ -30,7 +30,7 @@
  * GitHub history for details.
  */
 
-package org.opensearch.common.compress;
+package org.opensearch.core.compress;
 
 /**
  * Exception indicating that we were expecting something compressed, which