Add Filtered and Composing request loggers (apache#3469)

* Add Filtered and Composing request loggers

Add Filtered and Composite Request loggers
- enables users to filter request logs for slow queries.

fix test

* review comments

* review comment

* remove unused import

docs/content/configuration/index.md

@@ -92,7 +92,7 @@ All nodes that can serve queries can also log the query requests they see.
 
 |Property|Description|Default|
 |--------|-----------|-------|
-|`druid.request.logging.type`|Choices: noop, file, emitter, slf4j. How to log every query request.|noop|
+|`druid.request.logging.type`|Choices: noop, file, emitter, slf4j, filtered, composing. How to log every query request.|noop|
 
 Note that, you can enable sending all the HTTP requests to log by setting  "io.druid.jetty.RequestLog" to DEBUG level. See [Logging](../configuration/logging.html)
 
@@ -134,6 +134,22 @@ MDC fields populated with `setMDC`:
 |`resultOrdering`|The ordering of results|
 |`descending`|If the query is a descending query|
 
+#### Filtered Request Logging
+Filtered Request Logger filters requests based on a configurable query/time threshold. Only request logs where query/time is above the threshold are emitted.
+
+|Property|Description|Default|
+|--------|-----------|-------|
+|`druid.request.logging.queryTimeThresholdMs`|Threshold value for query/time in milliseconds.|0 i.e no filtering|
+|`druid.request.logging.delegate`|Delegate request logger to log requests.|none|
+
+#### Composite Request Logging
+Composite Request Logger emits request logs to multiple request loggers.
+
+|Property|Description|Default|
+|--------|-----------|-------|
+|`druid.request.logging.loggerProviders`|List of request loggers for emitting request logs.|none|
+
+
 ### Enabling Metrics
 
 Druid nodes periodically emit metrics and different metrics monitors can be included. Each node can overwrite the default list of monitors.

server/src/main/java/io/druid/guice/QueryableModule.java

@@ -25,8 +25,10 @@
 import com.google.inject.util.Providers;
 import io.druid.initialization.DruidModule;
 import io.druid.query.QuerySegmentWalker;
+import io.druid.server.log.ComposingRequestLoggerProvider;
 import io.druid.server.log.EmittingRequestLoggerProvider;
 import io.druid.server.log.FileRequestLoggerProvider;
+import io.druid.server.log.FilteredRequestLoggerProvider;
 import io.druid.server.log.LoggingRequestLoggerProvider;
 import io.druid.server.log.RequestLogger;
 import io.druid.server.log.RequestLoggerProvider;
@@ -54,7 +56,9 @@ public List<Module> getJacksonModules()
             .registerSubtypes(
                 EmittingRequestLoggerProvider.class,
                 FileRequestLoggerProvider.class,
-                LoggingRequestLoggerProvider.class
+                LoggingRequestLoggerProvider.class,
+                ComposingRequestLoggerProvider.class,
+                FilteredRequestLoggerProvider.class
             )
     );
   }

server/src/main/java/io/druid/server/log/ComposingRequestLoggerProvider.java

@@ -0,0 +1,84 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Metamarkets licenses this file
+ * to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package io.druid.server.log;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import com.google.common.base.Throwables;
+import com.google.common.collect.Lists;
+import io.druid.server.RequestLogLine;
+
+import javax.validation.constraints.NotNull;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ */
+@JsonTypeName("composing")
+public class ComposingRequestLoggerProvider implements RequestLoggerProvider
+{
+  @JsonProperty
+  @NotNull
+  private final List<RequestLoggerProvider> loggerProviders = Lists.newArrayList();
+
+  @Override
+  public RequestLogger get()
+  {
+    final List<RequestLogger> loggers = new ArrayList<>();
+    for (RequestLoggerProvider loggerProvider : loggerProviders) {
+      loggers.add(loggerProvider.get());
+    }
+    return new ComposingRequestLogger(loggers);
+  }
+
+  public static class ComposingRequestLogger implements RequestLogger
+  {
+    private final List<RequestLogger> loggers;
+
+    public ComposingRequestLogger(List<RequestLogger> loggers)
+    {
+      this.loggers = loggers;
+    }
+
+    @Override
+    public void log(RequestLogLine requestLogLine) throws IOException
+    {
+      Exception exception = null;
+      for (RequestLogger logger : loggers) {
+        try {
+          logger.log(requestLogLine);
+        }
+        catch (Exception e) {
+          if (exception == null) {
+            exception = e;
+          } else {
+            exception.addSuppressed(e);
+          }
+        }
+      }
+      if (exception != null) {
+        Throwables.propagateIfInstanceOf(exception, IOException.class);
+        throw Throwables.propagate(exception);
+      }
+    }
+  }
+
+}

