[Multi-Tenancy Test]WireConnectionSharingInBenchmark

sdk/cosmos/azure-cosmos-benchmark/src/main/java/com/azure/cosmos/benchmark/AsyncBenchmark.java

@@ -92,7 +92,8 @@ abstract class AsyncBenchmark<T> {
                 .preferredRegions(cfg.getPreferredRegionsList())
                 .consistencyLevel(cfg.getConsistencyLevel())
                 .userAgentSuffix(cfg.getApplicationName())
-                .contentResponseOnWriteEnabled(cfg.isContentResponseOnWriteEnabled());
+                .contentResponseOnWriteEnabled(cfg.isContentResponseOnWriteEnabled())
+                .connectionSharingAcrossClientsEnabled(cfg.isConnectionSharingAcrossClientsEnabled());
 
         clientBuilderAccessor
             .setRegionScopedSessionCapturingEnabled(benchmarkSpecificClientBuilder, cfg.isRegionScopedSessionContainerEnabled());
@@ -252,7 +253,14 @@ uuid, new PartitionKey(partitionKey), PojoizedJson.class)
             }
         }
 
-        docsToRead = Flux.merge(Flux.fromIterable(createDocumentObservables), 100).collectList().block();
+        if (createDocumentObservables.isEmpty()) {
+            docsToRead = new ArrayList<>();
+        } else {
+            int prePopConcurrency = Math.max(1, Math.min(cfg.getConcurrency(), 100));
+            docsToRead = Flux.merge(Flux.fromIterable(createDocumentObservables), prePopConcurrency)
+                .collectList()
+                .block();
+        }
         logger.info("Finished pre-populating {} documents", cfg.getNumberOfPreCreatedDocuments());
 
         init();

sdk/cosmos/azure-cosmos-benchmark/src/main/java/com/azure/cosmos/benchmark/BenchmarkConfig.java

@@ -34,6 +34,7 @@ public class BenchmarkConfig {
     private boolean suppressCleanup = false;
     private boolean gcBetweenCycles = true;
     private boolean enableJvmStats = false;
+    private boolean enableNettyHttpMetrics = false;
 
     // -- Reporting --
     private String reportingDirectory;
@@ -81,6 +82,7 @@ public static BenchmarkConfig fromConfiguration(Configuration cfg) throws IOExce
 
         config.gcBetweenCycles = cfg.isGcBetweenCycles();
         config.enableJvmStats = cfg.isEnableJvmStats();
+        config.enableNettyHttpMetrics = cfg.isEnableNettyHttpMetrics();
 
         // Reporting
         config.reportingDirectory = cfg.getReportingDirectory() != null
@@ -127,6 +129,7 @@ public static BenchmarkConfig fromConfiguration(Configuration cfg) throws IOExce
     public boolean isSuppressCleanup() { return suppressCleanup; }
     public boolean isGcBetweenCycles() { return gcBetweenCycles; }
     public boolean isEnableJvmStats() { return enableJvmStats; }
+    public boolean isEnableNettyHttpMetrics() { return enableNettyHttpMetrics; }
 
     public String getReportingDirectory() { return reportingDirectory; }
     public int getPrintingInterval() { return printingInterval; }

sdk/cosmos/azure-cosmos-benchmark/src/main/java/com/azure/cosmos/benchmark/BenchmarkOrchestrator.java

@@ -13,6 +13,9 @@
 import com.codahale.metrics.jvm.GarbageCollectorMetricSet;
 import com.codahale.metrics.jvm.MemoryUsageGaugeSet;
 import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Metrics;
+
+import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
 import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -123,24 +126,46 @@ public void run(BenchmarkConfig config) throws Exception {
                 config.getResultUploadDatabase(), config.getResultUploadContainer());
         }
 
+        // Netty HTTP connection pool metrics reporter (only when enabled)
+        NettyHttpMetricsReporter nettyMetricsReporter = null;
+        SimpleMeterRegistry nettyHttpMeterRegistry = null;
+        // Add a SimpleMeterRegistry to globalRegistry when netty metrics are enabled,
+        if (config.isEnableNettyHttpMetrics() && config.getReportingDirectory() != null) {
+            nettyHttpMeterRegistry = new SimpleMeterRegistry();
+            Metrics.addRegistry(nettyHttpMeterRegistry);
+            logger.info("SimpleMeterRegistry added to globalRegistry for Reactor Netty pool gauge backing");
+
+            Path nettyMetricsDir = Paths.get(config.getReportingDirectory());
+            nettyMetricsReporter = new NettyHttpMetricsReporter(nettyHttpMeterRegistry, nettyMetricsDir);
+            nettyMetricsReporter.start(config.getPrintingInterval(), TimeUnit.SECONDS);
+        }
+
         reporter.report();
         logger.info("[LIFECYCLE] PRE_CREATE timestamp={}", Instant.now());
         logger.info("BenchmarkConfig: {}", config);
 
         // ======== Lifecycle loop ========
