feat: add health check endpoint

ksql-rest-app/src/main/java/io/confluent/ksql/rest/healthcheck/HealthcheckAgent.java

@@ -0,0 +1,106 @@
+/*
+ * Copyright 2019 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (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.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.ksql.rest.healthcheck;
+
+import com.google.common.collect.ImmutableList;
+import io.confluent.ksql.rest.client.RestResponse;
+import io.confluent.ksql.rest.entity.HealthcheckResponse;
+import io.confluent.ksql.rest.entity.HealthcheckResponseDetail;
+import io.confluent.ksql.rest.entity.KsqlEntityList;
+import io.confluent.ksql.rest.server.KsqlRestConfig;
+import io.confluent.ksql.services.ServiceContext;
+import io.confluent.rest.RestConfig;
+import java.net.URI;
+import java.net.URL;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import org.apache.kafka.common.config.ConfigException;
+
+public class HealthcheckAgent {
+
+  private static final List<Check> DEFAULT_CHECKS = ImmutableList.of(
+      new Check("metastore", "list streams; list tables; list queries;"),
+      new Check("kafka", "list topics extended;")
+  );
+
+  private final ServiceContext serviceContext;
+  private final URI serverEndpoint;
+
+  public HealthcheckAgent(final ServiceContext serviceContext, final KsqlRestConfig restConfig) {
+    this.serviceContext = Objects.requireNonNull(serviceContext, "serviceContext");
+    this.serverEndpoint = getServerAddress(restConfig);
+  }
+
+  public HealthcheckResponse checkHealth() {
+    final Map<String, HealthcheckResponseDetail> results = DEFAULT_CHECKS.stream()
+        .collect(Collectors.toMap(
+            Check::getName,
+            check -> new HealthcheckResponseDetail(isSuccessful(check.getKsqlStatement()))
+        ));
+    final boolean allHealthy = results.values().stream()
+        .map(HealthcheckResponseDetail::getIsHealthy)
+        .reduce(Boolean::logicalAnd)
+        .orElse(true);
+    return new HealthcheckResponse(allHealthy, results);
+  }
+
+  private boolean isSuccessful(final String ksqlStatement) {
+    final RestResponse<KsqlEntityList> response =
+        serviceContext.getKsqlClient().makeKsqlRequest(serverEndpoint, ksqlStatement);
+    return response.isSuccessful();
+  }
+
+  private static URI getServerAddress(final KsqlRestConfig restConfig) {
+    final List<String> listeners = restConfig.getList(RestConfig.LISTENERS_CONFIG);
+    final String address = listeners.stream()
+        .map(String::trim)
+        .findFirst()
+        .orElseThrow(() -> invalidAddressException(listeners, "value cannot be empty"));
+
+    try {
+      return new URL(address).toURI();
+    } catch (final Exception e) {
+      throw invalidAddressException(listeners, e.getMessage());
+    }
+  }
+
+  private static RuntimeException invalidAddressException(
+      final List<String> serverAddresses,
+      final String message
+  ) {
+    return new ConfigException(RestConfig.LISTENERS_CONFIG, serverAddresses, message);
+  }
+
+  private static class Check {
+    private final String name;
+    private final String ksqlStatement;
+
+    Check(final String name, final String ksqlStatement) {
+      this.name = Objects.requireNonNull(name, "name");
+      this.ksqlStatement = Objects.requireNonNull(ksqlStatement, "ksqlStatement");
+    }
+
+    String getName() {
+      return name;
+    }
+
+    String getKsqlStatement() {
+      return ksqlStatement;
+    }
+  }
+}

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/KsqlRestApplication.java

@@ -45,6 +45,7 @@
 import io.confluent.ksql.rest.server.computation.StatementExecutor;
 import io.confluent.ksql.rest.server.context.KsqlRestServiceContextBinder;
 import io.confluent.ksql.rest.server.filters.KsqlAuthorizationFilter;
+import io.confluent.ksql.rest.server.resources.HealthcheckResource;
 import io.confluent.ksql.rest.server.resources.KsqlConfigurable;
 import io.confluent.ksql.rest.server.resources.KsqlExceptionMapper;
 import io.confluent.ksql.rest.server.resources.KsqlResource;
@@ -200,6 +201,7 @@ public void setupResources(final Configurable<?> config, final KsqlRestConfig ap
     config.register(statusResource);
     config.register(ksqlResource);
     config.register(streamedQueryResource);
+    config.register(HealthcheckResource.create(serviceContext, this.config));
     config.register(new KsqlExceptionMapper());
     config.register(new ServerStateDynamicBinding(serverState));
   }

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/KsqlRestConfig.java

@@ -71,6 +71,12 @@ public class KsqlRestConfig extends RestConfig {
       "Whether or not to set KsqlUncaughtExceptionHandler as the UncaughtExceptionHandler "
           + "for all threads in the application (this can be overridden). Default is false.";
 
+  public static final String KSQL_HEALTHCHECK_INTERVAL_MS_CONFIG =
+      KSQL_CONFIG_PREFIX + "healthcheck.interval.ms";
+  private static final String KSQL_HEALTHCHECK_INTERVAL_MS_DOC =
+      "Minimum time between consecutive healthcheck evaluations. Healthcheck queries before "
+          + "the interval has elapsed will receive cached responses.";
+
   private static final ConfigDef CONFIG_DEF;
 
   static {
@@ -102,14 +108,20 @@ public class KsqlRestConfig extends RestConfig {
         KSQL_SERVER_PRECONDITIONS,
         Type.LIST,
         "",
-       Importance.LOW,
-       KSQL_SERVER_PRECONDITIONS_DOC
+        Importance.LOW,
+        KSQL_SERVER_PRECONDITIONS_DOC
     ).define(
         KSQL_SERVER_ENABLE_UNCAUGHT_EXCEPTION_HANDLER,
         ConfigDef.Type.BOOLEAN,
         false,
         Importance.LOW,
         KSQL_SERVER_UNCAUGHT_EXCEPTION_HANDLER_DOC
+    ).define(
+        KSQL_HEALTHCHECK_INTERVAL_MS_CONFIG,
+        Type.LONG,
+        5000L,
+        Importance.LOW,
+        KSQL_HEALTHCHECK_INTERVAL_MS_DOC
     );
   }
 

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/filters/KsqlAuthorizationFilter.java

@@ -16,6 +16,7 @@
 package io.confluent.ksql.rest.server.filters;
 
 import io.confluent.ksql.rest.Errors;
+import io.confluent.ksql.rest.server.resources.HealthcheckResource;
 import io.confluent.ksql.rest.server.resources.ServerMetadataResource;
 import io.confluent.ksql.security.KsqlAuthorizationProvider;
 import java.lang.reflect.Method;
@@ -39,8 +40,10 @@
 public class KsqlAuthorizationFilter implements ContainerRequestFilter  {
   private static final Logger log = LoggerFactory.getLogger(KsqlAuthorizationFilter.class);
 
-  private static final Set<String> PATHS_WITHOUT_AUTHORIZATION =
-      getPathsFrom(ServerMetadataResource.class);
+  private static final Set<String> PATHS_WITHOUT_AUTHORIZATION = getPathsFrom(
+      ServerMetadataResource.class,
+      HealthcheckResource.class
+  );
 
   private final KsqlAuthorizationProvider authorizationProvider;
 
@@ -75,17 +78,20 @@ private boolean requiresAuthorization(final String path) {
     return !PATHS_WITHOUT_AUTHORIZATION.contains(path);
   }
 
-  private static Set<String> getPathsFrom(final Class<?> resourceClass) {
+  private static Set<String> getPathsFrom(final Class<?> ...resourceClass) {
+
     final Set<String> paths = new HashSet<>();
-    final String mainPath = StringUtils.stripEnd(
-        resourceClass.getAnnotation(Path.class).value(), "/"
-    );
+    for (final Class<?> clazz : resourceClass) {
+      final String mainPath = StringUtils.stripEnd(
+          clazz.getAnnotation(Path.class).value(), "/"
+      );
 
-    paths.add(mainPath);
-    for (Method m : resourceClass.getMethods()) {
-      if (m.isAnnotationPresent(Path.class)) {
-        paths.add(mainPath + "/"
-            + StringUtils.strip(m.getAnnotation(Path.class).value(), "/"));
+      paths.add(mainPath);
+      for (Method m : clazz.getMethods()) {
+        if (m.isAnnotationPresent(Path.class)) {
+          paths.add(mainPath + "/"
+              + StringUtils.strip(m.getAnnotation(Path.class).value(), "/"));
+        }
       }
     }
 

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/resources/HealthcheckResource.java

@@ -0,0 +1,109 @@
+/*
+ * Copyright 2019 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (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.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.ksql.rest.server.resources;
+
+import com.google.common.annotations.VisibleForTesting;
+import io.confluent.ksql.rest.entity.HealthcheckResponse;
+import io.confluent.ksql.rest.entity.Versions;
+import io.confluent.ksql.rest.healthcheck.HealthcheckAgent;
+import io.confluent.ksql.rest.server.KsqlRestConfig;
+import io.confluent.ksql.services.ServiceContext;
+import java.time.Duration;
+import java.util.Objects;
+import java.util.function.Supplier;
+import javax.ws.rs.GET;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+
+@Path("/healthcheck")
+@Produces({Versions.KSQL_V1_JSON, MediaType.APPLICATION_JSON})
+public class HealthcheckResource {
+  private final HealthcheckAgent healthcheckAgent;
+  private final Duration healthcheckInterval;
+  private final Supplier<Long> currentTimeSupplier;
+  private final ResponseCache responseCache;
+
+  @VisibleForTesting
+  HealthcheckResource(
+      final HealthcheckAgent healthcheckAgent,
+      final Duration healthcheckInterval,
+      final Supplier<Long> currentTimeSupplier
+  ) {
+    this.healthcheckAgent = Objects.requireNonNull(healthcheckAgent, "healthcheckAgent");
+    this.healthcheckInterval = Objects.requireNonNull(healthcheckInterval, "healthcheckInterval");
+    this.currentTimeSupplier = Objects.requireNonNull(currentTimeSupplier, "currentTimeSupplier");
+    this.responseCache = new ResponseCache(currentTimeSupplier);
+  }
+
+  @GET
+  public Response checkHealth() {
+    return Response.ok(getResponse()).build();
+  }
+
+  private HealthcheckResponse getResponse() {
+    if (responseCache.isEmpty() || timeSinceLastResponse().compareTo(healthcheckInterval) > 0) {
+      final HealthcheckResponse response = healthcheckAgent.checkHealth();
+      responseCache.cache(response);
+      return response;
+    } else {
+      return responseCache.lastResponse();
+    }
+  }
+
+  private Duration timeSinceLastResponse() {
+    return Duration.ofMillis(currentTimeSupplier.get() - responseCache.lastTimestamp());
+  }
+
+  public static HealthcheckResource create(
+      final ServiceContext serviceContext,
+      final KsqlRestConfig restConfig
+  ) {
+    return new HealthcheckResource(
+        new HealthcheckAgent(serviceContext, restConfig),
+        Duration.ofMillis(restConfig.getLong(KsqlRestConfig.KSQL_HEALTHCHECK_INTERVAL_MS_CONFIG)),
+        System::currentTimeMillis
+    );
+  }
+
+  private static class ResponseCache {
+    private final Supplier<Long> currentTimeSupplier;
+    private HealthcheckResponse response;
+    private long timestamp;
+
+    ResponseCache(final Supplier<Long> currentTimeSupplier) {
+      this.currentTimeSupplier = Objects.requireNonNull(currentTimeSupplier, "currentTimeSupplier");
+    }
+
+    void cache(final HealthcheckResponse response) {
+      this.response = response;
+      this.timestamp = currentTimeSupplier.get();
+    }
+
+    boolean isEmpty() {
+      return response == null;
+    }
+
+    HealthcheckResponse lastResponse() {
+      return response;
+    }
+
+    long lastTimestamp() {
+      return timestamp;
+    }
+  }
+}