watch mkdir pod logs (#16162)

Signed-off-by: Michal Vala <mvala@redhat.com>

infrastructures/kubernetes/src/main/java/org/eclipse/che/workspace/infrastructure/kubernetes/KubernetesInternalRuntime.java

@@ -14,7 +14,6 @@
 import static java.lang.String.format;
 import static java.util.Collections.emptyMap;
 import static java.util.stream.Collectors.toMap;
-import static org.eclipse.che.api.workspace.shared.Constants.DEBUG_WORKSPACE_START;
 import static org.eclipse.che.workspace.infrastructure.kubernetes.util.TracingSpanConstants.CHECK_SERVERS;
 import static org.eclipse.che.workspace.infrastructure.kubernetes.util.TracingSpanConstants.WAIT_MACHINES_START;
 import static org.eclipse.che.workspace.infrastructure.kubernetes.util.TracingSpanConstants.WAIT_RUNNING_ASYNC;
@@ -213,7 +212,8 @@ protected void internalStart(Map<String, String> startOptions) throws Infrastruc
       volumesStrategy.prepare(
           context.getEnvironment(),
           context.getIdentity(),
-          startSynchronizer.getStartTimeoutMillis());
+          startSynchronizer.getStartTimeoutMillis(),
+          startOptions);
 
       startSynchronizer.checkFailure();
 
