Fix empty logs and status messages for mmless ingestion (#15527)

* Fix empty logs and status messages for mmless ingestion

* Add tests

Author: George Shiqi Wu

extensions-contrib/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycle.java

@@ -288,7 +288,11 @@ private TaskStatus getTaskStatus(long duration)
             TaskStatus.class
         );
       } else {
-        taskStatus = TaskStatus.failure(taskId.getOriginalTaskId(), "task status not found");
+        log.info(
+            "Peon for task [%s] did not push its task status. Check k8s logs and events for the pod to see what happened.",
+            taskId
+        );
+        taskStatus = TaskStatus.failure(taskId.getOriginalTaskId(), "Peon did not report status successfully.");
       }
     }
     catch (IOException e) {
@@ -329,6 +333,15 @@ protected void saveLogs()
           FileUtils.copyInputStreamToFile(logWatch.getOutput(), file.toFile());
         } else {
           log.debug("Log stream not found for %s", taskId.getOriginalTaskId());
+          FileUtils.writeStringToFile(
+              file.toFile(),
+              StringUtils.format(
+                  "Peon for task [%s] did not report any logs. Check k8s metrics and events for the pod to see what happened.",
+                  taskId
+              ),
+              Charset.defaultCharset()
+          );
+
         }
         taskLogs.pushTaskLog(taskId.getOriginalTaskId(), file.toFile());
       }

extensions-contrib/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycleTest.java

@@ -276,16 +276,15 @@ public void test_join_withoutJob_returnsFailedTaskStatus() throws IOException
         EasyMock.anyLong(),
         EasyMock.eq(TimeUnit.MILLISECONDS)
     )).andReturn(new JobResponse(null, PeonPhase.FAILED));
-    EasyMock.expect(kubernetesClient.getPeonLogWatcher(k8sTaskId)).andReturn(Optional.of(logWatch));
+    EasyMock.expect(kubernetesClient.getPeonLogWatcher(k8sTaskId)).andReturn(Optional.absent());
     EasyMock.expect(taskLogs.streamTaskStatus(ID)).andReturn(Optional.absent());
+
     taskLogs.pushTaskLog(EasyMock.eq(ID), EasyMock.anyObject(File.class));
     EasyMock.expectLastCall();
     stateListener.stateChanged(KubernetesPeonLifecycle.State.RUNNING, ID);
     EasyMock.expectLastCall().once();
     stateListener.stateChanged(KubernetesPeonLifecycle.State.STOPPED, ID);
     EasyMock.expectLastCall().once();
-    logWatch.close();
-    EasyMock.expectLastCall();
 
     replayAll();
 
@@ -295,7 +294,7 @@ public void test_join_withoutJob_returnsFailedTaskStatus() throws IOException
 
     Assert.assertTrue(taskStatus.isFailure());
     Assert.assertEquals(ID, taskStatus.getId());
-    Assert.assertEquals("task status not found", taskStatus.getErrorMsg());
+    Assert.assertEquals("Peon did not report status successfully.", taskStatus.getErrorMsg());
     Assert.assertEquals(KubernetesPeonLifecycle.State.STOPPED, peonLifecycle.getState());
   }
 
@@ -436,16 +435,14 @@ public void test_join_withoutTaskStatus_returnsFailedTaskStatus() throws IOExcep
         EasyMock.anyLong(),
         EasyMock.eq(TimeUnit.MILLISECONDS)
     )).andReturn(new JobResponse(job, PeonPhase.SUCCEEDED));
-    EasyMock.expect(kubernetesClient.getPeonLogWatcher(k8sTaskId)).andReturn(Optional.of(logWatch));
+    EasyMock.expect(kubernetesClient.getPeonLogWatcher(k8sTaskId)).andReturn(Optional.absent());
     EasyMock.expect(taskLogs.streamTaskStatus(ID)).andReturn(Optional.absent());
     taskLogs.pushTaskLog(EasyMock.eq(ID), EasyMock.anyObject(File.class));
     EasyMock.expectLastCall();
     stateListener.stateChanged(KubernetesPeonLifecycle.State.RUNNING, ID);
     EasyMock.expectLastCall().once();
     stateListener.stateChanged(KubernetesPeonLifecycle.State.STOPPED, ID);
     EasyMock.expectLastCall().once();
-    logWatch.close();
-    EasyMock.expectLastCall();
 
     Assert.assertEquals(KubernetesPeonLifecycle.State.NOT_STARTED, peonLifecycle.getState());
 
@@ -457,7 +454,7 @@ public void test_join_withoutTaskStatus_returnsFailedTaskStatus() throws IOExcep
 
     Assert.assertTrue(taskStatus.isFailure());
     Assert.assertEquals(ID, taskStatus.getId());
-    Assert.assertEquals("task status not found", taskStatus.getErrorMsg());
+    Assert.assertEquals("Peon did not report status successfully.", taskStatus.getErrorMsg());
     Assert.assertEquals(KubernetesPeonLifecycle.State.STOPPED, peonLifecycle.getState());
   }