package com.hubspot.singularity.scheduler;

import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;

import com.google.common.math.Stats;
import com.google.inject.Inject;
import com.google.inject.name.Named;
import com.hubspot.singularity.SingularityMainModule;
import com.hubspot.singularity.config.SingularityConfiguration;

public class SingularityStatusUpdateDeltaPoller extends SingularityLeaderOnlyPoller {
  private final SingularityConfiguration configuration;
  private final Map<String, Map<Long, Long>> statusUpdateDeltas;
  private final AtomicBoolean shortCircuitForStatusUpdateDelay;

  @Inject
  public SingularityStatusUpdateDeltaPoller(SingularityConfiguration configuration,
                                            @Named(SingularityMainModule.STATUS_UPDATE_DELTAS) Map<String, Map<Long, Long>> statusUpdateDeltas,
                                            @Named(SingularityMainModule.STATUS_UPDATE_SHORT_CIRCUIT) AtomicBoolean shortCircuitForStatusUpdateDelay) {
    super(5L, TimeUnit.SECONDS);
    this.configuration = configuration;
    this.statusUpdateDeltas = statusUpdateDeltas;
    this.shortCircuitForStatusUpdateDelay = shortCircuitForStatusUpdateDelay;
  }

  @Override
  public void runActionOnPoll() {
    long now = System.currentTimeMillis();
    Set<String> requestIds = statusUpdateDeltas.keySet();
    int overThreshold = 0;
    for (String requestId : requestIds) {
      Map<Long, Long> requestDeltas = statusUpdateDeltas.get(requestId);
      List<Long> toRemove = requestDeltas.keySet()
          .stream()
          .filter((e) -> e < now - 30000)
          .collect(Collectors.toList());
      toRemove.forEach(requestDeltas::remove);
      if (requestDeltas.size() > 0 && Stats.meanOf(requestDeltas.values()) > configuration.getDelayPollersWhenDeltaOverMs()) {
        overThreshold++;
      }
    }

    if ((overThreshold * 100.0 / requestIds.size()) > configuration.getDelayPollersWhenPercentOfRequestsOverUpdateDelta()) {
      shortCircuitForStatusUpdateDelay.set(true);
    } else {
      shortCircuitForStatusUpdateDelay.set(false);
    }
  }
}

@@ -86,30 +86,8 @@
 ap;%0A
-import java.util.Set;%0A
 impo
@@ -1398,137 +1398,40 @@
     
-Set%3CString%3E requestIds = statusUpdateDeltas.keySet();%0A    int overThreshold = 0;%0A    for (String requestId : requestIds) %7B%0A      
+int overThreshold = 0;%0A    for (
 Map%3C
@@ -1446,23 +1446,16 @@
 ng%3E 
-requestD
+d
 eltas 
-=
+:
  sta
@@ -1474,23 +1474,19 @@
 tas.
-get(requestId);
+values()) %7B
 %0A   
@@ -1510,24 +1510,17 @@
 emove = 
-requestD
+d
 eltas.ke
@@ -1652,24 +1652,17 @@
 forEach(
-requestD
+d
 eltas::r
@@ -1679,24 +1679,17 @@
     if (
-requestD
+d
 eltas.si
@@ -1717,16 +1717,9 @@
 nOf(
-requestD
+d
 elta
@@ -1859,18 +1859,35 @@
 0 / 
-requestIds
+statusUpdateDeltas.keySet()
 .siz

fix iterator