Merge pull request #1028 from mbarbero/kubelimits-racecondition

JENKINS-66484 - Tentative fix of a race condition in KubernetesProvisioningLimits

Author: Vlatombe

src/main/java/org/csanchez/jenkins/plugins/kubernetes/KubernetesProvisioningLimits.java

@@ -1,6 +1,15 @@
 package org.csanchez.jenkins.plugins.kubernetes;
 
-import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import javax.annotation.Nonnull;
+
+import org.kohsuke.accmod.Restricted;
+import org.kohsuke.accmod.restrictions.NoExternalUse;
+
 import hudson.Extension;
 import hudson.ExtensionList;
 import hudson.init.InitMilestone;
@@ -10,16 +19,6 @@
 import jenkins.metrics.api.Metrics;
 import jenkins.model.Jenkins;
 import jenkins.model.NodeListener;
-import org.kohsuke.accmod.Restricted;
-import org.kohsuke.accmod.restrictions.NoExternalUse;
-
-import javax.annotation.Nonnull;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.logging.Level;
-import java.util.logging.Logger;
 
 /**
  * Implements provisioning limits for clouds and pod templates
@@ -31,26 +30,28 @@ public final class KubernetesProvisioningLimits {
     /**
      * Tracks current number of kubernetes agents per pod template
      */
-    private final Map<String, AtomicInteger> podTemplateCounts = Collections.synchronizedMap(new HashMap<>());
+    private final Map<String, Integer> podTemplateCounts = new HashMap<>();
 
     /**
      * Tracks current number of kubernetes agents per kubernetes cloud
      */
-    private final Map<String, AtomicInteger> cloudCounts = Collections.synchronizedMap(new HashMap<>());
+    private final Map<String, Integer> cloudCounts = new HashMap<>();
 
     @Initializer(after = InitMilestone.SYSTEM_CONFIG_LOADED)
     public static void init() {
         // We don't want anything to be provisioned while we do the initial count.
         Queue.withLock(() -> {
             final KubernetesProvisioningLimits instance = get();
-            Jenkins.get().getNodes()
-                    .stream()
-                    .filter(KubernetesSlave.class::isInstance)
-                    .map(KubernetesSlave.class::cast)
-                    .forEach(node -> {
-                instance.getGlobalCount(node.getCloudName()).addAndGet(node.getNumExecutors());
-                instance.getPodTemplateCount(node.getTemplateId()).addAndGet(node.getNumExecutors());
-            });
+            synchronized(instance) {
+                Jenkins.get().getNodes()
+                        .stream()
+                        .filter(KubernetesSlave.class::isInstance)
+                        .map(KubernetesSlave.class::cast)
+                        .forEach(node -> {
+                    instance.cloudCounts.put(node.getCloudName(), instance.getGlobalCount(node.getCloudName()) + node.getNumExecutors());
+                    instance.podTemplateCounts.put(node.getTemplateId(), instance.getPodTemplateCount(node.getTemplateId()) + node.getNumExecutors());
+                });
+            }
         });
     }
 
