Add maximum pending kubernetes pod count field (#11)

* Add pod state as part of kubernetes instance
* Do not allow creating kubernetes instance if the maximum pending pod limit has reached
* Add maximum pod count field in plugin settings metadata
* Add maximum pod count field in plugin settings view
* Use maximumPendingKubernetesPodCount value from the plugin settings
* Add test to verify the maximum pending pod count scenario

Author: GaneshSPatil

src/main/java/cd/go/contrib/elasticagent/KubernetesAgentInstances.java

@@ -29,6 +29,7 @@
 import java.util.Date;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Semaphore;
 
 import static cd.go.contrib.elasticagent.KubernetesPlugin.LOG;
 import static cd.go.contrib.elasticagent.executors.GetProfileMetadataExecutor.SPECIFIED_USING_POD_CONFIGURATION;
@@ -38,6 +39,8 @@ public class KubernetesAgentInstances implements AgentInstances<KubernetesInstan
     private final ConcurrentHashMap<String, KubernetesInstance> instances = new ConcurrentHashMap<>();
     public Clock clock = Clock.DEFAULT;
     private boolean refreshed;
+    final Semaphore semaphore = new Semaphore(0, true);
+
     private KubernetesClientFactory factory;
     private KubernetesInstanceFactory kubernetesInstanceFactory;
 
@@ -56,6 +59,26 @@ public KubernetesAgentInstances(KubernetesClientFactory factory, KubernetesInsta
 
     @Override
     public KubernetesInstance create(CreateAgentRequest request, PluginSettings settings, PluginRequest pluginRequest) throws Exception {
+        final Integer maxAllowedContainers = settings.getMaximumPendingAgentsCount();
+        synchronized (instances) {
+            doWithLockOnSemaphore(new SetupSemaphore(maxAllowedContainers, instances, semaphore));
+
+            if (semaphore.tryAcquire()) {
+                return createKubernetesInstance(request, settings, pluginRequest);
+            } else {
+                LOG.warn(String.format("The number of pending kubernetes pods is currently at the maximum permissible limit (%d). Total kubernetes pods (%d). Not creating any more containers.", maxAllowedContainers, instances.size()));
+                return null;
+            }
+        }
+    }
+
+    private void doWithLockOnSemaphore(Runnable runnable) {
+        synchronized (semaphore) {
+            runnable.run();
+        }
+    }
+
+    private KubernetesInstance createKubernetesInstance(CreateAgentRequest request, PluginSettings settings, PluginRequest pluginRequest) throws Exception {
         JobIdentifier jobIdentifier = request.jobIdentifier();
         if (isAgentCreatedForJob(jobIdentifier.getJobId())) {
             LOG.warn("[Create Agent Request] Request for creating an agent for Job Identifier [" + jobIdentifier + "] has already been scheduled. Skipping current request.");

src/main/java/cd/go/contrib/elasticagent/KubernetesInstance.java

@@ -28,13 +28,15 @@ public class KubernetesInstance {
     private final String name;
     private final Map<String, String> properties;
     private final Long jobId;
+    private final PodState state;
 
-    public KubernetesInstance(DateTime createdAt, String environment, String name, Map<String, String> properties, Long jobId) {
+    public KubernetesInstance(DateTime createdAt, String environment, String name, Map<String, String> properties, Long jobId, PodState state) {
         this.createdAt = createdAt.withZone(DateTimeZone.UTC);
         this.environment = environment;
         this.name = name;
         this.properties = properties;
         this.jobId = jobId;
+        this.state = state;
     }
 
     public void terminate(KubernetesClient client) {
@@ -60,4 +62,8 @@ public Map<String, String> getInstanceProperties() {
     public Long jobId() {
         return jobId;
     }
+
+    public boolean isPending() {
+        return this.state.equals(PodState.Pending);
+    }
 }

src/main/java/cd/go/contrib/elasticagent/KubernetesInstanceFactory.java

@@ -120,7 +120,7 @@ public KubernetesInstance fromKubernetesPod(Pod elasticAgentPod) {
             }
             String environment = metadata.getLabels().get(Constants.ENVIRONMENT_LABEL_KEY);
             Long jobId = Long.valueOf(metadata.getLabels().get(Constants.JOB_ID_LABEL_KEY));
-            kubernetesInstance = new KubernetesInstance(createdAt, environment, metadata.getName(), metadata.getAnnotations(), jobId);
+            kubernetesInstance = new KubernetesInstance(createdAt, environment, metadata.getName(), metadata.getAnnotations(), jobId, PodState.fromPod(elasticAgentPod));
         } catch (ParseException e) {
             throw new RuntimeException(e);
         }

src/main/java/cd/go/contrib/elasticagent/PluginSettings.java

@@ -31,6 +31,10 @@ public class PluginSettings {
     @SerializedName("auto_register_timeout")
     private String autoRegisterTimeout;
 
+    @Expose
+    @SerializedName("pending_pods_count")
+    private String pendingPodsCount;
+
     @Expose
     @SerializedName("kubernetes_cluster_url")
     private String kubernetesClusterUrl;
@@ -49,7 +53,6 @@ public class PluginSettings {
 
     private Period autoRegisterPeriod;
 
-
     public static PluginSettings fromJSON(String json) {
         return GSON.fromJson(json, PluginSettings.class);
     }
@@ -58,12 +61,13 @@ public static PluginSettings fromJSON(String json) {
     public boolean equals(Object o) {
         if (this == o) return true;
         if (!(o instanceof PluginSettings)) return false;
-
         PluginSettings that = (PluginSettings) o;
 
         if (goServerUrl != null ? !goServerUrl.equals(that.goServerUrl) : that.goServerUrl != null) return false;
         if (autoRegisterTimeout != null ? !autoRegisterTimeout.equals(that.autoRegisterTimeout) : that.autoRegisterTimeout != null)
             return false;
+        if (pendingPodsCount != null ? !pendingPodsCount.equals(that.pendingPodsCount) : that.pendingPodsCount != null)
+            return false;
         if (kubernetesClusterUrl != null ? !kubernetesClusterUrl.equals(that.kubernetesClusterUrl) : that.kubernetesClusterUrl != null)
             return false;
         return autoRegisterPeriod != null ? autoRegisterPeriod.equals(that.autoRegisterPeriod) : that.autoRegisterPeriod == null;
@@ -73,6 +77,7 @@ public boolean equals(Object o) {
     public int hashCode() {
         int result = goServerUrl != null ? goServerUrl.hashCode() : 0;
         result = 31 * result + (autoRegisterTimeout != null ? autoRegisterTimeout.hashCode() : 0);
+        result = 31 * result + (pendingPodsCount != null ? pendingPodsCount.hashCode() : 0);
         result = 31 * result + (kubernetesClusterUrl != null ? kubernetesClusterUrl.hashCode() : 0);
         result = 31 * result + (autoRegisterPeriod != null ? autoRegisterPeriod.hashCode() : 0);
         return result;
@@ -92,6 +97,14 @@ String getAutoRegisterTimeout() {
         return autoRegisterTimeout;
     }
 
+    public Integer getMaximumPendingAgentsCount() {
+        if (pendingPodsCount == null) {
+            pendingPodsCount = "10";
+        }
+
+        return Integer.valueOf(pendingPodsCount);
+    }
+
     public String getGoServerUrl() {
         return goServerUrl;
     }

src/main/java/cd/go/contrib/elasticagent/PodState.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2017 ThoughtWorks, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package cd.go.contrib.elasticagent;
+
+import io.fabric8.kubernetes.api.model.Pod;
+import io.fabric8.kubernetes.api.model.PodStatus;
+
+public enum PodState {
+    Running,
+    Pending;
+
+    public static PodState fromPod(Pod pod) {
+        PodStatus status = pod.getStatus();
+        if (status == null) {
+            return Pending;
+        }
+
+        return (status.getPhase() != null && status.getPhase().equals("Running")) ? Running : Pending;
+    }
+}

src/main/java/cd/go/contrib/elasticagent/SetupSemaphore.java

@@ -0,0 +1,64 @@
+/*
+ * Copyright 2017 ThoughtWorks, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package cd.go.contrib.elasticagent;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Semaphore;
+
+class SetupSemaphore implements Runnable {
+    private final Integer maxAllowedPendingPods;
+    private final Map<String, KubernetesInstance> instances;
+    private final Semaphore semaphore;
+
+    public SetupSemaphore(Integer maxAllowedPendingPods, Map<String, KubernetesInstance> instances, Semaphore semaphore) {
+        this.maxAllowedPendingPods = maxAllowedPendingPods;
+        this.instances = instances;
+        this.semaphore = semaphore;
+    }
+
+    @Override
+    public void run() {
+        List<KubernetesInstance> pendingInstances = getPendingInstances(instances);
+        int totalPendingPods = pendingInstances.size();
+        int availablePermits = maxAllowedPendingPods - totalPendingPods;
+
+        if (availablePermits <= 0) {
+            // no more capacity available.
+            semaphore.drainPermits();
+        } else {
+            int semaphoreValueDifference = availablePermits - semaphore.availablePermits();
+            if (semaphoreValueDifference > 0) {
+                semaphore.release(semaphoreValueDifference);
+            } else if (semaphoreValueDifference < 0) {
+                semaphore.tryAcquire(Math.abs(semaphoreValueDifference));
+            }
+        }
+    }
+
+    private List<KubernetesInstance> getPendingInstances(Map<String, KubernetesInstance> instances) {
+        ArrayList<KubernetesInstance> pendingInstances = new ArrayList<>();
+        for (KubernetesInstance kubernetesInstance : instances.values()) {
+            if (kubernetesInstance.isPending()) {
+                pendingInstances.add(kubernetesInstance);
+            }
+        }
+
+        return pendingInstances;
+    }
+}

src/main/java/cd/go/contrib/elasticagent/executors/GetPluginConfigurationExecutor.java

@@ -32,15 +32,17 @@
 public class GetPluginConfigurationExecutor implements RequestExecutor {
     public static final Field GO_SERVER_URL = new GoServerUrlField("go_server_url", "Go Server URL", false, "0");
     public static final Field AUTOREGISTER_TIMEOUT = new PositiveNumberField("auto_register_timeout", "Agent auto-register Timeout (in minutes)", "10", true, false, "1");
-    public static final Field KUBERNETES_CLUSTER_URL = new SecureURLField("kubernetes_cluster_url", "Kubernetes Cluster URL", true, "2");
-    public static final Field KUBERNETES_CLUSTER_USERNAME = new Field("kubernetes_cluster_username", "Kubernetes Cluster Username", null, false, false, "3");
-    public static final Field KUBERNETES_CLUSTER_PASSWORD = new Field("kubernetes_cluster_password", "Kubernetes Cluster Password", null, false, true, "4");
-    public static final Field KUBERNETES_CLUSTER_CA_CERT = new Field("kubernetes_cluster_ca_cert", "Kubernetes Cluster CA Certificate", null, false, true, "5");
+    public static final Field MAXIMUM_PENDING_PODS_COUNT = new PositiveNumberField("pending_pods_count", "Maximum Pending Kuberneted Pods Count", "10", true, false, "2");
+    public static final Field KUBERNETES_CLUSTER_URL = new SecureURLField("kubernetes_cluster_url", "Kubernetes Cluster URL", true, "3");
+    public static final Field KUBERNETES_CLUSTER_USERNAME = new Field("kubernetes_cluster_username", "Kubernetes Cluster Username", null, false, false, "4");
+    public static final Field KUBERNETES_CLUSTER_PASSWORD = new Field("kubernetes_cluster_password", "Kubernetes Cluster Password", null, false, true, "5");
+    public static final Field KUBERNETES_CLUSTER_CA_CERT = new Field("kubernetes_cluster_ca_cert", "Kubernetes Cluster CA Certificate", null, false, true, "6");
     public static final Map<String, Field> FIELDS = new LinkedHashMap<>();
 
     static {
         FIELDS.put(GO_SERVER_URL.key(), GO_SERVER_URL);
         FIELDS.put(AUTOREGISTER_TIMEOUT.key(), AUTOREGISTER_TIMEOUT);
+        FIELDS.put(MAXIMUM_PENDING_PODS_COUNT.key(), MAXIMUM_PENDING_PODS_COUNT);
 
         FIELDS.put(KUBERNETES_CLUSTER_URL.key(), KUBERNETES_CLUSTER_URL);
 

src/main/resources/plugin-settings.template.html

@@ -68,6 +68,12 @@
         <span class="form_error" ng-show="GOINPUTNAME[auto_register_timeout].$error.server">{{GOINPUTNAME[auto_register_timeout].$error.server}}</span>
     </div>
 
+    <div class="form_item_block">
+        <label>Maximum Pending Kubernetes Pods Count<span class='asterix'>*</span></label>
+        <input type="text" ng-model="pending_pods_count" ng-required="true"/>
+        <span class="form_error" ng-show="GOINPUTNAME[pending_pods_count].$error.server">{{GOINPUTNAME[pending_pods_count].$error.server}}</span>
+    </div>
+
     <div class="form_item_block">
         <label>Kubernetes Cluster URL:<span class='asterix'>*</span></label>
         <input type="text" ng-model="kubernetes_cluster_url" ng-required="true"/>
@@ -99,4 +105,4 @@
             Do not provide <code> -----BEGIN * </code> and <code> -----END * </code> in your certificate data.
         </label>
     </div>
-</div>
+</div>

src/test/java/cd/go/contrib/elasticagent/KubernetesAgentInstancesTest.java

@@ -19,7 +19,6 @@
 import cd.go.contrib.elasticagent.model.JobIdentifier;
 import cd.go.contrib.elasticagent.requests.CreateAgentRequest;
 import io.fabric8.kubernetes.client.KubernetesClient;
-import io.fabric8.kubernetes.client.dsl.internal.PodOperationsImpl;
 import org.joda.time.DateTime;
 import org.junit.Before;
 import org.junit.Test;
@@ -58,14 +57,15 @@ public void setUp() throws Exception {
         initMocks(this);
         testProperties = new HashMap<>();
         when(mockCreateAgentRequest.properties()).thenReturn(testProperties);
+        when(mockPluginSettings.getMaximumPendingAgentsCount()).thenReturn(10);
         when(factory.kubernetes(mockPluginSettings)).thenReturn(mockKubernetesClient);
         JobIdentifier jobId = new JobIdentifier("test", 1L, "Test pipeline", "test name", "1", "test job", 100L);
         when(mockCreateAgentRequest.jobIdentifier()).thenReturn(jobId);
     }
 
     @Test
     public void shouldCreateKubernetesPodUsingPodYamlAndCacheCreatedInstance() throws Exception {
-        KubernetesInstance kubernetesInstance = new KubernetesInstance(new DateTime(), "test", "test-agent", new HashMap<>(), 100L);
+        KubernetesInstance kubernetesInstance = new KubernetesInstance(new DateTime(), "test", "test-agent", new HashMap<>(), 100L, PodState.Running);
         when(mockKubernetesInstanceFactory.create(mockCreateAgentRequest, mockPluginSettings, mockKubernetesClient, mockPluginRequest, true)).
                 thenReturn(kubernetesInstance);
 
@@ -78,7 +78,7 @@ public void shouldCreateKubernetesPodUsingPodYamlAndCacheCreatedInstance() throw
 
     @Test
     public void shouldCreateKubernetesPodAndCacheCreatedInstance() throws Exception {
-        KubernetesInstance kubernetesInstance = new KubernetesInstance(new DateTime(), "test", "test-agent", new HashMap<>(), 100L);
+        KubernetesInstance kubernetesInstance = new KubernetesInstance(new DateTime(), "test", "test-agent", new HashMap<>(), 100L, PodState.Running);
         when(mockKubernetesInstanceFactory.create(mockCreateAgentRequest, mockPluginSettings, mockKubernetesClient, mockPluginRequest, false)).
                 thenReturn(kubernetesInstance);
         testProperties.put("SpecifiedUsingPodConfiguration", "false");
@@ -89,7 +89,7 @@ public void shouldCreateKubernetesPodAndCacheCreatedInstance() throws Exception
 
     @Test
     public void shouldNotCreatePodWhenOutstandingRequestsExistForJobs() throws Exception {
-        KubernetesInstance kubernetesInstance = new KubernetesInstance(new DateTime(), "test", "test-agent", new HashMap<>(), 100L);
+        KubernetesInstance kubernetesInstance = new KubernetesInstance(new DateTime(), "test", "test-agent", new HashMap<>(), 100L, PodState.Running);
         when(mockKubernetesInstanceFactory.create(mockCreateAgentRequest, mockPluginSettings, mockKubernetesClient, mockPluginRequest, false)).
                 thenReturn(kubernetesInstance);
         testProperties.put("SpecifiedUsingPodConfiguration", "false");
@@ -104,4 +104,28 @@ public void shouldNotCreatePodWhenOutstandingRequestsExistForJobs() throws Excep
         agentInstances.create(mockCreateAgentRequest, mockPluginSettings, mockPluginRequest);
         verify(mockKubernetesInstanceFactory, times(0)).create(any(), any(), any(), any(), any());
     }
-}
+
+    @Test
+    public void shouldNotCreatePodsWhenOutstandingLimitOfPendingKubernetesPodsHasReached() throws Exception {
+        //set maximum pending pod count to 1
+        when(mockPluginSettings.getMaximumPendingAgentsCount()).thenReturn(1);
+
+        //pending kubernetes pod
+        KubernetesInstance kubernetesInstance = new KubernetesInstance(new DateTime(), "test", "test-agent", new HashMap<>(), 100L, PodState.Pending);
+        when(mockKubernetesInstanceFactory.create(mockCreateAgentRequest, mockPluginSettings, mockKubernetesClient, mockPluginRequest, false)).
+                thenReturn(kubernetesInstance);
+        testProperties.put("SpecifiedUsingPodConfiguration", "false");
+
+        //first create agent request
+        KubernetesAgentInstances agentInstances = new KubernetesAgentInstances(factory, mockKubernetesInstanceFactory);
+        JobIdentifier jobId = new JobIdentifier("test", 1L, "Test pipeline", "test name", "1", "test job", 100L);
+        when(mockCreateAgentRequest.jobIdentifier()).thenReturn(jobId);
+        agentInstances.create(mockCreateAgentRequest, mockPluginSettings, mockPluginRequest);
+        verify(mockKubernetesInstanceFactory, times(1)).create(any(), any(), any(), any(), any());
+        reset(mockKubernetesInstanceFactory);
+
+        //second create agent request
+        agentInstances.create(mockCreateAgentRequest, mockPluginSettings, mockPluginRequest);
+        verify(mockKubernetesInstanceFactory, times(0)).create(any(), any(), any(), any(), any());
+    }
+}

src/test/java/cd/go/contrib/elasticagent/PluginSettingsTest.java

@@ -27,11 +27,13 @@ public void shouldDeserializeFromJSON() throws Exception {
         PluginSettings pluginSettings = PluginSettings.fromJSON("{" +
                 "\"go_server_url\": \"https://foo.go.cd/go\", " +
                 "\"auto_register_timeout\": \"10\", " +
+                "\"pending_pods_count\": \"10\", " +
                 "\"kubernetes_cluster_url\": \"https://cloud.example.com\" " +
                 "}");
 
         assertThat(pluginSettings.getGoServerUrl(), is("https://foo.go.cd/go"));
         assertThat(pluginSettings.getAutoRegisterTimeout(), is("10"));
+        assertThat(pluginSettings.getMaximumPendingAgentsCount(), is(10));
         assertThat(pluginSettings.getKubernetesClusterUrl(), is("https://cloud.example.com"));
     }
 }