feat: run tests in parallel

core/src/main/java/io/javaoperatorsdk/jenvtest/KubeAPIServerConfigBuilder.java

@@ -67,7 +67,7 @@ public KubeAPIServerConfig build() {
       }
     }
     if (waitForEtcdHealthCheckOnStartup == null) {
-      var waitForEtcdHealthCheckOnStartup = System.getenv(JENVTEST_API_SERVER_VERSION_ENV_VAR);
+      var waitForEtcdHealthCheckOnStartup = System.getenv(JENVTEST_WAIT_FOR_ETCD_HEALTH_CHECK);
       if (waitForEtcdHealthCheckOnStartup != null) {
         this.waitForEtcdHealthCheckOnStartup =
             Boolean.parseBoolean(waitForEtcdHealthCheckOnStartup);

core/src/main/java/io/javaoperatorsdk/jenvtest/binary/BinaryDownloader.java

@@ -17,6 +17,7 @@
 import io.javaoperatorsdk.jenvtest.JenvtestException;
 import io.javaoperatorsdk.jenvtest.Utils;
 import io.javaoperatorsdk.jenvtest.binary.repo.BinaryRepo;
+import io.javaoperatorsdk.jenvtest.lock.LockFile;
 
 public class BinaryDownloader {
 
@@ -32,7 +33,6 @@ public BinaryDownloader(String jenvtestDir, OSInfo osInfoProvider) {
     this.jenvtestDir = jenvtestDir;
     this.osInfoProvider = osInfoProvider;
     this.binaryRepo = new BinaryRepo(osInfoProvider);
-
   }
 
   BinaryDownloader(String jenvtestDir, BinaryRepo binaryRepo, OSInfo osInfoProvider) {
@@ -43,14 +43,30 @@ public BinaryDownloader(String jenvtestDir, OSInfo osInfoProvider) {
 
   public File download(String version) {
     log.info("Downloading binaries with version: {}", version);
-    var tempFile = binaryRepo.downloadVersionToTempFile(version);
-    File dir = createDirForBinaries(version);
-    extractFiles(tempFile, dir);
-    var deleted = tempFile.delete();
-    if (!deleted) {
-      log.warn("Unable to delete temp file: {}", tempFile.getPath());
+    var downloadDir = new File(jenvtestDir, BinaryManager.BINARY_LIST_DIR);
+    downloadDir.mkdirs();
+    LockFile lock =
+        new LockFile(version + ".lock", downloadDir.getPath());
+    var dirForVersion = dirForVersion(version);
+    if (lock.tryLock()) {
+      if (dirForVersion.exists()) {
+        return dirForVersion;
+      }
+      var tempFile = binaryRepo.downloadVersionToTempFile(version);
+      File dir = createDirForBinaries(version);
+      extractFiles(tempFile, dir);
+      var deleted = tempFile.delete();
+      if (!deleted) {
+        log.warn("Unable to delete temp file: {}", tempFile.getPath());
+      }
+      lock.releaseLock();
+      return dir;
+    } else {
+      log.debug("Waiting for lock to be deleted for version: {}", version);
+      lock.waitUntilLockDeleted();
+      log.debug("Lock deleted for version: {}", version);
+      return dirForVersion;
     }
-    return dir;
   }
 
   public File downloadLatest() {
@@ -99,14 +115,18 @@ private File extractEntry(TarArchiveEntry entry, File dir, TarArchiveInputStream
   }
 
   private File createDirForBinaries(String version) {
-    var dir = new File(jenvtestDir, BinaryManager.BINARY_LIST_DIR + File.separator
-        + version + Utils.platformSuffix(osInfoProvider));
+    var dir = dirForVersion(version);
     if (!dir.mkdirs()) {
       throw new JenvtestException("Cannot created director: " + dir.getPath());
     }
     return dir;
   }
 
+  private File dirForVersion(String version) {
+    return new File(jenvtestDir, BinaryManager.BINARY_LIST_DIR + File.separator
+        + version + Utils.platformSuffix(osInfoProvider));
+  }
+
   public String findLatestVersion() {
     var allRelevantVersions =
         listAllRelevantVersions().sorted(Utils.SEMVER_COMPARATOR).collect(Collectors.toList());

core/src/main/java/io/javaoperatorsdk/jenvtest/cert/CertManager.java

@@ -28,6 +28,7 @@
 import org.slf4j.LoggerFactory;
 
 import io.javaoperatorsdk.jenvtest.JenvtestException;
+import io.javaoperatorsdk.jenvtest.lock.LockFile;
 
 public class CertManager {
 
@@ -46,16 +47,36 @@ public CertManager(String jenvtestDir) {
   }
 
   public void createCertificatesIfNeeded() {
-    generateAPIServerCertificates();
-    generateUserCertificates();
+    if (certFilesPresent()) {
+      return;
+    }
+    // locking is for parallel execution
+    LockFile lockFile = new LockFile("cert.lock", jenvtestDir);
+    if (lockFile.tryLock()) {
+      if (certFilesPresent()) {
+        return;
+      }
+      try {
+        generateAPIServerCertificates();
+        generateUserCertificates();
+      } finally {
+        lockFile.releaseLock();
+      }
+    } else {
+      lockFile.waitUntilLockDeleted();
+    }
+  }
+
+  private boolean certFilesPresent() {
+    var apiCert = new File(jenvtestDir, API_SERVER_CERT_NAME);
+    var apiKey = new File(jenvtestDir, API_SERVER_KEY_NAME);
+    var clientCert = new File(jenvtestDir, CLIENT_CERT_NAME);
+    var clientKey = new File(jenvtestDir, CLIENT_KEY_NAME);
+
+    return apiCert.exists() && apiKey.exists() && clientCert.exists() && clientKey.exists();
   }
 
   private void generateAPIServerCertificates() {
-    var cert = new File(jenvtestDir, API_SERVER_CERT_NAME);
-    var key = new File(jenvtestDir, API_SERVER_KEY_NAME);
-    if (cert.exists() && key.exists()) {
-      return;
-    }
     log.info("Generating API Server certificates");
     generateKeyAndCertificate("CN=example.org", new File(jenvtestDir, API_SERVER_KEY_NAME),
         new File(jenvtestDir, API_SERVER_CERT_NAME),
@@ -70,11 +91,6 @@ private GeneralName dns(String dns) {
   }
 
   private void generateUserCertificates() {
-    var cert = new File(jenvtestDir, CLIENT_CERT_NAME);
-    var key = new File(jenvtestDir, CLIENT_KEY_NAME);
-    if (cert.exists() && key.exists()) {
-      return;
-    }
     log.info("Generating Client certificates");
     generateKeyAndCertificate("O=system:masters,CN=jenvtest",
         new File(jenvtestDir, CLIENT_KEY_NAME),

core/src/main/java/io/javaoperatorsdk/jenvtest/lock/LockFile.java

@@ -0,0 +1,71 @@
+package io.javaoperatorsdk.jenvtest.lock;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.*;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.javaoperatorsdk.jenvtest.JenvtestException;
+
+public class LockFile {
+
+  private static final Logger log = LoggerFactory.getLogger(LockFile.class);
+
+  private final String dir;
+  private final String lockFileName;
+
+  public LockFile(String lockFileName, String dir) {
+    this.dir = dir;
+    this.lockFileName = lockFileName;
+  }
+
+  public boolean tryLock() {
+    File file = new File(dir, lockFileName);
+    try {
+      return file.createNewFile();
+    } catch (IOException e) {
+      throw new JenvtestException(e);
+    }
+  }
+
+  public void releaseLock() {
+    File file = new File(dir, lockFileName);
+    try {
+      Files.deleteIfExists(file.toPath());
+    } catch (IOException e) {
+      throw new JenvtestException(e);
+    }
+  }
+
+  public void waitUntilLockDeleted() {
+    var file = new File(dir);
+    var path = file.toPath();
+
+    try (final WatchService watchService = FileSystems.getDefault().newWatchService()) {
+      path.register(watchService, StandardWatchEventKinds.ENTRY_DELETE);
+      while (true) {
+        final WatchKey wk = watchService.take();
+        for (WatchEvent<?> event : wk.pollEvents()) {
+          final Path changed = (Path) event.context();
+          log.info("!! Event path: {}  event: {}", changed, event);
+          if (changed.endsWith(lockFileName)) {
+            return;
+          }
+        }
+        // reset the key
+        boolean valid = wk.reset();
+        if (!valid) {
+          log.warn("Watch key no longer valid");
+        }
+      }
+    } catch (IOException e) {
+      throw new JenvtestException(e);
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+      throw new JenvtestException(e);
+    }
+  }
+
+}

core/src/main/java/io/javaoperatorsdk/jenvtest/process/EtcdProcess.java

@@ -79,7 +79,7 @@ public int startEtcd() {
   }
 
   private void waitUntilEtcdHealthy(int port) {
-    new ProcessReadinessChecker(port, "health", "etcd", false).waitUntilReady();
+    new ProcessReadinessChecker().waitUntilReady(port, "health", "etcd", false);
   }
 
   public void cleanEtcdData() {

core/src/main/java/io/javaoperatorsdk/jenvtest/process/KubeAPIServerProcess.java

@@ -17,6 +17,7 @@ public class KubeAPIServerProcess {
   private static final Logger log = LoggerFactory.getLogger(KubeAPIServerProcess.class);
   private static final Logger apiLog = LoggerFactory.getLogger(KubeAPIServerProcess.class
       .getName() + ".APIServerProcessLogs");
+  public static final String KUBE_API_SERVER = "Kube API Server";
 
   private final CertManager certManager;
   private final BinaryManager binaryManager;
@@ -85,7 +86,9 @@ private List<String> createCommand(File apiServerBinary, int apiServerPort, int
   }
 
   public void waitUntilReady() {
-    new ProcessReadinessChecker(apiServerPort, "readyz", "Kube API Server", true).waitUntilReady();
+    var readinessChecker = new ProcessReadinessChecker();
+    readinessChecker.waitUntilReady(apiServerPort, "readyz", KUBE_API_SERVER, true);
+    readinessChecker.waitUntilDefaultNamespaceAvailable(apiServerPort, binaryManager, certManager);
   }
 
   public void stopApiServer() {

core/src/main/java/io/javaoperatorsdk/jenvtest/process/ProcessReadinessChecker.java

@@ -14,6 +14,7 @@
 import java.security.cert.X509Certificate;
 import java.time.LocalTime;
 import java.time.temporal.ChronoUnit;
+import java.util.function.BooleanSupplier;
 
 import javax.net.ssl.SSLContext;
 import javax.net.ssl.SSLEngine;
@@ -24,34 +25,57 @@
 import org.slf4j.LoggerFactory;
 
 import io.javaoperatorsdk.jenvtest.JenvtestException;
+import io.javaoperatorsdk.jenvtest.binary.BinaryManager;
+import io.javaoperatorsdk.jenvtest.cert.CertManager;
+
+import static io.javaoperatorsdk.jenvtest.process.KubeAPIServerProcess.KUBE_API_SERVER;
 
 public class ProcessReadinessChecker {
+
   private static final Logger log = LoggerFactory.getLogger(ProcessReadinessChecker.class);
 
-  public static final int STARTUP_TIMEOUT = 10_000;
-  public static final int POLLING_INTERVAL = 150;
+  public static final int STARTUP_TIMEOUT = 60_000;
+  public static final int POLLING_INTERVAL = 200;
+
 
-  private final int port;
-  private final String readyCheckPath;
-  private final String processName;
-  private final boolean useTLS;
+  public void waitUntilDefaultNamespaceAvailable(int apiServerPort,
+      BinaryManager binaryManager,
+      CertManager certManager) {
+    pollWithTimeout(() -> defaultNamespaceExists(apiServerPort, binaryManager, certManager),
+        KUBE_API_SERVER);
+  }
 
+  private boolean defaultNamespaceExists(int apiServerPort, BinaryManager binaryManager,
+      CertManager certManager) {
+    try {
+      Process process = new ProcessBuilder(binaryManager.binaries().getKubectl().getPath(),
+          "--client-certificate=" + certManager.getClientCertPath(),
+          "--client-key=" + certManager.getClientKeyPath(),
+          "--certificate-authority=" + certManager.getAPIServerCertPath(),
+          "--server=https://127.0.0.1:" + apiServerPort,
+          "--request-timeout=5s",
+          "get", "ns", "default").start();
+      return process.waitFor() == 0;
+    } catch (IOException e) {
+      throw new JenvtestException(e);
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+      throw new JenvtestException(e);
+    }
+  }
 
-  public ProcessReadinessChecker(int port, String readyCheckPath, String processName,
+  public void waitUntilReady(int port, String readyCheckPath, String processName,
       boolean useTLS) {
-    this.port = port;
-    this.readyCheckPath = readyCheckPath;
-    this.processName = processName;
-    this.useTLS = useTLS;
+    var client = getHttpClient();
+    var request = getHttpRequest(useTLS, readyCheckPath, port);
+    pollWithTimeout(() -> ready(client, request, processName, port), processName);
   }
 
-  public void waitUntilReady() {
+  private static void pollWithTimeout(BooleanSupplier predicate, String processName) {
     try {
-      var client = getHttpClient();
-      var request = getHttpRequest();
       var startedAt = LocalTime.now();
       while (true) {
-        if (ready(client, request)) {
+        if (predicate.getAsBoolean()) {
           return;
         }
         if (LocalTime.now().isAfter(startedAt.plus(STARTUP_TIMEOUT, ChronoUnit.MILLIS))) {
@@ -65,7 +89,7 @@ public void waitUntilReady() {
     }
   }
 
-  private boolean ready(HttpClient client, HttpRequest request) {
+  private boolean ready(HttpClient client, HttpRequest request, String processName, int port) {
     try {
       var response = client.send(request, HttpResponse.BodyHandlers.ofString());
       log.debug("Ready Response message:{} code: {} for {} on Port: {}", response.body(),
@@ -84,7 +108,7 @@ private boolean ready(HttpClient client, HttpRequest request) {
     }
   }
 
-  private HttpRequest getHttpRequest() {
+  private HttpRequest getHttpRequest(boolean useTLS, String readyCheckPath, int port) {
     try {
       return HttpRequest.newBuilder()
           .uri(new URI((useTLS ? "https" : "http") + "://127.0.0.1:" + port + "/" + readyCheckPath))
@@ -128,7 +152,7 @@ public void checkServerTrusted(
 
                 @Override
                 public void checkClientTrusted(X509Certificate[] chain, String authType,
-                    SSLEngine engine) throws CertificateException {
+                    SSLEngine engine) {
 
                 }
 

core/src/test/java/io/javaoperatorsdk/jenvtest/sample/KubeApiServerTest.java

@@ -26,18 +26,6 @@ void apiServerWithSpecificVersion() {
         .build()));
   }
 
-  @Test
-  void usingKubeConfigFileToInitClient() {
-    var kubeApi = new KubeAPIServer(KubeAPIServerConfigBuilder.anAPIServerConfig()
-        .withUpdateKubeConfig(true)
-        .build());
-    kubeApi.start();
-
-    var client = new KubernetesClientBuilder().build();
-
-    TestUtils.simpleTest(client);
-  }
-
   @Test
   void canWaitForEtcdHealthCheckOnStartup() {
     var kubeApi = new KubeAPIServer(KubeAPIServerConfigBuilder.anAPIServerConfig()