/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.spark.ui.jobs

import java.util.Date

import scala.collection.mutable.HashMap
import scala.xml.Node

import org.apache.spark.scheduler.{StageInfo, TaskInfo}
import org.apache.spark.ui.{WebUI, UIUtils}
import org.apache.spark.util.Utils

/** Page showing list of all ongoing and recently finished stages */
private[ui] class StageTable(stages: Seq[StageInfo], parent: JobProgressUI) {
  private val basePath = parent.basePath
  private lazy val listener = parent.listener
  private lazy val isFairScheduler = parent.isFairScheduler

  def toNodeSeq: Seq[Node] = {
    listener.synchronized {
      stageTable(stageRow, stages)
    }
  }

  /** Special table that merges two header cells. */
  private def stageTable[T](makeRow: T => Seq[Node], rows: Seq[T]): Seq[Node] = {
    <table class="table table-bordered table-striped table-condensed sortable">
      <thead>
        <th>Stage Id</th>
        {if (isFairScheduler) {<th>Pool Name</th>} else {}}
        <th>Description</th>
        <th>Submitted</th>
        <th>Duration</th>
        <th>Tasks: Succeeded/Total</th>
        <th>Shuffle Read</th>
        <th>Shuffle Write</th>
      </thead>
      <tbody>
        {rows.map(r => makeRow(r))}
      </tbody>
    </table>
  }

  private def makeProgressBar(started: Int, completed: Int, failed: String, total: Int): Seq[Node] =
  {
    val completeWidth = "width: %s%%".format((completed.toDouble/total)*100)
    val startWidth = "width: %s%%".format((started.toDouble/total)*100)

    <div class="progress">
      <span style="text-align:center; position:absolute; width:100%; left:0;">
        {completed}/{total} {failed}
      </span>
      <div class="bar bar-completed" style={completeWidth}></div>
      <div class="bar bar-running" style={startWidth}></div>
    </div>
  }

  /** Render an HTML row that represents a stage */
  private def stageRow(s: StageInfo): Seq[Node] = {
    val poolName = listener.stageIdToPool.get(s.stageId)
    val nameLink =
      <a href={"%s/stages/stage?id=%s".format(UIUtils.prependBaseUri(basePath), s.stageId)}>
        {s.name}
      </a>
    val description = listener.stageIdToDescription.get(s.stageId)
      .map(d => <div><em>{d}</em></div><div>{nameLink}</div>).getOrElse(nameLink)
    val submissionTime = s.submissionTime match {
      case Some(t) => WebUI.formatDate(new Date(t))
      case None => "Unknown"
    }
    val finishTime = s.completionTime.getOrElse(System.currentTimeMillis)
    val duration = s.submissionTime.map { t =>
      if (finishTime > t) finishTime - t else System.currentTimeMillis - t
    }
    val formattedDuration = duration.map(d => parent.formatDuration(d)).getOrElse("Unknown")
    val startedTasks =
      listener.stageIdToTasksActive.getOrElse(s.stageId, HashMap[Long, TaskInfo]()).size
    val completedTasks = listener.stageIdToTasksComplete.getOrElse(s.stageId, 0)
    val failedTasks = listener.stageIdToTasksFailed.getOrElse(s.stageId, 0) match {
      case f if f > 0 => "(%s failed)".format(f)
      case _ => ""
    }
    val totalTasks = s.numTasks
    val shuffleReadSortable = listener.stageIdToShuffleRead.getOrElse(s.stageId, 0L)
    val shuffleRead = shuffleReadSortable match {
      case 0 => ""
      case b => Utils.bytesToString(b)
    }
    val shuffleWriteSortable = listener.stageIdToShuffleWrite.getOrElse(s.stageId, 0L)
    val shuffleWrite = shuffleWriteSortable match {
      case 0 => ""
      case b => Utils.bytesToString(b)
    }

    <tr>
      <td>{s.stageId}</td>
      {if (isFairScheduler) {
      <td>
        <a href={"%s/stages/pool?poolname=%s"
          .format(UIUtils.prependBaseUri(basePath), poolName.get)}>
          {poolName.get}
        </a>
      </td>
    }}
      <td>{description}</td>
      <td valign="middle">{submissionTime}</td>
      <td sorttable_customkey={duration.getOrElse(-1).toString}>{formattedDuration}</td>
      <td class="progress-cell">
        {makeProgressBar(startedTasks, completedTasks, failedTasks, totalTasks)}
      </td>
      <td sorttable_customekey={shuffleReadSortable.toString}>{shuffleRead}</td>
      <td sorttable_customekey={shuffleWriteSortable.toString}>{shuffleWrite}</td>
    </tr>
  }
}