adds ability to purge old task history from DB

SingularityService/pom.xml

@@ -352,6 +352,17 @@
       <artifactId>curator-test</artifactId>
       <scope>test</scope>
     </dependency>
+
+    <dependency>
+      <groupId>com.h2database</groupId>
+      <artifactId>h2</artifactId>
+      <scope>test</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.liquibase</groupId>
+      <artifactId>liquibase-core</artifactId>
+    </dependency>
 
     <dependency>
       <groupId>org.apache.directory.api</groupId>

SingularityService/singularity.sql → SingularityService/singularity_test.sql

@@ -1,15 +1,11 @@
-CREATE DATABASE IF NOT EXISTS singularity CHARACTER SET = UTF8;
-
-USE singularity;
-
 CREATE TABLE requestHistory (
   requestId VARCHAR(100) NOT NULL,
   createdAt TIMESTAMP NOT NULL DEFAULT '1971-01-01 00:00:01',
   requestState VARCHAR(25) NOT NULL,
   user VARCHAR(100) NULL,
   request BLOB NOT NULL,
   PRIMARY KEY (requestId, createdAt)
-) ENGINE=InnoDB;
+);
 
 CREATE TABLE deployHistory (
   requestId VARCHAR(100) NOT NULL,
@@ -20,17 +16,12 @@ CREATE TABLE deployHistory (
   deployState VARCHAR(25) NOT NULL,
   bytes BLOB NOT NULL,
   PRIMARY KEY (requestId, deployId),
-  INDEX (requestId, createdAt)
-) ENGINE=InnoDB;
+);
 
 CREATE TABLE taskHistory (
   taskId VARCHAR(200) PRIMARY KEY,
   requestId VARCHAR(100) NOT NULL,
   updatedAt TIMESTAMP NOT NULL DEFAULT '1971-01-01 00:00:01',
   lastTaskStatus VARCHAR(25) NULL,
   bytes BLOB NOT NULL,
-  INDEX (requestId, updatedAt)
-) ENGINE=InnoDB;
-
-CREATE USER 'singularity'@'%' IDENTIFIED BY '';
-GRANT ALL PRIVILEGES ON singularity.* TO 'singularity'@'%';
+);

SingularityService/src/main/java/com/hubspot/singularity/SingularityMainModule.java

@@ -38,6 +38,7 @@
 import com.hubspot.mesos.JavaUtils;
 import com.hubspot.mesos.client.MesosClient;
 import com.hubspot.singularity.config.CustomExecutorConfiguration;
+import com.hubspot.singularity.config.HistoryPurgingConfiguration;
 import com.hubspot.singularity.config.MesosConfiguration;
 import com.hubspot.singularity.config.S3Configuration;
 import com.hubspot.singularity.config.S3GroupOverrideConfiguration;
@@ -251,6 +252,12 @@ public Optional<S3Configuration> s3Configuration(final SingularityConfiguration
     return config.getS3Configuration();
   }
 