@@ -633,17 +633,10 @@ protected void listenEvents() throws InfrastructureException {
 
   private void watchLogsIfDebugEnabled(Map<String, String> startOptions)
       throws InfrastructureException {
-    if (startOptions == null || startOptions.isEmpty()) {
-      LOG.debug(
-          "'startOptions' is null or empty so we won't watch the container logs for workspace '{}'",
+    if (LogWatcher.shouldWatchLogs(startOptions)) {
+      LOG.info(
+          "Debug workspace startup. Will watch the logs of '{}'",
           getContext().getIdentity().getWorkspaceId());
-      return;
-    }
-    boolean shouldWatchContainerStartupLogs =
-        Boolean.parseBoolean(
-            startOptions.getOrDefault(DEBUG_WORKSPACE_START, Boolean.FALSE.toString()));
-
-    if (shouldWatchContainerStartupLogs) {
       // get all the pods we care about
       Set<String> podNames =
           machines

infrastructures/kubernetes/src/main/java/org/eclipse/che/workspace/infrastructure/kubernetes/namespace/log/LogWatcher.java

@@ -11,6 +11,7 @@
  */
 package org.eclipse.che.workspace.infrastructure.kubernetes.namespace.log;
 
+import static org.eclipse.che.api.workspace.shared.Constants.DEBUG_WORKSPACE_START;
 import static org.eclipse.che.api.workspace.shared.Constants.DEBUG_WORKSPACE_START_LOG_LIMIT_BYTES;
 
 import io.fabric8.kubernetes.client.KubernetesClient;
@@ -198,4 +199,20 @@ public static long getLogLimitBytes(Map<String, String> startOptions) {
       }
     }
   }
+
+  /**
+   * Takes `startOptions` map and tells whether it's set so we should watch the logs. To return
+   * true, flag must be stored under {@link
+   * org.eclipse.che.api.workspace.shared.Constants#DEBUG_WORKSPACE_START} key.
+   *
+   * @param startOptions options where we'll try to find log watch flag
+   * @return true if we should watch the logs, false otherwise
+   */
+  public static boolean shouldWatchLogs(Map<String, String> startOptions) {
+    if (startOptions == null || startOptions.isEmpty()) {
+      return false;
+    }
+    return Boolean.parseBoolean(
+        startOptions.getOrDefault(DEBUG_WORKSPACE_START, Boolean.FALSE.toString()));
+  }
 }

infrastructures/kubernetes/src/main/java/org/eclipse/che/workspace/infrastructure/kubernetes/namespace/pvc/CommonPVCStrategy.java

@@ -172,7 +172,11 @@ public void provision(KubernetesEnvironment k8sEnv, RuntimeIdentity identity)
 
   @Override
   @Traced
-  public void prepare(KubernetesEnvironment k8sEnv, RuntimeIdentity identity, long timeoutMillis)
+  public void prepare(
+      KubernetesEnvironment k8sEnv,
+      RuntimeIdentity identity,
+      long timeoutMillis,
+      Map<String, String> startOptions)
       throws InfrastructureException {
     String workspaceId = identity.getWorkspaceId();
 
@@ -215,10 +219,7 @@ public void prepare(KubernetesEnvironment k8sEnv, RuntimeIdentity identity, long
             commonPVC.getAdditionalProperties().remove(format(SUBPATHS_PROPERTY_FMT, workspaceId));
     if (preCreateDirs && subpaths != null) {
       pvcSubPathHelper.createDirs(
-          workspaceId,
-          identity.getInfrastructureNamespace(),
-          commonPVC.getMetadata().getName(),
-          subpaths);
+          identity, workspaceId, commonPVC.getMetadata().getName(), startOptions, subpaths);
     }
 
     log.debug("Preparing PVC done for workspace '{}'", workspaceId);

infrastructures/kubernetes/src/main/java/org/eclipse/che/workspace/infrastructure/kubernetes/namespace/pvc/PVCSubPathHelper.java

@@ -25,6 +25,8 @@
 import io.fabric8.kubernetes.api.model.PodBuilder;
 import io.fabric8.kubernetes.api.model.PodStatus;
 import java.util.Arrays;
+import java.util.Collections;
+import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -33,14 +35,19 @@
 import javax.inject.Inject;
 import javax.inject.Named;
 import javax.inject.Singleton;
+import org.eclipse.che.api.core.model.workspace.runtime.RuntimeIdentity;
 import org.eclipse.che.api.workspace.server.spi.InfrastructureException;
 import org.eclipse.che.commons.lang.concurrent.LoggingUncaughtExceptionHandler;
 import org.eclipse.che.commons.lang.concurrent.ThreadLocalPropagateContext;
 import org.eclipse.che.commons.observability.ExecutorServiceWrapper;
 import org.eclipse.che.workspace.infrastructure.kubernetes.namespace.KubernetesDeployments;
 import org.eclipse.che.workspace.infrastructure.kubernetes.namespace.KubernetesNamespaceFactory;
+import org.eclipse.che.workspace.infrastructure.kubernetes.namespace.log.LogWatchTimeouts;
+import org.eclipse.che.workspace.infrastructure.kubernetes.namespace.log.LogWatcher;
+import org.eclipse.che.workspace.infrastructure.kubernetes.namespace.log.PodLogToEventPublisher;
 import org.eclipse.che.workspace.infrastructure.kubernetes.provision.SecurityContextProvisioner;
 import org.eclipse.che.workspace.infrastructure.kubernetes.util.Containers;
+import org.eclipse.che.workspace.infrastructure.kubernetes.util.RuntimeEventsPublisher;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -78,6 +85,7 @@ public class PVCSubPathHelper {
   private final String jobMemoryLimit;
   private final KubernetesNamespaceFactory factory;
   private final ExecutorService executor;
+  private final RuntimeEventsPublisher eventsPublisher;
 
   private final SecurityContextProvisioner securityContextProvisioner;
 
@@ -87,11 +95,13 @@ public class PVCSubPathHelper {
       @Named("che.infra.kubernetes.pvc.jobs.image") String jobImage,
       KubernetesNamespaceFactory factory,
       SecurityContextProvisioner securityContextProvisioner,
-      ExecutorServiceWrapper executorServiceWrapper) {
+      ExecutorServiceWrapper executorServiceWrapper,
+      RuntimeEventsPublisher eventPublisher) {
     this.jobMemoryLimit = jobMemoryLimit;
     this.jobImage = jobImage;
     this.factory = factory;
     this.securityContextProvisioner = securityContextProvisioner;
+    this.eventsPublisher = eventPublisher;
     this.executor =
         executorServiceWrapper.wrap(
             Executors.newFixedThreadPool(
@@ -108,16 +118,20 @@ public class PVCSubPathHelper {
    * Performs create workspace directories job by given paths and waits until it finished.
    *
    * @param workspaceId workspace identifier
-   * @param namespace
    * @param dirs workspace directories to create
    */
-  void createDirs(String workspaceId, String namespace, String pvcName, String... dirs) {
+  void createDirs(
+      RuntimeIdentity identity,
+      String workspaceId,
+      String pvcName,
+      Map<String, String> startOptions,
+      String... dirs) {
     LOG.debug(
         "Preparing PVC `{}` for workspace `{}`. Directories to create: {}",
         pvcName,
         workspaceId,
         Arrays.toString(dirs));
-    execute(workspaceId, namespace, pvcName, MKDIR_COMMAND_BASE, dirs);
+    execute(identity, workspaceId, pvcName, MKDIR_COMMAND_BASE, startOptions, dirs);
   }
 
   /**
@@ -140,6 +154,24 @@ CompletableFuture<Void> removeDirsAsync(
         executor);
   }
 
+  @VisibleForTesting
+  void execute(
+      RuntimeIdentity identity,
+      String workspaceId,
+      String pvcName,
+      String[] commandBase,
+      Map<String, String> startOptions,
+      String... arguments) {
+    execute(
+        identity,
+        workspaceId,
+        identity.getInfrastructureNamespace(),
+        pvcName,
+        commandBase,
+        startOptions,
+        arguments);
+  }
+
   /**
    * Executes the job with the specified arguments.
    *
@@ -154,6 +186,17 @@ void execute(
       String pvcName,
       String[] commandBase,
       String... arguments) {
+    execute(null, workspaceId, namespace, pvcName, commandBase, Collections.emptyMap(), arguments);
+  }
+
+  private void execute(
+      RuntimeIdentity identity,
+      String workspaceId,
+      String namespace,
+      String pvcName,
+      String[] commandBase,
+      Map<String, String> startOptions,
+      String... arguments) {
     final String jobName = commandBase[0];
     final String podName = jobName + '-' + workspaceId;
     final String[] command = buildCommand(commandBase, arguments);
@@ -164,6 +207,7 @@ void execute(
     try {
       deployments = factory.access(workspaceId, namespace).deployments();
       deployments.create(pod);
+      watchLogsIfDebugEnabled(deployments, pod, identity, startOptions);
       final Pod finished = deployments.wait(podName, WAIT_POD_TIMEOUT_MIN, POD_PREDICATE::apply);
       PodStatus finishedStatus = finished.getStatus();
       if (POD_PHASE_FAILED.equals(finishedStatus.getPhase())) {
@@ -179,8 +223,10 @@ void execute(
           Arrays.toString(command),
           workspaceId,
           ex.getMessage());
+      deployments.stopWatch(true);
     } finally {
       if (deployments != null) {
+        deployments.stopWatch();
         try {
           deployments.delete(podName);
         } catch (InfrastructureException ignored) {
@@ -189,6 +235,21 @@ void execute(
     }
   }
 
+  private void watchLogsIfDebugEnabled(
+      KubernetesDeployments deployment,
+      Pod pod,
+      RuntimeIdentity identity,
+      Map<String, String> startOptions)
+      throws InfrastructureException {
+    if (LogWatcher.shouldWatchLogs(startOptions)) {
+      deployment.watchLogs(
+          new PodLogToEventPublisher(eventsPublisher, identity),
+          LogWatchTimeouts.AGGRESSIVE,
+          Collections.singleton(pod.getMetadata().getName()),
+          LogWatcher.getLogLimitBytes(startOptions));
+    }
+  }
+
   /**
    * Builds the command by given base and paths.
    *

infrastructures/kubernetes/src/main/java/org/eclipse/che/workspace/infrastructure/kubernetes/namespace/pvc/UniqueWorkspacePVCStrategy.java

@@ -127,7 +127,11 @@ public void provision(KubernetesEnvironment k8sEnv, RuntimeIdentity identity)
 
   @Traced
   @Override
-  public void prepare(KubernetesEnvironment k8sEnv, RuntimeIdentity identity, long timeoutMillis)
+  public void prepare(
+      KubernetesEnvironment k8sEnv,
+      RuntimeIdentity identity,
+      long timeoutMillis,
+      Map<String, String> startOptions)
       throws InfrastructureException {
     String workspaceId = identity.getWorkspaceId();
 

infrastructures/kubernetes/src/main/java/org/eclipse/che/workspace/infrastructure/kubernetes/namespace/pvc/WorkspaceVolumesStrategy.java

@@ -11,6 +11,7 @@
  */
 package org.eclipse.che.workspace.infrastructure.kubernetes.namespace.pvc;
 
+import java.util.Map;
 import org.eclipse.che.api.core.model.workspace.Workspace;
 import org.eclipse.che.api.core.model.workspace.runtime.RuntimeIdentity;
 import org.eclipse.che.api.workspace.server.spi.InfrastructureException;
@@ -45,7 +46,11 @@ void provision(KubernetesEnvironment k8sEnv, RuntimeIdentity identity)
    * @param timeoutMillis timeout in milliseconds
    * @throws InfrastructureException when any error while preparation occurs
    */
-  void prepare(KubernetesEnvironment k8sEnv, RuntimeIdentity identity, long timeoutMillis)
+  void prepare(
+      KubernetesEnvironment k8sEnv,
+      RuntimeIdentity identity,
+      long timeoutMillis,
+      Map<String, String> startOptions)
       throws InfrastructureException;
 
   /**

infrastructures/kubernetes/src/main/java/org/eclipse/che/workspace/infrastructure/kubernetes/wsplugins/PluginBrokerManager.java

@@ -117,7 +117,7 @@ public List<ChePlugin> getTooling(
 
     ListenBrokerEvents listenBrokerEvents = getListenEventPhase(workspaceId, brokersResult);
     PrepareStorage prepareStorage =
-        getPrepareStoragePhase(identity, startSynchronizer, brokerEnvironment);
+        getPrepareStoragePhase(identity, startSynchronizer, brokerEnvironment, startOptions);
     WaitBrokerResult waitBrokerResult = getWaitBrokerPhase(workspaceId, brokersResult);
     DeployBroker deployBroker =
         getDeployBrokerPhase(
@@ -134,9 +134,10 @@ private ListenBrokerEvents getListenEventPhase(String workspaceId, BrokersResult
   private PrepareStorage getPrepareStoragePhase(
       RuntimeIdentity identity,
       StartSynchronizer startSynchronizer,
-      KubernetesEnvironment brokerEnvironment) {
+      KubernetesEnvironment brokerEnvironment,
+      Map<String, String> startOptions) {
     return new PrepareStorage(
-        identity, brokerEnvironment, volumesStrategy, startSynchronizer, tracer);
+        identity, brokerEnvironment, volumesStrategy, startSynchronizer, tracer, startOptions);
   }
 
   private DeployBroker getDeployBrokerPhase(

infrastructures/kubernetes/src/main/java/org/eclipse/che/workspace/infrastructure/kubernetes/wsplugins/brokerphases/DeployBroker.java

@@ -12,7 +12,6 @@
 package org.eclipse.che.workspace.infrastructure.kubernetes.wsplugins.brokerphases;
 
 import static java.lang.String.format;
-import static org.eclipse.che.api.workspace.shared.Constants.DEBUG_WORKSPACE_START;
 import static org.eclipse.che.workspace.infrastructure.kubernetes.util.TracingSpanConstants.DEPLOY_BROKER_PHASE;
 import static org.slf4j.LoggerFactory.getLogger;
 
@@ -157,17 +156,9 @@ public List<ChePlugin> execute() throws InfrastructureException {
 
   private void watchLogsIfDebugEnabled(Map<String, String> startOptions, Pod pluginBrokerPod)
       throws InfrastructureException {
-    if (startOptions == null || startOptions.isEmpty()) {
+    if (LogWatcher.shouldWatchLogs(startOptions)) {
       LOG.debug(
-          "'startOptions' is null or empty so we won't watch the plugin broker pod logs for workspace '{}'",
-          runtimeId.getWorkspaceId());
-      return;
-    }
-    boolean shouldWatchContainerStartupLogs =
-        Boolean.parseBoolean(
-            startOptions.getOrDefault(DEBUG_WORKSPACE_START, Boolean.FALSE.toString()));
-
-    if (shouldWatchContainerStartupLogs) {
+          "Will watch the logs of plugin broker of workspace '{}'", runtimeId.getWorkspaceId());
       namespace
           .deployments()
           .watchLogs(

infrastructures/kubernetes/src/main/java/org/eclipse/che/workspace/infrastructure/kubernetes/wsplugins/brokerphases/PrepareStorage.java

@@ -17,6 +17,7 @@
 import io.opentracing.Span;
 import io.opentracing.Tracer;
 import java.util.List;
+import java.util.Map;
 import org.eclipse.che.api.core.model.workspace.runtime.RuntimeIdentity;
 import org.eclipse.che.api.workspace.server.spi.InfrastructureException;
 import org.eclipse.che.api.workspace.server.wsplugins.model.ChePlugin;
@@ -40,18 +41,21 @@ public class PrepareStorage extends BrokerPhase {
   private final WorkspaceVolumesStrategy volumesStrategy;
   private final StartSynchronizer startSynchronizer;
   private final Tracer tracer;
+  private final Map<String, String> startOptions;
 
   public PrepareStorage(
       RuntimeIdentity identity,
       KubernetesEnvironment brokerEnvironment,
       WorkspaceVolumesStrategy volumesStrategy,
       StartSynchronizer startSynchronizer,
-      Tracer tracer) {
+      Tracer tracer,
+      Map<String, String> startOptions) {
     this.identity = identity;
     this.brokerEnvironment = brokerEnvironment;
     this.volumesStrategy = volumesStrategy;
     this.startSynchronizer = startSynchronizer;
     this.tracer = tracer;
+    this.startOptions = startOptions;
   }
 
   @Override
@@ -61,7 +65,7 @@ public List<ChePlugin> execute() throws InfrastructureException {
 
     try {
       volumesStrategy.prepare(
-          brokerEnvironment, identity, startSynchronizer.getStartTimeoutMillis());
+          brokerEnvironment, identity, startSynchronizer.getStartTimeoutMillis(), startOptions);
     } catch (InfrastructureException e) {
       TracingTags.setErrorStatus(tracingSpan, e);
       throw e;