-        runLifecycleLoop(config, registry, reporter);
-
-        // Cleanup reporters
-        reporter.report();
-        reporter.stop();
-        if (resultReporter != null) {
-            resultReporter.report();
-            resultReporter.stop();
-        }
-        if (resultUploaderClient != null) {
-            resultUploaderClient.close();
+        try {
+            runLifecycleLoop(config, registry, reporter);
+        } finally {
+            // Cleanup reporters
+            reporter.report();
+            reporter.stop();
+            if (resultReporter != null) {
+                resultReporter.report();
+                resultReporter.stop();
+            }
+            if (resultUploaderClient != null) {
+                resultUploaderClient.close();
+            }
+            if (nettyMetricsReporter != null) {
+                nettyMetricsReporter.stop();
+            }
+            if (nettyHttpMeterRegistry != null) {
+                Metrics.removeRegistry(nettyHttpMeterRegistry);
+            }
+            clearGlobalSystemProperties();
         }
-        clearGlobalSystemProperties();
     }
 
     // ======== Lifecycle loop (create -> run -> close -> settle x N) ========
@@ -348,6 +373,7 @@ private void clearGlobalSystemProperties() {
         System.clearProperty("COSMOS.E2E_TIMEOUT_ERROR_HIT_THRESHOLD_FOR_PPAF");
         System.clearProperty("COSMOS.E2E_TIMEOUT_ERROR_HIT_TIME_WINDOW_IN_SECONDS_FOR_PPAF");
         System.clearProperty("COSMOS.MIN_CONNECTION_POOL_SIZE_PER_ENDPOINT");
+        System.clearProperty("COSMOS.NETTY_HTTP_CLIENT_METRICS_ENABLED");
     }
 
     private void setGlobalSystemProperties(BenchmarkConfig config) {
@@ -373,6 +399,11 @@ private void setGlobalSystemProperties(BenchmarkConfig config) {
                 String.valueOf(config.getMinConnectionPoolSizePerEndpoint()));
         }
 