+  @Provides
+  @Singleton
+  public HistoryPurgingConfiguration historyPurgingConfiguration(final SingularityConfiguration config) {
+    return config.getHistoryPurgingConfiguration();
+  }
+
   private JadeTemplate getJadeTemplate(String name) throws IOException {
     Parser parser = new Parser("templates/" + name, JadeTemplateLoader.JADE_LOADER);
     Node root = parser.parse();

SingularityService/src/main/java/com/hubspot/singularity/config/HistoryPurgingConfiguration.java

@@ -0,0 +1,65 @@
+package com.hubspot.singularity.config;
+
+import com.google.common.base.Optional;
+
+public class HistoryPurgingConfiguration {
+
+  private int deleteTaskHistoryAfterDays = 365;
+
+  private int deleteTaskHistoryAfterTasksPerRequest = 10000;
+
+  private boolean deleteTaskHistoryBytesInsteadOfEntireRow = true;
+
+  private int checkTaskHistoryEveryHours = 24;
+
+  private boolean enabled = false;
+
+  private Optional<Integer> absentIfNotOverOne(int value) {
+    if (value < 1) {
+      return Optional.absent();
+    }
+    return Optional.of(value);
+  }
+
+  public Optional<Integer> getDeleteTaskHistoryAfterDays() {
+    return absentIfNotOverOne(deleteTaskHistoryAfterDays);
+  }
+
+  public void setDeleteTaskHistoryAfterDays(int deleteTaskHistoryAfterDays) {
+    this.deleteTaskHistoryAfterDays = deleteTaskHistoryAfterDays;
+  }
+
+  public Optional<Integer> getDeleteTaskHistoryAfterTasksPerRequest() {
+    return absentIfNotOverOne(deleteTaskHistoryAfterTasksPerRequest);
+  }
+
+  public void setDeleteTaskHistoryAfterTasksPerRequest(int deleteTaskHistoryAfterTasksPerRequest) {
+    this.deleteTaskHistoryAfterTasksPerRequest = deleteTaskHistoryAfterTasksPerRequest;
+  }
+
+  public boolean isDeleteTaskHistoryBytesInsteadOfEntireRow() {
+    return deleteTaskHistoryBytesInsteadOfEntireRow;
+  }
+
+  public void setDeleteTaskHistoryBytesInsteadOfEntireRow(boolean deleteTaskHistoryBytesInsteadOfEntireRow) {
+    this.deleteTaskHistoryBytesInsteadOfEntireRow = deleteTaskHistoryBytesInsteadOfEntireRow;
+  }
+
+  public int getCheckTaskHistoryEveryHours() {
+    return checkTaskHistoryEveryHours;
+  }
+
+  public void setCheckTaskHistoryEveryHours(int checkTaskHistoryEveryHours) {
+    this.checkTaskHistoryEveryHours = checkTaskHistoryEveryHours;
+  }
+
+  public boolean isEnabledAndValid() {
+    return enabled && checkTaskHistoryEveryHours > 1 && (getDeleteTaskHistoryAfterDays().isPresent() || getDeleteTaskHistoryAfterTasksPerRequest().isPresent());
+  }
+
+  public void setEnabled(boolean enabled) {
+    this.enabled = enabled;
+  }
+
+
+}

SingularityService/src/main/java/com/hubspot/singularity/config/SingularityConfiguration.java

@@ -117,6 +117,10 @@ public class SingularityConfiguration extends Configuration {
 
   private int maxRequestIdSize = 100;
 
+  @JsonProperty("historyPurging")
+  @Valid
+  private HistoryPurgingConfiguration historyPurgingConfiguration = new HistoryPurgingConfiguration();
+
   @JsonProperty("mesos")
   @Valid
   private MesosConfiguration mesosConfiguration;
@@ -742,4 +746,12 @@ public void setAuthConfiguration(AuthConfiguration authConfiguration) {
     this.authConfiguration = authConfiguration;
   }
 
+  public HistoryPurgingConfiguration getHistoryPurgingConfiguration() {
+    return historyPurgingConfiguration;
+  }
+
+  public void setHistoryPurgingConfiguration(HistoryPurgingConfiguration historyPurgingConfiguration) {
+    this.historyPurgingConfiguration = historyPurgingConfiguration;
+  }
+
 }

SingularityService/src/main/java/com/hubspot/singularity/data/history/HistoryJDBI.java

@@ -12,6 +12,7 @@
 import com.hubspot.singularity.SingularityDeployHistory;
 import com.hubspot.singularity.SingularityRequestHistory;
 import com.hubspot.singularity.SingularityTaskIdHistory;
+import com.hubspot.singularity.data.history.SingularityMappers.SingularityRequestIdCount;
 
 @UseStringTemplate3StatementLocator
 public interface HistoryJDBI {
@@ -43,5 +44,18 @@ public interface HistoryJDBI {
   @SqlQuery("SELECT DISTINCT requestId FROM requestHistory WHERE requestId LIKE CONCAT(:requestIdLike, '%') LIMIT :limitStart, :limitCount")
   List<String> getRequestHistoryLike(@Bind("requestIdLike") String requestIdLike, @Bind("limitStart") Integer limitStart, @Bind("limitCount") Integer limitCount);
 
+  @SqlQuery("SELECT requestId, COUNT(*) as count FROM taskHistory WHERE updatedAt \\< :updatedAt GROUP BY requestId")
+  List<SingularityRequestIdCount> getRequestIdCounts(@Bind("updatedAt") Date updatedAt);
+
+  @SqlQuery("SELECT MIN(updatedAt) from (SELECT updatedAt FROM taskHistory WHERE requestId = :requestId ORDER BY updatedAt DESC LIMIT :limit) as alias")
+  Date getMinUpdatedAtWithLimitForRequest(@Bind("requestId") String requestId, @Bind("limit") Integer limit);
+
+  @SqlUpdate("UPDATE taskHistory SET bytes = '' WHERE requestId = :requestId AND updatedAt \\< :updatedAtBefore")
+  void updateTaskHistoryNullBytesForRequestBefore(@Bind("requestId") String requestId, @Bind("updatedAtBefore") Date updatedAtBefore);
+
+  @SqlUpdate("DELETE FROM taskHistory WHERE requestId = :requestId AND updatedAt \\< :updatedAtBefore")
+  void deleteTaskHistoryForRequestBefore(@Bind("requestId") String requestId, @Bind("updatedAtBefore") Date updatedAtBefore);
+
   void close();
+
 }

SingularityService/src/main/java/com/hubspot/singularity/data/history/HistoryManager.java

@@ -1,12 +1,14 @@
 package com.hubspot.singularity.data.history;
 
+import java.util.Date;
 import java.util.List;
 
 import com.google.common.base.Optional;
 import com.hubspot.singularity.SingularityDeployHistory;
 import com.hubspot.singularity.SingularityRequestHistory;
 import com.hubspot.singularity.SingularityTaskHistory;
 import com.hubspot.singularity.SingularityTaskIdHistory;
+import com.hubspot.singularity.data.history.SingularityMappers.SingularityRequestIdCount;
 
 public interface HistoryManager {
 
@@ -32,4 +34,8 @@ public enum OrderDirection {
 
   List<String> getRequestHistoryLike(String requestIdLike, Integer limitStart, Integer limitCount);
 
+  List<SingularityRequestIdCount> getRequestIdCounts(Date before);
+
+  void purgeTaskHistory(String requestId, int count, Optional<Integer> limit, Optional<Date> purgeBefore, boolean deleteRowInsteadOfUpdate);
+
 }

SingularityService/src/main/java/com/hubspot/singularity/data/history/JDBIHistoryManager.java

@@ -3,6 +3,9 @@
 import java.util.Date;
 import java.util.List;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import com.google.common.base.Optional;
 import com.google.inject.Inject;
 import com.hubspot.singularity.DeployState;
@@ -11,17 +14,18 @@
 import com.hubspot.singularity.SingularityRequestHistory;
 import com.hubspot.singularity.SingularityTaskHistory;
 import com.hubspot.singularity.SingularityTaskIdHistory;
+import com.hubspot.singularity.data.history.SingularityMappers.SingularityRequestIdCount;
 import com.hubspot.singularity.data.transcoders.Transcoder;
 
 public class JDBIHistoryManager implements HistoryManager {
 
+  private static final Logger LOG = LoggerFactory.getLogger(JDBIHistoryManager.class);
+
   private final HistoryJDBI history;
   private final Transcoder<SingularityTaskHistory> taskHistoryTranscoder;
   private final Transcoder<SingularityDeployHistory> deployHistoryTranscoder;
   private final Transcoder<SingularityRequest> singularityRequestTranscoder;
 
-  // TODO jdbi timeouts / exceptions
-
   @Inject
   public JDBIHistoryManager(HistoryJDBI history, Transcoder<SingularityTaskHistory> taskHistoryTranscoder, Transcoder<SingularityDeployHistory> deployHistoryTranscoder,
       Transcoder<SingularityRequest> singularityRequestTranscoder) {
@@ -111,4 +115,38 @@ public Optional<SingularityTaskHistory> getTaskHistory(String taskId) {
     return Optional.of(taskHistoryTranscoder.fromBytes(historyBytes));
   }
 
+  @Override
+  public List<SingularityRequestIdCount> getRequestIdCounts(Date before) {
+    return history.getRequestIdCounts(before);
+  }
+
+  @Override
+  public void purgeTaskHistory(String requestId, int count, Optional<Integer> limit, Optional<Date> purgeBefore, boolean deleteRowInsteadOfUpdate) {
+    if (limit.isPresent() && count > limit.get()) {
+      Date beforeBasedOnLimit = history.getMinUpdatedAtWithLimitForRequest(requestId, limit.get());
+
+      if (deleteRowInsteadOfUpdate) {
+        LOG.info("Deleting task history for {} above {} items (before {})", requestId, limit.get(), beforeBasedOnLimit);
+
+        history.deleteTaskHistoryForRequestBefore(requestId, beforeBasedOnLimit);
+      } else {
+        LOG.info("Purging task history bytes for {} above {} items (before {})", requestId, limit.get(), beforeBasedOnLimit);
+
+        history.updateTaskHistoryNullBytesForRequestBefore(requestId, beforeBasedOnLimit);
+      }
+    }
+
+    if (purgeBefore.isPresent()) {
+      if (deleteRowInsteadOfUpdate) {
+        LOG.info("Deleting task history for {} before {}", requestId, purgeBefore.get());
+
+        history.deleteTaskHistoryForRequestBefore(requestId, purgeBefore.get());
+      } else {
+        LOG.info("Purging task history bytes for {} before {}", requestId, purgeBefore.get());
+
+        history.updateTaskHistoryNullBytesForRequestBefore(requestId, purgeBefore.get());
+      }
+    }
+  }
+
 }

SingularityService/src/main/java/com/hubspot/singularity/data/history/NoopHistoryManager.java

@@ -1,6 +1,7 @@
 package com.hubspot.singularity.data.history;
 
 import java.util.Collections;
+import java.util.Date;
 import java.util.List;
 
 import javax.inject.Inject;
@@ -10,6 +11,7 @@
 import com.hubspot.singularity.SingularityRequestHistory;
 import com.hubspot.singularity.SingularityTaskHistory;
 import com.hubspot.singularity.SingularityTaskIdHistory;
+import com.hubspot.singularity.data.history.SingularityMappers.SingularityRequestIdCount;
 
 public class NoopHistoryManager implements HistoryManager {
 
@@ -62,4 +64,14 @@ public List<String> getRequestHistoryLike(String requestIdLike, Integer limitSta
     return Collections.emptyList();
   }
 
+  @Override
+  public List<SingularityRequestIdCount> getRequestIdCounts(Date before) {
+    return Collections.emptyList();
+  }
+
+  @Override
+  public void purgeTaskHistory(String requestId, int count, Optional<Integer> limit, Optional<Date> purgeBefore, boolean deleteRowInsteadOfUpdate) {
+    throw new UnsupportedOperationException("NoopHistoryManager can not update/delete");
+  }
+
 }

SingularityService/src/main/java/com/hubspot/singularity/data/history/SingularityHistoryModule.java

@@ -1,9 +1,6 @@
 package com.hubspot.singularity.data.history;
 
 import static com.google.common.base.Preconditions.checkNotNull;
-import io.dropwizard.db.DataSourceFactory;
-import io.dropwizard.jdbi.DBIFactory;
-import io.dropwizard.setup.Environment;
 
 import java.util.Set;
 
@@ -12,7 +9,6 @@
 import org.skife.jdbi.v2.DBI;
 import org.skife.jdbi.v2.tweak.ResultSetMapper;
 
-import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Optional;
 import com.google.common.collect.ImmutableSet;
 import com.google.inject.AbstractModule;
@@ -25,6 +21,10 @@
 import com.google.inject.multibindings.Multibinder;
 import com.hubspot.singularity.config.SingularityConfiguration;
 
+import io.dropwizard.db.DataSourceFactory;
+import io.dropwizard.jdbi.DBIFactory;
+import io.dropwizard.setup.Environment;
+
 public class SingularityHistoryModule extends AbstractModule {
 
   private final Optional<DataSourceFactory> configuration;
@@ -34,21 +34,17 @@ public SingularityHistoryModule(SingularityConfiguration configuration) {
     this.configuration = configuration.getDatabaseConfiguration();
   }
 
-  @VisibleForTesting
-  public SingularityHistoryModule() {
-    this.configuration = Optional.absent();
-  }
-
   @Override
   public void configure() {
-
     Multibinder<ResultSetMapper<?>> resultSetMappers = Multibinder.newSetBinder(binder(), new TypeLiteral<ResultSetMapper<?>>() {});
 
     resultSetMappers.addBinding().to(SingularityMappers.SingularityBytesMapper.class).in(Scopes.SINGLETON);
     resultSetMappers.addBinding().to(SingularityMappers.SingularityRequestIdMapper.class).in(Scopes.SINGLETON);
     resultSetMappers.addBinding().to(SingularityMappers.SingularityRequestHistoryMapper.class).in(Scopes.SINGLETON);
     resultSetMappers.addBinding().to(SingularityMappers.SingularityTaskIdHistoryMapper.class).in(Scopes.SINGLETON);
     resultSetMappers.addBinding().to(SingularityMappers.SingularityDeployHistoryLiteMapper.class).in(Scopes.SINGLETON);
+    resultSetMappers.addBinding().to(SingularityMappers.SingularityRequestIdCountMapper.class).in(Scopes.SINGLETON);
+    resultSetMappers.addBinding().to(SingularityMappers.DateMapper.class).in(Scopes.SINGLETON);
 
     bind(TaskHistoryHelper.class).in(Scopes.SINGLETON);
     bind(RequestHistoryHelper.class).in(Scopes.SINGLETON);