Monitors waiting time for executions (#132)

* Monitors waiting time for executions - Monitors via executor polling - Adds waitingSeconds info to execution model - Adds a progressbar component to display waiting / running times * Moves monitoring logic to executor & reformats * Makes Executor public again.
criteo · Jul 28, 2017 · 10d9f0e · 10d9f0e
1 parent 48fbb36
commit 10d9f0e
Show file tree

Hide file tree

Showing 7 changed files with 104 additions and 21 deletions.
diff --git a/core/src/main/javascript/app/pages/Execution.js b/core/src/main/javascript/app/pages/Execution.js
@@ -8,6 +8,8 @@ import ExitFullscreenIcon from "react-icons/lib/md/fullscreen-exit";
 import AutoScrollIcon from "react-icons/lib/md/arrow-downward";
 import BreakIcon from "react-icons/lib/md/keyboard-control";
 import CalendarIcon from "react-icons/lib/md/date-range";
+import ReactTooltip from "react-tooltip";
+
 import moment from "moment";
 
 import Window from "../components/Window";
@@ -43,6 +45,52 @@ type State = {
   autoScroll: boolean
 };
 
+type ProgressBarProps = {
+  classes: any,
+  totalTimeSeconds: number,
+  waitingTimeSeconds: number
+};
+
+const ProgressBarComponent = ({
+  classes,
+  totalTimeSeconds,
+  waitingTimeSeconds
+}: ProgressBarProps) => {
+  const totalWidth = 200;
+  const height = 8;
+  const barWidth = totalTimeSeconds !== 0
+    ? waitingTimeSeconds / totalTimeSeconds * totalWidth
+    : 0;
+
+  const tooltip =
+    `Waiting : ${moment.utc(waitingTimeSeconds * 1000).format("HH:mm:ss")} / ` +
+    `Running : ${moment
+      .utc((totalTimeSeconds - waitingTimeSeconds) * 1000)
+      .format("HH:mm:ss")}`;
+
+  return (
+    <span className={classes.main}>
+      <svg width={totalWidth} height={height} className={classes.svg}>
+        <g data-tip={tooltip}>
+          <rect width={totalWidth} height={height} fill="#00BCD4" />
+          <rect width={barWidth} height={height} fill="#ff9800" />
+        </g>
+      </svg>
+      <ReactTooltip className={classes.tooltip} effect="float" html={true} />
+    </span>
+  );
+};
+
+const ProgressBar = injectSheet({
+  main: {
+    "margin-left": "10px"
+  },
+  svg: {
+    lineHeight: "18px",
+    display: "inline-block"
+  }
+})(ProgressBarComponent);
+
 class Execution extends React.Component {
   props: Props;
   state: State;
@@ -241,13 +289,23 @@ class Execution extends React.Component {
                       <dt key="duration">Duration:</dt>,
                       <dd key="duration_">
                         {data.endTime
-                          ? moment
-                              .utc(
-                                moment(data.endTime).diff(
-                                  moment(data.startTime)
+                          ? [
+                              moment
+                                .utc(
+                                  moment(data.endTime).diff(
+                                    moment(data.startTime)
+                                  )
                                 )
-                              )
-                              .format("HH:mm:ss")
+                                .format("HH:mm:ss"),
+                              <ProgressBar
+                                totalTimeSeconds={
+                                  moment(data.endTime).diff(
+                                    moment(data.startTime)
+                                  ) / 1000
+                                }
+                                waitingTimeSeconds={data.waitingSeconds}
+                              />
+                            ]
                           : <Clock time={data.startTime} humanize={false} />}
                       </dd>
                     ]

diff --git a/core/src/main/javascript/datamodel.js b/core/src/main/javascript/datamodel.js
@@ -24,7 +24,8 @@ export type ExecutionLog = {
   failing?: {
     failedExecutions: Array<ExecutionLog>,
     nextRetry: ?string
-  }
+  },
+  waitingSeconds : number
 };
 
 export type Paginated<A> = {

diff --git a/core/src/main/scala/com/criteo/cuttle/App.scala b/core/src/main/scala/com/criteo/cuttle/App.scala
@@ -69,7 +69,8 @@ private[cuttle] object App {
               "failedExecutions" -> Json.fromValues(failedExecutions.map(_.asJson(executionLogEncoder))),
               "nextRetry" -> nextRetry.asJson
             )
-        }.asJson
+        }.asJson,
+        "waitingSeconds" -> execution.waitingSeconds.asJson
       )
   }
 