server/src/main/java/io/druid/server/log/FilteredRequestLoggerProvider.java

@@ -0,0 +1,69 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Metamarkets licenses this file
+ * to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package io.druid.server.log;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import io.druid.server.RequestLogLine;
+
+import javax.validation.constraints.NotNull;
+import java.io.IOException;
+
+/**
+ */
+@JsonTypeName("filtered")
+public class FilteredRequestLoggerProvider implements RequestLoggerProvider
+{
+  @JsonProperty
+  @NotNull
+  private RequestLoggerProvider delegate = null;
+
+  @JsonProperty
+  private long queryTimeThresholdMs = 0;
+
+  @Override
+  public RequestLogger get()
+  {
+    return new FilteredRequestLogger(delegate.get(), queryTimeThresholdMs);
+  }
+
+  public static class FilteredRequestLogger implements RequestLogger
+  {
+
+    private final long queryTimeThresholdMs;
+    private final RequestLogger logger;
+
+    public FilteredRequestLogger(RequestLogger logger, long queryTimeThresholdMs)
+    {
+      this.logger = logger;
+      this.queryTimeThresholdMs = queryTimeThresholdMs;
+    }
+
+    @Override
+    public void log(RequestLogLine requestLogLine) throws IOException
+    {
+      Object queryTime = requestLogLine.getQueryStats().getStats().get("query/time");
+      if (queryTime != null && ((Number) queryTime).longValue() >= queryTimeThresholdMs) {
+        logger.log(requestLogLine);
+      }
+    }
+  }
+
+}

server/src/test/java/io/druid/server/log/FilteredRequestLoggerTest.java

@@ -0,0 +1,73 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Metamarkets licenses this file
+ * to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package io.druid.server.log;
+
+import com.google.common.collect.ImmutableMap;
+import io.druid.server.QueryStats;
+import io.druid.server.RequestLogLine;
+import org.easymock.EasyMock;
+import org.junit.Test;
+
+import java.io.IOException;
+
+public class FilteredRequestLoggerTest
+{
+  @Test
+  public void testFilterBelowThreshold() throws IOException
+  {
+    RequestLogger delegate = EasyMock.createStrictMock(RequestLogger.class);
+    delegate.log((RequestLogLine) EasyMock.anyObject());
+    EasyMock.expectLastCall().andThrow(new IOException());
+    FilteredRequestLoggerProvider.FilteredRequestLogger logger = new FilteredRequestLoggerProvider.FilteredRequestLogger(
+        delegate,
+        1000
+    );
+    RequestLogLine requestLogLine = EasyMock.createMock(RequestLogLine.class);
+    EasyMock.expect(requestLogLine.getQueryStats())
+            .andReturn(new QueryStats(ImmutableMap.<String, Object>of("query/time", 100)))
+            .once();
+    EasyMock.replay(requestLogLine, delegate);
+    logger.log(requestLogLine);
+  }
+
+  @Test
+  public void testNotFilterAboveThreshold() throws IOException
+  {
+    RequestLogger delegate = EasyMock.createStrictMock(RequestLogger.class);
+    delegate.log((RequestLogLine) EasyMock.anyObject());
+    EasyMock.expectLastCall().times(2);
+    FilteredRequestLoggerProvider.FilteredRequestLogger logger = new FilteredRequestLoggerProvider.FilteredRequestLogger(
+        delegate,
+        1000
+    );
+    RequestLogLine requestLogLine = EasyMock.createMock(RequestLogLine.class);
+    EasyMock.expect(requestLogLine.getQueryStats())
+            .andReturn(new QueryStats(ImmutableMap.<String, Object>of("query/time", 10000)))
+            .once();
+    EasyMock.expect(requestLogLine.getQueryStats())
+            .andReturn(new QueryStats(ImmutableMap.<String, Object>of("query/time", 1000)))
+            .once();
+    EasyMock.replay(requestLogLine, delegate);
+    logger.log(requestLogLine);
+    logger.log(requestLogLine);
+
+    EasyMock.verify(requestLogLine, delegate);
+  }
+}