+        if (config.isEnableNettyHttpMetrics()) {
+            System.setProperty("COSMOS.NETTY_HTTP_CLIENT_METRICS_ENABLED", "true");
+            logger.info("Reactor Netty HTTP connection pool metrics enabled");
+        }
+
         logger.info("Global system properties set (circuit breaker: {}, PPAF: {}, minConnPoolSize: {})",
             config.isPartitionLevelCircuitBreakerEnabled(),
             config.isPerPartitionAutomaticFailoverRequired(),

sdk/cosmos/azure-cosmos-benchmark/src/main/java/com/azure/cosmos/benchmark/Configuration.java

@@ -73,6 +73,9 @@ public class Configuration {
     @Parameter(names = "-maxConnectionPoolSize", description = "Max Connection Pool Size")
     private Integer maxConnectionPoolSize = 1000;
 
+    @Parameter(names = "-connectionSharingAcrossClientsEnabled", description = "Enable connection sharing across CosmosClient instances (Gateway mode). Reduces connection count for multi-tenant scenarios.")
+    private boolean connectionSharingAcrossClientsEnabled = false;
+
     @Parameter(names = "-diagnosticsThresholdDuration", description = "Latency threshold for printing diagnostics", converter = DurationConverter.class)
     private Duration diagnosticsThresholdDuration = Duration.ofSeconds(60);
 
@@ -92,6 +95,9 @@ public class Configuration {
     @Parameter(names = "-enableJvmStats", description = "Enables JVM Stats")
     private boolean enableJvmStats;
 
+    @Parameter(names = "-enableNettyHttpMetrics", description = "Enables Reactor Netty HTTP client metrics (connection pool gauges via COSMOS.NETTY_HTTP_CLIENT_METRICS_ENABLED)")
+    private boolean enableNettyHttpMetrics;
+
     @Parameter(names = "-throughput", description = "provisioned throughput for test container")
     private int throughput = 100000;
 
@@ -488,6 +494,10 @@ public Integer getMaxConnectionPoolSize() {
         return maxConnectionPoolSize;
     }
 
+    public boolean isConnectionSharingAcrossClientsEnabled() {
+        return connectionSharingAcrossClientsEnabled;
+    }
+
     public ConnectionMode getConnectionMode() {
         return connectionMode;
     }
@@ -540,6 +550,10 @@ public boolean isEnableJvmStats() {
         return enableJvmStats;
     }
 
+    public boolean isEnableNettyHttpMetrics() {
+        return enableNettyHttpMetrics;
+    }
+
     public MeterRegistry getAzureMonitorMeterRegistry() {
         String instrumentationKey = System.getProperty("azure.cosmos.monitoring.azureMonitor.instrumentationKey",
             StringUtils.defaultString(Strings.emptyToNull(

sdk/cosmos/azure-cosmos-benchmark/src/main/java/com/azure/cosmos/benchmark/NettyHttpMetricsReporter.java

@@ -0,0 +1,153 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.cosmos.benchmark;
+
+import io.micrometer.core.instrument.Gauge;
+import io.micrometer.core.instrument.Meter;
+import io.micrometer.core.instrument.MeterRegistry;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.time.Instant;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Periodically samples Reactor Netty connection pool metrics from a Micrometer
+ * {@link MeterRegistry} and writes them to a CSV file.
+ *
+ * <p>Metrics captured (when {@code ConnectionProvider.metrics(true)} is enabled):</p>
+ * <ul>
+ *   <li>{@code reactor.netty.connection.provider.total.connections}</li>
+ *   <li>{@code reactor.netty.connection.provider.active.connections}</li>
+ *   <li>{@code reactor.netty.connection.provider.idle.connections}</li>
+ *   <li>{@code reactor.netty.connection.provider.pending.connections}</li>
+ *   <li>{@code reactor.netty.connection.provider.max.connections}</li>
+ *   <li>{@code reactor.netty.connection.provider.max.pending.connections}</li>
+ * </ul>
+ *
+ * <p>CSV columns: timestamp, metric, pool_id, pool_name, remote_address, value</p>
+ */
+public class NettyHttpMetricsReporter {
+
+    private static final Logger logger = LoggerFactory.getLogger(NettyHttpMetricsReporter.class);
+    private static final String METRIC_PREFIX = "reactor.netty.connection.provider";
+    private static final String CSV_HEADER = "timestamp,metric,pool_id,pool_name,remote_address,value";
+
+    private final MeterRegistry registry;
+    private final Path outputFile;
+    private final ScheduledExecutorService scheduler;
+    private BufferedWriter writer;
+
+    /**
+     * @param registry   the Micrometer registry to query (typically {@code Metrics.globalRegistry})
+     * @param outputDir  directory to write the CSV file into
+     */
+    public NettyHttpMetricsReporter(MeterRegistry registry, Path outputDir) {
+        this.registry = registry;
+        this.outputFile = outputDir.resolve("netty-pool-metrics.csv");
+        this.scheduler = Executors.newSingleThreadScheduledExecutor(r -> {
+            Thread t = new Thread(r, "netty-metrics-reporter");
+            t.setDaemon(true);
+            return t;
+        });
+    }
+
+    /**
+     * Start periodic reporting.
+     *
+     * @param interval reporting interval
+     * @param unit     time unit
+     */
+    public void start(long interval, TimeUnit unit) {
+        try {
+            Files.createDirectories(outputFile.getParent());
+            writer = Files.newBufferedWriter(outputFile,
+                StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING);
+            writer.write(CSV_HEADER);
+            writer.newLine();
+            writer.flush();
+        } catch (IOException e) {
+            logger.error("Failed to create netty pool metrics CSV: {}", outputFile, e);
+            return;
+        }
+
+        scheduler.scheduleAtFixedRate(this::report, interval, interval, unit);
+        logger.info("NettyHttpMetricsReporter started -> {} (every {}s)", outputFile, unit.toSeconds(interval));
+    }
+
+    /**
+     * Write a single snapshot of all pool metrics to CSV.
+     */
+    public void report() {
+        if (writer == null) return;
+
+        String timestamp = Instant.now().toString();
+        int count = 0;
+
+        try {
+            for (Meter meter : registry.getMeters()) {
+                String name = meter.getId().getName();
+                if (!name.startsWith(METRIC_PREFIX)) continue;
+
+                // Only report gauge-type metrics (connections counts)
+                if (!(meter instanceof Gauge)) continue;
+
+                double value = ((Gauge) meter).value();
+                String poolId = meter.getId().getTag("id");
+                String poolName = meter.getId().getTag("name");
+                String remoteAddr = meter.getId().getTag("remote.address");
+
+                // Strip the common prefix for shorter metric names in CSV
+                String shortName = name.substring(METRIC_PREFIX.length() + 1);
+
+                writer.write(String.format("%s,%s,%s,%s,%s,%.0f",
+                    timestamp, shortName,
+                    poolId != null ? poolId : "",
+                    poolName != null ? poolName : "",
+                    remoteAddr != null ? remoteAddr : "",
+                    value));
+                writer.newLine();
+                count++;
+            }
+            writer.flush();
+        } catch (IOException e) {
+            logger.warn("Failed to write netty pool metrics", e);
+        }
+
+        if (count > 0) {
+            logger.debug("NettyHttpMetricsReporter: wrote {} pool metrics", count);
+        }
+    }
+
+    /**
+     * Stop the reporter and close the CSV file.
+     */
+    public void stop() {
+        scheduler.shutdown();
+        try {
+            scheduler.awaitTermination(5, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        }
+
+        // Final snapshot
+        report();
+
+        if (writer != null) {
+            try {
+                writer.close();
+                logger.info("NettyHttpMetricsReporter stopped. Output: {}", outputFile);
+            } catch (IOException e) {
+                logger.warn("Failed to close netty pool metrics CSV", e);
+            }
+        }
+    }
+}