diff --git a/core/src/main/scala/com/criteo/cuttle/Cuttle.scala b/core/src/main/scala/com/criteo/cuttle/Cuttle.scala
@@ -20,6 +20,7 @@ class CuttleProject[S <: Scheduling] private[cuttle] (
   ) = {
     val xa = Database.connect(databaseConfig)
     val executor = new Executor[S](platforms, xa)(retryStrategy)
+
     Server.listen(port = httpPort, onError = { e =>
       e.printStackTrace()
       InternalServerError(e.getMessage)

diff --git a/core/src/main/scala/com/criteo/cuttle/Database.scala b/core/src/main/scala/com/criteo/cuttle/Database.scala
@@ -62,6 +62,7 @@ private[cuttle] object Database {
         end_time    DATETIME NOT NULL,
         context_id  VARCHAR(1000) NOT NULL,
         success     BOOLEAN NOT NULL,
+        waiting_seconds INT NOT NULL,
         PRIMARY KEY (id)
       ) ENGINE = INNODB;
 
@@ -128,8 +129,8 @@ private[cuttle] trait Queries {
     for {
       contextId <- logContext
       result <- sql"""
-        INSERT INTO executions (id, job, start_time, end_time, success, context_id)
-        VALUES (${e.id}, ${e.job}, ${e.startTime}, ${e.endTime}, ${e.status}, ${contextId})
+        INSERT INTO executions (id, job, start_time, end_time, success, context_id, waiting_seconds)
+        VALUES (${e.id}, ${e.job}, ${e.startTime}, ${e.endTime}, ${e.status}, ${contextId}, ${e.waitingSeconds})
         """.update.run
     } yield result
 
@@ -158,29 +159,29 @@ private[cuttle] trait Queries {
       case _ => sql"ORDER BY end_time, id DESC"
     }
     (sql"""
-      SELECT executions.id, job, start_time, end_time, contexts.json AS context, success
+      SELECT executions.id, job, start_time, end_time, contexts.json AS context, success, executions.waiting_seconds
       FROM executions INNER JOIN (""" ++ contextQuery ++ sql""") contexts
       ON executions.context_id = contexts.id WHERE """ ++ Fragments.in(
       fr"job",
       NonEmptyList.fromListUnsafe(jobs.toList)) ++ orderBy ++ sql""" LIMIT $limit OFFSET $offset""")
-      .query[(String, String, Instant, Instant, Json, ExecutionStatus)]
+      .query[(String, String, Instant, Instant, Json, ExecutionStatus, Int)]
       .list
       .map(_.map {
-        case (id, job, startTime, endTime, context, status) =>
-          ExecutionLog(id, job, Some(startTime), Some(endTime), context, status)
+        case (id, job, startTime, endTime, context, status, waitingSeconds) =>
+          ExecutionLog(id, job, Some(startTime), Some(endTime), context, status, waitingSeconds = waitingSeconds)
       })
   }
 
   def getExecutionById(contextQuery: Fragment, id: String): ConnectionIO[Option[ExecutionLog]] =
     (sql"""
-      SELECT executions.id, job, start_time, end_time, contexts.json AS context, success
+      SELECT executions.id, job, start_time, end_time, contexts.json AS context, success, executions.waiting_seconds
       FROM executions INNER JOIN (""" ++ contextQuery ++ sql""") contexts
       ON executions.context_id = contexts.id WHERE executions.id = $id""")
-      .query[(String, String, Instant, Instant, Json, ExecutionStatus)]
+      .query[(String, String, Instant, Instant, Json, ExecutionStatus, Int)]
       .option
       .map(_.map {
-        case (id, job, startTime, endTime, context, status) =>
-          ExecutionLog(id, job, Some(startTime), Some(endTime), context, status)
+        case (id, job, startTime, endTime, context, status, waitingSeconds) =>
+          ExecutionLog(id, job, Some(startTime), Some(endTime), context, status, waitingSeconds = waitingSeconds)
       })
 
   def unpauseJob[S <: Scheduling](job: Job[S]): ConnectionIO[Int] =

diff --git a/core/src/main/scala/com/criteo/cuttle/Executor.scala b/core/src/main/scala/com/criteo/cuttle/Executor.scala
@@ -55,7 +55,8 @@ private[cuttle] case class ExecutionLog(
   endTime: Option[Instant],
   context: Json,
   status: ExecutionStatus,
-  failing: Option[FailingJob] = None
+  failing: Option[FailingJob] = None,
+  waitingSeconds: Int
 )
 
 private object ExecutionCancelledException extends RuntimeException("Execution cancelled")
@@ -72,6 +73,7 @@ case class Execution[S <: Scheduling](
   private[cuttle] val cancelSignal = Promise[Nothing]
   def isCancelled = cancelSignal.isCompleted
   val cancelled = cancelSignal.future
+  var waitingSeconds = 0
   def onCancelled(thunk: () => Unit) = cancelled.andThen {
     case Failure(_) =>
       thunk()
@@ -91,10 +93,15 @@ case class Execution[S <: Scheduling](
       startTime,
       None,
       context.toJson,
-      status
+      status,
+      waitingSeconds = waitingSeconds
     )
 
   private[cuttle] val isParked = new AtomicBoolean(false)
+
+  private[cuttle] def updateWaitingTime(seconds: Int): Unit =
+    waitingSeconds += seconds
+
   def park(duration: FiniteDuration): Future[Unit] =
     if (isParked.get) {
       sys.error(s"Already parked")
@@ -147,6 +154,8 @@ class Executor[S <: Scheduling] private[cuttle] (
   private val recentFailures = TMap.empty[(Job[S], S#Context), (Option[Execution[S]], FailingJob)]
   private val timer = new Timer("com.criteo.cuttle.Executor.timer")
 
+  startMonitoringExecutions()
+
   private def flagWaitingExecutions(executions: Seq[Execution[S]]): Seq[(Execution[S], ExecutionStatus)] = {
     val waitings = platforms.flatMap(_.waiting).toSet
     executions.map { execution =>
@@ -520,4 +529,16 @@ class Executor[S <: Scheduling] private[cuttle] (
         }
       ))
   }
+
+  private def startMonitoringExecutions() = {
+    val SC = fs2.Scheduler.fromFixedDaemonPool(1, "com.criteo.cuttle.platforms.ExecutionMonitor.SC")
+
+    val intervalSeconds = 1
+
+    SC.scheduleAtFixedRate(intervalSeconds.second) {
+      runningExecutions
+        .filter({ case (_, s) => s == ExecutionStatus.ExecutionWaiting })
+        .foreach({ case (e, _) => e.updateWaitingTime(intervalSeconds) })
+    }
+  }
 }
diff --git a/timeseries/src/main/scala/com/criteo/cuttle/timeseries/TimeSeriesApp.scala b/timeseries/src/main/scala/com/criteo/cuttle/timeseries/TimeSeriesApp.scala
@@ -108,7 +108,7 @@ private[timeseries] trait TimeSeriesApp { self: TimeSeriesScheduler =>
             (lo, hi) <- grid.split(interval)
           } yield {
             val context = TimeSeriesContext(grid.truncate(lo), grid.ceil(hi), maybeBackfill)
-            ExecutionLog("", job.id, None, None, context.asJson, ExecutionTodo, None)
+            ExecutionLog("", job.id, None, None, context.asJson, ExecutionTodo, None, 0)
           }
         val throttledExecutions = executor.allFailingExecutions
           .filter(e => e.job == job && e.context.toInterval.intersects(requestedInterval))