Merge pull request #2270 from HubSpot/parallel_persist

ssalinas · web-flow · commit 327959190645 · 2022-02-23T10:05:55.000-05:00
Run persisters in parallel
diff --git a/SingularityService/src/main/java/com/hubspot/singularity/config/SingularityConfiguration.java b/SingularityService/src/main/java/com/hubspot/singularity/config/SingularityConfiguration.java
@@ -261,6 +261,8 @@ public class SingularityConfiguration extends Configuration {
 
   private long persistHistoryEverySeconds = TimeUnit.MINUTES.toSeconds(2);
 
+  private int historyPollerConcurrency = 5;
+
   private int maxPendingImmediatePersists = 200;
 
   private long reconcileAgentsEveryMinutes = TimeUnit.HOURS.toMinutes(1);
@@ -2112,6 +2114,14 @@ public void setRequestCacheTtl(int requestCacheTtlInSeconds) {
     this.requestCacheTtlInSeconds = requestCacheTtlInSeconds;
   }
 
+  public int getHistoryPollerConcurrency() {
+    return historyPollerConcurrency;
+  }
+
+  public void setHistoryPollerConcurrency(int historyPollerConcurrency) {
+    this.historyPollerConcurrency = historyPollerConcurrency;
+  }
+
   public boolean skipPersistingTooLongTaskIds() {
     return skipPersistingTooLongTaskIds;
   }
diff --git a/SingularityService/src/main/java/com/hubspot/singularity/data/history/SingularityDeployHistoryPersister.java b/SingularityService/src/main/java/com/hubspot/singularity/data/history/SingularityDeployHistoryPersister.java
@@ -6,15 +6,19 @@
 import com.hubspot.singularity.SingularityDeleteResult;
 import com.hubspot.singularity.SingularityDeployHistory;
 import com.hubspot.singularity.SingularityDeployKey;
+import com.hubspot.singularity.SingularityManagedThreadPoolFactory;
 import com.hubspot.singularity.SingularityRequestDeployState;
+import com.hubspot.singularity.async.CompletableFutures;
 import com.hubspot.singularity.config.SingularityConfiguration;
 import com.hubspot.singularity.data.DeployManager;
 import com.hubspot.singularity.mesos.SingularitySchedulerLock;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
@@ -42,12 +46,13 @@ public SingularityDeployHistoryPersister(
     DeployManager deployManager,
     HistoryManager historyManager,
     SingularitySchedulerLock schedulerLock,
+    SingularityManagedThreadPoolFactory managedThreadPoolFactory,
     @Named(SingularityHistoryModule.PERSISTER_LOCK) ReentrantLock persisterLock,
     @Named(
       SingularityHistoryModule.LAST_DEPLOY_PERSISTER_SUCCESS
     ) AtomicLong lastPersisterSuccess
   ) {
-    super(configuration, persisterLock, lastPersisterSuccess);
+    super(configuration, persisterLock, lastPersisterSuccess, managedThreadPoolFactory);
     this.schedulerLock = schedulerLock;
     this.deployManager = deployManager;
     this.historyManager = historyManager;
@@ -72,75 +77,85 @@ public void runActionOnPoll() {
           Collectors.groupingBy(SingularityDeployKey::getRequestId, Collectors.toList())
         );
 
+      List<CompletableFuture<Void>> futures = new ArrayList<>();
       for (String requestId : deployManager
         .getAllRequestDeployStatesByRequestId()
         .keySet()) {
-        LOG.debug("Checking deploy histories to persist for request {}", requestId);
-        schedulerLock.runWithRequestLock(
-          () -> {
-            Optional<SingularityRequestDeployState> deployState = deployManager.getRequestDeployState(
-              requestId
-            );
-
-            if (!deployState.isPresent()) {
-              LOG.warn("No request deploy state for {}", requestId);
-              return;
-            }
-
-            int i = 0;
-            List<SingularityDeployHistory> deployHistories = allDeployIdsByRequest
-              .getOrDefault(requestId, Collections.emptyList())
-              .stream()
-              .map(
-                deployKey ->
-                  deployManager.getDeployHistory(
-                    deployKey.getRequestId(),
-                    deployKey.getDeployId(),
-                    true
-                  )
-              )
-              .filter(Optional::isPresent)
-              .map(Optional::get)
-              .sorted(
-                Comparator
-                  .comparingLong(
-                    SingularityDeployHistory::getCreateTimestampForCalculatingHistoryAge
-                  )
-                  .reversed()
-              )
-              .collect(Collectors.toList());
-
-            for (SingularityDeployHistory deployHistory : deployHistories) {
-              numTotal.increment();
-              if (
-                !shouldTransferDeploy(
-                  requestId,
-                  deployState.get(),
-                  deployHistory.getDeployMarker().getDeployId()
-                )
-              ) {
-                continue;
-              }
-
-              LOG.info(
-                "Persisting deploy {} for request {}",
-                deployHistory.getDeployMarker().getDeployId(),
-                requestId
+        futures.add(
+          CompletableFuture.runAsync(
+            () -> {
+              LOG.debug("Checking deploy histories to persist for request {}", requestId);
+              schedulerLock.runWithRequestLock(
+                () -> {
+                  Optional<SingularityRequestDeployState> deployState = deployManager.getRequestDeployState(
+                    requestId
+                  );
+
+                  if (!deployState.isPresent()) {
+                    LOG.warn("No request deploy state for {}", requestId);
+                    return;
+                  }
+
+                  int i = 0;
+                  List<SingularityDeployHistory> deployHistories = allDeployIdsByRequest
+                    .getOrDefault(requestId, Collections.emptyList())
+                    .stream()
+                    .map(
+                      deployKey ->
+                        deployManager.getDeployHistory(
+                          deployKey.getRequestId(),
+                          deployKey.getDeployId(),
+                          true
+                        )
+                    )
+                    .filter(Optional::isPresent)
+                    .map(Optional::get)
+                    .sorted(
+                      Comparator
+                        .comparingLong(
+                          SingularityDeployHistory::getCreateTimestampForCalculatingHistoryAge
+                        )
+                        .reversed()
+                    )
+                    .collect(Collectors.toList());
+
+                  for (SingularityDeployHistory deployHistory : deployHistories) {
+                    numTotal.increment();
+                    if (
+                      !shouldTransferDeploy(
+                        requestId,
+                        deployState.get(),
+                        deployHistory.getDeployMarker().getDeployId()
+                      )
+                    ) {
+                      continue;
+                    }
+
+                    LOG.info(
+                      "Persisting deploy {} for request {}",
+                      deployHistory.getDeployMarker().getDeployId(),
+                      requestId
+                    );
+                    if (moveToHistoryOrCheckForPurge(deployHistory, i++)) {
+                      numTransferred.increment();
+                    } else {
+                      LOG.error("Deploy History Persister failed on {}", deployHistory);
+                      persisterSuccess.getAndSet(false);
+                    }
+                  }
+                },
+                requestId,
+                getClass().getSimpleName(),
+                SingularitySchedulerLock.Priority.LOW
               );
-              if (moveToHistoryOrCheckForPurge(deployHistory, i++)) {
-                numTransferred.increment();
-              } else {
-                LOG.error("Deploy History Persister failed on {}", deployHistory);
-                persisterSuccess.getAndSet(false);
-              }
-            }
-          },
-          requestId,
-          getClass().getSimpleName(),
-          SingularitySchedulerLock.Priority.LOW
+            },
+            persisterExecutor
+          )
         );
       }
 
+      CompletableFutures.allOf(futures).join();
+
       LOG.info(
         "Transferred {} out of {} deploys in {}",
         numTransferred,
diff --git a/SingularityService/src/main/java/com/hubspot/singularity/data/history/SingularityHistoryPersister.java b/SingularityService/src/main/java/com/hubspot/singularity/data/history/SingularityHistoryPersister.java
@@ -3,9 +3,11 @@
 import com.hubspot.mesos.JavaUtils;
 import com.hubspot.singularity.SingularityDeleteResult;
 import com.hubspot.singularity.SingularityHistoryItem;
+import com.hubspot.singularity.SingularityManagedThreadPoolFactory;
 import com.hubspot.singularity.config.SingularityConfiguration;
 import com.hubspot.singularity.scheduler.SingularityLeaderOnlyPoller;
 import java.util.Optional;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.ReentrantLock;
@@ -20,18 +22,24 @@ public abstract class SingularityHistoryPersister<T extends SingularityHistoryIt
 
   protected final SingularityConfiguration configuration;
   protected final ReentrantLock persisterLock;
-
+  protected final ExecutorService persisterExecutor;
   protected final AtomicLong lastPersisterSuccess;
 
   public SingularityHistoryPersister(
     SingularityConfiguration configuration,
     ReentrantLock persisterLock,
-    AtomicLong lastPersisterSuccess
+    AtomicLong lastPersisterSuccess,
+    SingularityManagedThreadPoolFactory managedThreadPoolFactory
   ) {
     super(configuration.getPersistHistoryEverySeconds(), TimeUnit.SECONDS);
     this.configuration = configuration;
     this.persisterLock = persisterLock;
     this.lastPersisterSuccess = lastPersisterSuccess;
+    this.persisterExecutor =
+      managedThreadPoolFactory.get(
+        "persister",
+        configuration.getHistoryPollerConcurrency()
+      );
   }
 
   @Override
diff --git a/SingularityService/src/main/java/com/hubspot/singularity/data/history/SingularityRequestHistoryPersister.java b/SingularityService/src/main/java/com/hubspot/singularity/data/history/SingularityRequestHistoryPersister.java
@@ -6,7 +6,9 @@
 import com.hubspot.mesos.JavaUtils;
 import com.hubspot.singularity.SingularityDeleteResult;
 import com.hubspot.singularity.SingularityHistoryItem;
+import com.hubspot.singularity.SingularityManagedThreadPoolFactory;
 import com.hubspot.singularity.SingularityRequestHistory;
+import com.hubspot.singularity.async.CompletableFutures;
 import com.hubspot.singularity.config.SingularityConfiguration;
 import com.hubspot.singularity.data.RequestManager;
 import com.hubspot.singularity.data.history.SingularityRequestHistoryPersister.SingularityRequestHistoryParent;
@@ -16,12 +18,14 @@
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.ReentrantLock;
 import javax.inject.Singleton;
+import javax.ws.rs.HEAD;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -42,12 +46,13 @@ public SingularityRequestHistoryPersister(
     RequestManager requestManager,
     HistoryManager historyManager,
     SingularitySchedulerLock lock,
+    SingularityManagedThreadPoolFactory managedThreadPoolFactory,
     @Named(SingularityHistoryModule.PERSISTER_LOCK) ReentrantLock persisterLock,
     @Named(
       SingularityHistoryModule.LAST_REQUEST_PERSISTER_SUCCESS
     ) AtomicLong lastPersisterSuccess
   ) {
-    super(configuration, persisterLock, lastPersisterSuccess);
+    super(configuration, persisterLock, lastPersisterSuccess, managedThreadPoolFactory);
     this.requestManager = requestManager;
     this.historyManager = historyManager;
     this.lock = lock;
@@ -157,22 +162,38 @@ public void runActionOnPoll() {
       Collections.sort(requestHistoryParents, Collections.reverseOrder()); // createdAt descending
 
       AtomicInteger i = new AtomicInteger();
+      List<CompletableFuture<Void>> futures = new ArrayList<>();
       for (SingularityRequestHistoryParent requestHistoryParent : requestHistoryParents) {
-        lock.runWithRequestLock(
-          () -> {
-            if (moveToHistoryOrCheckForPurge(requestHistoryParent, i.getAndIncrement())) {
-              numHistoryTransferred.getAndAdd(requestHistoryParent.history.size());
-            } else {
-              LOG.error("Request History Persister failed on {}", requestHistoryParent);
-              persisterSuccess.getAndSet(false);
-            }
-          },
-          requestHistoryParent.requestId,
-          "request history purger",
-          SingularitySchedulerLock.Priority.LOW
+        futures.add(
+          CompletableFuture.runAsync(
+            () ->
+              lock.runWithRequestLock(
+                () -> {
+                  if (
+                    moveToHistoryOrCheckForPurge(
+                      requestHistoryParent,
+                      i.getAndIncrement()
+                    )
+                  ) {
+                    numHistoryTransferred.getAndAdd(requestHistoryParent.history.size());
+                  } else {
+                    LOG.error(
+                      "Request History Persister failed on {}",
+                      requestHistoryParent
+                    );
+                    persisterSuccess.getAndSet(false);
+                  }
+                },
+                requestHistoryParent.requestId,
+                "request history purger",
+                SingularitySchedulerLock.Priority.LOW
+              ),
+            persisterExecutor
+          )
         );
       }
 
+      CompletableFutures.allOf(futures).join();
       LOG.info(
         "Transferred {} history updates for {} requests in {}",
         numHistoryTransferred,
diff --git a/SingularityService/src/main/java/com/hubspot/singularity/data/history/SingularityTaskHistoryPersister.java b/SingularityService/src/main/java/com/hubspot/singularity/data/history/SingularityTaskHistoryPersister.java
