control-service: logs endpoint doesn't hang

projects/control-service/projects/pipelines_control_service/src/integration-test/java/com/vmware/taurus/datajobs/it/DataJobViewLogsIT.java

@@ -0,0 +1,87 @@
+/*
+ * Copyright 2021-2023 VMware, Inc.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package com.vmware.taurus.datajobs.it;
+
+import com.vmware.taurus.ControlplaneApplication;
+import com.vmware.taurus.controlplane.model.data.DataJobExecution;
+import com.vmware.taurus.datajobs.it.common.BaseIT;
+import com.vmware.taurus.datajobs.it.common.DataJobDeploymentExtension;
+import com.vmware.taurus.datajobs.it.common.JobExecutionUtil;
+import com.vmware.taurus.service.execution.JobExecutionService;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.tuple.ImmutablePair;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.testcontainers.shaded.org.awaitility.Awaitility;
+
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import static com.vmware.taurus.datajobs.it.common.JobExecutionUtil.testDataJobExecutionRead;
+
+@Slf4j
+@SpringBootTest(
+    webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
+    classes = ControlplaneApplication.class)
+public class DataJobViewLogsIT extends BaseIT {
+
+  @Autowired private JobExecutionService executionService;
+
+  // simple_job_read_logs.zip contains a job that would run for more than 30 minutes if allowed
+  @RegisterExtension
+  static DataJobDeploymentExtension dataJobDeploymentExtension =
+      new DataJobDeploymentExtension("simple_job_read_logs.zip");
+
+  @Test
+  public void testJobLogsViewing(
+      String jobName, String username, String deploymentId, String teamName) throws Exception {
+    // manually start job execution
+    ImmutablePair<String, String> executeDataJobResult =
+        JobExecutionUtil.executeDataJob(jobName, teamName, username, deploymentId, mockMvc);
+    String executionId = executeDataJobResult.getRight();
+    String opId = executeDataJobResult.getLeft();
+    testDataJobExecutionRead(
+        executionId,
+        DataJobExecution.StatusEnum.RUNNING,
+        opId,
+        jobName,
+        teamName,
+        username,
+        mockMvc);
+    Awaitility.await()
+        .atMost(6, TimeUnit.MINUTES)
+        .ignoreExceptionsMatching(
+            (e) ->
+                // It is only by looking 3 exceptions deep that we can tell the reason the logs read
+                // failed.
+                e.getCause()
+                    .getCause()
+                    // If 400 is in the error message then the pod is not up yet and we should keep
+                    // trying to poll for the logs.
+                    .getMessage()
+                    .contains("400"))
+        .until(
+            () -> {
+              getLogsFast(jobName, teamName, executionId);
+              return true;
+            });
+  }
+
+  /** We are testing that the server returns logs quickly. */
+  private void getLogsFast(String jobName, String teamName, String executionId)
+      throws InterruptedException, ExecutionException, TimeoutException {
+    Executors.newSingleThreadExecutor()
+        .submit(
+            () -> {
+              executionService.getJobExecutionLogs(teamName, jobName, executionId, 0);
+            })
+        .get(5, TimeUnit.SECONDS);
+  }
+}

projects/control-service/projects/pipelines_control_service/src/main/java/com/vmware/taurus/service/KubernetesService.java

@@ -31,6 +31,7 @@
 import io.kubernetes.client.util.Yaml;
 import lombok.*;
 import net.javacrumbs.shedlock.spring.annotation.EnableSchedulerLock;
+import okhttp3.Response;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.tuple.ImmutablePair;
 import org.apache.commons.lang3.tuple.Pair;
@@ -914,18 +915,15 @@ public Optional<String> getJobLogs(String jobName, Integer tailLines)
       var pods = listJobPods(job.get());
       if (pods.size() > 0) {
         var pod = pods.get(pods.size() - 1);
-        PodLogs podLogs = new PodLogs(client);
 
         try (BufferedReader br =
             new BufferedReader(
                 new InputStreamReader(
-                    podLogs.streamNamespacedPodLog(
+                    readNamespacedPodLog(
                         pod.getMetadata().getNamespace(),
                         pod.getMetadata().getName(),
                         pod.getSpec().getContainers().get(0).getName(),
-                        null,
-                        tailLines,
-                        true),
+                        tailLines),
                     Charsets.UTF_8))) {
           String logs = br.lines().collect(Collectors.joining(System.lineSeparator()));
           return Optional.of(logs);
@@ -935,6 +933,28 @@ public Optional<String> getJobLogs(String jobName, Integer tailLines)
     return Optional.empty();
   }
 
+  /**
+   * This function is copy and pasted from the kuberenetes class PodLogs. the only difference is
+   * that we set follow=false
+   */
+  private InputStream readNamespacedPodLog(
+      String namespace, String name, String container, Integer tailLines)
+      throws ApiException, IOException {
+    Response response =
+        new CoreV1Api(client)
+            .readNamespacedPodLogCall(
+                name, namespace, container, false, null, null, "false", false, null, tailLines,
+                true, null)
+            .execute();
+    if (!response.isSuccessful()) {
+      if (response.body() != null) {
+        response.close();
+      }
+      throw new ApiException(response.code(), "Logs request failed: " + response.code());
+    }
+    return response.body().byteStream();
+  }
+
   public JobStatusCondition watchJob(
       String jobName, int timeoutSeconds, Consumer<JobStatus> watcher)
       throws ApiException, IOException, InterruptedException {