Add Zstd support for EVCache

evcache-core/build.gradle

@@ -43,6 +43,7 @@ dependencies {
         api group:"joda-time",                    name:"joda-time",                        version:"latest.release"
         api group:"javax.annotation",             name:"javax.annotation-api",             version:"latest.release"
         api group:"com.github.fzakaria",          name:"ascii85",                          version:"latest.release"
+        api group:"com.github.luben",             name:"zstd-jni",                         version:"1.5.7-11"
 
         testImplementation group:"org.testng",               name:"testng",                           version:"7.5"
         testImplementation group:"com.beust",                name:"jcommander",                       version:"1.72"

evcache-core/src/main/java/com/netflix/evcache/EVCacheSerializingTranscoder.java

@@ -22,32 +22,44 @@
 
 package com.netflix.evcache;
 
+import com.github.luben.zstd.Zstd;
+import com.github.luben.zstd.ZstdInputStream;
 import com.netflix.evcache.config.EVCacheTranscoderProperties;
+import com.netflix.evcache.config.EVCacheTranscoderProperties.CompressionAlgorithm;
 import com.netflix.evcache.metrics.EVCacheMetricsFactory;
 import com.netflix.evcache.pool.EVCacheValue;
 import com.netflix.evcache.pool.EVCacheValueSerde;
 import com.netflix.evcache.util.EVCacheConfig;
 import com.netflix.spectator.api.BasicTag;
+import com.netflix.spectator.api.DistributionSummary;
 import com.netflix.spectator.api.Tag;
-import com.netflix.spectator.api.Timer;
 import net.spy.memcached.CachedData;
 import net.spy.memcached.transcoders.BaseSerializingTranscoder;
 import net.spy.memcached.transcoders.Transcoder;
 import net.spy.memcached.transcoders.TranscoderUtils;
 
-import java.time.Duration;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
 import java.util.ArrayList;
 import java.util.Date;
+import java.util.EnumMap;
 import java.util.List;
-import java.util.concurrent.TimeUnit;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Transcoder that serializes and compresses objects.
  */
 public class EVCacheSerializingTranscoder extends BaseSerializingTranscoder implements
         Transcoder<Object> {
 
+    private static final Logger log = LoggerFactory.getLogger(EVCacheSerializingTranscoder.class);
+
     // General flags
     static final int SERIALIZED = 1;
     static final int COMPRESSED = 2;
@@ -63,12 +75,15 @@ public class EVCacheSerializingTranscoder extends BaseSerializingTranscoder impl
     static final int SPECIAL_DOUBLE = (7 << 8);
     static final int SPECIAL_BYTEARRAY = (8 << 8);
 
-    static final String COMPRESSION = "COMPRESSION_METRIC";
+    public static final int DEFAULT_ZSTD_COMPRESSION_LEVEL = 3;
+
+    private static final int ZSTD_MAGIC = 0xFD2FB528;
 
     private final TranscoderUtils tu = new TranscoderUtils(true);
-    private Timer timer;
+    protected final String appName;
+    protected EVCacheTranscoderProperties transcoderProperties;
 
-    protected final EVCacheTranscoderProperties transcoderProperties;
+    private final EnumMap<CompressionAlgorithm, DistributionSummary> compressionRatioSummaries;
 
     /**
      * Get a serializing transcoder with the default max data size.
@@ -94,7 +109,22 @@ public EVCacheSerializingTranscoder(int max) {
      */
     public EVCacheSerializingTranscoder(int max, EVCacheTranscoderProperties properties) {
         super(max);
+        this.appName = properties.getAppName();
         this.transcoderProperties = properties;
+        this.compressionRatioSummaries = buildCompressionRatioSummaries(appName);
+    }
+
+    private static EnumMap<CompressionAlgorithm, DistributionSummary> buildCompressionRatioSummaries(String appName) {
+        EnumMap<CompressionAlgorithm, DistributionSummary> summaries = new EnumMap<>(CompressionAlgorithm.class);
+        for (CompressionAlgorithm algo : CompressionAlgorithm.values()) {
+            List<Tag> tagList = new ArrayList<>(2);
+            tagList.add(new BasicTag(EVCacheMetricsFactory.COMPRESSION_TYPE, algo.name().toLowerCase()));
+            if (appName != null && !appName.isEmpty()) {
+                tagList.add(new BasicTag(EVCacheMetricsFactory.CACHE, appName));
+            }
+            summaries.put(algo, EVCacheMetricsFactory.getInstance().getDistributionSummary(EVCacheMetricsFactory.COMPRESSION_RATIO, tagList));
+        }
+        return summaries;
     }
 
     @Override
@@ -146,7 +176,7 @@ public Object decode(CachedData d) {
                     rv = data;
                     break;
                 default:
-                    getLogger().warn("Undecodeable with flags %x", flags);
+                    log.warn("Undecodeable with flags {}", Integer.toHexString(flags));
             }
         } else {
             rv = decodeString(data);
@@ -194,19 +224,19 @@ public CachedData encode(Object o) {
         }
         assert b != null;
         if (b.length > compressionThreshold) {
+            int originalLength = b.length;
             byte[] compressed = compress(b);
-            if (compressed.length < b.length) {
-                getLogger().trace("Compressed %s from %d to %d",
-                        o.getClass().getName(), b.length, compressed.length);
+            if (compressed.length < originalLength) {
+                if (log.isTraceEnabled()) {
+                    log.trace("Compressed {} from {} to {}",
+                            o.getClass().getName(), originalLength, compressed.length);
+                }
                 b = compressed;
                 flags |= COMPRESSED;
-            } else {
-                getLogger().debug("Compression increased the size of %s from %d to %d",
-                        o.getClass().getName(), b.length, compressed.length);
+            } else if (log.isDebugEnabled()) {
+                log.debug("Compression increased the size of {} from {} to {}",
+                        o.getClass().getName(), originalLength, compressed.length);
             }
-
-            long compression_ratio = Math.round((double) compressed.length / b.length * 100);
-            updateTimerWithCompressionRatio(compression_ratio);
         }
         return new CachedData(flags, b, getMaxSize());
     }
@@ -227,14 +257,89 @@ protected Object deserialize(byte[] in) {
         return super.deserialize(in);
     }
 
-    private void updateTimerWithCompressionRatio(long ratio_percentage) {
-        if(timer == null) {
-            final List<Tag> tagList = new ArrayList<Tag>(1);
-            tagList.add(new BasicTag(EVCacheMetricsFactory.COMPRESSION_TYPE, "gzip"));
-            timer = EVCacheMetricsFactory.getInstance().getPercentileTimer(EVCacheMetricsFactory.COMPRESSION_RATIO, tagList, Duration.ofMillis(100));
-        };
+    @Override
+    protected byte[] compress(byte[] in) {
+        if (in == null) throw new NullPointerException("Can't compress null");
+
+        CompressionAlgorithm compressionAlgorithm = transcoderProperties.getCompressionAlgorithmProperty().get();
+        byte[] compressed;
+        switch (compressionAlgorithm) {
+            case ZSTD:
+                int zstdLevel = transcoderProperties.getZstdCompressionLevelProperty().get();
+                if (log.isDebugEnabled()) {
+                    log.debug("algorithm: {}, level: {}, appName: {}", compressionAlgorithm, zstdLevel, appName);
+                }
+                compressed = Zstd.compress(in, zstdLevel);
+                break;
+            case GZIP:
+                if (log.isDebugEnabled()) {
+                    log.debug("algorithm: {}, appName: {}", compressionAlgorithm, appName);
+                }
+                compressed = super.compress(in);
+                break;
+            default:
+                throw new IllegalArgumentException("Unsupported compression algorithm: " + compressionAlgorithm);
+        }
 
-        timer.record(ratio_percentage, TimeUnit.MILLISECONDS);
+        if (compressed != null) {
+            long ratioPerCent = Math.round((double) compressed.length / in.length * 100.0);
+            recordCompressionRatio(ratioPerCent, compressionAlgorithm);
+        }
+
+        return compressed;
     }
 
+    @Override
+    protected byte[] decompress(byte[] in) {
+        if (in == null || in.length == 0) return in;
+        if (isZstdCompressed(in)) return decompressZstd(in);
+        return super.decompress(in);
+    }
+
+    private boolean isZstdCompressed(byte[] data) {
+        if (data == null || data.length < 4) return false;
+        int magic = ByteBuffer.wrap(data, 0, 4).order(ByteOrder.LITTLE_ENDIAN).getInt();
+        return magic == ZSTD_MAGIC;
+    }
+
+    private byte[] decompressZstd(byte[] in) {
+        long originalSize = Zstd.getFrameContentSize(in);
+        if (originalSize > Integer.MAX_VALUE) {
+            throw new RuntimeException("Zstd decompressed size exceeds int range: " + originalSize);
+        }
+        if (originalSize > 0) {
+            // Fast path: frame carries a content-size header (compress() above always does).
+            return Zstd.decompress(in, (int) originalSize);
+        }
+        // Slow path: declared size is 0, unknown (-1), or invalid (-2) — stream-decode and let
+        // ZstdInputStream surface any frame errors.
+        log.warn("Zstd frame missing content-size header (getFrameContentSize={}); falling back to stream decode. appName={}", originalSize, appName);
+        ZstdInputStream zis = null;
+        try {
+             zis = new ZstdInputStream(new ByteArrayInputStream(in));
+            return readAll(zis);
+        } catch (IOException e) {
+            log.error("Error reading Zstd input stream", e);
+            return null;
+        } finally {
+            try { if (zis != null) zis.close(); } catch (IOException ignored) {}
+        }
+    }
+
+    private static byte[] readAll(InputStream in) throws IOException {
+        ByteArrayOutputStream out = new ByteArrayOutputStream();
+        byte[] buf = new byte[8192];
+        int n;
+        while ((n = in.read(buf)) != -1) {
+            out.write(buf, 0, n);
+        }
+        return out.toByteArray();
+    }
+
+    private void recordCompressionRatio(long ratioPerCent, CompressionAlgorithm compressionAlgorithm) {
+        DistributionSummary summary = compressionRatioSummaries.get(compressionAlgorithm);
+        if (summary != null) {
+            summary.record(ratioPerCent);
+        }
+    }
 }

evcache-core/src/main/java/com/netflix/evcache/config/EVCacheTranscoderProperties.java

@@ -2,6 +2,7 @@
 
 import com.netflix.archaius.api.Property;
 import com.netflix.archaius.api.PropertyRepository;
+import net.spy.memcached.compat.SpyObject;
 
 /**
  * Properties related to {@link com.netflix.evcache.EVCacheTranscoder}
@@ -21,16 +22,22 @@
  *   {@code .get()}.
  *
  */
-public final class EVCacheTranscoderProperties {
+public final class EVCacheTranscoderProperties extends SpyObject {
 
     public static final boolean DEFAULT_BINARY_SERIALIZATION_ENABLED = false;
     public static final int DEFAULT_MAX_DATA_SIZE_BYTES = 20 * 1024 * 1024;
     public static final int DEFAULT_COMPRESSION_THRESHOLD_BYTES = 120;
+    public static final CompressionAlgorithm DEFAULT_COMPRESSION_ALGORITHM = CompressionAlgorithm.GZIP;
+    public static final int DEFAULT_COMPRESSION_ZSTD_LEVEL = 3;
+
+    public enum CompressionAlgorithm { GZIP, ZSTD }
 
     public enum Key {
         BINARY_SERIALIZATION_ENABLED("binary.serialization.enabled", "default.evcache.binary.serialization.enabled"),
         MAX_DATA_SIZE_BYTES("max.data.size", "default.evcache.max.data.size"),
-        COMPRESSION_THRESHOLD_BYTES("compression.threshold", "default.evcache.compression.threshold");
+        COMPRESSION_THRESHOLD_BYTES("compression.threshold", "default.evcache.compression.threshold"),
+        COMPRESSION_ALGORITHM("compression.algorithm", "default.evcache.compression.algorithm"),
+        COMPRESSION_ZSTD_LEVEL("compression.zstd.level", "default.evcache.compression.zstd.level");
 
         final String appKeySuffix;
         final String globalKey;
@@ -46,6 +53,8 @@ public enum Key {
     private final boolean binarySerializationEnabled;
     private final int maxDataSizeBytes;
     private final int compressionThresholdBytes;
+    private final Property<CompressionAlgorithm> compressionAlgorithmProperty;
+    private final Property<Integer> zstdCompressionLevelProperty;
 
     /**
      * Construct the bundle and snapshot every property via the three-level resolution chain.
@@ -68,6 +77,11 @@ public EVCacheTranscoderProperties(String appName, PropertyRepository propertyRe
                 Key.MAX_DATA_SIZE_BYTES, Integer.class, DEFAULT_MAX_DATA_SIZE_BYTES).get();
         this.compressionThresholdBytes = getProperty(appName, propertyRepository,
                 Key.COMPRESSION_THRESHOLD_BYTES, Integer.class, DEFAULT_COMPRESSION_THRESHOLD_BYTES).get();
+        this.compressionAlgorithmProperty = getProperty(appName, propertyRepository,
+                Key.COMPRESSION_ALGORITHM, String.class, DEFAULT_COMPRESSION_ALGORITHM.name())
+                .map(this::parseCompressionAlgorithm);
+        this.zstdCompressionLevelProperty = getProperty(appName, propertyRepository,
+                Key.COMPRESSION_ZSTD_LEVEL, Integer.class, DEFAULT_COMPRESSION_ZSTD_LEVEL);
     }
 
     public String getAppName() {
@@ -86,6 +100,40 @@ public int getCompressionThresholdBytes() {
         return compressionThresholdBytes;
     }
 
+    /**
+     * Live-updating {@link Property} handle for the transcoder compression algorithm. Calling
+     * {@code .get()} returns the current {@link CompressionAlgorithm} value; underlying storage
+     * is a String property (case-insensitive) so ops can set the FP as {@code "gzip"} or
+     * {@code "ZSTD"} interchangeably. Handle is resolved once at bundle construction and
+     * shared across callers — {@code .get()} on it always observes the latest FP value.
+     */
+    public Property<CompressionAlgorithm> getCompressionAlgorithmProperty() {
+        return compressionAlgorithmProperty;
+    }
+
+    /**
+     * Live-updating {@link Property} handle for the zstd compression level. Resolved once at
+     * bundle construction.
+     */
+    public Property<Integer> getZstdCompressionLevelProperty() {
+        return zstdCompressionLevelProperty;
+    }
+
+    /**
+     * Parse an FP algorithm string (case-insensitive) into a {@link CompressionAlgorithm}. An
+     * unrecognized value falls back to {@link #DEFAULT_COMPRESSION_ALGORITHM} rather than
+     * propagating a {@code null} (which would NPE the compression switch at encode time) — a
+     * typo'd fast property degrades to the default instead of taking down writes.
+     */
+    private CompressionAlgorithm parseCompressionAlgorithm(String value) {
+        try {
+            return CompressionAlgorithm.valueOf(value.toUpperCase());
+        } catch (IllegalArgumentException | NullPointerException e) {
+            getLogger().warn("Unrecognized compression algorithm '%s'; falling back to %s", value, DEFAULT_COMPRESSION_ALGORITHM);
+            return DEFAULT_COMPRESSION_ALGORITHM;
+        }
+    }
+
     private static <T> Property<T> getProperty(String appName, PropertyRepository propertyRepository,
                                            Key key, Class<T> type, T defaultValue) {
         if (appName == null || appName.isEmpty()) {