docs/server2/cron_servlet.py

# Copyright 2013 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 posixpath
import traceback

from app_yaml_helper import AppYamlHelper
from appengine_wrappers import IsDeadlineExceededError, logservice
from branch_utility import BranchUtility
from compiled_file_system import CompiledFileSystem
from data_source_registry import CreateDataSources
from environment import GetAppVersion, IsDevServer
from extensions_paths import EXAMPLES, PUBLIC_TEMPLATES, STATIC_DOCS
from file_system_util import CreateURLsFromPaths
from future import Future
from gcs_file_system_provider import CloudStorageFileSystemProvider
from github_file_system_provider import GithubFileSystemProvider
from host_file_system_provider import HostFileSystemProvider
from object_store_creator import ObjectStoreCreator
from render_servlet import RenderServlet
from server_instance import ServerInstance
from servlet import Servlet, Request, Response
from special_paths import SITE_VERIFICATION_FILE
from timer import Timer, TimerClosure


class _SingletonRenderServletDelegate(RenderServlet.Delegate):
  def __init__(self, server_instance):
    self._server_instance = server_instance

  def CreateServerInstance(self):
    return self._server_instance

class _CronLogger(object):
  '''Wraps the logging.* methods to prefix them with 'cron' and flush
  immediately. The flushing is important because often these cron runs time
  out and we lose the logs.
  '''
  def info(self, msg, *args):    self._log(logging.info, msg, args)
  def warning(self, msg, *args): self._log(logging.warning, msg, args)
  def error(self, msg, *args):   self._log(logging.error, msg, args)

  def _log(self, logfn, msg, args):
    try:
      logfn('cron: %s' % msg, *args)
    finally:
      logservice.flush()

_cronlog = _CronLogger()

def _RequestEachItem(title, items, request_callback):
  '''Runs a task |request_callback| named |title| for each item in |items|.
  |request_callback| must take an item and return a servlet response.
  Returns true if every item was successfully run, false if any return a
  non-200 response or raise an exception.
  '''
  _cronlog.info('%s: starting', title)
  success_count, failure_count = 0, 0
  timer = Timer()
  try:
    for i, item in enumerate(items):
      def error_message(detail):
        return '%s: error rendering %s (%s of %s): %s' % (
            title, item, i + 1, len(items), detail)
      try:
        response = request_callback(item)
        if response.status == 200:
          success_count += 1
        else:
          _cronlog.error(error_message('response status %s' % response.status))
          failure_count += 1
      except Exception as e:
        _cronlog.error(error_message(traceback.format_exc()))
        failure_count += 1
        if IsDeadlineExceededError(e): raise
  finally:
    _cronlog.info('%s: rendered %s of %s with %s failures in %s',
        title, success_count, len(items), failure_count,
        timer.Stop().FormatElapsed())
  return success_count == len(items)

