feat: Adds logging for every request to ksqlDB (#6615)

* feat: Adds logging for every request that comes in

docs/operate-and-deploy/installation/server-config/config-reference.md

@@ -612,6 +612,19 @@ property has the value `KSQL_PROCESSING_LOG`.
 Toggles whether or not the processing log should include rows in log
 messages. By default, this property has the value `false`.
 
+### ksql.logging.server.skipped.response.codes
+
+A comma-separated list of HTTP response codes to skip during server
+request logging. This is useful for ignoring certain 4XX errors that you
+might not want to show up in the logs.
+
+### ksql.logging.server.rate.limited.request.paths
+
+A list of `path:rate_limit` pairs, to limit the rate of server request
+logging.  This is useful for requests that are coming in at a high rate,
+such as for pull queries. This setting enables seeing the logs when the request rate is low
+and dropping them when they go over the threshold.
+
 ksqlDB-Connect Settings
 -----------------------
 

ksqldb-common/src/main/java/io/confluent/ksql/configdef/ConfigValidators.java

@@ -17,10 +17,12 @@
 
 import static java.util.Objects.requireNonNull;
 
+import io.confluent.ksql.util.KsqlConfig;
 import java.net.URL;
 import java.util.Arrays;
 import java.util.EnumSet;
 import java.util.List;
+import java.util.Map;
 import java.util.function.Function;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
@@ -154,6 +156,41 @@ public static Validator oneOrMore() {
     };
   }
 
