Ref #27090: parse k8s events + re-write retry to be faster

Author: LEDfan

src/main/java/eu/openanalytics/containerproxy/backend/docker/DockerSwarmBackend.java

@@ -182,7 +182,7 @@ protected Container startContainer(ContainerSpec spec, Proxy proxy) throws Excep
 		container.getParameters().put(PARAM_SERVICE_ID, serviceId);
 
 		// Give the service some time to start up and launch a container.
-		boolean containerFound = Retrying.retry(i -> {
+		boolean containerFound = Retrying.retry((i, maxAttempts) -> {
 			try {
 				Task serviceTask = dockerClient
 						.listTasks(Task.Criteria.builder().serviceName(serviceName).build())
@@ -194,7 +194,7 @@ protected Container startContainer(ContainerSpec spec, Proxy proxy) throws Excep
 				throw new RuntimeException("Failed to inspect swarm service tasks", e);
 			}
 			return (container.getId() != null);
-		}, 30, 2000, true);
+		}, 60_000, true);
 
 		if (!containerFound) {
 			dockerClient.removeService(serviceId);

src/main/java/eu/openanalytics/containerproxy/backend/kubernetes/KubernetesBackend.java

@@ -297,58 +297,30 @@ protected Container startContainer(ContainerSpec spec, Proxy proxy) throws Excep
 		Pod startedPod = kubeClient.pods().inNamespace(effectiveKubeNamespace).create(patchedPod);
 
 		int totalWaitMs = Integer.parseInt(environment.getProperty("proxy.kubernetes.pod-wait-time", "60000"));
-		int maxTries = totalWaitMs / 1000;
-		Retrying.retry(i -> {
-				if (!Readiness.getInstance().isReady(kubeClient.resource(startedPod).fromServer().get())) {
-					if (i > 1 && log != null) log.debug(String.format("Container not ready yet, trying again (%d/%d)", i, maxTries));
-					return false;
-				}
-				return true;
+		boolean podReady = Retrying.retry((currentAttempt, maxAttempts) -> {
+			if (!Readiness.getInstance().isReady(kubeClient.resource(startedPod).fromServer().get())) {
+				if (currentAttempt > 0 && log != null) log.debug(String.format("Container not ready yet, trying again (%d/%d)", currentAttempt, maxAttempts));
+				return false;
 			}
-		, maxTries, 1000);
-		if (!Readiness.getInstance().isReady(kubeClient.resource(startedPod).fromServer().get())) {
-			Pod pod = kubeClient.resource(startedPod).fromServer().get();
-			container.getParameters().put(PARAM_POD, pod);
-			proxy.getContainers().add(container);
-
-			proxyStatusService.containerStartFailed(proxy, container);
-			throw new ContainerProxyException("Container did not become ready in time");
-		}
-		proxyStatusService.containerStarted(proxy, container);
-		Pod pod = kubeClient.resource(startedPod).fromServer().get();
-
-		// TODO check k8s compatibility
-		List<Event> events = kubeClient.v1().events().withInvolvedObject(new ObjectReferenceBuilder()
-				.withKind("Pod")
-				.withName(pod.getMetadata().getName())
-				.withNamespace(pod.getMetadata().getNamespace())
-				.build()).list().getItems();
-
-		LocalDateTime pullingTime = null;
-		LocalDateTime pulledTime = null;
-		LocalDateTime scheduledTime = null;
-
-		for (Event event : events) {
-			if (event.getCount() != null && event.getCount() >  1) {
-				// ignore events which happened multiple time as we are unable to properly process them
-				continue;
-			}
-			if (event.getReason().equalsIgnoreCase("Pulling")) {
-				pullingTime = OffsetDateTime.parse(event.getLastTimestamp()).atZoneSameInstant(ZoneId.systemDefault()).toLocalDateTime();
-			} else if (event.getReason().equalsIgnoreCase("Pulled")) {
-				pulledTime = OffsetDateTime.parse(event.getLastTimestamp()).atZoneSameInstant(ZoneId.systemDefault()).toLocalDateTime();
-			} else if (event.getReason().equalsIgnoreCase("Scheduled")) {
-				scheduledTime = OffsetDateTime.parse(event.getEventTime().getTime()).atZoneSameInstant(ZoneId.systemDefault()).toLocalDateTime();
+			return true;
+		}, totalWaitMs);
+
+		if (!podReady) {
+			// check a final time whether the pod is ready
+			if (!Readiness.getInstance().isReady(kubeClient.resource(startedPod).fromServer().get())) {
+				Pod pod = kubeClient.resource(startedPod).fromServer().get();
+				container.getParameters().put(PARAM_POD, pod);
+				proxy.getContainers().add(container);
+
+				proxyStatusService.containerStartFailed(proxy, container);
+				throw new ContainerProxyException("Container did not become ready in time");
 			}
 		}
 
-		if (pullingTime != null && pulledTime != null) {
-			proxyStatusService.imagePulled(proxy, container, pullingTime, pulledTime);
-		}
+		proxyStatusService.containerStarted(proxy, container);
+		Pod pod = kubeClient.resource(startedPod).fromServer().get();
 
-		if (scheduledTime != null) {
-			proxyStatusService.containerScheduled(proxy, container,  scheduledTime);
-		}
+		parseKubernetesEvents(proxy, container, pod);
 
 		Service service = null;
 		if (isUseInternalNetwork()) {
@@ -374,7 +346,7 @@ protected Container startContainer(ContainerSpec spec, Proxy proxy) throws Excep
 						.build());
 
 			// Workaround: waitUntilReady appears to be buggy.
-			Retrying.retry(i -> isServiceReady(kubeClient.resource(startupService).fromServer().get()), 60, 1000);
+			Retrying.retry((currentAttempts, maxAttempts) -> isServiceReady(kubeClient.resource(startupService).fromServer().get()), 60_000);
 
 			service = kubeClient.resource(startupService).fromServer().get();
 		}
@@ -400,6 +372,56 @@ protected Container startContainer(ContainerSpec spec, Proxy proxy) throws Excep
 		return container;
 	}
 
+	private LocalDateTime getEventTime(Event event) {
+		if (event.getEventTime() != null && event.getEventTime().getTime() != null) {
+			return OffsetDateTime.parse(event.getEventTime().getTime()).atZoneSameInstant(ZoneId.systemDefault()).toLocalDateTime();
+		}
+
+		if (event.getFirstTimestamp() != null) {
+			return OffsetDateTime.parse(event.getFirstTimestamp()).atZoneSameInstant(ZoneId.systemDefault()).toLocalDateTime();
+		}
+
+		if (event.getLastTimestamp() != null) {
+			return OffsetDateTime.parse(event.getLastTimestamp()).atZoneSameInstant(ZoneId.systemDefault()).toLocalDateTime();
+		}
+
+		return null;
+	}
+
+	private void parseKubernetesEvents(Proxy proxy, Container container, Pod pod) {
+		List<Event> events = kubeClient.v1().events().withInvolvedObject(new ObjectReferenceBuilder()
+				.withKind("Pod")
+				.withName(pod.getMetadata().getName())
+				.withNamespace(pod.getMetadata().getNamespace())
+				.build()).list().getItems();
+
+		LocalDateTime pullingTime = null;
+		LocalDateTime pulledTime = null;
+		LocalDateTime scheduledTime = null;
+
+		for (Event event : events) {
+			if (event.getCount() != null && event.getCount() >  1) {
+				// ignore events which happened multiple time as we are unable to properly process them
+				continue;
+			}
+			if (event.getReason().equalsIgnoreCase("Pulling")) {
+				pullingTime = getEventTime(event);
+			} else if (event.getReason().equalsIgnoreCase("Pulled")) {
+				pulledTime = getEventTime(event);
+			} else if (event.getReason().equalsIgnoreCase("Scheduled")) {
+				scheduledTime = getEventTime(event);
+			}
+		}
+
+		if (pullingTime != null && pulledTime != null) {
+			proxyStatusService.imagePulled(proxy, container, pullingTime, pulledTime);
+		}
+
+		if (scheduledTime != null) {
+			proxyStatusService.containerScheduled(proxy, container,  scheduledTime);
+		}
+	}
+
 	private JsonPatch readPatchFromSpec(ContainerSpec containerSpec, Proxy proxy) throws JsonMappingException, JsonProcessingException {
 		String patchAsString = proxy.getSpec().getKubernetesPodPatch();
 		if (patchAsString == null) {

src/main/java/eu/openanalytics/containerproxy/model/runtime/ProxyStartupLog.java

@@ -1,3 +1,23 @@
+/**
+ * ContainerProxy
+ *
+ * Copyright (C) 2016-2021 Open Analytics
+ *
+ * ===========================================================================
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the Apache License as published by
+ * The Apache Software Foundation, either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * Apache License for more details.
+ *
+ * You should have received a copy of the Apache License
+ * along with this program.  If not, see <http://www.apache.org/licenses/>
+ */
 package eu.openanalytics.containerproxy.model.runtime;
 
 import java.time.Duration;

src/main/java/eu/openanalytics/containerproxy/service/AppRecoveryService.java

@@ -123,15 +123,14 @@ public void recoverRunningApps() throws Exception {
 				Proxy proxy = proxies.get(proxyId);
 				Container container = new Container();
 				container.setId(containerInfo.getContainerId());
-				// TODO
-//				container.setIndex(spec.getIndex());
 				container.setParameters(containerInfo.getParameters());
 				ContainerSpec containerSpec = proxy.getSpec().getContainerSpec(containerInfo.getImage());
 				if (containerSpec == null) {
 					log.warn(String.format("Found existing container (%s) but not corresponding container spec.", containerInfo.getContainerId()));
 					continue;
 				}
 				container.setSpec(containerSpec);
+				container.setIndex(containerSpec.getIndex());
 				proxy.addContainer(container);
 				proxy.setStatus(ProxyStatus.Up);
 

src/main/java/eu/openanalytics/containerproxy/service/ProxyStatusService.java

@@ -1,3 +1,23 @@
+/**
+ * ContainerProxy
+ *
+ * Copyright (C) 2016-2021 Open Analytics
+ *
+ * ===========================================================================
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the Apache License as published by
+ * The Apache Software Foundation, either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * Apache License for more details.
+ *
+ * You should have received a copy of the Apache License
+ * along with this program.  If not, see <http://www.apache.org/licenses/>
+ */
 package eu.openanalytics.containerproxy.service;
 
 import eu.openanalytics.containerproxy.model.runtime.Container;

src/main/java/eu/openanalytics/containerproxy/util/Retrying.java

@@ -20,32 +20,61 @@
  */
 package eu.openanalytics.containerproxy.util;
 
-import java.util.function.IntPredicate;
-
 public class Retrying {
 
-	public static boolean retry(IntPredicate job, int tries, int waitTime) {
-		return retry(job, tries, waitTime, false);
-	}
-	
-	public static boolean retry(IntPredicate job, int tries, int waitTime, boolean retryOnException) {
-		boolean retVal = false;
-		RuntimeException exception = null;
-		for (int currentTry = 1; currentTry <= tries; currentTry++) {
-			try {
-				if (job.test(currentTry)) {
-					retVal = true;
-					exception = null;
-					break;
-				}
-			} catch (RuntimeException e) {
-				if (retryOnException) exception = e;
-				else throw e;
-			}
-			try { Thread.sleep(waitTime); } catch (InterruptedException ignore) {}
-		}
-		if (exception == null) return retVal;
-		else throw exception;
-		
-	}
+    @FunctionalInterface
+    public interface Attempt {
+        boolean attempt(int currentAttempt, int maxAttempts);
+    }
+
+    public static boolean retry(Attempt job, int maxDelay) {
+        return retry(job, maxDelay, false);
+    }
+
+    public static boolean retry(Attempt job, int maxDelay, boolean retryOnException) {
+        boolean retVal = false;
+        RuntimeException exception = null;
+        int maxAttempts = numberOfAttempts(maxDelay);
+        for (int currentAttempt = 0; currentAttempt < maxAttempts; currentAttempt++) {
+            delay(currentAttempt); // delay here so that we don't delay for the last iteration
+            try {
+                if (job.attempt(currentAttempt, maxAttempts)) {
+                    retVal = true;
+                    exception = null;
+                    break;
+                }
+            } catch (RuntimeException e) {
+                if (retryOnException) exception = e;
+                else throw e;
+            }
+        }
+        if (exception == null) return retVal;
+        else throw exception;
+    }
+
+    public static void delay(Integer attempt) {
+        try {
+            if (attempt == 0) {
+            } else if (attempt <= 5) {
+                Thread.sleep(200);
+            } else if (attempt <= 10) {
+                Thread.sleep(400);
+            } else {
+                Thread.sleep(2_000);
+            }
+        } catch (InterruptedException ignore) {
+        }
+    }
+
+    public static int numberOfAttempts(Integer maxDelay) {
+        if (maxDelay < 2_000) {
+            throw new IllegalArgumentException("The maximum delay should at least be 2000ms");
+        }
+        // it takes 11 attempts to have a delay of 3 000ms
+        return (int) Math.ceil((maxDelay - 3_000) / 2_000.0) + 11;
+    }
+
 }
+
+
+

src/test/java/eu/openanalytics/containerproxy/test/unit/RetryingTest.java

@@ -0,0 +1,80 @@
+/**
+ * ContainerProxy
+ *
+ * Copyright (C) 2016-2021 Open Analytics
+ *
+ * ===========================================================================
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the Apache License as published by
+ * The Apache Software Foundation, either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * Apache License for more details.
+ *
+ * You should have received a copy of the Apache License
+ * along with this program.  If not, see <http://www.apache.org/licenses/>
+ */
+package eu.openanalytics.containerproxy.test.unit;
+
+import eu.openanalytics.containerproxy.ContainerProxyApplication;
+import eu.openanalytics.containerproxy.test.proxy.PropertyOverrideContextInitializer;
+import eu.openanalytics.containerproxy.test.proxy.TestProxyService;
+import eu.openanalytics.containerproxy.util.Retrying;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.ActiveProfiles;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit.jupiter.SpringExtension;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.concurrent.atomic.AtomicInteger;
+
+@SpringBootTest(classes = {TestProxyService.TestConfiguration.class, ContainerProxyApplication.class})
+@ExtendWith(SpringExtension.class)
+@ActiveProfiles("test")
+@ContextConfiguration(initializers = PropertyOverrideContextInitializer.class)
+public class RetryingTest {
+
+    @Test
+    public void testNumberOfAttempts() {
+        Assertions.assertEquals(Retrying.numberOfAttempts(3_000), 11);
+        Assertions.assertEquals(Retrying.numberOfAttempts(10_000), 15);
+        Assertions.assertEquals(Retrying.numberOfAttempts(60_000), 40); // compared to 30 when not using the faster checks
+    }
+
+    @Test
+    public void testFastDelay() {
+        Instant start = Instant.now();
+        AtomicInteger called = new AtomicInteger(0);
+        Retrying.retry((i, m) -> {
+            called.incrementAndGet();
+            return false;
+        }, 3_000);
+        long time = Duration.between(start, Instant.now()).toMillis();
+        Assertions.assertTrue(time > 3_000);
+        Assertions.assertTrue(time < 5_000);
+        Assertions.assertEquals(11, called.get());
+    }
+
+    @Test
+    public void testSlowDelay() {
+        Instant start = Instant.now();
+        AtomicInteger called = new AtomicInteger(0);
+        Retrying.retry((i, m) -> {
+            called.incrementAndGet();
+            return false;
+        }, 10_000);
+        long time = Duration.between(start, Instant.now()).toMillis();
+        Assertions.assertTrue(time > 10_000);
+        Assertions.assertTrue(time < 12_000);
+        Assertions.assertEquals(15, called.get());
+    }
+
+}