@@ -67,26 +68,24 @@ public static KubernetesProvisioningLimits get() {
      * @param podTemplate the pod template used to schedule the agent
      * @param numExecutors the number of executors (pretty much always 1)
      */
-    @SuppressFBWarnings(value="JLM_JSR166_UTILCONCURRENT_MONITORENTER", justification = "Trust me here")
-    public boolean register(@Nonnull KubernetesCloud cloud, @Nonnull PodTemplate podTemplate, int numExecutors) {
-        AtomicInteger globalCount = getGlobalCount(cloud.name);
-        AtomicInteger podTemplateCount = getPodTemplateCount(podTemplate.getId());
-        synchronized (globalCount) {
-            synchronized (podTemplateCount) {
-                if (globalCount.get() + numExecutors <= cloud.getContainerCap()) {
-                    if (podTemplateCount.get() + numExecutors <= podTemplate.getInstanceCap()) {
-                        int g = globalCount.addAndGet(numExecutors);
-                        int p = podTemplateCount.addAndGet(numExecutors);
-                        LOGGER.log(Level.FINEST, () -> cloud.name + " global limit: " + g + "/" + cloud.getContainerCap());
-                        LOGGER.log(Level.FINEST, () -> podTemplate.getName() + " template limit: " + p + "/" + podTemplate.getInstanceCap());
-                        return true;
-                    } else {
-                        Metrics.metricRegistry().counter(MetricNames.REACHED_POD_CAP).inc();
-                    }
-                } else {
-                    Metrics.metricRegistry().counter(MetricNames.REACHED_GLOBAL_CAP).inc();
-                }
+    public synchronized boolean register(@Nonnull KubernetesCloud cloud, @Nonnull PodTemplate podTemplate, int numExecutors) {
+        int newGlobalCount = getGlobalCount(cloud.name) + numExecutors;
+        if (newGlobalCount <= cloud.getContainerCap()) {
+            int newPodTemplateCount = getPodTemplateCount(podTemplate.getId()) + numExecutors;
+            if (newPodTemplateCount <= podTemplate.getInstanceCap()) {
+                cloudCounts.put(cloud.name, newGlobalCount);
+                LOGGER.log(Level.FINEST, () -> cloud.name + " global limit: " + newGlobalCount + "/" + cloud.getContainerCap());
+
+                podTemplateCounts.put(podTemplate.getId(), newPodTemplateCount);
+                LOGGER.log(Level.FINEST, () -> podTemplate.getName() + " template limit: " + newPodTemplateCount + "/" + podTemplate.getInstanceCap());
+                return true;
+            } else {
+                LOGGER.log(Level.FINEST, () -> podTemplate.getName() + " template limit reached: " + getPodTemplateCount(podTemplate.getId()) + "/" + podTemplate.getInstanceCap() + ". Cannot add " + numExecutors + " more!");
+                Metrics.metricRegistry().counter(MetricNames.REACHED_POD_CAP).inc();
             }
+        } else {
+            LOGGER.log(Level.FINEST, () -> cloud.name + " global limit reached: " + getGlobalCount(cloud.name) + "/" + cloud.getContainerCap() + ". Cannot add " + numExecutors + " more!");
+            Metrics.metricRegistry().counter(MetricNames.REACHED_GLOBAL_CAP).inc();
         }
         return false;
     }
@@ -97,40 +96,32 @@ public boolean register(@Nonnull KubernetesCloud cloud, @Nonnull PodTemplate pod
      * @param podTemplate the pod template used to schedule the agent
      * @param numExecutors the number of executors (pretty much always 1)
      */
-    @SuppressFBWarnings(value="JLM_JSR166_UTILCONCURRENT_MONITORENTER", justification = "Trust me here")
-    public void unregister(@Nonnull KubernetesCloud cloud, @Nonnull PodTemplate podTemplate, int numExecutors) {
-        AtomicInteger globalCount = getGlobalCount(cloud.name);
-        AtomicInteger podTemplateCount = getPodTemplateCount(podTemplate.getId());
-        synchronized (globalCount) {
-            synchronized (podTemplateCount) {
-                int newGlobalCount = globalCount.addAndGet(numExecutors * -1);
-                int newPodTemplateCount = podTemplateCount.addAndGet(numExecutors * -1);
-                if (newGlobalCount < 0) {
-                    LOGGER.log(Level.WARNING, "Global count for " + cloud.name + " went below zero. There is likely a bug in kubernetes-plugin");
-                    globalCount.set(0);
-                } else {
-                    LOGGER.log(Level.FINEST, () -> cloud.name + " global limit: " + newGlobalCount + "/" + cloud.getContainerCap());
-                }
-                if (newPodTemplateCount < 0) {
-                    LOGGER.log(Level.WARNING, "Pod template count for " + podTemplate.getId() + " went below zero. There is likely a bug in kubernetes-plugin");
-                    podTemplateCount.set(0);
-                } else {
-                    LOGGER.log(Level.FINEST, () -> podTemplate.getName() + " template limit: " + newPodTemplateCount + "/" + podTemplate.getInstanceCap());
-                }
-            }
+    public synchronized void unregister(@Nonnull KubernetesCloud cloud, @Nonnull PodTemplate podTemplate, int numExecutors) {
+        int newGlobalCount = getGlobalCount(cloud.name) - numExecutors;
+        if (newGlobalCount < 0) {
+            LOGGER.log(Level.WARNING, "Global count for " + cloud.name + " went below zero. There is likely a bug in kubernetes-plugin");
+        }
+        cloudCounts.put(cloud.name, Math.max(0, newGlobalCount));
+        LOGGER.log(Level.FINEST, () -> cloud.name + " global limit: " + Math.max(0, newGlobalCount) + "/" + cloud.getContainerCap());
+
+        int newPodTemplateCount = getPodTemplateCount(podTemplate.getId()) - numExecutors;
+        if (newPodTemplateCount < 0) {
+            LOGGER.log(Level.WARNING, "Pod template count for " + podTemplate.getName() + " went below zero. There is likely a bug in kubernetes-plugin");
         }
+        podTemplateCounts.put(podTemplate.getId(), Math.max(0, newPodTemplateCount));
+        LOGGER.log(Level.FINEST, () -> podTemplate.getName() + " template limit: " + Math.max(0, newPodTemplateCount) + "/" + podTemplate.getInstanceCap());
     }
 
     @Nonnull
     @Restricted(NoExternalUse.class)
-    AtomicInteger getGlobalCount(String name) {
-        return cloudCounts.computeIfAbsent(name, k -> new AtomicInteger());
+    int getGlobalCount(String cloudName) {
+        return cloudCounts.getOrDefault(cloudName, 0);
     }
 
     @Nonnull
     @Restricted(NoExternalUse.class)
-    AtomicInteger getPodTemplateCount(String id) {
-        return podTemplateCounts.computeIfAbsent(id, k -> new AtomicInteger());
+    int getPodTemplateCount(String podTemplate) {
+        return podTemplateCounts.getOrDefault(podTemplate, 0);
     }
 
     @Extension

src/test/java/org/csanchez/jenkins/plugins/kubernetes/KubernetesProvisioningLimitsTest.java

@@ -1,21 +1,24 @@
 package org.csanchez.jenkins.plugins.kubernetes;
 
-import org.junit.Rule;
-import org.junit.Test;
-import org.jvnet.hudson.test.JenkinsRule;
-import org.jvnet.hudson.test.LoggerRule;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
 
-import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.Callable;
+import java.util.concurrent.CompletionService;
+import java.util.concurrent.ExecutorCompletionService;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
 import java.util.logging.Level;
 
-import static org.junit.Assert.assertEquals;
+import org.junit.Rule;
+import org.junit.Test;
+import org.jvnet.hudson.test.JenkinsRule;
+import org.jvnet.hudson.test.LoggerRule;
 
 public class KubernetesProvisioningLimitsTest {
+
     @Rule
     public JenkinsRule j = new JenkinsRule();
 
@@ -25,40 +28,63 @@ public class KubernetesProvisioningLimitsTest {
     @Test
     public void lotsOfCloudsAndTemplates() throws InterruptedException {
         ThreadLocalRandom testRandom = ThreadLocalRandom.current();
-        for (int i = 1; i <= 10; i++) {
+        for (int i = 1; i < 4; i++) {
             KubernetesCloud cloud = new KubernetesCloud("kubernetes-" + i);
-            cloud.setContainerCap(testRandom.nextInt(40)+1);
-            for (int j = 1; j < 10; j++) {
+            cloud.setContainerCap(testRandom.nextInt(4)+1);
+            for (int j = 1; j < 4; j++) {
                 PodTemplate pt = new PodTemplate();
+                pt.setName(cloud.name + "-podTemplate-" + j);
                 pt.setInstanceCap(testRandom.nextInt(4)+1);
                 cloud.addTemplate(pt);
             }
             j.jenkins.clouds.add(cloud);
         }
-        ExecutorService EXEC = Executors.newFixedThreadPool(100);
-        List<Callable<Void>> tasks = new ArrayList<>();
+
+        ExecutorService threadPool = Executors.newCachedThreadPool();
+        System.out.println(threadPool.getClass().getName());
+        CompletionService<Void> ecs = new ExecutorCompletionService<>(threadPool);
+        KubernetesProvisioningLimits kubernetesProvisioningLimits = KubernetesProvisioningLimits.get();
+
+        List<KubernetesCloud> clouds = j.jenkins.clouds.getAll(KubernetesCloud.class);
         for (int k = 0 ; k < 1000; k++) {
-            Callable<Void> c = () -> {
+            ecs.submit(() -> {
                 ThreadLocalRandom random = ThreadLocalRandom.current();
-                List<KubernetesCloud> clouds = j.jenkins.clouds.getAll(KubernetesCloud.class);
                 KubernetesCloud cloud = clouds.get(random.nextInt(clouds.size()));
                 List<PodTemplate> templates = cloud.getTemplates();
                 PodTemplate podTemplate = templates.get(random.nextInt(templates.size()));
-                if (KubernetesProvisioningLimits.get().register(cloud, podTemplate, 1)) {
-                    Thread.sleep(random.nextInt(3) * 1000);
-                    KubernetesProvisioningLimits.get().unregister(cloud, podTemplate, 1);
+                while (!kubernetesProvisioningLimits.register(cloud, podTemplate, 1)) {
+                    try {
+                        Thread.sleep(8);
+                    } catch(InterruptedException e) {
+                        Thread.currentThread().interrupt();
+                        break;
+                    }
                 }
-                return null;
-            };
-            tasks.add(c);
+
+                ecs.submit(() -> {
+                    kubernetesProvisioningLimits.unregister(cloud, podTemplate, 1);
+                }, null);
+            }, null);
         }
-        EXEC.invokeAll(tasks);
+
+        while (ecs.poll(20, TimeUnit.SECONDS) != null) {
+            try {
+                Thread.sleep(8);
+            } catch(InterruptedException e) {
+                Thread.currentThread().interrupt();
+                break;
+            }
+        }
+
+        threadPool.shutdown();
+        assertTrue(threadPool.awaitTermination(60, TimeUnit.SECONDS));
+        assertTrue(threadPool.shutdownNow().size() == 0);
 
         // Check that every count is back to 0
         for (KubernetesCloud cloud : j.jenkins.clouds.getAll(KubernetesCloud.class)) {
-            assertEquals(0, KubernetesProvisioningLimits.get().getGlobalCount(cloud.name).get());
+            assertEquals(0, KubernetesProvisioningLimits.get().getGlobalCount(cloud.name));
             for (PodTemplate template : cloud.getTemplates()) {
-                assertEquals(0, KubernetesProvisioningLimits.get().getPodTemplateCount(template.getId()).get());
+                assertEquals(0, KubernetesProvisioningLimits.get().getPodTemplateCount(template.getId()));
             }
         }
     }