[controller] Add a lightweight controller health endpoint

internal/venice-common/src/main/java/com/linkedin/venice/controllerapi/ControllerRoute.java

@@ -113,6 +113,7 @@
 public enum ControllerRoute implements VeniceDimensionInterface {
   // Fallback route
   UNKNOWN_ROUTE("", HttpMethod.GET, Collections.emptyList()),
+  HEALTH("/health", HttpMethod.GET, Collections.emptyList()),
   // Request a topic that writer should produce to
   REQUEST_TOPIC("/request_topic", HttpMethod.POST, Arrays.asList(NAME, PUSH_TYPE, PUSH_JOB_ID), PUSH_IN_SORTED_ORDER),
   // Do an empty push into a new version for this store

internal/venice-common/src/test/java/com/linkedin/venice/controllerapi/ControllerRouteDimensionTest.java

@@ -1,7 +1,9 @@
 package com.linkedin.venice.controllerapi;
 
+import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertTrue;
 
+import com.linkedin.venice.HttpMethod;
 import com.linkedin.venice.stats.dimensions.VeniceDimensionTestFixture;
 import com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions;
 import com.linkedin.venice.utils.CollectionUtils;
@@ -10,6 +12,16 @@
 
 
 public class ControllerRouteDimensionTest {
+  @Test
+  public void testHealthRouteRequiresNoParameters() {
+    assertEquals(ControllerRoute.HEALTH.getPath(), "/health");
+    assertEquals(ControllerRoute.HEALTH.getHttpMethod(), HttpMethod.GET);
+    assertTrue(ControllerRoute.HEALTH.getParams().isEmpty());
+    assertTrue(ControllerRoute.HEALTH.getOptionalParams().isEmpty());
+    assertEquals(ControllerRoute.valueOfPath("/health"), ControllerRoute.HEALTH);
+    assertEquals(ControllerRoute.valueOfPath("/healthcheck"), ControllerRoute.UNKNOWN_ROUTE);
+  }
+
   @Test
   public void testUpdateStoreVersionStorageModeRequiresCluster() {
     assertTrue(ControllerRoute.UPDATE_STORE_VERSION_STORAGE_MODE.getParams().contains(ControllerApiConstants.CLUSTER));
@@ -19,6 +31,7 @@ public void testUpdateStoreVersionStorageModeRequiresCluster() {
   public void testDimensionInterface() {
     Map<ControllerRoute, String> expectedValues = CollectionUtils.<ControllerRoute, String>mapBuilder()
         .put(ControllerRoute.UNKNOWN_ROUTE, "unknown_route")
+        .put(ControllerRoute.HEALTH, "health")
         .put(ControllerRoute.REQUEST_TOPIC, "request_topic")
         .put(ControllerRoute.EMPTY_PUSH, "empty_push")
         .put(ControllerRoute.END_OF_PUSH, "end_of_push")

internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/server/TestAdminSparkWithMocks.java

@@ -32,16 +32,19 @@
 import com.linkedin.venice.utils.SslUtils;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
 import org.apache.commons.io.IOUtils;
 import org.apache.http.HttpResponse;
 import org.apache.http.HttpStatus;
 import org.apache.http.NameValuePair;
 import org.apache.http.client.entity.UrlEncodedFormEntity;
+import org.apache.http.client.methods.HttpGet;
 import org.apache.http.client.methods.HttpPost;
 import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
 import org.apache.http.message.BasicNameValuePair;
+import org.apache.http.util.EntityUtils;
 import org.mockito.Mockito;
 import org.testng.Assert;
 import org.testng.annotations.BeforeMethod;
@@ -65,6 +68,29 @@ public void setUp() {
     requestHandler = new VeniceControllerRequestHandler(dependencies);
   }
 
+  @Test(timeOut = 30000)
+  public void testMockServerHealthTracksStartupAndShutdown() throws Exception {
+    AdminSparkServer server = Mockito
+        .spy(ServiceFactory.getMockAdminSparkServer(admin, "test_cluster", Collections.emptyList(), requestHandler));
+    try (CloseableHttpAsyncClient httpClient = HttpClientUtils.getMinimalHttpClient(1, 1, Optional.empty())) {
+      httpClient.start();
+      HttpGet healthRequest = new HttpGet("http://localhost:" + server.getPort() + ControllerRoute.HEALTH.getPath());
+      HttpResponse response = httpClient.execute(healthRequest, null).get();
+      Assert.assertEquals(response.getStatusLine().getStatusCode(), HttpStatus.SC_OK);
+      Assert.assertEquals(EntityUtils.toString(response.getEntity()), "OK");
+      Mockito.doAnswer(invocation -> {
+        Mockito.doCallRealMethod().when(server).stopInner();
+        HttpResponse stoppingResponse = httpClient.execute(healthRequest, null).get();
+        Assert.assertEquals(stoppingResponse.getStatusLine().getStatusCode(), HttpStatus.SC_SERVICE_UNAVAILABLE);
+        Assert.assertEquals(EntityUtils.toString(stoppingResponse.getEntity()), "NOT_READY");
+        return invocation.callRealMethod();
+      }).when(server).stopInner();
+      server.stop();
+    } finally {
+      server.stop();
+    }
+  }
+
   @Test
   public void testGetRealTimeTopicForStreamPushJobUsesAdmin() throws Exception {
     // setup server with mock admin, note returns topic "store_rt"

internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/integration/utils/ServiceFactory.java

@@ -19,6 +19,7 @@
 import com.linkedin.venice.exceptions.VeniceException;
 import com.linkedin.venice.pubsub.PubSubClientsFactory;
 import com.linkedin.venice.pubsub.PubSubTopicRepository;
+import com.linkedin.venice.serialization.avro.AvroProtocolDefinition;
 import com.linkedin.venice.utils.ExceptionUtils;
 import com.linkedin.venice.utils.PropertyBuilder;
 import com.linkedin.venice.utils.ReflectUtils;
@@ -41,6 +42,7 @@
 import java.util.Properties;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -158,6 +160,8 @@ public static AdminSparkServer getMockAdminSparkServer(
     return getService("MockAdminSparkServer", (serviceName) -> {
       Set<String> clusters = new HashSet<>();
       clusters.add(cluster);
+      // The mock listener has no controller owner; its health follows its own lifecycle.
+      AtomicBoolean apiReady = new AtomicBoolean(false);
       AdminSparkServer server = new AdminSparkServer(
           TestUtils.getFreePort(),
           admin,
@@ -171,7 +175,21 @@ public static AdminSparkServer getMockAdminSparkServer(
           null,
           false,
           new PubSubTopicRepository(),
-          requestHandler);
+          requestHandler,
+          AvroProtocolDefinition.PUSH_JOB_DETAILS.getSerializer(),
+          apiReady::get) {
+        @Override
+        public synchronized void start() {
+          super.start();
+          apiReady.set(isRunning());
+        }
+
+        @Override
+        public synchronized void stop() throws Exception {
+          apiReady.set(false);
+          super.stop();
+        }
+      };
       server.start();
       return server;
     });

services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceController.java

@@ -1,6 +1,7 @@
 package com.linkedin.venice.controller;
 
 import static com.linkedin.venice.ConfigKeys.ZOOKEEPER_ADDRESS;
+import static com.linkedin.venice.controller.ParentControllerRegionState.ACTIVE;
 
 import com.linkedin.d2.balancer.D2Client;
 import com.linkedin.venice.SSLConfig;
@@ -73,6 +74,7 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.atomic.AtomicReference;
 import org.apache.avro.Schema;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -115,6 +117,30 @@ public static List<Class<? extends ModuleMetricEntityInterface>> getMetricEntity
   public static final Collection<MetricEntity> CONTROLLER_SERVICE_METRIC_ENTITIES =
       ModuleMetricEntityInterface.getUniqueMetricEntities(getMetricEntityEnumClasses());
 
+  static final class ApiReadiness {
+    private enum Phase {
+      NOT_READY, READY, STOPPING
+    }
+
+    private final AtomicReference<Phase> phase = new AtomicReference<>(Phase.NOT_READY);
+
+    void markReady() {
+      // Restart creates a new owner; stopping must defeat any late startup completion.
+      phase.compareAndSet(Phase.NOT_READY, Phase.READY);
+    }
+
+    void markStopping() {
+      phase.set(Phase.STOPPING);
+    }
+
+    boolean isReady() {
+      return phase.get() == Phase.READY;
+    }
+  }
+
+  private final ApiReadiness apiReadiness = new ApiReadiness();
+  private final boolean apiRegionEligible;
+
   // services
   private final VeniceControllerService controllerService;
   private final AdminSparkServer adminServer;
@@ -205,6 +231,9 @@ public VeniceController(
 
   public VeniceController(VeniceControllerContext ctx) {
     this.multiClusterConfigs = new VeniceControllerMultiClusterConfig(ctx.getPropertiesList());
+    VeniceControllerClusterConfig commonConfig = multiClusterConfigs.getCommonConfig();
+    // ACTIVE/PASSIVE is a parent-region failover role, not this host's cluster leadership.
+    this.apiRegionEligible = !commonConfig.isParent() || commonConfig.getParentControllerRegionState() == ACTIVE;
     this.logContext = multiClusterConfigs.getLogContext();
     this.metricsRepository = ctx.getMetricsRepository();
     this.serviceDiscoveryAnnouncers = ctx.getServiceDiscoveryAnnouncers();
@@ -292,7 +321,18 @@ AdminSparkServer createAdminServer(boolean secure) {
         multiClusterConfigs.getCommonConfig().isDisableParentRequestTopicForStreamPushes(),
         pubSubTopicRepository,
         secure ? secureRequestHandler : unsecureRequestHandler,
-        controllerService.getPushJobDetailsSerializer());
+        controllerService.getPushJobDetailsSerializer(),
+        this::isApiReady);
+  }
+
+  /**
+   * Requires successful startup, no shutdown, running core/HTTP services, and an active region for parent controllers.
+   * HTTPS is required only when configured; eligible standbys do not need cluster leadership.
+   */
+  private boolean isApiReady() {
+    // Check lifecycle state last so shutdown during service checks makes this probe unready.
+    return apiRegionEligible && controllerService.isRunning() && adminServer.isRunning()
+        && (!sslEnabled || secureAdminServer.isRunning()) && apiReadiness.isReady();
   }
 
   private TopicCleanupService createTopicCleanupService() {
@@ -472,6 +512,7 @@ public void start() {
       adminSecureGrpcServer.start();
     }
     LOGGER.info("Controller is started.");
+    apiReadiness.markReady();
   }
 
   private void initializeSystemSchema(Admin admin) {
@@ -574,6 +615,7 @@ private static void createAndExecuteSchemaInitializer(
    * Causes venice controller and its associated services to stop executing.
    */
   public void stop() {
+    apiReadiness.markStopping();
     // unregister from service discovery first
     asyncRetryingServiceDiscoveryAnnouncer.unregister();
     // TODO: we may want a dependency structure so we ensure services are shutdown in the correct order.