Addressed comments, and added base URL prepending to links in description

Author: tdas

core/src/main/scala/org/apache/spark/ui/UIUtils.scala

@@ -20,7 +20,9 @@ package org.apache.spark.ui
 import java.text.SimpleDateFormat
 import java.util.{Locale, Date}
 
-import scala.xml.{Node, Text, Unparsed}
+import scala.util.control.NonFatal
+import scala.xml._
+import scala.xml.transform.{RewriteRule, RuleTransformer}
 
 import org.apache.spark.Logging
 import org.apache.spark.ui.scope.RDDOperationGraph
@@ -395,4 +397,43 @@ private[spark] object UIUtils extends Logging {
     </script>
   }
 
+  /**
+   * Convert a description string to HTML. It will try to parse the string as HTML and sanitize
+   * any links. If that fails, then whole string will treated as a simple text.
+   */
+  def makeDescription(desc: String, basePathUri: String): NodeSeq = {
+    import scala.language.postfixOps
+
+    // If the description can be parsed as HTML and has only relative links, then render
+    // as HTML, otherwise render as escaped string
+    try {
+      // Try to load the description as unescaped HTML
+      val xml = XML.loadString("<span class=\"description-input\" " +
+        s"title=${"\"" + Utility.escape(desc) + "\""}>$desc</span>")
+      val allLinks = xml \\ "_" flatMap { _.attributes } filter { _.key == "href" }
+      val areAllLinksRelative = allLinks.forall { _.value.toString.startsWith ("/") }
+
+      // If all the links are relative then, transform the links to absolute links
+      // with basePathUri
+      if (areAllLinksRelative) {
+        val rule = new RewriteRule() {
+          override def transform(n: Node): Seq[Node] = {
+            n match {
+              case e: Elem if e \ "@href" nonEmpty =>
+                val relativePath = e.attribute("href").get.toString
+                val fullUri = s"${basePathUri.stripSuffix("/")}/${relativePath.stripPrefix("/")}"
+                e % Attribute(null, "href", fullUri, Null)
+              case _ => n
+            }
+          }
+        }
+        new RuleTransformer(rule).transform(xml)
+      } else {
+        <span class="description-input" title={desc}> {desc} </span>
+      }
+    } catch {
+      case NonFatal(e) =>
+        <span class="description-input" title={desc}> {desc} </span>
+    }
+  }
 }

core/src/main/scala/org/apache/spark/ui/jobs/AllJobsPage.scala

@@ -21,7 +21,6 @@ import java.util.Date
 import javax.servlet.http.HttpServletRequest
 
 import scala.collection.mutable.{HashMap, ListBuffer}
-import scala.util.control.NonFatal
 import scala.xml._
 
 import org.apache.spark.JobExecutionStatus
@@ -225,27 +224,7 @@ private[ui] class AllJobsPage(parent: JobsTab) extends WebUIPage("") {
       }
       val formattedDuration = duration.map(d => UIUtils.formatDuration(d)).getOrElse("Unknown")
       val formattedSubmissionTime = job.submissionTime.map(UIUtils.formatDate).getOrElse("Unknown")
-      val jobDescription = {
-        val d = lastStageDescription
-        // If the description can be parsed as HTML and has only relative links, then render
-        // as HTML, otherwise render as escaped string
-        try {
-          // Try to load the description as unescaped HTML
-          val xml = XML.loadString("<span class=\"description-input\" " +
-            s"title=${"\"" + Utility.escape(d) + "\""}>$d</span>")
-          val allLinks = xml \\ "_" flatMap { _.attributes } filter { _.key == "href" }
-          val areAllLinksRelative = allLinks.forall { _.value.toString.startsWith ("/") }
-          if (areAllLinksRelative) {
-            xml
-          } else {
-            <span class="description-input" title={d}> {d} </span>
-          }
-        } catch {
-          case NonFatal(e) =>
-            <span class="description-input" title={d}> {d} </span>
-        }
-      }
-
+      val jobDescription = UIUtils.makeDescription(lastStageDescription, parent.basePath)
 
       val detailUrl =
         "%s/jobs/job?id=%s".format(UIUtils.prependBaseUri(parent.basePath), job.jobId)

core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala

@@ -19,8 +19,7 @@ package org.apache.spark.ui.jobs
 
 import java.util.Date
 
-import scala.util.control.NonFatal
-import scala.xml.{Utility, Node, Text, XML}
+import scala.xml.{Node, Text}
 
 import org.apache.commons.lang3.StringEscapeUtils
 
@@ -116,23 +115,7 @@ private[ui] class StageTableBase(
       stageData <- listener.stageIdToData.get((s.stageId, s.attemptId))
       desc <- stageData.description
     } yield {
-      // If the description can be parsed as HTML and has only relative links, then render
-      // as HTML, otherwise render as escaped string
-      try {
-        // Try to load the description as unescaped HTML
-        val xml = XML.loadString("<span class=\"description-input\" " +
-          s"title=${"\"" + Utility.escape(desc) + "\""}>$desc</span>")
-        val allLinks = xml \\ "_" flatMap { _.attributes } filter { _.key == "href" }
-        val areAllLinksRelative = allLinks.forall { _.value.toString.startsWith ("/") }
-        if (areAllLinksRelative) {
-          xml
-        } else {
-          <span class="description-input" title={desc}> {desc} </span>
-        }
-      } catch {
-        case NonFatal(e) =>
-          <span class="description-input" title={desc}> {desc} </span>
-      }
+      UIUtils.makeDescription(desc, basePathUri)
     }
     <div>{stageDesc.getOrElse("")} {killLink} {nameLink} {details}</div>
   }

streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala

@@ -198,7 +198,9 @@ class StreamingContext private[streaming] (
 
   private var state: StreamingContextState = INITIALIZED
 
-  private[streaming] val startSite = new AtomicReference[CallSite](null)
+  private val startSite = new AtomicReference[CallSite](null)
+
+  private[streaming] def getStartSite(): CallSite = startSite.get()
 
   private var shutdownHookRef: AnyRef = _
 
@@ -735,7 +737,7 @@ object StreamingContext extends Logging {
         throw new IllegalStateException(
           "Only one StreamingContext may be started in this JVM. " +
             "Currently running StreamingContext was started at" +
-            activeContext.get.startSite.get.longForm)
+            activeContext.get.getStartSite().longForm)
       }
     }
   }

streaming/src/main/scala/org/apache/spark/streaming/scheduler/ReceiverTracker.scala

@@ -555,7 +555,7 @@ class ReceiverTracker(ssc: StreamingContext, skipReceiverLaunch: Boolean = false
         }
       receiverRDD.setName(s"Receiver $receiverId")
       ssc.sparkContext.setJobDescription(s"Streaming job running receiver $receiverId")
-      ssc.sparkContext.setCallSite(ssc.startSite.get)
+      ssc.sparkContext.setCallSite(ssc.getStartSite())
 
       val future = ssc.sparkContext.submitJob[Receiver[_], Unit, Unit](
         receiverRDD, startReceiverFunc, Seq(0), (_, _) => Unit, ())