Add a new scoring mode accounting for max historical usage of all tasks

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

@@ -23,6 +23,8 @@ public class RequestUtilization {
   private long maxDiskBytesUsed = 0;
   private long minDiskBytesUsed = 0;
 
+  private double cpuBurstRating = 0;
+
   @JsonCreator
   public RequestUtilization(@JsonProperty("requestId") String requestId,
                             @JsonProperty("deployId") String deployId) {
@@ -40,6 +42,7 @@ public RequestUtilization addMemBytesReserved(long memBytes) {
     return this;
   }
 
+  // This is a running total, not the current usage
   public RequestUtilization addCpuUsed(double cpu) {
     this.cpuUsed += cpu;
     return this;
@@ -93,6 +96,11 @@ public int getNumTasks() {
     return numTasks;
   }
 
+  // 0 -> 1, where 0 is never over-utilized, or only short bursts and 1 is consistently overutilized
+  public double getCpuBurstRating() {
+    return cpuBurstRating;
+  }
+
   @JsonIgnore
   public double getAvgMemBytesUsed() {
     return memBytesUsed / (double) numTasks;
@@ -170,16 +178,30 @@ public RequestUtilization setMinDiskBytesUsed(long minDiskBytesUsed) {
     return this;
   }
 
+  public RequestUtilization setCpuBurstRating(double cpuBurstRating) {
+    this.cpuBurstRating = cpuBurstRating;
+    return this;
+  }
+
   @Override
   public String toString() {
     return "RequestUtilization{" +
-        "requestId=" + requestId +
-        ", deployId=" + deployId +
+        "requestId='" + requestId + '\'' +
+        ", deployId='" + deployId + '\'' +
         ", memBytesUsed=" + memBytesUsed +
         ", memBytesReserved=" + memBytesReserved +
         ", cpuUsed=" + cpuUsed +
         ", cpuReserved=" + cpuReserved +
+        ", diskBytesUsed=" + diskBytesUsed +
+        ", diskBytesReserved=" + diskBytesReserved +
         ", numTasks=" + numTasks +
+        ", maxMemBytesUsed=" + maxMemBytesUsed +
+        ", minMemBytesUsed=" + minMemBytesUsed +
+        ", maxCpuUsed=" + maxCpuUsed +
+        ", minCpuUsed=" + minCpuUsed +
+        ", maxDiskBytesUsed=" + maxDiskBytesUsed +
+        ", minDiskBytesUsed=" + minDiskBytesUsed +
+        ", cpuBurstRating=" + cpuBurstRating +
         '}';
   }
 }

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

@@ -12,7 +12,7 @@ public enum ResourceUsageType {
     CPU_USED, MEMORY_BYTES_USED, DISK_BYTES_USED, CPU_FREE, MEMORY_BYTES_FREE, DISK_BYTES_FREE
   }
 
-  public static final long BYTES_PER_MEGABYTE = 1000L * 1000L;
+  public static final long BYTES_PER_MEGABYTE = 1024L * 1024L;
 
   private final double cpusUsed;
   private final double cpusReserved;

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

@@ -1,5 +1,5 @@
 package com.hubspot.singularity;
 
 public enum  SingularityUsageScoringStrategy {
-  SPREAD_TASK_USAGE, SPREAD_SYSTEM_USAGE
+  SPREAD_TASK_USAGE, SPREAD_SYSTEM_USAGE, PROBABLE_MAX_USAGE
 }

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

@@ -59,6 +59,10 @@ public class MesosConfiguration {
   private int offersConcurrencyLimit = 100;
   private SingularityUsageScoringStrategy scoringStrategy = SingularityUsageScoringStrategy.SPREAD_TASK_USAGE;
   private MachineLoadMetric scoreUsingSystemLoad = MachineLoadMetric.LOAD_5;
+  private double longRunningFreeResourceWeight = 0.5;
+  private double longRunningUsedResourceWeight = 0.5;
+  private double nonLonRunningFreeResourceWeight = 0.75;
+  private double nonLongRunningUsedResourceWeight = 0.25;
 
   public int getMaxNumInstancesPerRequest() {
     return maxNumInstancesPerRequest;
@@ -291,4 +295,36 @@ public MachineLoadMetric getScoreUsingSystemLoad() {
   public void setScoreUsingSystemLoad(MachineLoadMetric scoreUsingSystemLoad) {
     this.scoreUsingSystemLoad = scoreUsingSystemLoad;
   }
+
+  public double getLongRunningFreeResourceWeight() {
+    return longRunningFreeResourceWeight;
+  }
+
+  public void setLongRunningFreeResourceWeight(double longRunningFreeResourceWeight) {
+    this.longRunningFreeResourceWeight = longRunningFreeResourceWeight;
+  }
+
+  public double getLongRunningUsedResourceWeight() {
+    return longRunningUsedResourceWeight;
+  }
+
+  public void setLongRunningUsedResourceWeight(double longRunningUsedResourceWeight) {
+    this.longRunningUsedResourceWeight = longRunningUsedResourceWeight;
+  }
+
+  public double getNonLonRunningFreeResourceWeight() {
+    return nonLonRunningFreeResourceWeight;
+  }
+
+  public void setNonLonRunningFreeResourceWeight(double nonLonRunningFreeResourceWeight) {
+    this.nonLonRunningFreeResourceWeight = nonLonRunningFreeResourceWeight;
+  }
+
+  public double getNonLongRunningUsedResourceWeight() {
+    return nonLongRunningUsedResourceWeight;
+  }
+
+  public void setNonLongRunningUsedResourceWeight(double nonLongRunningUsedResourceWeight) {
+    this.nonLongRunningUsedResourceWeight = nonLongRunningUsedResourceWeight;
+  }
 }

SingularityService/src/main/java/com/hubspot/singularity/mesos/SingularityMesosOfferScheduler.java

@@ -26,6 +26,7 @@
 import com.hubspot.singularity.RequestUtilization;
 import com.hubspot.singularity.SingularityDeployStatistics;
 import com.hubspot.singularity.SingularityPendingTaskId;
+import com.hubspot.singularity.SingularitySlaveUsage;
 import com.hubspot.singularity.SingularitySlaveUsage.ResourceUsageType;
 import com.hubspot.singularity.SingularitySlaveUsageWithId;
 import com.hubspot.singularity.SingularityTask;
@@ -42,6 +43,7 @@
 import com.hubspot.singularity.data.UsageManager;
 import com.hubspot.singularity.helpers.MesosUtils;
 import com.hubspot.singularity.helpers.SingularityMesosTaskHolder;
+import com.hubspot.singularity.mesos.SingularitySlaveUsageWithCalculatedScores.MaxProbableUsage;
 import com.hubspot.singularity.scheduler.SingularityLeaderCache;
 import com.hubspot.singularity.scheduler.SingularityScheduler;
 
@@ -56,6 +58,7 @@ public class SingularityMesosOfferScheduler {
   private final SingularityMesosTaskPrioritizer taskPrioritizer;
   private final SingularityScheduler scheduler;
   private final SingularityConfiguration configuration;
+  private final MesosConfiguration mesosConfiguration;
   private final SingularityMesosTaskBuilder mesosTaskBuilder;
   private final SingularitySlaveAndRackManager slaveAndRackManager;
   private final SingularitySlaveAndRackHelper slaveAndRackHelper;
@@ -94,6 +97,7 @@ public SingularityMesosOfferScheduler(MesosConfiguration mesosConfiguration,
     this.taskManager = taskManager;
     this.scheduler = scheduler;
     this.configuration = configuration;
+    this.mesosConfiguration = mesosConfiguration;
     this.mesosTaskBuilder = mesosTaskBuilder;
     this.slaveAndRackManager = slaveAndRackManager;
     this.taskSizeOptimizer = taskSizeOptimizer;
@@ -180,6 +184,9 @@ public Collection<SingularityOfferHolder> checkOffers(final Collection<Offer> of
     }
 
     final AtomicInteger tasksScheduled = new AtomicInteger(0);
+    Map<String, RequestUtilization> requestUtilizations = usageManager.getRequestUtilizations();
+    List<SingularityTaskId> activeTaskIds = taskManager.getActiveTaskIds();
+
     final Map<String, SingularitySlaveUsageWithCalculatedScores> currentSlaveUsagesBySlaveId = usageManager.getCurrentSlaveUsages(
         offerHolders.values()
             .stream()
@@ -188,14 +195,16 @@ public Collection<SingularityOfferHolder> checkOffers(final Collection<Offer> of
         .parallelStream()
         .collect(Collectors.toMap(
             SingularitySlaveUsageWithId::getSlaveId,
-            (usageWithId) -> new SingularitySlaveUsageWithCalculatedScores(usageWithId, configuration.getMesosConfiguration().getScoringStrategy(), configuration.getMesosConfiguration().getScoreUsingSystemLoad())
+            (usageWithId) -> new SingularitySlaveUsageWithCalculatedScores(
+                  usageWithId,
+                  configuration.getMesosConfiguration().getScoringStrategy(),
+                  configuration.getMesosConfiguration().getScoreUsingSystemLoad(),
+                  getMaxProbableUsageForSlave(activeTaskIds, requestUtilizations, offerHolders.get(usageWithId.getSlaveId()).getSanitizedHost())
+            )
         ));
 
     LOG.trace("Found slave usages {}", currentSlaveUsagesBySlaveId);
 
-    Map<String, RequestUtilization> requestUtilizations = usageManager.getRequestUtilizations();
-
-
     Map<String, Integer> tasksPerOfferHost = new ConcurrentHashMap<>();
 
     for (SingularityTaskRequestHolder taskRequestHolder : sortedTaskRequestHolders) {
@@ -233,6 +242,34 @@ public Collection<SingularityOfferHolder> checkOffers(final Collection<Offer> of
     return offerHolders.values();
   }
 
+  private MaxProbableUsage getMaxProbableUsageForSlave(List<SingularityTaskId> activeTaskIds, Map<String, RequestUtilization> requestUtilizations, String sanitizedHostname) {
+    double cpu = 0;
+    double memBytes = 0;
+    double diskBytes = 0;
+    for (SingularityTaskId taskId : activeTaskIds) {
+      if (taskId.getSanitizedHost().equals(sanitizedHostname)) {
+        if (requestUtilizations.containsKey(taskId.getRequestId())) {
+          RequestUtilization utilization = requestUtilizations.get(taskId.getRequestId());
+          // To account for cpu bursts, tend towards max usage if the app is consistently over-utilizing cpu, tend towards avg if it is over-utilized in short bursts
+          cpu += (utilization.getMaxCpuUsed() - utilization.getAvgCpuUsed()) * utilization.getCpuBurstRating() + utilization.getAvgCpuUsed();
+          memBytes += utilization.getMaxMemBytesUsed();
+          diskBytes += utilization.getMaxDiskBytesUsed();
+        } else {
+          Optional<SingularityTask> maybeTask = taskManager.getTask(taskId);
+          if (maybeTask.isPresent()) {
+            Resources resources = maybeTask.get().getTaskRequest().getPendingTask().getResources()
+                .or(maybeTask.get().getTaskRequest().getDeploy().getResources())
+                .or(defaultResources);
+            cpu += resources.getCpus();
+            memBytes += resources.getDiskMb() * SingularitySlaveUsage.BYTES_PER_MEGABYTE;
+            diskBytes += resources.getDiskMb() * SingularitySlaveUsage.BYTES_PER_MEGABYTE;
+          }
+        }
+      }
+    }
+    return new MaxProbableUsage(cpu, memBytes, diskBytes);
+  }
+
   private boolean isOfferFull(SingularityOfferHolder offerHolder) {
     return configuration.getMaxTasksPerOffer() > 0 && offerHolder.getAcceptedTasks().size() >= configuration.getMaxTasksPerOffer();
   }
@@ -339,15 +376,15 @@ private double scoreLongRunningTask(SingularitySlaveUsageWithCalculatedScores sl
         1 - slaveUsageWithScores.getLongRunningMemUsedScore(), slaveUsageWithScores.getMemFreeScore(),
         1 - slaveUsageWithScores.getLongRunningCpusUsedScore(), slaveUsageWithScores.getCpusFreeScore(),
         1 - slaveUsageWithScores.getLongRunningDiskUsedScore(), slaveUsageWithScores.getDiskFreeScore(),
-        0.50, 0.50);
+        mesosConfiguration.getLongRunningFreeResourceWeight(), mesosConfiguration.getLongRunningUsedResourceWeight());
   }
 
   private double scoreNonLongRunningTask(SingularityTaskRequest taskRequest, SingularitySlaveUsageWithCalculatedScores slaveUsageWithScores) {
     Optional<SingularityDeployStatistics> statistics = deployManager.getDeployStatistics(taskRequest.getRequest().getId(), taskRequest.getDeploy().getId());
     final double epsilon = 0.0001;
 
-    double freeResourceWeight = 0.75;
-    double usedResourceWeight = 0.25;
+    double freeResourceWeight = mesosConfiguration.getNonLonRunningFreeResourceWeight();
+    double usedResourceWeight = mesosConfiguration.getNonLongRunningUsedResourceWeight();
 
     if (statistics.isPresent() && statistics.get().getAverageRuntimeMillis().isPresent()) {
       final double maxNonLongRunningUsedResourceWeight = configuration.getMaxNonLongRunningUsedResourceWeight();

SingularityService/src/main/java/com/hubspot/singularity/mesos/SingularitySlaveUsageWithCalculatedScores.java

@@ -8,6 +8,7 @@
 class SingularitySlaveUsageWithCalculatedScores {
   private final SingularitySlaveUsage slaveUsage;
   private final MachineLoadMetric systemLoadMetric;
+  private final MaxProbableUsage maxProbableTaskUsage;
   private boolean missingUsageData;
   private double longRunningCpusUsedScore;
   private double longRunningMemUsedScore;
@@ -24,9 +25,10 @@ class SingularitySlaveUsageWithCalculatedScores {
   private double estimatedAddedMemoryBytesReserved = 0;
   private double estimatedAddedDiskBytesReserved = 0;
 
-  public SingularitySlaveUsageWithCalculatedScores(SingularitySlaveUsage slaveUsage, SingularityUsageScoringStrategy scoringStrategy, MachineLoadMetric systemLoadMetric) {
+  public SingularitySlaveUsageWithCalculatedScores(SingularitySlaveUsage slaveUsage, SingularityUsageScoringStrategy scoringStrategy, MachineLoadMetric systemLoadMetric, MaxProbableUsage maxProbableTaskUsage) {
     this.slaveUsage = slaveUsage;
     this.systemLoadMetric = systemLoadMetric;
+    this.maxProbableTaskUsage = maxProbableTaskUsage;
     if (missingUsageData(slaveUsage)) {
       this.missingUsageData = true;
       setScores(0, 0, 0, 0, 0, 0);
@@ -71,6 +73,15 @@ void setScores(SingularityUsageScoringStrategy scoringStrategy) {
         double diskFreeScore = 1 - ((slaveUsage.getDiskMbReserved() + (estimatedAddedDiskBytesReserved / SingularitySlaveUsage.BYTES_PER_MEGABYTE)) / slaveUsage.getDiskMbTotal().get());
         setScores(longRunningCpusUsedScore, longRunningMemUsedScore, longRunningDiskUsedScore, cpusFreeScore, memFreeScore, diskFreeScore);
         break;
+      case PROBABLE_MAX_USAGE:
+        double probableMaxCpuFreeScore = Math.max(
+            0,
+            1 - (getMaxProbableCpuWithEstimatedUsage() / slaveUsage.getSystemCpusTotal())
+        );
+        double probableMaxMemFreeScore = 1 - (getMaxProbableMemBytesWithEstimatedUsage() / slaveUsage.getSystemMemTotalBytes());
+        double probableMaxDiskFreeScore = 1 - (getMaxProbableDiskBytesWithEstimatedUsage() / slaveUsage.getSlaveDiskTotal());
+        setScores(longRunningCpusUsedScore, longRunningMemUsedScore, longRunningDiskUsedScore, probableMaxCpuFreeScore, probableMaxMemFreeScore, probableMaxDiskFreeScore);
+        break;
       case SPREAD_SYSTEM_USAGE:
       default:
         double systemCpuFreeScore = Math.max(0, 1 - ((getSystemLoadMetric() + estimatedAddedCpusUsage) / slaveUsage.getSystemCpusTotal()));
@@ -80,6 +91,18 @@ void setScores(SingularityUsageScoringStrategy scoringStrategy) {
     }
   }
 
+  private double getMaxProbableCpuWithEstimatedUsage() {
+    return Math.max(getSystemLoadMetric(), maxProbableTaskUsage.getCpu()) + estimatedAddedCpusUsage;
+  }
+
+  private double getMaxProbableMemBytesWithEstimatedUsage() {
+    return Math.max(slaveUsage.getSystemMemTotalBytes() - slaveUsage.getSystemMemFreeBytes(), maxProbableTaskUsage.getMemBytes()) + estimatedAddedMemoryBytesUsage;
+  }
+
+  private double getMaxProbableDiskBytesWithEstimatedUsage() {
+    return Math.max(slaveUsage.getSlaveDiskUsed(), maxProbableTaskUsage.getDiskBytes()) + estimatedAddedDiskBytesUsage;
+  }
+
   boolean isMissingUsageData() {
     return missingUsageData;
   }
@@ -148,6 +171,30 @@ private double getSystemLoadMetric() {
     }
   }
 
+  static class MaxProbableUsage {
+    private final double cpu;
+    private final double memBytes;
+    private final double diskBytes;
+
+    public MaxProbableUsage(double cpu, double memBytes, double diskBytes) {
+      this.cpu = cpu;
+      this.memBytes = memBytes;
+      this.diskBytes = diskBytes;
+    }
+
+    public double getCpu() {
+      return cpu;
+    }
+
+    public double getMemBytes() {
+      return memBytes;
+    }
+
+    public double getDiskBytes() {
+      return diskBytes;
+    }
+  }
+
   @Override
   public String toString() {
     return "SingularitySlaveUsageWithCalculatedScores{" +

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

@@ -1,6 +1,7 @@
 package com.hubspot.singularity.scheduler;
 
 import java.util.ArrayList;
+import java.util.Comparator;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -414,8 +415,11 @@ private void updateRequestUtilization(Map<String, RequestUtilization> utilizatio
     }
 
     List<SingularityTaskUsage> pastTaskUsagesCopy = copyUsages(pastTaskUsages, latestUsage, task);
+    pastTaskUsagesCopy.sort(Comparator.comparingDouble(SingularityTaskUsage::getTimestamp));
     int numTasks = pastTaskUsagesCopy.size() - 1;
 
+    int numCpuOverages = 0;
+
     for (int i = 0; i < numTasks; i++) {
       SingularityTaskUsage olderUsage = pastTaskUsagesCopy.get(i);
       SingularityTaskUsage newerUsage = pastTaskUsagesCopy.get(i + 1);
@@ -428,13 +432,19 @@ private void updateRequestUtilization(Map<String, RequestUtilization> utilizatio
       curMaxDiskBytesUsed = Math.max(newerUsage.getDiskTotalBytes(), curMaxDiskBytesUsed);
       curMinDiskBytesUsed = Math.min(newerUsage.getDiskTotalBytes(), curMinDiskBytesUsed);
 
+      if (cpusUsed > cpuReservedForTask) {
+        numCpuOverages ++;
+      }
+
       requestUtilization
           .addCpuUsed(cpusUsed)
           .addMemBytesUsed(newerUsage.getMemoryTotalBytes())
           .addDiskBytesUsed(newerUsage.getDiskTotalBytes())
           .incrementTaskCount();
     }
 
+    double cpuBurstRating = pastTaskUsagesCopy.size() > 0 ? numCpuOverages / (double) pastTaskUsagesCopy.size() : 1;
+
     requestUtilization
         .addMemBytesReserved((long) (memoryMbReservedForTask * SingularitySlaveUsage.BYTES_PER_MEGABYTE * numTasks))
         .addCpuReserved(cpuReservedForTask * numTasks)
@@ -444,7 +454,8 @@ private void updateRequestUtilization(Map<String, RequestUtilization> utilizatio
         .setMaxMemBytesUsed(curMaxMemBytesUsed)
         .setMinMemBytesUsed(curMinMemBytesUsed)
         .setMaxDiskBytesUsed(curMaxDiskBytesUsed)
-        .setMinDiskBytesUsed(curMinDiskBytesUsed);
+        .setMinDiskBytesUsed(curMinDiskBytesUsed)
+        .setCpuBurstRating(cpuBurstRating);
 
     utilizationPerRequestId.put(requestId, requestUtilization);
   }
@@ -548,7 +559,7 @@ private SingularityClusterUtilization getClusterUtilization(Map<String, RequestU
     }
 
     double avgUnderUtilizedCpu = numRequestsWithUnderUtilizedCpu != 0 ? totalUnderUtilizedCpu / numRequestsWithUnderUtilizedCpu : 0;
-    double avgOverUtilizedCpu = numRequestsWithOverUtilizedCpu != 0? totalOverUtilizedCpu / numRequestsWithOverUtilizedCpu : 0;
+    double avgOverUtilizedCpu = numRequestsWithOverUtilizedCpu != 0 ? totalOverUtilizedCpu / numRequestsWithOverUtilizedCpu : 0;
     long avgUnderUtilizedMemBytes = numRequestsWithUnderUtilizedMemBytes != 0 ? totalUnderUtilizedMemBytes / numRequestsWithUnderUtilizedMemBytes : 0;
     long avgUnderUtilizedDiskBytes = numRequestsWithUnderUtilizedDiskBytes != 0 ? totalUnderUtilizedDiskBytes / numRequestsWithUnderUtilizedDiskBytes : 0;