[server] Add new connection setup latency metric (linkedin#2554)

Track the time from channel initialization (including SSL factory
creation) to SSL handshake completion as a Tehuti percentile sensor
with P50/P95/P99, Avg, and Max. This enables measuring per-connection
overhead during connection storms.

Author: huangminchn

services/venice-server/src/main/java/com/linkedin/venice/listener/HttpChannelInitializer.java

@@ -201,6 +201,7 @@ interface ChannelPipelineConsumer {
   @Override
   public void initChannel(SocketChannel ch) {
     if (sslFactory.isPresent()) {
+      ch.attr(ServerConnectionStatsHandler.CHANNEL_INIT_START_TS).set(System.nanoTime());
       SslInitializer sslInitializer = new SslInitializer(SslUtils.toAlpiniSSLFactory(sslFactory.get()), false);
       if (sslHandshakeExecutor != null) {
         sslInitializer

services/venice-server/src/main/java/com/linkedin/venice/listener/ServerConnectionStatsHandler.java

@@ -3,10 +3,12 @@
 import com.linkedin.venice.authorization.IdentityParser;
 import com.linkedin.venice.stats.ServerConnectionStats;
 import com.linkedin.venice.utils.DaemonThreadFactory;
+import com.linkedin.venice.utils.LatencyUtils;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
 import io.netty.handler.ssl.SslHandler;
+import io.netty.handler.ssl.SslHandshakeCompletionEvent;
 import io.netty.util.Attribute;
 import io.netty.util.AttributeKey;
 import java.security.cert.Certificate;
@@ -27,6 +29,7 @@
 public class ServerConnectionStatsHandler extends ChannelInboundHandlerAdapter {
   private static final Logger LOGGER = LogManager.getLogger(ServerConnectionStatsHandler.class);
   public static final AttributeKey<Boolean> CHANNEL_ACTIVATED = AttributeKey.valueOf("channelActivated");
+  public static final AttributeKey<Long> CHANNEL_INIT_START_TS = AttributeKey.valueOf("channelInitStartTs");
   private final IdentityParser identityParser;
   private final ServerConnectionStats serverConnectionStats;
   private final String routerPrincipalName;
@@ -97,6 +100,17 @@ public void channelActive(ChannelHandlerContext ctx) throws Exception {
     super.channelActive(ctx);
   }
 
+  @Override
+  public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
+    if (evt instanceof SslHandshakeCompletionEvent && ((SslHandshakeCompletionEvent) evt).isSuccess()) {
+      Long initStartTs = ctx.channel().attr(CHANNEL_INIT_START_TS).getAndSet(null);
+      if (initStartTs != null) {
+        serverConnectionStats.recordNewConnectionSetupLatency(LatencyUtils.getElapsedTimeFromNSToMS(initStartTs));
+      }
+    }
+    super.userEventTriggered(ctx, evt);
+  }
+
   @Override
   public void channelInactive(ChannelHandlerContext ctx) throws Exception {
     Attribute<Boolean> activated = ctx.channel().attr(CHANNEL_ACTIVATED);

services/venice-server/src/main/java/com/linkedin/venice/stats/ServerConnectionStats.java

@@ -12,10 +12,12 @@ public class ServerConnectionStats extends AbstractVeniceStats {
   public static final String ROUTER_CONNECTION_COUNT_GAUGE = "router_connection_count";
   public static final String CLIENT_CONNECTION_REQUEST = "client_connection_request";
   public static final String CLIENT_CONNECTION_COUNT_GAUGE = "client_connection_count";
+  public static final String NEW_CONNECTION_SETUP_LATENCY = "new_connection_setup_latency";
 
   private final Sensor routerConnectionRequestSensor;
   private final Sensor clientConnectionRequestSensor;
   private final Sensor connectionRequestSensor;
+  private final Sensor newConnectionSetupLatencySensor;
 
   private final AtomicLong routerConnectionCount = new AtomicLong();
   private final AtomicLong clientConnectionCount = new AtomicLong();
@@ -29,6 +31,9 @@ public ServerConnectionStats(MetricsRepository metricsRepository, String name) {
         new AsyncGauge((ignored, ignored2) -> clientConnectionCount.get(), CLIENT_CONNECTION_COUNT_GAUGE));
     clientConnectionRequestSensor = registerSensorIfAbsent(CLIENT_CONNECTION_REQUEST, new OccurrenceRate());
     connectionRequestSensor = registerSensorIfAbsent("connection_request", new OccurrenceRate());
+    newConnectionSetupLatencySensor = registerSensorIfAbsent(
+        NEW_CONNECTION_SETUP_LATENCY,
+        TehutiUtils.getPercentileStatWithAvgAndMax(getName(), NEW_CONNECTION_SETUP_LATENCY));
   }
 
   public void incrementRouterConnectionCount() {
@@ -52,4 +57,11 @@ public void decrementClientConnectionCount() {
   public void newConnectionRequest() {
     connectionRequestSensor.record();
   }
+
+  /**
+   * Record the latency from the start of channel initialization to SSL handshake completion.
+   */
+  public void recordNewConnectionSetupLatency(double latencyMs) {
+    newConnectionSetupLatencySensor.record(latencyMs);
+  }
 }

services/venice-server/src/test/java/com/linkedin/venice/listener/HttpChannelInitializerTest.java

@@ -17,6 +17,7 @@
 import io.grpc.ServerInterceptor;
 import io.netty.channel.ChannelPipeline;
 import io.netty.channel.socket.SocketChannel;
+import io.netty.util.Attribute;
 import io.tehuti.metrics.MetricsRepository;
 import java.util.List;
 import java.util.Optional;
@@ -102,6 +103,7 @@ public void testInitChannelWithSSLExecutor() {
     ChannelPipeline channelPipeline = mock(ChannelPipeline.class);
     SocketChannel ch = mock(SocketChannel.class);
     doReturn(channelPipeline).when(ch).pipeline();
+    doReturn(mock(Attribute.class)).when(ch).attr(any());
     doReturn(channelPipeline).when(channelPipeline).addLast(any());
     HttpChannelInitializer initializer = new HttpChannelInitializer(
         storeMetadataRepository,

services/venice-server/src/test/java/com/linkedin/venice/listener/ServerConnectionStatsHandlerTest.java

@@ -1,5 +1,6 @@
 package com.linkedin.venice.listener;
 
+import static org.mockito.ArgumentMatchers.anyDouble;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.mock;
@@ -8,18 +9,21 @@
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertTrue;
 
 import com.linkedin.venice.stats.ServerConnectionStats;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelPipeline;
 import io.netty.handler.ssl.SslHandler;
+import io.netty.handler.ssl.SslHandshakeCompletionEvent;
 import io.netty.util.Attribute;
 import java.security.cert.Certificate;
 import java.security.cert.X509Certificate;
 import javax.net.ssl.SSLEngine;
 import javax.net.ssl.SSLSession;
 import javax.security.auth.x500.X500Principal;
+import org.mockito.ArgumentCaptor;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
 
@@ -97,4 +101,56 @@ public void testChannelRegisteredUnregisteredWithSslHandler() throws Exception {
     verify(serverConnectionStats, times(1)).decrementClientConnectionCount();
     verify(serverConnectionStats, times(2)).newConnectionRequest();
   }
+
+  @Test
+  public void testConnectionSetupLatencyRecordedOnHandshakeSuccess() throws Exception {
+    ServerConnectionStats serverConnectionStats = mock(ServerConnectionStats.class);
+    ServerConnectionStatsHandler handler =
+        new ServerConnectionStatsHandler(null, serverConnectionStats, "venice-router");
+
+    // Simulate initChannel setting the start timestamp
+    Attribute<Long> initStartTsAttr = mock(Attribute.class);
+    long startTs = System.nanoTime();
+    when(channel.attr(ServerConnectionStatsHandler.CHANNEL_INIT_START_TS)).thenReturn(initStartTsAttr);
+    when(initStartTsAttr.getAndSet(null)).thenReturn(startTs);
+
+    // Fire a successful handshake event
+    handler.userEventTriggered(context, SslHandshakeCompletionEvent.SUCCESS);
+
+    // Verify latency was recorded with a non-negative value
+    ArgumentCaptor<Double> latencyCaptor = ArgumentCaptor.forClass(Double.class);
+    verify(serverConnectionStats, times(1)).recordNewConnectionSetupLatency(latencyCaptor.capture());
+    assertTrue(latencyCaptor.getValue() >= 0, "Latency should be non-negative");
+  }
+
+  @Test
+  public void testConnectionSetupLatencyNotRecordedOnHandshakeFailure() throws Exception {
+    ServerConnectionStats serverConnectionStats = mock(ServerConnectionStats.class);
+    ServerConnectionStatsHandler handler =
+        new ServerConnectionStatsHandler(null, serverConnectionStats, "venice-router");
+
+    // Fire a failed handshake event
+    SslHandshakeCompletionEvent failedEvent =
+        new SslHandshakeCompletionEvent(new javax.net.ssl.SSLHandshakeException("test failure"));
+    handler.userEventTriggered(context, failedEvent);
+
+    // Verify latency was NOT recorded
+    verify(serverConnectionStats, never()).recordNewConnectionSetupLatency(anyDouble());
+  }
+
+  @Test
+  public void testConnectionSetupLatencyNotRecordedWhenTimestampMissing() throws Exception {
+    ServerConnectionStats serverConnectionStats = mock(ServerConnectionStats.class);
+    ServerConnectionStatsHandler handler =
+        new ServerConnectionStatsHandler(null, serverConnectionStats, "venice-router");
+
+    // No timestamp was set (e.g., non-SSL path)
+    Attribute<Long> initStartTsAttr = mock(Attribute.class);
+    when(channel.attr(ServerConnectionStatsHandler.CHANNEL_INIT_START_TS)).thenReturn(initStartTsAttr);
+    when(initStartTsAttr.getAndSet(null)).thenReturn(null);
+
+    handler.userEventTriggered(context, SslHandshakeCompletionEvent.SUCCESS);
+
+    verify(serverConnectionStats, never()).recordNewConnectionSetupLatency(anyDouble());
+  }
 }