Richer logging and better error handling in driver pod watch (apache#154

)

* pod-watch progress around watch events

* Simplify return

* comments

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/LoggingPodStatusWatcher.scala

@@ -50,35 +50,47 @@ private[kubernetes] class LoggingPodStatusWatcher(podCompletedFuture: CountDownL
   }
 
   private var pod: Option[Pod] = Option.empty
-  private var prevPhase: String = null
   private def phase: String = pod.map(_.getStatus().getPhase()).getOrElse("unknown")
+  private def status: String = pod.map(_.getStatus().getContainerStatuses().toString())
+    .getOrElse("unknown")
 
   override def eventReceived(action: Action, pod: Pod): Unit = {
     this.pod = Option(pod)
-
-    logShortStatus()
-    if (prevPhase != phase) {
-      logLongStatus()
-    }
-    prevPhase = phase
-
-    if (phase == "Succeeded" || phase == "Failed") {
-      podCompletedFuture.countDown()
-      scheduler.shutdown()
+    action match {
+      case Action.DELETED =>
+        closeWatch()
+
+      case Action.ERROR =>
+        closeWatch()
+
+      case _ =>
+        logLongStatus()
+        if (hasCompleted()) {
+          closeWatch()
+        }
     }
   }
 
   override def onClose(e: KubernetesClientException): Unit = {
-    scheduler.shutdown()
-    logDebug(s"Stopped watching application $appId with last-observed phase $phase")
+    logDebug(s"Stopping watching application $appId with last-observed phase $phase")
+    closeWatch()
   }
 
   private def logShortStatus() = {
     logInfo(s"Application status for $appId (phase: $phase)")
   }
 
   private def logLongStatus() = {
-    logInfo("Phase changed, new state: " + pod.map(formatPodState(_)).getOrElse("unknown"))
+    logInfo("State changed, new state: " + pod.map(formatPodState(_)).getOrElse("unknown"))
+  }
+
+  private def hasCompleted(): Boolean = {
+    phase == "Succeeded" || phase == "Failed"
+  }
+
+  private def closeWatch(): Unit = {
+    podCompletedFuture.countDown()
+    scheduler.shutdown()
   }
 
   private def formatPodState(pod: Pod): String = {
@@ -103,7 +115,8 @@ private[kubernetes] class LoggingPodStatusWatcher(podCompletedFuture: CountDownL
             .asScala
             .map(_.getImage)
             .mkString(", ")),
-      ("phase", pod.getStatus.getPhase())
+      ("phase", pod.getStatus.getPhase()),
+      ("status", pod.getStatus.getContainerStatuses().toString)
     )
 
     // Use more loggable format if value is null or empty