Merge pull request #1947 from noddi/feature/count-parameter-history-e…

…ndpoints Add count parameter to history endpoints
apache · Nov 12, 2015 · 4f46d45 · 4f46d45
2 parents 148153b + 6e5d2c6
commit 4f46d45
Show file tree

Hide file tree

Showing 10 changed files with 417 additions and 6 deletions.
diff --git a/common/src/main/java/io/druid/audit/AuditManager.java b/common/src/main/java/io/druid/audit/AuditManager.java
@@ -62,4 +62,21 @@ public interface AuditManager
    */
   public List<AuditEntry> fetchAuditHistory(String type, Interval interval);
 
+  /**
+   * Provides last N entries of audit history for given key, type
+   * @param key
+   * @param type
+   * @param limit
+   * @return list of AuditEntries satisfying the passed parameters
+   */
+  public List<AuditEntry> fetchAuditHistory(String key, String type, int limit);
+
+  /**
+   * Provides last N entries of audit history for given type
+   * @param type type of auditEntry
+   * @param limit
+   * @return list of AuditEntries satisfying the passed parameters
+   */
+  public List<AuditEntry> fetchAuditHistory(String type, int limit);
+
 }
diff --git a/docs/content/configuration/coordinator.md b/docs/content/configuration/coordinator.md
@@ -90,3 +90,9 @@ http://<COORDINATOR_IP>:<PORT>/druid/coordinator/v1/config/history?interval=<int
 ```
 
 default value of interval can be specified by setting `druid.audit.manager.auditHistoryMillis` (1 week if not configured) in coordinator runtime.properties
+
+To view last <n> entries of the audit history of coordinator dynamic config issue a GET request to the URL -
+
+```
+http://<COORDINATOR_IP>:<PORT>/druid/coordinator/v1/config/history?count=<n>
+```
diff --git a/docs/content/configuration/indexing-service.md b/docs/content/configuration/indexing-service.md
@@ -169,6 +169,12 @@ http://<OVERLORD_IP>:<port>/druid/indexer/v1/worker/history?interval=<interval>
 
 default value of interval can be specified by setting `druid.audit.manager.auditHistoryMillis` (1 week if not configured) in overlord runtime.properties.
 
+To view last <n> entries of the audit history of worker config issue a GET request to the URL -
+
+```
+http://<OVERLORD_IP>:<port>/druid/indexer/v1/worker/history?count=<n>
+```
+
 #### Worker Select Strategy
 
 ##### Fill Capacity

diff --git a/docs/content/design/coordinator.md b/docs/content/design/coordinator.md
@@ -193,6 +193,10 @@ Returns all rules for a specified datasource and includes default datasource.
 
  Returns audit history of rules for a specified datasource. default value of interval can be specified by setting `druid.audit.manager.auditHistoryMillis` (1 week if not configured) in coordinator runtime.properties
 
+* `/druid/coordinator/v1/rules/{dataSourceName}/history?count=<n>`
+
+ Returns last <n> entries of audit history of rules for a specified datasource.
+
 #### Intervals
 
 * `/druid/coordinator/v1/intervals`

diff --git a/indexing-service/src/main/java/io/druid/indexing/overlord/http/OverlordResource.java b/indexing-service/src/main/java/io/druid/indexing/overlord/http/OverlordResource.java
@@ -29,6 +29,7 @@
 import com.google.common.util.concurrent.SettableFuture;
 import com.google.inject.Inject;
 import com.metamx.common.logger.Logger;
+
 import io.druid.audit.AuditInfo;
 import io.druid.audit.AuditManager;
 import io.druid.common.config.JacksonConfigManager;
@@ -46,6 +47,7 @@
 import io.druid.metadata.EntryExistsException;
 import io.druid.tasklogs.TaskLogStreamer;
 import io.druid.timeline.DataSegment;
+
 import org.joda.time.DateTime;
 import org.joda.time.Interval;
 
@@ -62,6 +64,8 @@
 import javax.ws.rs.core.Context;
 import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
+import javax.ws.rs.core.Response.Status;
+
 import java.io.IOException;
 import java.util.Collection;
 import java.util.List;
@@ -219,10 +223,28 @@ public Response setWorkerConfig(
   @Path("/worker/history")
   @Produces(MediaType.APPLICATION_JSON)
   public Response getWorkerConfigHistory(
-      @QueryParam("interval") final String interval
+      @QueryParam("interval") final String interval,
+      @QueryParam("count") final Integer count
   )
   {
     Interval theInterval = interval == null ? null : new Interval(interval);
+    if (theInterval == null && count != null) {
+      try {
+        return Response.ok(
+            auditManager.fetchAuditHistory(
+                WorkerBehaviorConfig.CONFIG_KEY,
+                WorkerBehaviorConfig.CONFIG_KEY,
+                count
+            )
+        )
+                       .build();
+      }
+      catch (IllegalArgumentException e) {
+        return Response.status(Response.Status.BAD_REQUEST)
+                       .entity(ImmutableMap.<String, Object>of("error", e.getMessage()))
+                       .build();
+      }
+    }
     return Response.ok(
         auditManager.fetchAuditHistory(
             WorkerBehaviorConfig.CONFIG_KEY,

diff --git a/server/src/main/java/io/druid/server/audit/SQLAuditManager.java b/server/src/main/java/io/druid/server/audit/SQLAuditManager.java
@@ -22,16 +22,19 @@
 import com.google.inject.Inject;
 import com.metamx.emitter.service.ServiceEmitter;
 import com.metamx.emitter.service.ServiceMetricEvent;
+
 import io.druid.audit.AuditEntry;
 import io.druid.audit.AuditManager;
 import io.druid.guice.ManageLifecycle;
 import io.druid.guice.annotations.Json;
 import io.druid.metadata.MetadataStorageTablesConfig;
 import io.druid.metadata.SQLMetadataConnector;
+
 import org.joda.time.DateTime;
 import org.joda.time.Interval;
 import org.skife.jdbi.v2.Handle;
 import org.skife.jdbi.v2.IDBI;
+import org.skife.jdbi.v2.Query;
 import org.skife.jdbi.v2.StatementContext;
 import org.skife.jdbi.v2.tweak.HandleCallback;
 import org.skife.jdbi.v2.tweak.ResultSetMapper;
@@ -40,6 +43,7 @@
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.List;
+import java.util.Map;
 
 @ManageLifecycle
 public class SQLAuditManager implements AuditManager
@@ -166,6 +170,14 @@ private Interval getIntervalOrDefault(Interval interval)
     return theInterval;
   }
 
+  private int getLimit(int limit) throws IllegalArgumentException
+  {
+    if (limit < 1) {
+      throw new IllegalArgumentException("Limit must be greater than zero!");
+    }
+    return limit;
+  }
+
   @Override
   public List<AuditEntry> fetchAuditHistory(final String type, Interval interval)
   {
@@ -207,4 +219,63 @@ public AuditEntry map(int index, ResultSet r, StatementContext ctx)
     );
   }
 
+  @Override
+  public List<AuditEntry> fetchAuditHistory(final String key, final String type, int limit)
+      throws IllegalArgumentException
+  {
+    return fetchAuditHistoryLastEntries(key, type, limit);
+  }
+
+  @Override
+  public List<AuditEntry> fetchAuditHistory(final String type, int limit)
+      throws IllegalArgumentException
+  {
+    return fetchAuditHistoryLastEntries(null, type, limit);
+  }
+
+  private List<AuditEntry> fetchAuditHistoryLastEntries(final String key, final String type, int limit)
+      throws IllegalArgumentException
+  {
+    final int theLimit = getLimit(limit);
+    String queryString = String.format("SELECT payload FROM %s WHERE type = :type", getAuditTable());
+    if (key != null) {
+      queryString += " and audit_key = :audit_key";
+    }
+    queryString += " ORDER BY created_date DESC";
+    final String theQueryString = queryString;
+
+    return dbi.withHandle(
+        new HandleCallback<List<AuditEntry>>()
+        {
+          @Override
+          public List<AuditEntry> withHandle(Handle handle) throws Exception
+          {
+            Query<Map<String, Object>> query = handle.createQuery(theQueryString);
+            if (key != null) {
+              query.bind("audit_key", key);
+            }
+            return query.bind("type", type)
+                        .setMaxRows(theLimit)
+                        .map(
+                            new ResultSetMapper<AuditEntry>()
+                            {
+                              @Override
+                              public AuditEntry map(int index, ResultSet r, StatementContext ctx)
+                                  throws SQLException
+                              {
+                                try {
+                                  return jsonMapper.readValue(r.getBytes("payload"), AuditEntry.class);
+                                }
+                                catch (IOException e) {
+                                  throw new SQLException(e);
+                                }
+                              }
+                            }
+                          )
+                          .list();
+          }
+        }
+        );
+  }
+
 }
diff --git a/server/src/main/java/io/druid/server/http/CoordinatorDynamicConfigsResource.java b/server/src/main/java/io/druid/server/http/CoordinatorDynamicConfigsResource.java
@@ -21,8 +21,11 @@
 import io.druid.audit.AuditManager;
 import io.druid.common.config.JacksonConfigManager;
 import io.druid.server.coordinator.CoordinatorDynamicConfig;
+
 import org.joda.time.Interval;
 
+import com.google.common.collect.ImmutableMap;
+
 import javax.inject.Inject;
 import javax.servlet.http.HttpServletRequest;
 import javax.ws.rs.Consumes;
@@ -90,10 +93,28 @@ public Response setDynamicConfigs(final CoordinatorDynamicConfig dynamicConfig,
   @Path("/history")
   @Produces(MediaType.APPLICATION_JSON)
   public Response getDatasourceRuleHistory(
-      @QueryParam("interval") final String interval
+      @QueryParam("interval") final String interval,
+      @QueryParam("count") final Integer count
   )
   {
     Interval theInterval = interval == null ? null : new Interval(interval);
+    if (theInterval == null && count != null) {
+      try {
+        return Response.ok(
+            auditManager.fetchAuditHistory(
+                CoordinatorDynamicConfig.CONFIG_KEY,
+                CoordinatorDynamicConfig.CONFIG_KEY,
+                count
+            )
+        )
+                       .build();
+      }
+      catch (IllegalArgumentException e) {
+        return Response.status(Response.Status.BAD_REQUEST)
+                       .entity(ImmutableMap.<String, Object>of("error", e.getMessage()))
+                       .build();
+      }
+    }
     return Response.ok(
         auditManager.fetchAuditHistory(
             CoordinatorDynamicConfig.CONFIG_KEY,

diff --git a/server/src/main/java/io/druid/server/http/RulesResource.java b/server/src/main/java/io/druid/server/http/RulesResource.java
@@ -19,10 +19,12 @@
 
 import com.google.common.collect.ImmutableMap;
 import com.google.inject.Inject;
+
 import io.druid.audit.AuditInfo;
 import io.druid.audit.AuditManager;
 import io.druid.metadata.MetadataRuleManager;
 import io.druid.server.coordinator.rules.Rule;
+
 import org.joda.time.Interval;
 
 import javax.servlet.http.HttpServletRequest;
@@ -38,6 +40,8 @@
 import javax.ws.rs.core.Context;
 import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
+import javax.ws.rs.core.Response.Status;
+
 import java.util.List;
 
 /**
@@ -108,10 +112,22 @@ public Response setDatasourceRules(
   @Produces(MediaType.APPLICATION_JSON)
   public Response getDatasourceRuleHistory(
       @PathParam("dataSourceName") final String dataSourceName,
-      @QueryParam("interval") final String interval
+      @QueryParam("interval") final String interval,
+      @QueryParam("count") final Integer count
   )
   {
     Interval theInterval = interval == null ? null : new Interval(interval);
+    if (theInterval == null && count != null) {
+      try {
+        return Response.ok(auditManager.fetchAuditHistory(dataSourceName, "rules", count))
+                       .build();
+      }
+      catch (IllegalArgumentException e) {
+        return Response.status(Response.Status.BAD_REQUEST)
+                       .entity(ImmutableMap.<String, Object>of("error", e.getMessage()))
+                       .build();
+      }
+    }
     return Response.ok(auditManager.fetchAuditHistory(dataSourceName, "rules", theInterval))
                    .build();
   }