+  public static Validator intList() {
+    return (name, val) -> {
+      if (!(val instanceof List)) {
+        throw new ConfigException(name, val, "Must be a list");
+      }
+      @SuppressWarnings("unchecked")
+      final List<String> list = (List<String>) val;
+      list.forEach(intStr -> {
+        try {
+          Integer.parseInt(intStr);
+        } catch (NumberFormatException e) {
+          throw new ConfigException(name, intStr, "Not an integer");
+        }
+      });
+    };
+  }
+
+  public static Validator mapWithDoubleValue() {
+    return (name, val) -> {
+      if (!(val instanceof String)) {
+        throw new ConfigException(name, val, "Must be a string");
+      }
+
+      final String str = (String) val;
+      final Map<String, String> map = KsqlConfig.parseStringAsMap(name, str);
+      map.forEach((k, valueStr) -> {
+        try {
+          Double.parseDouble(valueStr);
+        } catch (NumberFormatException e) {
+          throw new ConfigException(name, valueStr, "Not a double");
+        }
+      });
+    };
+  }
+
   public static final class ValidCaseInsensitiveString implements Validator {
 
     private final List<String> validStrings;

ksqldb-common/src/main/java/io/confluent/ksql/util/KsqlConfig.java

@@ -1108,6 +1108,10 @@ public KsqlConfig overrideBreakingConfigsWithOriginalValues(final Map<String, ?>
 
   public Map<String, String> getStringAsMap(final String key) {
     final String value = getString(key).trim();
+    return parseStringAsMap(key, value);
+  }
+
+  public static Map<String, String> parseStringAsMap(final String key, final String value) {
     try {
       return value.equals("")
           ? Collections.emptyMap()

ksqldb-rest-app/src/main/java/io/confluent/ksql/api/server/LoggingHandler.java

@@ -0,0 +1,154 @@
+/*
+ * Copyright 2020 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.api.server;
+
+import static io.confluent.ksql.rest.server.KsqlRestConfig.KSQL_LOGGING_SERVER_RATE_LIMITED_REQUEST_PATHS_CONFIG;
+import static io.confluent.ksql.rest.server.KsqlRestConfig.KSQL_LOGGING_SERVER_SKIPPED_RESPONSE_CODES_CONFIG;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.util.concurrent.RateLimiter;
+import io.confluent.ksql.api.auth.ApiUser;
+import io.confluent.ksql.rest.server.KsqlRestConfig;
+import io.vertx.core.Handler;
+import io.vertx.core.http.HttpMethod;
+import io.vertx.core.http.HttpVersion;
+import io.vertx.ext.web.RoutingContext;
+import io.vertx.ext.web.impl.Utils;
+import java.time.Clock;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class LoggingHandler implements Handler<RoutingContext> {
+
+  private static final Logger LOG = LoggerFactory.getLogger(LoggingHandler.class);
+  static final String HTTP_HEADER_USER_AGENT = "User-Agent";
+
+  private final Set<Integer> skipResponseCodes;
+  private final Map<String, Double> rateLimitedPaths;
+  private final Logger logger;
+  private final Clock clock;
+  private final Function<Double, RateLimiter> rateLimiterFactory;
+
+  private final Map<String, RateLimiter> rateLimiters = new ConcurrentHashMap<>();
+
+  public LoggingHandler(final Server server) {
+    this(server, LOG, Clock.systemUTC(), RateLimiter::create);
+  }
+
+  @VisibleForTesting
+  LoggingHandler(
+      final Server server,
+      final Logger logger,
+      final Clock clock,
+      final Function<Double, RateLimiter> rateLimiterFactory) {
+    this.skipResponseCodes = getSkipResponseCodes(server.getConfig());
+    this.rateLimitedPaths = getSkipRequestPaths(server.getConfig());
+    this.logger = logger;
+    this.clock = clock;
+    this.rateLimiterFactory = rateLimiterFactory;
+  }
+
+  @Override
+  public void handle(final RoutingContext routingContext) {
+    routingContext.addEndHandler(ar -> {
+      // After the response is complete, log results here.
+      if (skipResponseCodes.contains(routingContext.response().getStatusCode())) {
+        return;
+      }
+      if (rateLimitedPaths.containsKey(routingContext.request().path())) {
+        final String path = routingContext.request().path();
+        final double rateLimit = rateLimitedPaths.get(path);
+        rateLimiters.computeIfAbsent(path, (k) -> rateLimiterFactory.apply(rateLimit));
+        if (!rateLimiters.get(path).tryAcquire()) {
+          return;
+        }
+      }
+      final long contentLength = routingContext.request().response().bytesWritten();
+      final HttpVersion version = routingContext.request().version();
+      final HttpMethod method = routingContext.request().method();
+      final String uri = routingContext.request().uri();
+      final int status = routingContext.request().response().getStatusCode();
+      final long requestBodyLength = routingContext.request().bytesRead();
+      final String versionFormatted;
+      switch (version) {
+        case HTTP_1_0:
+          versionFormatted = "HTTP/1.0";
+          break;
+        case HTTP_1_1:
+          versionFormatted = "HTTP/1.1";
+          break;
+        case HTTP_2:
+          versionFormatted = "HTTP/2.0";
+          break;
+        default:
+          versionFormatted = "-";
+      }
+      final String name = Optional.ofNullable((ApiUser) routingContext.user())
+          .map(u -> u.getPrincipal().getName())
+          .orElse("-");
+      final String userAgent = Optional.ofNullable(
+          routingContext.request().getHeader(HTTP_HEADER_USER_AGENT)).orElse("-");
+      final String timestamp = Utils.formatRFC1123DateTime(clock.millis());
+      final String message = String.format(
+          "%s - %s [%s] \"%s %s %s\" %d %d \"-\" \"%s\" %d",
+          routingContext.request().remoteAddress().host(),
+          name,
+          timestamp,
+          method,
+          uri,
+          versionFormatted,
+          status,
+          contentLength,
+          userAgent,
+          requestBodyLength);
+      doLog(status, message);
+    });
+    routingContext.next();
+  }
+
+  private static Set<Integer> getSkipResponseCodes(final KsqlRestConfig config) {
+    // Already validated as all ints
+    return config.getList(KSQL_LOGGING_SERVER_SKIPPED_RESPONSE_CODES_CONFIG)
+        .stream()
+        .map(Integer::parseInt).collect(ImmutableSet.toImmutableSet());
+  }
+
+  private static Map<String, Double> getSkipRequestPaths(final KsqlRestConfig config) {
+    // Already validated as having double values
+    return config.getStringAsMap(KSQL_LOGGING_SERVER_RATE_LIMITED_REQUEST_PATHS_CONFIG)
+        .entrySet().stream()
+        .collect(ImmutableMap.toImmutableMap(Entry::getKey,
+            entry -> Double.parseDouble(entry.getValue())));
+  }
+
+  private void doLog(final int status, final String message) {
+    if (status >= 500) {
+      logger.error(message);
+    } else if (status >= 400) {
+      logger.warn(message);
+    } else {
+      logger.info(message);
+    }
+  }
+}

ksqldb-rest-app/src/main/java/io/confluent/ksql/api/server/ServerVerticle.java

@@ -112,6 +112,8 @@ int actualPort() {
   private Router setupRouter() {
     final Router router = Router.router(vertx);
 
+    router.route().handler(new LoggingHandler(server));
+
     KsqlCorsHandler.setupCorsHandler(server, router);
 
     // /chc endpoints need to be before server state handler but after CORS handler as they

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

@@ -15,13 +15,16 @@
 
 package io.confluent.ksql.rest.server;
 
+import static io.confluent.ksql.configdef.ConfigValidators.intList;
+import static io.confluent.ksql.configdef.ConfigValidators.mapWithDoubleValue;
 import static io.confluent.ksql.configdef.ConfigValidators.oneOrMore;
 import static io.confluent.ksql.configdef.ConfigValidators.zeroOrPositive;
 
 import com.google.common.annotations.VisibleForTesting;
 import io.confluent.ksql.configdef.ConfigValidators;
 import io.confluent.ksql.rest.DefaultErrorMessages;
 import io.confluent.ksql.rest.ErrorMessages;
+import io.confluent.ksql.util.KsqlConfig;
 import io.confluent.ksql.util.KsqlException;
 import io.confluent.ksql.util.KsqlServerException;
 import io.vertx.core.http.ClientAuth;
@@ -330,6 +333,16 @@ public class KsqlRestConfig extends AbstractConfig {
       "The key store certificate alias to be used for internal client requests. If not set, "
           + "the system will fall back on the Vert.x default choice";
 
+  public static final String KSQL_LOGGING_SERVER_SKIPPED_RESPONSE_CODES_CONFIG =
+      KSQL_CONFIG_PREFIX + "logging.server.skipped.response.codes";
+  private static final String KSQL_LOGGING_SERVER_SKIPPED_RESPONSE_CODES_DOC =
+      "A list of HTTP response codes to skip during server request logging";
+
+  public static final String KSQL_LOGGING_SERVER_RATE_LIMITED_REQUEST_PATHS_CONFIG =
+      KSQL_CONFIG_PREFIX + "logging.server.rate.limited.request.paths";
+  private static final String KSQL_LOGGING_SERVER_RATE_LIMITED_REQUEST_PATHS_DOC =
+      "A list of path:rate_limit pairs, to rate limit the server request logging";
+
   private static final ConfigDef CONFIG_DEF;
 
   static {
@@ -626,6 +639,20 @@ public class KsqlRestConfig extends AbstractConfig {
             KSQL_AUTHENTICATION_PLUGIN_DEFAULT,
             ConfigDef.Importance.LOW,
             KSQL_AUTHENTICATION_PLUGIN_DOC
+        ).define(
+            KSQL_LOGGING_SERVER_SKIPPED_RESPONSE_CODES_CONFIG,
+            Type.LIST,
+            "",
+            intList(),
+            ConfigDef.Importance.LOW,
+            KSQL_LOGGING_SERVER_SKIPPED_RESPONSE_CODES_DOC
+        ).define(
+            KSQL_LOGGING_SERVER_RATE_LIMITED_REQUEST_PATHS_CONFIG,
+            Type.STRING,
+            "",
+            mapWithDoubleValue(),
+            ConfigDef.Importance.LOW,
+            KSQL_LOGGING_SERVER_RATE_LIMITED_REQUEST_PATHS_DOC
         );
   }
 
@@ -924,4 +951,8 @@ private static String getLocalHostName() {
     }
   }
 
+  public Map<String, String> getStringAsMap(final String key) {
+    final String value = getString(key).trim();
+    return KsqlConfig.parseStringAsMap(key, value);
+  }
 }