Refresh Kubernetes instances before creating agents (#13)

* Refresh KubernetesInstances on each create-agent call
* At the time of the pod creation, the state of pod will always be pending as it goes through a series of kubernetes queues before its running.
* Refresh the state of KubernetesInstance as the pod state has changed since it has been scheduled.
* Before each create-agent-request, refresh the state of all pods, and based on the refreshed pending/running state information, make a decision for scheduling a new pod.

* Use created Kubernetes Pod for KubernetesInstance creation instead of using the reference pod
* Once the pod has been created by the library, it returns the current state of the newly created pod, use newly created pod object for any future references

Author: GaneshSPatil

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

@@ -38,7 +38,6 @@
 public class KubernetesAgentInstances implements AgentInstances<KubernetesInstance> {
     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;
@@ -149,22 +148,17 @@ public Agents instancesCreatedAfterTimeout(PluginSettings settings, Agents agent
 
     @Override
     public void refreshAll(PluginRequest pluginRequest) throws Exception {
-        LOG.debug("Refreshing Elastic agents.");
+        LOG.debug("[Refresh Instances]. Syncing k8s elastic agent pod information");
         KubernetesClient client = factory.kubernetes(pluginRequest.getPluginSettings());
         PodList list = client.pods().inNamespace(Constants.KUBERNETES_NAMESPACE_KEY).list();
 
-        if (!refreshed) {
-            LOG.debug("Syncing k8s elastic agent pod information");
-            for (Pod pod : list.getItems()) {
-                Map<String, String> podLabels = pod.getMetadata().getLabels();
-                if (podLabels != null) {
-                    if (StringUtils.equals(Constants.KUBERNETES_POD_KIND_LABEL_VALUE, podLabels.get(Constants.KUBERNETES_POD_KIND_LABEL_KEY))) {
-                        register(kubernetesInstanceFactory.fromKubernetesPod(pod));
-                    }
+        for (Pod pod : list.getItems()) {
+            Map<String, String> podLabels = pod.getMetadata().getLabels();
+            if (podLabels != null) {
+                if (StringUtils.equals(Constants.KUBERNETES_POD_KIND_LABEL_VALUE, podLabels.get(Constants.KUBERNETES_POD_KIND_LABEL_KEY))) {
+                    register(kubernetesInstanceFactory.fromKubernetesPod(pod));
                 }
             }
-
-            refreshed = true;
         }
     }
 

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

@@ -112,8 +112,8 @@ private static void setAnnotations(Pod pod, CreateAgentRequest request) {
 
     private KubernetesInstance createKubernetesPod(KubernetesClient client, Pod elasticAgentPod) {
         LOG.info(String.format("[Create Agent] Creating K8s pod with spec:%s", elasticAgentPod.toString()));
-        client.pods().inNamespace(Constants.KUBERNETES_NAMESPACE_KEY).create(elasticAgentPod);
-        return fromKubernetesPod(elasticAgentPod);
+        Pod pod = client.pods().inNamespace(Constants.KUBERNETES_NAMESPACE_KEY).create(elasticAgentPod);
+        return fromKubernetesPod(pod);
     }
 
     public KubernetesInstance fromKubernetesPod(Pod elasticAgentPod) {

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

@@ -70,7 +70,6 @@ public GoPluginApiResponse handle(GoPluginApiRequest request) throws UnhandledRe
                     refreshInstances();
                     return CreateAgentRequest.fromJSON(request.requestBody()).executor(agentInstances, pluginRequest).execute();
                 case REQUEST_SHOULD_ASSIGN_WORK:
-                    refreshInstances();
                     return ShouldAssignWorkRequest.fromJSON(request.requestBody()).executor(agentInstances).execute();
                 case REQUEST_SERVER_PING:
                     refreshInstances();

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

@@ -17,18 +17,20 @@
 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) {
+        if (pod.getStatus() == null) {
             return Pending;
         }
 
-        return (status.getPhase() != null && status.getPhase().equals("Running")) ? Running : Pending;
+        if ("Running".equals(pod.getStatus().getPhase())) {
+            return Running;
+        }
+
+        return Pending;
     }
 }

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

@@ -20,12 +20,13 @@
 import com.google.gson.Gson;
 import io.fabric8.kubernetes.api.model.*;
 import io.fabric8.kubernetes.client.KubernetesClient;
-import io.fabric8.kubernetes.client.dsl.internal.NodeOperationsImpl;
 import io.fabric8.kubernetes.client.dsl.internal.PodOperationsImpl;
 import org.junit.Before;
 import org.junit.Test;
 import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
 
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -54,9 +55,6 @@ public class KubernetesAgentInstancesIntegrationTest {
     @Mock
     private KubernetesClient mockKubernetesClient;
 
-    @Mock
-    private NodeOperationsImpl nodes;
-
     @Mock
     private PodOperationsImpl pods;
 
@@ -67,6 +65,15 @@ public void setUp() throws Exception {
         when(mockedKubernetesClientFactory.kubernetes(any())).thenReturn(mockKubernetesClient);
 
         when(pods.inNamespace(Constants.KUBERNETES_NAMESPACE_KEY)).thenReturn(pods);
+
+        when(pods.create(any())).thenAnswer(new Answer<Pod>() {
+            @Override
+            public Pod answer(InvocationOnMock invocation) throws Throwable {
+                Object[] args = invocation.getArguments();
+                return (Pod) args[0];
+            }
+        });
+
         when(pods.list()).thenReturn(new PodList());
         when(mockKubernetesClient.pods()).thenReturn(pods);
 

src/test/java/cd/go/contrib/elasticagent/executors/ServerPingRequestExecutorTest.java

@@ -32,6 +32,8 @@
 import org.junit.Test;
 import org.mockito.ArgumentMatcher;
 import org.mockito.Mock;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
 
 import java.util.*;
 
@@ -62,12 +64,20 @@ public void setUp() throws Exception {
         when(mockedClient.pods()).thenReturn(mockedOperation);
         when(mockedOperation.inNamespace(Constants.KUBERNETES_NAMESPACE_KEY)).thenReturn(mockedNamespaceOperation);
 
-        when(mockedNamespaceOperation.create(any(Pod.class))).thenReturn(mockedPod);
+        when(mockedNamespaceOperation.create(any(Pod.class))).thenAnswer(new Answer<Pod>() {
+            @Override
+            public Pod answer(InvocationOnMock invocation) throws Throwable {
+                Object[] args = invocation.getArguments();
+                return (Pod) args[0];
+            }
+        });
+
         when(mockedNamespaceOperation.withName(anyString())).thenReturn(podResource);
         when(podResource.get()).thenReturn(mockedPod);
 
         objectMetadata = new ObjectMeta();
         objectMetadata.setCreationTimestamp(getSimpleDateFormat().format(new Date()));
+
         when(mockedPod.getMetadata()).thenReturn(objectMetadata);
     }
 

src/test/java/cd/go/contrib/elasticagent/executors/ShouldAssignWorkRequestExecutorTest.java

@@ -31,6 +31,8 @@
 import org.junit.Before;
 import org.junit.Test;
 import org.mockito.Mock;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -40,7 +42,6 @@
 import static org.hamcrest.Matchers.is;
 import static org.junit.Assert.assertThat;
 import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 import static org.mockito.MockitoAnnotations.initMocks;
 
@@ -64,7 +65,13 @@ public void setUp() throws Exception {
         when(factory.kubernetes(any(PluginSettings.class))).thenReturn(mockedClient);
         when(mockedClient.pods()).thenReturn(mockedOperation);
         when(mockedOperation.inNamespace(KUBERNETES_NAMESPACE_KEY)).thenReturn(mockedNamespaceOperation);
-        when(mockedNamespaceOperation.create(any(Pod.class))).thenReturn(mock(Pod.class));
+        when(mockedNamespaceOperation.create(any(Pod.class))).thenAnswer(new Answer<Pod>() {
+            @Override
+            public Pod answer(InvocationOnMock invocation) throws Throwable {
+                Object[] args = invocation.getArguments();
+                return (Pod) args[0];
+            }
+        });
 
         agentInstances = new KubernetesAgentInstances(factory);
         properties.put("foo", "bar");