tools/android/loading Move task creation to background and add dashboard

tools/android/loading/cloud/frontend/clovis_frontend.py

 # Copyright 2016 The Chromium Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
 import logging
+import datetime
 import math
 import os
 import sys
 import time
 
 import cloudstorage
 import flask
 from google.appengine.api import (app_identity, taskqueue)
 from google.appengine.ext import deferred
 from oauth2client.client import GoogleCredentials
 
 import common.clovis_paths
 from common.clovis_task import ClovisTask
 import common.google_bigquery_helper
 import common.google_instance_helper
 from common.loading_trace_database import LoadingTraceDatabase
 import email_helper
+from frontend_job import FrontendJob
 from memory_logs import MemoryLogs
 
 
 # Global variables.
 logging.Formatter.converter = time.gmtime
 clovis_logger = logging.getLogger('clovis_frontend')
 clovis_logger.setLevel(logging.DEBUG)
 project_name = app_identity.get_application_id()
 instance_helper = common.google_instance_helper.GoogleInstanceHelper(
     credentials=GoogleCredentials.get_application_default(),
@@ skipping 59 matching lines @@
     email_address (str): Email address of the user to be notified.
     status (str): Status of the task, indicating the success or the cause of
                   failure.
     task_url (str): URL where the results of the task can be found.
   """
   email_helper.SendEmailTaskComplete(
       to_address=email_address, tag=tag, status=status, task_url=task_url,
       logger=clovis_logger)
   clovis_logger.info('Scheduling instance group destruction for tag: ' + tag)
   deferred.defer(DeleteInstanceGroup, tag)
+  FrontendJob.DeleteForTag(tag)
 
 
 def GetEstimatedTaskDurationInSeconds(task):
   """Returns an estimation of the time required to run the task.
 
   Args:
     task: (ClovisTask) The task.
 
   Returns:
     float: Time estimation in seconds, or -1 in case of failure.
@@ skipping 205 matching lines @@
 
   # Load the task from JSON.
   task = ClovisTask.FromJsonString(http_body_str)
   if not task:
     clovis_logger.error('Invalid JSON task.')
     return Render('Invalid JSON task:\n' + http_body_str, memory_logs)
 
   task_tag = task.BackendParams()['tag']
   clovis_logger.info('Start processing %s task with tag %s.' % (task.Action(),
                                                                 task_tag))
+  user_email = email_helper.GetUserEmail()
+
+  # Write the job to the datastore.
+  frontend_job = FrontendJob.CreateForTag(task_tag)
+  frontend_job.email = user_email
+  frontend_job.status = 'not_started'
+  frontend_job.clovis_task = task.ToJsonString()
+  frontend_job.put()
+
+  # Process the job on the queue, to avoid timeout issues.
+  deferred.defer(SpawnTasksOnBackgroundQueue, task_tag)
+
+  return Render(
+      flask.Markup(
+          '<a href="%s">See progress.</a>' % FrontendJob.GetJobURL(task_tag)),
+      memory_logs)
+
+
+def SpawnTasksOnBackgroundQueue(task_tag):
+  """Spawns Clovis tasks associated with task_tag from the backgound queue.
+
+  This function is mostly a wrapper around SpawnTasks() that catches exceptions.
+  It is assumed that a FrontendJob for task_tag exists.
+  """
+  memory_logs = MemoryLogs(clovis_logger)
+  memory_logs.Start()
+  clovis_logger.info('Spawning tasks on background queue.')
+
+  try:
+    frontend_job = FrontendJob.GetFromTag(task_tag)
+    frontend_job.status = 'will_start'
+    SpawnTasks(frontend_job)
+  except Exception:
+    pass  # Catch all exceptions so that AppEngine does not retry indefinitely.

[mattcary, 2016/06/09 15:40:06]

Log the exception so if tasks disappear we have a chance at figuring out why?

[droger, 2016/06/10 12:14:04]

Done.

+
+  # Update the task.
+  if frontend_job:
+    frontend_job.log = memory_logs.Flush()
+    frontend_job.put()
+
+
+def SpawnTasks(frontend_job):
+  """ Spawns Clovis tasks associated with the frontend job."""
+  user_email = frontend_job.email
+  task = ClovisTask.FromJsonString(frontend_job.clovis_task)
+  task_tag = task.BackendParams()['tag']
+
   # Compute the task directory.
+  frontend_job.status = 'building_task_dir'
   task_dir_components = []
-  user_email = email_helper.GetUserEmail()
   user_name = None
   if user_email:
     user_name = user_email[:user_email.find('@')]
   if user_name:
     task_dir_components.append(user_name)
   task_name = task.BackendParams().get('task_name')
   if task_name:
     task_dir_components.append(task_name)
   task_dir_components.append(task_tag)
   task_dir = os.path.join(task.Action(), '_'.join(task_dir_components))
 
   # Build the URL where the result will live.
+  frontend_job.status = 'building_task_url'

[mattcary, 2016/06/09 15:40:06]

Do you need to put these status update values, or will it magically update?

[droger, 2016/06/10 12:14:04]

You need to put.

This is only done at the end (line 368) and not every time the status changes.

I'm adding a couple more calls now, but not every time the status changes
because I'm worried that calling it too often could cause problems.

[mattcary, 2016/06/10 12:17:27]

Shouldn't we put after each return, so we have the current status?

This is only going to be executed from the push queue, and so is going to have <
10 queries/minute. I shouldn't think that would be a problem for updates in the
datastore.

[droger, 2016/06/10 12:58:27]

put() will be executed after each return, by the caller code:
this function is called on line 361, and put() is called line 368.

[mattcary, 2016/06/10 13:24:20]

Ah! I was just looking in SpawnTasks. Thanks.

   task_url = GetTaskURL(task, task_dir)
   if task_url:
     clovis_logger.info('Task result URL: ' + task_url)
+    frontend_job.task_url = task_url
   else:
-    return Render('Could not build the task URL.', memory_logs)
+    frontend_job.status = 'task_url_error'
+    return
 
   # Split the task in smaller tasks.
+  frontend_job.status = 'splitting_task'
   sub_tasks = SplitClovisTask(task)
   if not sub_tasks:
-    return Render('Task split failed.', memory_logs)
+    frontend_job.status = 'task_split_error'
+    return
 
   # Compute estimates for the work duration, in order to compute the instance
   # count and the timeout.
+  frontend_job.status = 'estimating_duration'
   sequential_duration_s = \
       GetEstimatedTaskDurationInSeconds(sub_tasks[0]) * len(sub_tasks)
   if sequential_duration_s <= 0:
-    return Render('Time estimation failed.', memory_logs)
+    frontend_job.status = 'time_estimation_error'
+    return
 
   # Compute the number of required instances if not specified.
-  if not task.BackendParams().get('instance_count'):
+  if task.BackendParams().get('instance_count') is None:
+    frontend_job.status = 'estimating_instance_count'
     target_parallel_duration_s = 1800.0 # 30 minutes.
     task.BackendParams()['instance_count'] = math.ceil(
         sequential_duration_s / target_parallel_duration_s)
 
   # Check the instance quotas.
   clovis_logger.info(
       'Requesting %i instances.' % task.BackendParams()['instance_count'])
+  frontend_job.status = 'checking_instance_quotas'
   max_instances = instance_helper.GetAvailableInstanceCount()
   if max_instances == -1:
-    return Render('Failed to count the available instances.', memory_logs)
+    frontend_job.status = 'instance_count_error'
+    return
   elif task.BackendParams()['instance_count'] == 0:
-    return Render('Cannot create instances, quota exceeded.', memory_logs)
+    frontend_job.status = 'no_instance_available_error'
+    return
   elif max_instances < task.BackendParams()['instance_count']:
     clovis_logger.warning(
         'Instance count limited by quota: %i available / %i requested.' % (
             max_instances, task.BackendParams()['instance_count']))
     task.BackendParams()['instance_count'] = max_instances
 
   # Compute the timeout if there is none specified.
-  expected_duration_h = sequential_duration_s / (
-      task.BackendParams()['instance_count'] * 3600.0)
+  expected_duration_s = sequential_duration_s / (
+      task.BackendParams()['instance_count'])
+  frontend_job.eta = datetime.datetime.now() + datetime.timedelta(
+      seconds=expected_duration_s)
   if not task.BackendParams().get('timeout_hours'):
     # Timeout is at least 1 hour.
-    task.BackendParams()['timeout_hours'] = max(1, 5 * expected_duration_h)
+    task.BackendParams()['timeout_hours'] = max(
+        1, 5 * expected_duration_s / 3600.0)
   clovis_logger.info(
       'Timeout delay: %.1f hours. ' % task.BackendParams()['timeout_hours'])
 
+  frontend_job.status = 'queueing_tasks'
   if not EnqueueTasks(sub_tasks, task_tag):
-    return Render('Task creation failed.', memory_logs)
+    frontend_job.status = 'task_creation_error'
+    return
 
   # Start polling the progress.
   clovis_logger.info('Creating worker polling task.')
   first_poll_delay_minutes = 10
   deferred.defer(PollWorkers, task_tag, time.time(),
                  task.BackendParams()['timeout_hours'], user_email,
                  task_url, _countdown=(60 * first_poll_delay_minutes))
 
   # Start the instances if required.
+  frontend_job.status = 'creating_instances'
   if not CreateInstanceTemplate(task, task_dir):
-    return Render('Instance template creation failed.', memory_logs)
+    frontend_job.status = 'instance_template_error'
+    return
   if not CreateInstances(task):
-    return Render('Instance creation failed.', memory_logs)
+    frontend_job.status = 'instance_creation_error'
+    return
 
-  return Render(flask.Markup(
-      'Success!<br>Your task %s has started.<br>'
-      'Expected duration: %.1f hours.<br>'
-      'You will be notified at %s when completed.') % (
-          task_tag, expected_duration_h, user_email),
-      memory_logs)
-
+  frontend_job.status = 'started'
 
 def EnqueueTasks(tasks, task_tag):
   """Enqueues a list of tasks in the Google Cloud task queue, for consumption by
   Google Compute Engine.
   """
   q = taskqueue.Queue('clovis-queue')
   # Add tasks to the queue by groups.
   # TODO(droger): This supports thousands of tasks, but maybe not millions.
   # Defer the enqueuing if it times out.
   group_size = 100
@@ skipping 26 matching lines @@
 @app.route('/form_sent', methods=['POST'])
 def StartFromForm():
   """HTML form endpoint."""
   data_stream = flask.request.files.get('json_task')
   if not data_stream:
     return Render('Failed, no content.')
   http_body_str = data_stream.read()
   return StartFromJsonString(http_body_str)
 
 
+@app.route('/list_jobs')
+def ShowTaskList():
+  """Shows a list of all active jobs."""
+  tags = FrontendJob.ListJobs()
+
+  if not tags:
+    Render('No active jobs.')
+
+  html = flask.Markup('Active tasks:<ul>')
+  for tag in tags:
+    html += flask.Markup(
+        '<li><a href="%s">%s</a></li>') % (FrontendJob.GetJobURL(tag), tag)
+  html += flask.Markup('</ul>')
+  return Render(html)
+
+
+@app.route(FrontendJob.SHOW_JOB_URL)
+def ShowTask():
+  """Shows basic information abour a job."""
+  tag = flask.request.args.get('tag')
+  if not tag:
+    return Render('Invalid task tag.')
+
+  frontend_job = FrontendJob.GetFromTag(tag)
+
+  if not frontend_job:
+    return Render('Task not found.')
+
+  message = flask.Markup('Task details:' + frontend_job.RenderAsHtml())
+
+  log = None
+  if frontend_job.log:
+    log = frontend_job.log.split('\n')
+
+  return flask.render_template('log.html', body=message, log=log)
+
+
 @app.errorhandler(404)
 def PageNotFound(e):  # pylint: disable=unused-argument
   """Return a custom 404 error."""
   return 'Sorry, Nothing at this URL.', 404
 
 
 @app.errorhandler(500)
 def ApplicationError(e):
   """Return a custom 500 error."""
   return 'Sorry, unexpected error: {}'.format(e), 499