store stats in ZK, expose in separate endpoint

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

@@ -50,8 +50,6 @@ public class SingularityState {
 
   private final Optional<Boolean> authDatastoreHealthy;
 
-  private final Optional<SingularityTaskReconciliationStatistics> lastTaskReconciliationStatistics;
-
   @JsonCreator
   public SingularityState(@JsonProperty("activeTasks") int activeTasks, @JsonProperty("activeRequests") int activeRequests, @JsonProperty("cooldownRequests") int cooldownRequests,
       @JsonProperty("pausedRequests") int pausedRequests, @JsonProperty("scheduledTasks") int scheduledTasks, @JsonProperty("pendingRequests") int pendingRequests, @JsonProperty("lbCleanupTasks") int lbCleanupTasks,
@@ -61,8 +59,7 @@ public SingularityState(@JsonProperty("activeTasks") int activeTasks, @JsonPrope
       @JsonProperty("lateTasks") int lateTasks, @JsonProperty("futureTasks") int futureTasks, @JsonProperty("maxTaskLag") long maxTaskLag, @JsonProperty("generatedAt") long generatedAt,
       @JsonProperty("overProvisionedRequestIds") List<String> overProvisionedRequestIds, @JsonProperty("underProvisionedRequestIds") List<String> underProvisionedRequestIds,
       @JsonProperty("overProvisionedRequests") int overProvisionedRequests, @JsonProperty("underProvisionedRequests") int underProvisionedRequests, @JsonProperty("finishedRequests") int finishedRequests,
-      @JsonProperty("unknownRacks") int unknownRacks, @JsonProperty("unknownSlaves") int unknownSlaves, @JsonProperty("authDatastoreHealthy") Optional<Boolean> authDatastoreHealthy,
-      @JsonProperty("lastTaskReconciliationStatistics") Optional<SingularityTaskReconciliationStatistics> lastTaskReconciliationStatistics) {
+      @JsonProperty("unknownRacks") int unknownRacks, @JsonProperty("unknownSlaves") int unknownSlaves, @JsonProperty("authDatastoreHealthy") Optional<Boolean> authDatastoreHealthy) {
     this.activeTasks = activeTasks;
     this.activeRequests = activeRequests;
     this.pausedRequests = pausedRequests;
@@ -94,7 +91,6 @@ public SingularityState(@JsonProperty("activeTasks") int activeTasks, @JsonPrope
     this.overProvisionedRequestIds = overProvisionedRequestIds;
     this.underProvisionedRequestIds = underProvisionedRequestIds;
     this.authDatastoreHealthy = authDatastoreHealthy;
-    this.lastTaskReconciliationStatistics = lastTaskReconciliationStatistics;
   }
 
   public int getFinishedRequests() {
@@ -235,10 +231,6 @@ public Optional<Boolean> getAuthDatastoreHealthy() {
     return authDatastoreHealthy;
   }
 
-  public Optional<SingularityTaskReconciliationStatistics> getLastTaskReconciliationStatistics() {
-    return lastTaskReconciliationStatistics;
-  }
-
   @Override
   public String toString() {
     return "SingularityState [activeTasks=" + activeTasks + ", pausedRequests=" + pausedRequests + ", activeRequests=" + activeRequests + ", cooldownRequests=" + cooldownRequests + ", scheduledTasks=" + scheduledTasks
@@ -247,7 +239,7 @@ public String toString() {
         + activeSlaves + ", deadSlaves=" + deadSlaves + ", decommissioningSlaves=" + decommissioningSlaves + ", unknownSlaves=" + unknownSlaves + ", activeRacks=" + activeRacks + ", deadRacks="
         + deadRacks + ", decommissioningRacks=" + decommissioningRacks + ", unknownRacks=" + unknownRacks + ", oldestDeploy=" + oldestDeploy + ", numDeploys=" + numDeploys + ", generatedAt="
         + generatedAt + ", hostStates=" + hostStates + ", overProvisionedRequestIds=" + overProvisionedRequestIds + ", underProvisionedRequestIds=" + underProvisionedRequestIds
-        + ", overProvisionedRequests=" + overProvisionedRequests + ", underProvisionedRequests=" + underProvisionedRequests + ", authDatastoreHealthy=" + authDatastoreHealthy + ", lastTaskReconciliationStatistics=" + lastTaskReconciliationStatistics + "]";
+        + ", overProvisionedRequests=" + overProvisionedRequests + ", underProvisionedRequests=" + underProvisionedRequests + ", authDatastoreHealthy=" + authDatastoreHealthy + "]";
   }
 
 }

SingularityService/src/main/java/com/hubspot/singularity/data/StateManager.java

@@ -21,6 +21,7 @@
 import com.google.inject.Inject;
 import com.hubspot.mesos.CounterMap;
 import com.hubspot.mesos.JavaUtils;
+import com.hubspot.singularity.SingularityCreateResult;
 import com.hubspot.singularity.SingularityHostState;
 import com.hubspot.singularity.SingularityPendingDeploy;
 import com.hubspot.singularity.SingularityPendingTaskId;
@@ -44,6 +45,7 @@ public class StateManager extends CuratorManager {
 
   private static final String ROOT_PATH = "/hosts";
   private static final String STATE_PATH = "/STATE";
+  private static final String TASK_RECONCILIATION_STATISTICS_PATH = STATE_PATH + "/taskReconciliation";
 
   private final RequestManager requestManager;
   private final TaskManager taskManager;
@@ -54,12 +56,12 @@ public class StateManager extends CuratorManager {
   private final Transcoder<SingularityHostState> hostStateTranscoder;
   private final SingularityConfiguration singularityConfiguration;
   private final SingularityAuthDatastore authDatastore;
-  private final SingularityTaskReconciliation taskReconciliation;
+  private final Transcoder<SingularityTaskReconciliationStatistics> taskReconciliationStatisticsTranscoder;
 
   @Inject
   public StateManager(CuratorFramework curatorFramework, SingularityConfiguration configuration, MetricRegistry metricRegistry, RequestManager requestManager, TaskManager taskManager,
       DeployManager deployManager, SlaveManager slaveManager, RackManager rackManager, Transcoder<SingularityState> stateTranscoder, Transcoder<SingularityHostState> hostStateTranscoder,
-      SingularityConfiguration singularityConfiguration, SingularityAuthDatastore authDatastore, SingularityTaskReconciliation taskReconciliation) {
+      SingularityConfiguration singularityConfiguration, SingularityAuthDatastore authDatastore, Transcoder<SingularityTaskReconciliationStatistics> taskReconciliationStatisticsTranscoder) {
     super(curatorFramework, configuration, metricRegistry);
 
     this.requestManager = requestManager;
@@ -71,7 +73,15 @@ public StateManager(CuratorFramework curatorFramework, SingularityConfiguration
     this.deployManager = deployManager;
     this.singularityConfiguration = singularityConfiguration;
     this.authDatastore = authDatastore;
-    this.taskReconciliation = taskReconciliation;
+    this.taskReconciliationStatisticsTranscoder = taskReconciliationStatisticsTranscoder;
+  }
+
+  public SingularityCreateResult saveTaskReconciliationStatistics(SingularityTaskReconciliationStatistics taskReconciliationStatistics) {
+    return save(TASK_RECONCILIATION_STATISTICS_PATH, taskReconciliationStatistics, taskReconciliationStatisticsTranscoder);
+  }
+
+  public Optional<SingularityTaskReconciliationStatistics> getTaskReconciliationStatistics() {
+    return getData(TASK_RECONCILIATION_STATISTICS_PATH, taskReconciliationStatisticsTranscoder);
   }
 
   public void save(SingularityHostState hostState) throws InterruptedException {
@@ -295,11 +305,9 @@ public SingularityState generateState(boolean includeRequestIds) {
 
     final Optional<Boolean> authDatastoreHealthy = authDatastore.isHealthy();
 
-    final Optional<SingularityTaskReconciliationStatistics> lastTaskReconciliationStatistics = taskReconciliation.getLastTaskReconciliationStatistics();
-
     return new SingularityState(activeTasks, numActiveRequests, cooldownRequests, numPausedRequests, scheduledTasks, pendingRequests, lbCleanupTasks, lbCleanupRequests, cleaningRequests, activeSlaves,
         deadSlaves, decommissioningSlaves, activeRacks, deadRacks, decommissioningRacks, cleaningTasks, states, oldestDeploy, numDeploys, scheduledTasksInfo.getNumLateTasks(),
         scheduledTasksInfo.getNumFutureTasks(), scheduledTasksInfo.getMaxTaskLag(), System.currentTimeMillis(), includeRequestIds ? overProvisionedRequestIds : null,
-            includeRequestIds ? underProvisionedRequestIds : null, overProvisionedRequestIds.size(), underProvisionedRequestIds.size(), numFinishedRequests, unknownRacks, unknownSlaves, authDatastoreHealthy, lastTaskReconciliationStatistics);
+            includeRequestIds ? underProvisionedRequestIds : null, overProvisionedRequestIds.size(), underProvisionedRequestIds.size(), numFinishedRequests, unknownRacks, unknownSlaves, authDatastoreHealthy);
   }
 }

SingularityService/src/main/java/com/hubspot/singularity/data/transcoders/SingularityTranscoderModule.java

@@ -36,6 +36,7 @@
 import com.hubspot.singularity.SingularityTaskId;
 import com.hubspot.singularity.SingularityTaskDestroyFrameworkMessage;
 import com.hubspot.singularity.SingularityTaskMetadata;
+import com.hubspot.singularity.SingularityTaskReconciliationStatistics;
 import com.hubspot.singularity.SingularityTaskShellCommandRequest;
 import com.hubspot.singularity.SingularityTaskShellCommandUpdate;
 import com.hubspot.singularity.SingularityTaskStatusHolder;
@@ -82,6 +83,7 @@ public void configure(final Binder binder) {
     bindTranscoder(binder).asJson(SingularityExpiringScale.class);
     bindTranscoder(binder).asJson(SingularityExpiringSkipHealthchecks.class);
     bindTranscoder(binder).asJson(SingularityTaskDestroyFrameworkMessage.class);
+    bindTranscoder(binder).asJson(SingularityTaskReconciliationStatistics.class);
 
     bindTranscoder(binder).asCompressedJson(SingularityDeployHistory.class);
     bindTranscoder(binder).asCompressedJson(SingularityDeploy.class);

SingularityService/src/main/java/com/hubspot/singularity/resources/StateResource.java

@@ -8,9 +8,11 @@
 import javax.ws.rs.QueryParam;
 import javax.ws.rs.core.MediaType;
 
+import com.google.common.base.Optional;
 import com.google.inject.Inject;
 import com.hubspot.singularity.SingularityService;
 import com.hubspot.singularity.SingularityState;
+import com.hubspot.singularity.SingularityTaskReconciliationStatistics;
 import com.hubspot.singularity.data.StateManager;
 import com.wordnik.swagger.annotations.Api;
 import com.wordnik.swagger.annotations.ApiOperation;
@@ -47,4 +49,11 @@ public List<String> getUnderProvisionedRequestIds(@QueryParam("skipCache") boole
   public List<String> getOverProvisionedRequestIds(@QueryParam("skipCache") boolean skipCache) {
     return stateManager.getState(skipCache, true).getOverProvisionedRequestIds();
   }
+
+  @GET
+  @Path("/task-reconciliation")
+  @ApiOperation("Retrieve information about the most recent task reconciliation")
+  public Optional<SingularityTaskReconciliationStatistics> getTaskReconciliationStatistics() {
+    return stateManager.getTaskReconciliationStatistics();
+  }
 }

SingularityService/src/main/java/com/hubspot/singularity/scheduler/SingularityTaskReconciliation.java

@@ -7,7 +7,6 @@
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicReference;
 
 import javax.inject.Singleton;
 
@@ -33,6 +32,7 @@
 import com.hubspot.singularity.SingularityTaskReconciliationStatistics;
 import com.hubspot.singularity.SingularityTaskStatusHolder;
 import com.hubspot.singularity.config.SingularityConfiguration;
+import com.hubspot.singularity.data.StateManager;
 import com.hubspot.singularity.data.TaskManager;
 import com.hubspot.singularity.mesos.SchedulerDriverSupplier;
 import com.hubspot.singularity.sentry.SingularityExceptionNotifier;
@@ -50,18 +50,19 @@ public class SingularityTaskReconciliation {
   private final SingularityAbort abort;
   private final SingularityExceptionNotifier exceptionNotifier;
   private final SchedulerDriverSupplier schedulerDriverSupplier;
-
-  private final AtomicReference<SingularityTaskReconciliationStatistics> taskReconciliationStatistics;
+  private final StateManager stateManager;
 
   @Inject
   public SingularityTaskReconciliation(SingularityManagedScheduledExecutorServiceFactory executorServiceFactory,
       SingularityExceptionNotifier exceptionNotifier,
       TaskManager taskManager,
+      StateManager stateManager,
       SingularityConfiguration configuration,
       @Named(SingularityMainModule.SERVER_ID_PROPERTY) String serverId,
       SingularityAbort abort,
       SchedulerDriverSupplier schedulerDriverSupplier) {
     this.taskManager = taskManager;
+    this.stateManager = stateManager;
     this.serverId = serverId;
 
     this.exceptionNotifier = exceptionNotifier;
@@ -71,8 +72,6 @@ public SingularityTaskReconciliation(SingularityManagedScheduledExecutorServiceF
 
     this.isRunningReconciliation = new AtomicBoolean(false);
     this.executorService = executorServiceFactory.get(getClass().getSimpleName());
-
-    this.taskReconciliationStatistics = new AtomicReference<>();
   }
 
   enum ReconciliationState {
@@ -84,10 +83,6 @@ boolean isReconciliationRunning() {
     return isRunningReconciliation.get();
   }
 
-  public Optional<SingularityTaskReconciliationStatistics> getLastTaskReconciliationStatistics() {
-    return Optional.fromNullable(taskReconciliationStatistics.get());
-  }
-
   public ReconciliationState startReconciliation() {
     final long taskReconciliationStartedAt = System.currentTimeMillis();
 
@@ -165,7 +160,7 @@ private void checkReconciliation(final SchedulerDriver driver, final long reconc
     if (taskStatuses.isEmpty()) {
       LOG.info("Task reconciliation ended after {} checks and {}", numTimes, JavaUtils.duration(reconciliationStart));
 
-      taskReconciliationStatistics.set(new SingularityTaskReconciliationStatistics(reconciliationStart, System.currentTimeMillis() - reconciliationStart, numTimes, remainingTaskCounts));
+      stateManager.saveTaskReconciliationStatistics(new SingularityTaskReconciliationStatistics(reconciliationStart, System.currentTimeMillis() - reconciliationStart, numTimes, remainingTaskCounts));
 
       isRunningReconciliation.set(false);