(WIP) Distributed in-memory cache for Singularity data

SingularityBase/src/main/java/com/hubspot/singularity/SingularityHostState.java

@@ -25,6 +25,7 @@ public class SingularityHostState {
 
   private final String mesosMaster;
   private final boolean mesosConnected;
+  private final long cacheLag;
 
   @JsonCreator
   public SingularityHostState(@JsonProperty("master") boolean master,
@@ -37,7 +38,8 @@ public SingularityHostState(@JsonProperty("master") boolean master,
                               @JsonProperty("mesosConnected") boolean mesosConnected,
                               @JsonProperty("offerCacheSize") int offerCacheSize,
                               @JsonProperty("availableCachedCpus") double availableCachedCpus,
-                              @JsonProperty("availableCachedMemory") double availableCachedMemory) {
+                              @JsonProperty("availableCachedMemory") double availableCachedMemory,
+                              @JsonProperty("cacheLag") long cacheLag) {
     this.master = master;
     this.uptime = uptime;
     this.driverStatus = driverStatus;
@@ -49,6 +51,7 @@ public SingularityHostState(@JsonProperty("master") boolean master,
     this.availableCachedCpus = availableCachedCpus;
     this.availableCachedMemory = availableCachedMemory;
     this.offerCacheSize = offerCacheSize;
+    this.cacheLag = cacheLag;
   }
 
   @Schema(description = "Address for this scheduler instance (host:port)")
@@ -106,6 +109,11 @@ public double getAvailableCachedMemory() {
     return availableCachedMemory;
   }
 
+  @Schema(description = "Last measured delay for the atomix cache")
+  public long getCacheLag() {
+    return cacheLag;
+  }
+
   @Override
   public String toString() {
     return "SingularityHostState{" +
@@ -120,6 +128,7 @@ public String toString() {
         ", hostname='" + hostname + '\'' +
         ", mesosMaster='" + mesosMaster + '\'' +
         ", mesosConnected=" + mesosConnected +
+        ", cacheLag='" + cacheLag + '\'' +
         '}';
   }
 }

SingularityBase/src/main/java/com/hubspot/singularity/SingularityScheduledTasksInfo.java

@@ -1,6 +1,5 @@
 package com.hubspot.singularity;
 
-import java.util.ArrayList;
 import java.util.List;
 
 public class SingularityScheduledTasksInfo {
@@ -9,9 +8,11 @@ public class SingularityScheduledTasksInfo {
   private final long maxTaskLag;
   private final long timestamp;
   private final List<SingularityPendingTaskId> lateTasks;
+  private final List<SingularityPendingTaskId> onDemandLateTasks;
 
-  private SingularityScheduledTasksInfo(List<SingularityPendingTaskId> lateTasks, int numFutureTasks, long maxTaskLag, long timestamp) {
+  public SingularityScheduledTasksInfo(List<SingularityPendingTaskId> lateTasks, List<SingularityPendingTaskId> onDemandLateTasks, int numFutureTasks, long maxTaskLag, long timestamp) {
     this.lateTasks = lateTasks;
+    this.onDemandLateTasks = onDemandLateTasks;
     this.numFutureTasks = numFutureTasks;
     this.maxTaskLag = maxTaskLag;
     this.timestamp = timestamp;
@@ -21,8 +22,8 @@ public List<SingularityPendingTaskId> getLateTasks() {
     return lateTasks;
   }
 
-  public int getNumLateTasks() {
-    return getLateTasks().size();
+  public List<SingularityPendingTaskId> getOnDemandLateTasks() {
+    return onDemandLateTasks;
   }
 
   public int getNumFutureTasks() {
@@ -36,28 +37,4 @@ public long getMaxTaskLag() {
   public long getTimestamp() {
     return timestamp;
   }
-
-  public static SingularityScheduledTasksInfo getInfo(List<SingularityPendingTask> pendingTasks, long millisDeltaForLateTasks) {
-    final long now = System.currentTimeMillis();
-
-    int numFutureTasks = 0;
-    long maxTaskLag = 0;
-    List<SingularityPendingTaskId> lateTasks = new ArrayList<>();
-
-    for (SingularityPendingTask pendingTask : pendingTasks) {
-      long delta = now - pendingTask.getPendingTaskId().getNextRunAt();
-
-      if (delta > millisDeltaForLateTasks) {
-        lateTasks.add(pendingTask.getPendingTaskId());
-      } else {
-        numFutureTasks++;
-      }
-
-      if (delta > maxTaskLag) {
-        maxTaskLag = delta;
-      }
-    }
-
-    return new SingularityScheduledTasksInfo(lateTasks, numFutureTasks, maxTaskLag, now);
-  }
 }

SingularityBase/src/main/java/com/hubspot/singularity/SingularitySlave.java

@@ -88,4 +88,40 @@ public String toString() {
         ", resources=" + resources +
         "} " + super.toString();
   }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    if (!super.equals(o)) {
+      return false;
+    }
+
+    SingularitySlave that = (SingularitySlave) o;
+
+    if (host != null ? !host.equals(that.host) : that.host != null) {
+      return false;
+    }
+    if (rackId != null ? !rackId.equals(that.rackId) : that.rackId != null) {
+      return false;
+    }
+    if (attributes != null ? !attributes.equals(that.attributes) : that.attributes != null) {
+      return false;
+    }
+    return resources != null ? resources.equals(that.resources) : that.resources == null;
+  }
+
+  @Override
+  public int hashCode() {
+    int result = super.hashCode();
+    result = 31 * result + (host != null ? host.hashCode() : 0);
+    result = 31 * result + (rackId != null ? rackId.hashCode() : 0);
+    result = 31 * result + (attributes != null ? attributes.hashCode() : 0);
+    result = 31 * result + (resources != null ? resources.hashCode() : 0);
+    return result;
+  }
 }

SingularityBase/src/main/java/com/hubspot/singularity/SingularityState.java

@@ -4,6 +4,7 @@
 import java.util.List;
 
 import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.google.common.base.Optional;
 
@@ -372,6 +373,14 @@ public long getLastHeartbeatAt() {
     return lastHeartbeatAt;
   }
 
+  @JsonIgnore
+  public SingularityState withHostStates(List<SingularityHostState> hostStates) {
+    return new SingularityState(activeTasks, launchingTasks, activeRequests, cooldownRequests, pausedRequests, scheduledTasks, pendingRequests, lbCleanupTasks, lbCleanupRequests,
+        cleaningRequests, activeSlaves, deadSlaves, decommissioningSlaves, activeRacks, deadRacks, decommissioningRacks, cleaningTasks, hostStates, oldestDeploy, numDeploys, oldestDeployStep,
+        activeDeploys, lateTasks, listLateTasks, onDemandLateTasks, onDemandListLateTasks, futureTasks, maxTaskLag, generatedAt, overProvisionedRequestIds, underProvisionedRequestIds, overProvisionedRequests,
+        underProvisionedRequests, finishedRequests, unknownRacks, unknownSlaves, authDatastoreHealthy, minimumPriorityLevel, avgStatusUpdateDelayMs, lastHeartbeatAt);
+  }
+
   @Override
   public String toString() {
     return "SingularityState{" +

SingularityClient/pom.xml

@@ -87,14 +87,26 @@
     </dependency>
 
     <dependency>
-      <groupId>junit</groupId>
-      <artifactId>junit</artifactId>
+      <groupId>org.assertj</groupId>
+      <artifactId>assertj-core</artifactId>
       <scope>test</scope>
     </dependency>
 
     <dependency>
-      <groupId>org.assertj</groupId>
-      <artifactId>assertj-core</artifactId>
+      <groupId>org.junit.jupiter</groupId>
+      <artifactId>junit-jupiter-api</artifactId>
+      <scope>test</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.junit.jupiter</groupId>
+      <artifactId>junit-jupiter-engine</artifactId>
+      <scope>test</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.junit.platform</groupId>
+      <artifactId>junit-platform-runner</artifactId>
       <scope>test</scope>
     </dependency>
 

SingularityClient/src/test/java/com/hubspot/singularity/client/SingularityClientModuleTest.java

@@ -2,7 +2,7 @@
 
 import java.util.Collections;
 
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
 
 import com.google.inject.AbstractModule;
 import com.google.inject.Guice;

SingularityClient/src/test/java/com/hubspot/singularity/client/SingularityClientTest.java

@@ -3,14 +3,13 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.*;
 import static org.mockito.internal.verification.VerificationModeFactory.times;
 
 import java.net.URI;
 
-import org.junit.Before;
-import org.junit.Test;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;
 import org.mockito.Captor;
 import org.mockito.Mock;
@@ -34,7 +33,7 @@ public class SingularityClientTest {
 
   private SingularityClient singularityClient;
 
-  @Before
+  @BeforeEach
   public void setUp() throws Exception {
     MockitoAnnotations.initMocks(this);
     singularityClient = buildClient();

SingularityExecutor/pom.xml

@@ -117,15 +117,29 @@
     </dependency>
 
     <dependency>
-      <groupId>junit</groupId>
-      <artifactId>junit</artifactId>
+      <groupId>org.assertj</groupId>
+      <artifactId>assertj-core</artifactId>
       <scope>test</scope>
     </dependency>
+
     <dependency>
-      <groupId>org.assertj</groupId>
-      <artifactId>assertj-core</artifactId>
+      <groupId>org.junit.jupiter</groupId>
+      <artifactId>junit-jupiter-api</artifactId>
+      <scope>test</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.junit.jupiter</groupId>
+      <artifactId>junit-jupiter-engine</artifactId>
       <scope>test</scope>
     </dependency>
+
+    <dependency>
+      <groupId>org.junit.platform</groupId>
+      <artifactId>junit-platform-runner</artifactId>
+      <scope>test</scope>
+    </dependency>
+
     <dependency>
       <groupId>org.mockito</groupId>
       <artifactId>mockito-core</artifactId>

SingularityExecutor/src/test/java/com/hubspot/singularity/executor/config/SingularityExecutorConfigurationTest.java

@@ -11,7 +11,7 @@
 
 import javax.validation.Validator;
 
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.github.jknack.handlebars.Handlebars;

SingularityS3Base/pom.xml

@@ -91,14 +91,26 @@
 
     <!-- Unit test dependencies -->
     <dependency>
-      <groupId>junit</groupId>
-      <artifactId>junit</artifactId>
+      <groupId>org.assertj</groupId>
+      <artifactId>assertj-core</artifactId>
       <scope>test</scope>
     </dependency>
 
     <dependency>
-      <groupId>org.assertj</groupId>
-      <artifactId>assertj-core</artifactId>
+      <groupId>org.junit.jupiter</groupId>
+      <artifactId>junit-jupiter-api</artifactId>
+      <scope>test</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.junit.jupiter</groupId>
+      <artifactId>junit-jupiter-engine</artifactId>
+      <scope>test</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.junit.platform</groupId>
+      <artifactId>junit-platform-runner</artifactId>
       <scope>test</scope>
     </dependency>
 

SingularityS3Base/src/test/java/com/hubspot/singularity/s3/base/ArtifactManagerTest.java

@@ -1,4 +1,4 @@
-package com.hubspot.singularity.executor.config;
+package com.hubspot.singularity.s3.base;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.fail;
@@ -13,30 +13,22 @@
 import java.util.Arrays;
 import java.util.List;
 
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.ExpectedException;
-import org.junit.rules.TemporaryFolder;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
 import org.slf4j.LoggerFactory;
 
 import com.google.common.base.Throwables;
 import com.hubspot.singularity.runner.base.configuration.SingularityRunnerBaseConfiguration;
 import com.hubspot.singularity.runner.base.sentry.SingularityRunnerExceptionNotifier;
-import com.hubspot.singularity.s3.base.ArtifactManager;
 import com.hubspot.singularity.s3.base.config.SingularityS3Configuration;
 
 public class ArtifactManagerTest {
 
   private ArtifactManager artifactManager;
 
-  @Rule
-  public TemporaryFolder cacheDir = new TemporaryFolder();
+  private File cacheDir = com.google.common.io.Files.createTempDir();
 
-  @Rule
-  public ExpectedException expectedException = ExpectedException.none();
-
-  @Before
+  @BeforeEach
   public void setup() {
     SingularityRunnerBaseConfiguration baseConfig = new SingularityRunnerBaseConfiguration();
     SingularityS3Configuration s3Config = new SingularityS3Configuration();
@@ -57,13 +49,13 @@ public void itAbsorbsFileAlreadyExistsExceptionsWhenCopyingDuplicateFiles() {
 
     assertThat(originalPath.toFile()).hasContent(String.join(System.lineSeparator(), lines));
 
-    Path copyPath = Paths.get(cacheDir.getRoot().toString() + "/copy.txt");
+    Path copyPath = Paths.get(cacheDir.toString() + "/copy.txt");
     assertThat(copyPath).doesNotExist();
-    artifactManager.copy(originalPath, cacheDir.getRoot().toPath(), "copy.txt");
+    artifactManager.copy(originalPath, cacheDir.toPath(), "copy.txt");
     assertThat(copyPath).exists();
 
     // A redundant copy operation should not throw.
-    artifactManager.copy(originalPath, cacheDir.getRoot().toPath(), "copy.txt");
+    artifactManager.copy(originalPath, cacheDir.toPath(), "copy.txt");
   }
 
   @Test
@@ -82,7 +74,7 @@ public void itPropagatesFileAlreadyExistsExceptionsWhenCopyingNonDuplicateFiles(
 
   public Path write(String fileName, List<String> lines) {
     try {
-      File file = cacheDir.newFile(fileName);
+      File file = cacheDir.toPath().resolve(fileName).toFile();
       Path path = file.toPath();
       Files.write(path, lines, Charset.forName("UTF-8"));
       return path;