[improve][broker] Optimize gzip compression for /metrics endpoint by sharing/caching compressed result

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/stats/prometheus/PrometheusMetricsServlet.java

@@ -39,6 +39,7 @@
 import org.slf4j.LoggerFactory;
 
 public class PrometheusMetricsServlet extends HttpServlet {
+    public static final String DEFAULT_METRICS_PATH = "/metrics";
     private static final long serialVersionUID = 1L;
     static final int HTTP_STATUS_OK_200 = 200;
     static final int HTTP_STATUS_INTERNAL_SERVER_ERROR_500 = 500;

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/GzipHandlerUtil.java

@@ -19,8 +19,10 @@
 package org.apache.pulsar.broker.web;
 
 import java.util.List;
+import org.eclipse.jetty.http.pathmap.PathSpecSet;
 import org.eclipse.jetty.server.Handler;
 import org.eclipse.jetty.server.handler.gzip.GzipHandler;
+import org.eclipse.jetty.util.IncludeExclude;
 
 public class GzipHandlerUtil {
     public static Handler wrapWithGzipHandler(Handler innerHandler, List<String> gzipCompressionExcludedPaths) {
@@ -45,4 +47,23 @@ public static boolean isGzipCompressionCompletelyDisabled(List<String> gzipCompr
                 && (gzipCompressionExcludedPaths.get(0).equals("^.*")
                 || gzipCompressionExcludedPaths.get(0).equals("^.*$"));
     }
+
+    /**
+     * Check if GZIP compression is enabled for the given endpoint.
+     * @param gzipCompressionExcludedPaths list of paths that should not be compressed
+     * @param endpoint the endpoint to check
+     * @return true if GZIP compression is enabled for the endpoint, false otherwise
+     */
+    public static boolean isGzipCompressionEnabledForEndpoint(List<String> gzipCompressionExcludedPaths,
+                                                              String endpoint) {
+        if (gzipCompressionExcludedPaths == null || gzipCompressionExcludedPaths.isEmpty()) {
+            return true;
+        }
+        if (isGzipCompressionCompletelyDisabled(gzipCompressionExcludedPaths)) {
+            return false;
+        }
+        IncludeExclude<String> paths = new IncludeExclude<>(PathSpecSet.class);
+        paths.exclude(gzipCompressionExcludedPaths.toArray(new String[0]));
+        return paths.test(endpoint);
+    }
 }

pulsar-broker-common/src/test/java/org/apache/pulsar/broker/web/GzipHandlerUtilTest.java

@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.web;
+
+import static org.testng.Assert.*;
+import java.util.Arrays;
+import org.testng.annotations.Test;
+
+public class GzipHandlerUtilTest {
+
+    @Test
+    public void testIsGzipCompressionEnabledForEndpoint() {
+        assertTrue(GzipHandlerUtil.isGzipCompressionEnabledForEndpoint(null, "/metrics"));
+        assertFalse(GzipHandlerUtil.isGzipCompressionEnabledForEndpoint(Arrays.asList("^.*"), "/metrics"));
+        assertFalse(GzipHandlerUtil.isGzipCompressionEnabledForEndpoint(Arrays.asList("^.*$"), "/metrics"));
+        assertFalse(GzipHandlerUtil.isGzipCompressionEnabledForEndpoint(Arrays.asList("/metrics"), "/metrics"));
+        assertTrue(GzipHandlerUtil.isGzipCompressionEnabledForEndpoint(Arrays.asList("/metrics"), "/metrics2"));
+        assertTrue(GzipHandlerUtil.isGzipCompressionEnabledForEndpoint(Arrays.asList("/admin", "/custom"), "/metrics"));
+    }
+}

pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java

@@ -110,6 +110,7 @@
 import org.apache.pulsar.broker.service.schema.SchemaStorageFactory;
 import org.apache.pulsar.broker.stats.MetricsGenerator;
 import org.apache.pulsar.broker.stats.PulsarBrokerOpenTelemetry;
+import org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsServlet;
 import org.apache.pulsar.broker.stats.prometheus.PrometheusRawMetricsProvider;
 import org.apache.pulsar.broker.stats.prometheus.PulsarPrometheusMetricsServlet;
 import org.apache.pulsar.broker.storage.ManagedLedgerStorage;
@@ -1040,7 +1041,7 @@ private void addWebServerHandlers(WebService webService,
                 true, attributeMap, true, Topics.class);
 
         // Add metrics servlet
-        webService.addServlet("/metrics",
+        webService.addServlet(PrometheusMetricsServlet.DEFAULT_METRICS_PATH,
                 new ServletHolder(metricsServlet),
                 config.isAuthenticateMetricsEndpoint(), attributeMap);
 

pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/PrometheusMetricsGenerator.java

@@ -30,6 +30,8 @@
 import java.io.OutputStream;
 import java.io.OutputStreamWriter;
 import java.io.Writer;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
 import java.nio.CharBuffer;
 import java.nio.charset.StandardCharsets;
 import java.time.Clock;
@@ -43,6 +45,8 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+import java.util.zip.CRC32;
+import java.util.zip.Deflater;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.bookkeeper.stats.NullStatsProvider;
 import org.apache.bookkeeper.stats.StatsProvider;
@@ -72,7 +76,7 @@ public class PrometheusMetricsGenerator implements AutoCloseable {
     private volatile boolean closed;
 
     public static class MetricsBuffer {
-        private final CompletableFuture<ByteBuf> bufferFuture;
+        private final CompletableFuture<ResponseBuffer> bufferFuture;
         private final long createTimeslot;
         private final AtomicInteger refCnt = new AtomicInteger(2);
 
@@ -81,7 +85,7 @@ public static class MetricsBuffer {
             createTimeslot = timeslot;
         }
 
-        public CompletableFuture<ByteBuf> getBufferFuture() {
+        public CompletableFuture<ResponseBuffer> getBufferFuture() {
             return bufferFuture;
         }
 
@@ -113,6 +117,151 @@ public void release() {
         }
     }
 
+    /**
+     * A wraps the response buffer and asynchronously provides a gzip compressed buffer when requested.
+     */
+    public static class ResponseBuffer {
+        private final ByteBuf uncompressedBuffer;
+        private boolean released = false;
+        private CompletableFuture<ByteBuf> compressedBuffer;
+
+        private ResponseBuffer(final ByteBuf uncompressedBuffer) {
+            this.uncompressedBuffer = uncompressedBuffer;
+        }
+
+        public ByteBuf getUncompressedBuffer() {
+            return uncompressedBuffer;
+        }
+
+        public synchronized CompletableFuture<ByteBuf> getCompressedBuffer(Executor executor) {
+            if (released) {
+                throw new IllegalStateException("Already released!");
+            }
+            if (compressedBuffer == null) {
+                compressedBuffer = new CompletableFuture<>();
+                ByteBuf retainedDuplicate = uncompressedBuffer.retainedDuplicate();
+                executor.execute(() -> {
+                    try {
+                        compressedBuffer.complete(compress(retainedDuplicate));
+                    } catch (Exception e) {
+                        compressedBuffer.completeExceptionally(e);
+                    } finally {
+                        retainedDuplicate.release();
+                    }
+                });
+            }
+            return compressedBuffer;
+        }
+
+        private ByteBuf compress(ByteBuf uncompressedBuffer) {
+            GzipByteBufferWriter gzipByteBufferWriter = new GzipByteBufferWriter(uncompressedBuffer.alloc(),
+                    uncompressedBuffer.readableBytes());
+            return gzipByteBufferWriter.compress(uncompressedBuffer);
+        }
+
+        public synchronized void release() {
+            released = true;
+            uncompressedBuffer.release();
+            if (compressedBuffer != null) {
+                compressedBuffer.whenComplete((byteBuf, throwable) -> {
+                    if (byteBuf != null) {
+                        byteBuf.release();
+                    }
+                });
+            }
+        }
+    }
+
+    /**
+     * Compress input nio buffers into gzip format with output in a Netty composite ByteBuf.
+     */
+    private static class GzipByteBufferWriter {
+        private static final byte[] GZIP_HEADER =
+                new byte[] {(byte) 0x1f, (byte) 0x8b, Deflater.DEFLATED, 0, 0, 0, 0, 0, 0, 0};
+        private final ByteBufAllocator bufAllocator;
+        private final Deflater deflater;
+        private final CRC32 crc;
+        private final int bufferSize;
+        private final CompositeByteBuf resultBuffer;
+        private ByteBuf backingCompressBuffer;
+        private ByteBuffer compressBuffer;
+
+        GzipByteBufferWriter(ByteBufAllocator bufAllocator, int readableBytes) {
+            deflater = new Deflater(Deflater.DEFAULT_COMPRESSION, true);
+            crc = new CRC32();
+            this.bufferSize = Math.max(Math.min(resolveChunkSize(bufAllocator), readableBytes), 8192);
+            this.bufAllocator = bufAllocator;
+            this.resultBuffer = bufAllocator.compositeDirectBuffer(readableBytes / bufferSize + 1);
+            allocateBuffer();
+        }
+
+        /**
+         * Compress the input Netty buffer and append it to the result buffer in gzip format.
+         * @param uncompressedBuffer
+         */
+        public ByteBuf compress(ByteBuf uncompressedBuffer) {
+            try {
+                ByteBuffer[] nioBuffers = uncompressedBuffer.nioBuffers();
+                for (int i = 0, nioBuffersLength = nioBuffers.length; i < nioBuffersLength; i++) {
+                    ByteBuffer nioBuffer = nioBuffers[i];
+                    compressAndAppend(nioBuffer, i == 0, i == nioBuffersLength - 1);
+                }
+                return resultBuffer;
+            } finally {
+                close();
+            }
+        }
+
+        private void compressAndAppend(ByteBuffer nioBuffer, boolean isFirst, boolean isLast) {
+            if (isFirst) {
+                // write gzip header
+                compressBuffer.put(GZIP_HEADER);
+            }
+            nioBuffer.mark();
+            crc.update(nioBuffer);
+            nioBuffer.reset();
+            deflater.setInput(nioBuffer);
+            if (isLast) {
+                deflater.finish();
+            }
+            while (!deflater.needsInput() && !deflater.finished()) {
+                int written = deflater.deflate(compressBuffer);
+                if (written == 0 && !compressBuffer.hasRemaining()) {
+                    backingCompressBuffer.setIndex(0, compressBuffer.position());
+                    resultBuffer.addComponent(true, backingCompressBuffer);
+                    allocateBuffer();
+                }
+            }
+            if (isLast) {
+                // write gzip footer, integer values are in little endian byte order
+                compressBuffer.order(ByteOrder.LITTLE_ENDIAN);
+                // write CRC32 checksum
+                compressBuffer.putInt((int) crc.getValue());
+                // write uncompressed size
+                compressBuffer.putInt(deflater.getTotalIn());
+                // append the last compressed buffer
+                backingCompressBuffer.setIndex(0, compressBuffer.position());
+                resultBuffer.addComponent(true, backingCompressBuffer);
+                backingCompressBuffer = null;
+                compressBuffer = null;
+            }
+        }
+
+        private void allocateBuffer() {
+            backingCompressBuffer = bufAllocator.directBuffer(bufferSize);
+            compressBuffer = backingCompressBuffer.nioBuffer(0, bufferSize);
+        }
+
+        private void close() {
+            if (deflater != null) {
+                deflater.end();
+            }
+            if (backingCompressBuffer != null) {
+                backingCompressBuffer.release();
+            }
+        }
+    }
+
     private final PulsarService pulsar;
     private final boolean includeTopicMetrics;
     private final boolean includeConsumerMetrics;
@@ -187,13 +336,7 @@ private ByteBuf allocateMultipartCompositeDirectBuffer() {
         // use composite buffer with pre-allocated buffers to ensure that the pooled allocator can be used
         // for allocating the buffers
         ByteBufAllocator byteBufAllocator = PulsarByteBufAllocator.DEFAULT;
-        int chunkSize;
-        if (byteBufAllocator instanceof PooledByteBufAllocator) {
-            PooledByteBufAllocator pooledByteBufAllocator = (PooledByteBufAllocator) byteBufAllocator;
-            chunkSize = Math.max(pooledByteBufAllocator.metric().chunkSize(), DEFAULT_INITIAL_BUFFER_SIZE);
-        } else {
-            chunkSize = DEFAULT_INITIAL_BUFFER_SIZE;
-        }
+        int chunkSize = resolveChunkSize(byteBufAllocator);
         CompositeByteBuf buf = byteBufAllocator.compositeDirectBuffer(
                 Math.max(MINIMUM_FOR_MAX_COMPONENTS, (initialBufferSize / chunkSize) + 1));
         int totalLen = 0;
@@ -204,6 +347,17 @@ private ByteBuf allocateMultipartCompositeDirectBuffer() {
         return buf;
     }
 
+    private static int resolveChunkSize(ByteBufAllocator byteBufAllocator) {
+        int chunkSize;
+        if (byteBufAllocator instanceof PooledByteBufAllocator) {
+            PooledByteBufAllocator pooledByteBufAllocator = (PooledByteBufAllocator) byteBufAllocator;
+            chunkSize = Math.max(pooledByteBufAllocator.metric().chunkSize(), DEFAULT_INITIAL_BUFFER_SIZE);
+        } else {
+            chunkSize = DEFAULT_INITIAL_BUFFER_SIZE;
+        }
+        return chunkSize;
+    }
+
     private static void generateBrokerBasicMetrics(PulsarService pulsar, SimpleTextOutputStream stream) {
         String clusterName = pulsar.getConfiguration().getClusterName();
         // generate managedLedgerCache metrics
@@ -335,10 +489,10 @@ public MetricsBuffer renderToBuffer(Executor executor, List<PrometheusRawMetrics
                     if (currentMetricsBuffer != null) {
                         currentMetricsBuffer.release();
                     }
-                    CompletableFuture<ByteBuf> bufferFuture = newMetricsBuffer.getBufferFuture();
+                    CompletableFuture<ResponseBuffer> bufferFuture = newMetricsBuffer.getBufferFuture();
                     executor.execute(() -> {
                         try {
-                            bufferFuture.complete(generate0(metricsProviders));
+                            bufferFuture.complete(new ResponseBuffer(generate0(metricsProviders)));
                         } catch (Exception e) {
                             bufferFuture.completeExceptionally(e);
                         } finally {