class CronServlet(Servlet):
  '''Servlet which runs a cron job.
  '''
  def __init__(self, request, delegate_for_test=None):
    Servlet.__init__(self, request)
    self._delegate = delegate_for_test or CronServlet.Delegate()

  class Delegate(object):
    '''CronServlet's runtime dependencies. Override for testing.
    '''
    def CreateBranchUtility(self, object_store_creator):
      return BranchUtility.Create(object_store_creator)

    def CreateHostFileSystemProvider(self,
                                     object_store_creator,
                                     max_trunk_revision=None):
      return HostFileSystemProvider(object_store_creator,
                                    max_trunk_revision=max_trunk_revision)

    def CreateGithubFileSystemProvider(self, object_store_creator):
      return GithubFileSystemProvider(object_store_creator)

    def CreateGCSFileSystemProvider(self, object_store_creator):
      return CloudStorageFileSystemProvider(object_store_creator)

    def GetAppVersion(self):
      return GetAppVersion()

  def Get(self):
    # Crons often time out, and if they do we need to make sure to flush the
    # logs before the process gets killed (Python gives us a couple of
    # seconds).
    #
    # So, manually flush logs at the end of the cron run. However, sometimes
    # even that isn't enough, which is why in this file we use _cronlog and
    # make it flush the log every time its used.
    logservice.AUTOFLUSH_ENABLED = False
    try:
      return self._GetImpl()
    except BaseException:
      _cronlog.error('Caught top-level exception! %s', traceback.format_exc())
    finally:
      logservice.flush()

  def _GetImpl(self):
    # Cron strategy:
    #
    # Find all public template files and static files, and render them. Most of
    # the time these won't have changed since the last cron run, so it's a
    # little wasteful, but hopefully rendering is really fast (if it isn't we
    # have a problem).
    _cronlog.info('starting')

    # This is returned every time RenderServlet wants to create a new
    # ServerInstance.
    #
    # TODO(kalman): IMPORTANT. This sometimes throws an exception, breaking
    # everything. Need retry logic at the fetcher level.
    server_instance = self._GetSafeServerInstance()
    trunk_fs = server_instance.host_file_system_provider.GetTrunk()

    def render(path):
      request = Request(path, self._request.host, self._request.headers)
      delegate = _SingletonRenderServletDelegate(server_instance)
      return RenderServlet(request, delegate).Get()

    def request_files_in_dir(path, prefix='', strip_ext=None):
      '''Requests every file found under |path| in this host file system, with
      a request prefix of |prefix|. |strip_ext| is an optional list of file
      extensions that should be stripped from paths before requesting.
      '''
      def maybe_strip_ext(name):
        if name == SITE_VERIFICATION_FILE or not strip_ext:
          return name
        base, ext = posixpath.splitext(name)
        return base if ext in strip_ext else name
      files = [maybe_strip_ext(name)
               for name, _ in CreateURLsFromPaths(trunk_fs, path, prefix)]
      return _RequestEachItem(path, files, render)

    results = []

    try:
      # Start running the hand-written Cron methods first; they can be run in
      # parallel. They are resolved at the end.
      def run_cron_for_future(target):
        title = target.__class__.__name__
        future, init_timer = TimerClosure(target.Cron)
        assert isinstance(future, Future), (
            '%s.Cron() did not return a Future' % title)
        def resolve():
          resolve_timer = Timer()
          try:
            future.Get()
          except Exception as e:
            _cronlog.error('%s: error %s' % (title, traceback.format_exc()))
            results.append(False)
            if IsDeadlineExceededError(e): raise
          finally:
            resolve_timer.Stop()
            _cronlog.info('%s took %s: %s to initialize and %s to resolve' %
                (title,
                 init_timer.With(resolve_timer).FormatElapsed(),
                 init_timer.FormatElapsed(),
                 resolve_timer.FormatElapsed()))
        return Future(callback=resolve)

      targets = (CreateDataSources(server_instance).values() +
                 [server_instance.content_providers,
                  server_instance.api_models])
      title = 'initializing %s parallel Cron targets' % len(targets)
      _cronlog.info(title)
      timer = Timer()
      try:
        cron_futures = [run_cron_for_future(target) for target in targets]
      finally:
        _cronlog.info('%s took %s' % (title, timer.Stop().FormatElapsed()))

      # Samples are too expensive to run on the dev server, where there is no
      # parallel fetch.
      #
      # XXX(kalman): Currently samples are *always* too expensive to fetch, so
      # disabling them for now. It won't break anything so long as we're still
      # not enforcing that everything gets cached for normal instances.
      if False:  # should be "not IsDevServer()":
        # Fetch each individual sample file.
        results.append(request_files_in_dir(EXAMPLES,
                                            prefix='extensions/examples'))

      # Resolve the hand-written Cron method futures.
      title = 'resolving %s parallel Cron targets' % len(targets)
      _cronlog.info(title)
      timer = Timer()
      try:
        for future in cron_futures:
          future.Get()
      finally:
        _cronlog.info('%s took %s' % (title, timer.Stop().FormatElapsed()))

    except:
      results.append(False)
      # This should never actually happen (each cron step does its own
      # conservative error checking), so re-raise no matter what it is.
      _cronlog.error('uncaught error: %s' % traceback.format_exc())
      raise
    finally:
      success = all(results)
      _cronlog.info('finished (%s)', 'success' if success else 'FAILED')
      return (Response.Ok('Success') if success else
              Response.InternalError('Failure'))

  def _GetSafeServerInstance(self):
    '''Returns a ServerInstance with a host file system at a safe revision,
    meaning the last revision that the current running version of the server
    existed.
    '''
    delegate = self._delegate

    # IMPORTANT: Get a ServerInstance pinned to the most recent revision, not
    # HEAD. These cron jobs take a while and run very frequently such that
    # there is usually one running at any given time, and eventually a file
    # that we're dealing with will change underneath it, putting the server in
    # an undefined state.
    server_instance_near_head = self._CreateServerInstance(
        self._GetMostRecentRevision())

    app_yaml_handler = AppYamlHelper(
        server_instance_near_head.object_store_creator,
        server_instance_near_head.host_file_system_provider)

    if app_yaml_handler.IsUpToDate(delegate.GetAppVersion()):
      return server_instance_near_head

    # The version in app.yaml is greater than the currently running app's.
    # The safe version is the one before it changed.
    safe_revision = app_yaml_handler.GetFirstRevisionGreaterThan(
        delegate.GetAppVersion()) - 1

    _cronlog.info('app version %s is out of date, safe is %s',
        delegate.GetAppVersion(), safe_revision)

    return self._CreateServerInstance(safe_revision)

  def _GetMostRecentRevision(self):
    '''Gets the revision of the most recent patch submitted to the host file
    system. This is similar to HEAD but it's a concrete revision so won't
    change as the cron runs.
    '''
    head_fs = (
        self._CreateServerInstance(None).host_file_system_provider.GetTrunk())
    return head_fs.Stat('').version

  def _CreateServerInstance(self, revision):
    '''Creates a ServerInstance pinned to |revision|, or HEAD if None.
    NOTE: If passed None it's likely that during the cron run patches will be
    submitted at HEAD, which may change data underneath the cron run.
    '''
    object_store_creator = ObjectStoreCreator(start_empty=True)
    branch_utility = self._delegate.CreateBranchUtility(object_store_creator)
    host_file_system_provider = self._delegate.CreateHostFileSystemProvider(
        object_store_creator, max_trunk_revision=revision)
    github_file_system_provider = self._delegate.CreateGithubFileSystemProvider(
        object_store_creator)
    gcs_file_system_provider = self._delegate.CreateGCSFileSystemProvider(
        object_store_creator)
    return ServerInstance(object_store_creator,
                          CompiledFileSystem.Factory(object_store_creator),
                          branch_utility,
                          host_file_system_provider,
                          github_file_system_provider,
                          gcs_file_system_provider)