files: #!/usr/bin/env python
  
  """ A service for handling TaskQueue request from application servers.
  It uses RabbitMQ and Celery to handle tasks. """
  
  import base64
  import datetime
  import hashlib
  import json
  import os
  import socket
  import sys
  import time
  import tq_lib
  
  from cassandra import OperationTimedOut
  from cassandra.cluster import SimpleStatement
  from cassandra.policies import FallthroughRetryPolicy
  from distutils.spawn import find_executable
  from queue import InvalidLeaseRequest
  from queue import PullQueue
  from queue import PushQueue
  from task import Task
  from tq_config import TaskQueueConfig
  from .unpackaged import APPSCALE_LIB_DIR
  from .unpackaged import APPSCALE_PYTHON_APPSERVER
  from .utils import (get_celery_queue_name,
                      get_queue_function_name,
                      logger)
  
  sys.path.append(APPSCALE_LIB_DIR)
  import appscale_info
  import constants
  import file_io
  import monit_app_configuration
  import monit_interface
  
  sys.path.append(APPSCALE_PYTHON_APPSERVER)
  from google.appengine.api import apiproxy_stub_map
  from google.appengine.api import datastore_errors
  from google.appengine.api import datastore_distributed
  from google.appengine.api.taskqueue import taskqueue_service_pb
  from google.appengine.ext import db
  from google.appengine.runtime import apiproxy_errors
  
  sys.path.append(TaskQueueConfig.CELERY_WORKER_DIR)
  
  # A policy that does not retry statements.
  NO_RETRIES = FallthroughRetryPolicy()
  
  
  def create_pull_queue_tables(cluster, session):
    """ Create the required tables for pull queues.
  
    Args:
      cluster: A cassandra-driver cluster.
      session: A cassandra-driver session.
    """
    logger.info('Trying to create pull_queue_tasks')
    create_table = """
      CREATE TABLE IF NOT EXISTS pull_queue_tasks (
        app text,
        queue text,
        id text,
        payload text,
        enqueued timestamp,
        lease_expires timestamp,
        retry_count int,
        tag text,
        PRIMARY KEY ((app, queue, id))
      )
    """
    statement = SimpleStatement(create_table, retry_policy=NO_RETRIES)
    try:
      session.execute(statement)
    except OperationTimedOut:
      logger.warning(
        'Encountered an operation timeout while creating pull_queue_tasks. '
        'Waiting 1 minute for schema to settle.')
      time.sleep(60)
      raise
  
    logger.info('Trying to create pull_queue_tasks_index')
    create_index_table = """
      CREATE TABLE IF NOT EXISTS pull_queue_tasks_index (
        app text,
        queue text,
        eta timestamp,
        id text,
        tag text,
        tag_exists boolean,
        PRIMARY KEY ((app, queue, eta), id)
      ) WITH gc_grace_seconds = 120
    """
    statement = SimpleStatement(create_index_table, retry_policy=NO_RETRIES)
    try:
      session.execute(statement)
    except OperationTimedOut:
      logger.warning(
        'Encountered an operation timeout while creating pull_queue_tasks_index.'
        ' Waiting 1 minute for schema to settle.')
      time.sleep(60)
      raise
  
    logger.info('Trying to create pull_queue_tags index')
    create_index = """
      CREATE INDEX IF NOT EXISTS pull_queue_tags ON pull_queue_tasks_index (tag);
    """
    session.execute(create_index)
  
    # This additional index is needed for groupByTag=true,tag=None queries
    # because Cassandra can only do '=' queries on secondary indices.
    logger.info('Trying to create pull_queue_tag_exists index')
    create_index = """
      CREATE INDEX IF NOT EXISTS pull_queue_tag_exists
      ON pull_queue_tasks_index (tag_exists);
    """
    session.execute(create_index)
  
    logger.info('Trying to create pull_queue_leases')
    create_leases_table = """
      CREATE TABLE IF NOT EXISTS pull_queue_leases (
        app text,
        queue text,
        leased timestamp,
        PRIMARY KEY ((app, queue, leased))
      ) WITH gc_grace_seconds = 120
    """
    statement = SimpleStatement(create_leases_table, retry_policy=NO_RETRIES)
    try:
      session.execute(statement)
    except OperationTimedOut:
      logger.warning(
        'Encountered an operation timeout while creating pull_queue_leases. '
        'Waiting 1 minute for schema to settle.')
      time.sleep(60)
      raise
  
  
  class TaskName(db.Model):
    """ A datastore model for tracking task names in order to prevent
    tasks with the same name from being enqueued repeatedly.
    
    Attributes:
      timestamp: The time the task was enqueued.
    """
    STORED_KIND_NAME = "__task_name__"
    timestamp = db.DateTimeProperty(auto_now_add=True)
    queue = db.StringProperty(required=True)
    state = db.StringProperty(required=True)
    endtime = db.DateTimeProperty()
    app_id = db.StringProperty(required=True)
  
    @classmethod
    def kind(cls):
      """ Kind name override. """
      return cls.STORED_KIND_NAME
  
  def setup_env():
    """ Sets required environment variables for GAE datastore library. """
    os.environ['AUTH_DOMAIN'] = "appscale.com"
    os.environ['USER_EMAIL'] = ""
    os.environ['USER_NICKNAME'] = ""
    os.environ['APPLICATION_ID'] = ""
  
  class DistributedTaskQueue():
    """ AppScale taskqueue layer for the TaskQueue API. """
  
    # Required start worker name tags.
    SETUP_WORKERS_TAGS = ['app_id']
  
    # Required stop worker name tags.
    STOP_WORKERS_TAGS = ['app_id']
  
    # The location of where celery logs go.
    LOG_DIR = "/var/log/appscale/celery_workers/"
  
    # The hard time limit of a running task in seconds, extra
    # time over the soft limit allows it to catch up to interrupts.
    HARD_TIME_LIMIT = 610
  
    #
  
  [... the rest is omitted ...]
files: AppTaskQueue/appscale/taskqueue/distributed_tq.py
files: #!/usr/bin/env python
  
  """ A service for handling TaskQueue request from application servers.
  It uses RabbitMQ and Celery to handle tasks. """
  
  import base64
  import datetime
  import hashlib
  import json
  import os
  import socket
  import sys
  import time
  import tq_lib
  
  from cassandra import OperationTimedOut
  from cassandra.cluster import SimpleStatement
  from cassandra.policies import FallthroughRetryPolicy
  from distutils.spawn import find_executable
  from queue import InvalidLeaseRequest
  from queue import PullQueue
  from queue import PushQueue
  from task import Task
  from tq_config import TaskQueueConfig
  from .unpackaged import APPSCALE_LIB_DIR
  from .unpackaged import APPSCALE_PYTHON_APPSERVER
  from .utils import logger
  
  sys.path.append(APPSCALE_LIB_DIR)
  import appscale_info
  import constants
  import file_io
  import monit_app_configuration
  import monit_interface
  
  sys.path.append(APPSCALE_PYTHON_APPSERVER)
  from google.appengine.api import apiproxy_stub_map
  from google.appengine.api import datastore_errors
  from google.appengine.api import datastore_distributed
  from google.appengine.api.taskqueue import taskqueue_service_pb
  from google.appengine.ext import db
  from google.appengine.runtime import apiproxy_errors
  
  sys.path.append(TaskQueueConfig.CELERY_WORKER_DIR)
  
  # A policy that does not retry statements.
  NO_RETRIES = FallthroughRetryPolicy()
  
  
  def create_pull_queue_tables(cluster, session):
    """ Create the required tables for pull queues.
  
    Args:
      cluster: A cassandra-driver cluster.
      session: A cassandra-driver session.
    """
    logger.info('Trying to create pull_queue_tasks')
    create_table = """
      CREATE TABLE IF NOT EXISTS pull_queue_tasks (
        app text,
        queue text,
        id text,
        payload text,
        enqueued timestamp,
        lease_expires timestamp,
        retry_count int,
        tag text,
        PRIMARY KEY ((app, queue, id))
      )
    """
    statement = SimpleStatement(create_table, retry_policy=NO_RETRIES)
    try:
      session.execute(statement)
    except OperationTimedOut:
      logger.warning(
        'Encountered an operation timeout while creating pull_queue_tasks. '
        'Waiting 1 minute for schema to settle.')
      time.sleep(60)
      raise
  
    logger.info('Trying to create pull_queue_tasks_index')
    create_index_table = """
      CREATE TABLE IF NOT EXISTS pull_queue_tasks_index (
        app text,
        queue text,
        eta timestamp,
        id text,
        tag text,
        tag_exists boolean,
        PRIMARY KEY ((app, queue, eta), id)
      ) WITH gc_grace_seconds = 120
    """
    statement = SimpleStatement(create_index_table, retry_policy=NO_RETRIES)
    try:
      session.execute(statement)
    except OperationTimedOut:
      logger.warning(
        'Encountered an operation timeout while creating pull_queue_tasks_index.'
        ' Waiting 1 minute for schema to settle.')
      time.sleep(60)
      raise
  
    logger.info('Trying to create pull_queue_tags index')
    create_index = """
      CREATE INDEX IF NOT EXISTS pull_queue_tags ON pull_queue_tasks_index (tag);
    """
    session.execute(create_index)
  
    # This additional index is needed for groupByTag=true,tag=None queries
    # because Cassandra can only do '=' queries on secondary indices.
    logger.info('Trying to create pull_queue_tag_exists index')
    create_index = """
      CREATE INDEX IF NOT EXISTS pull_queue_tag_exists
      ON pull_queue_tasks_index (tag_exists);
    """
    session.execute(create_index)
  
    logger.info('Trying to create pull_queue_leases')
    create_leases_table = """
      CREATE TABLE IF NOT EXISTS pull_queue_leases (
        app text,
        queue text,
        leased timestamp,
        PRIMARY KEY ((app, queue, leased))
      ) WITH gc_grace_seconds = 120
    """
    statement = SimpleStatement(create_leases_table, retry_policy=NO_RETRIES)
    try:
      session.execute(statement)
    except OperationTimedOut:
      logger.warning(
        'Encountered an operation timeout while creating pull_queue_leases. '
        'Waiting 1 minute for schema to settle.')
      time.sleep(60)
      raise
  
  
  class TaskName(db.Model):
    """ A datastore model for tracking task names in order to prevent
    tasks with the same name from being enqueued repeatedly.
    
    Attributes:
      timestamp: The time the task was enqueued.
    """
    STORED_KIND_NAME = "__task_name__"
    timestamp = db.DateTimeProperty(auto_now_add=True)
    queue = db.StringProperty(required=True)
    state = db.StringProperty(required=True)
    endtime = db.DateTimeProperty()
    app_id = db.StringProperty(required=True)
  
    @classmethod
    def kind(cls):
      """ Kind name override. """
      return cls.STORED_KIND_NAME
  
  def setup_env():
    """ Sets required environment variables for GAE datastore library. """
    os.environ['AUTH_DOMAIN'] = "appscale.com"
    os.environ['USER_EMAIL'] = ""
    os.environ['USER_NICKNAME'] = ""
    os.environ['APPLICATION_ID'] = ""
  
  class DistributedTaskQueue():
    """ AppScale taskqueue layer for the TaskQueue API. """
  
    # Required start worker name tags.
    SETUP_WORKERS_TAGS = ['app_id']
  
    # Required stop worker name tags.
    STOP_WORKERS_TAGS = ['app_id']
  
    # The location of where celery logs go.
    LOG_DIR = "/var/log/appscale/celery_workers/"
  
    # The hard time limit of a running task in seconds, extra
    # time over the soft limit allows it to catch up to interrupts.
    HARD_TIME_LIMIT = 610
  
    # The soft time limit of a running task.
    TASK_SOFT
  
  [... the rest is omitted ...]
files: AppTaskQueue/appscale/taskqueue/distributed_tq.py
files: """ AppScale TaskQueue configuration class. It deals with the configuration
  file given with an application 'queue.yaml' or 'queue.xml'. """
  
  import os
  import sys
  
  from queue import InvalidQueueConfiguration
  from queue import PullQueue
  from queue import PushQueue
  from unpackaged import APPSCALE_LIB_DIR
  from unpackaged import APPSCALE_PYTHON_APPSERVER
  from .utils import (get_celery_annotation_name,
                      get_celery_queue_name,
                      get_celery_worker_module_name,
                      logger)
  
  sys.path.append(APPSCALE_PYTHON_APPSERVER)
  from google.appengine.api import queueinfo
  from google.appengine.runtime import apiproxy_errors
  
  sys.path.append(APPSCALE_LIB_DIR)
  import appscale_info
  import file_io
  import xmltodict
  
  class TaskQueueConfig():
    """ Contains configuration of the TaskQueue system. """
  
    # Min concurrency per worker.
    MIN_CELERY_CONCURRENCY = 2
  
    # Max concurrency per worker.
    MAX_CELERY_CONCURRENCY = 20
  
    # The default YAML used if a queue.yaml or queue.xml is not supplied.
    DEFAULT_QUEUE_YAML = """
  queue:
  - name: default
    rate: 5/s
  """
  
    # Location of all celery configuration files.
    CELERY_CONFIG_DIR = '/etc/appscale/celery/configuration/'
  
    # Location of all celery workers scripts.
    CELERY_WORKER_DIR = '/etc/appscale/celery/workers/'
  
    # Location where celery workers back up state to.
    CELERY_STATE_DIR = '/opt/appscale/celery/'
  
    # Directory with the task templates.
    TEMPLATE_DIR = os.path.join(
      os.path.dirname(sys.modules['appscale.taskqueue'].__file__), 'templates')
  
    # The location of a header of a queue worker script.
    HEADER_LOC = os.path.join(TEMPLATE_DIR, 'header.py')
    
    # The location of the task template code.
    TASK_LOC = os.path.join(TEMPLATE_DIR, 'task.py')
  
    # XML configs use "-" while yaml uses "_". These are the tags which 
    # need to be converted to match the yaml tags.
    YAML_TO_XML_TAGS_TO_CONVERT = ['bucket-size', 'max-concurrent-request', 
                                   'retry-parameters', 'task-age-limit', 
                                   'total-storage-limit', 'user-email',
                                   'write-email', 'min-backoff-seconds',
                                   'max-backoff-seconds', 'max-doublings',
                                   'task-retry-limit', 'max-concurrent-requests']
  
    # Tags from queue.xml that are ignored.
    TAGS_TO_IGNORE = ['#text']
  
    def __init__(self, app_id, db_access=None):
      """ Configuration constructor. 
  
      Args:
        app_id: The application ID.
        db_access: A DatastoreProxy object.
      """
      self._app_id = app_id
      file_io.mkdir(self.CELERY_CONFIG_DIR)
      file_io.mkdir(self.CELERY_WORKER_DIR)
      self.db_access = db_access
      self.queues = self.load_queues_from_file()
  
    def get_queue_file_location(self, app_id):
      """ Gets the location of the queue.yaml or queue.xml file of a given
      application.
  
      Args:
        app_id: The application ID.
      Returns:
        The location of the queue.yaml or queue.xml file, and 
        an empty string if it could not be found.
      Raises:
        apiproxy_errors.ApplicationError if multiple invalid files are found.
      """
      app_dir = appscale_info.get_app_path(app_id)
  
      queue_yaml = 'queue.yaml'
      queue_xml = 'queue.xml'
      
      paths = []
      queue_yaml_detected = False
      queue_xml_detected = False
      for root, sub_dirs, files in os.walk(app_dir):
        for file in files:
          if file == queue_yaml:
            queue_yaml_detected = True
            paths.append(os.path.abspath(os.path.join(root, file)))
          if file == queue_xml:
            queue_xml_detected = True
            paths.append(os.path.abspath(os.path.join(root, file)))
  
      if not paths:
        return ""
  
      paths = sorted(paths, key = lambda k : len(k.split(os.sep)))
      if len(paths) == 1:
        return paths[0]
  
      # If multiple files were detected and it's Python, return
      # the shortest path.
      if queue_yaml_detected:
        return paths[0]
      
      # If multiple files were detected and it's Java, return
      # the first path that contains WEB-INF.
      for path in paths:
        if queue_xml in path and "WEB-INF" in path and queue_xml_detected:
          return path
      
      raise apiproxy_errors.\
        ApplicationError("Multiple unusable queue configuration files detected")
  
    def load_queues_from_file(self):
      """ Translates an application's queue configuration file to queue objects.
     
      Returns:
        A dictionary mapping queue names to Queue objects.
      Raises:
        ValueError: If queue_file is unable to get loaded.
      """
      using_default = False
      queue_file = ''
  
      try:
        queue_file = self.get_queue_file_location(self._app_id)
        try:
          info = file_io.read(queue_file)
          logger.info('Found queue file for {} in: {}'.
            format(self._app_id, queue_file))
        except IOError:
          logger.error(
            'No queue file found for {}, using default queue'.format(self._app_id))
          info = self.DEFAULT_QUEUE_YAML
          using_default = True
      except apiproxy_errors.ApplicationError as application_error:
        logger.error(application_error.message)
        info = self.DEFAULT_QUEUE_YAML
        using_default = True
  
      #TODO handle bad xml/yaml files.
      if queue_file.endswith('yaml')
  
  [... the rest is omitted ...]
files: AppTaskQueue/appscale/taskqueue/tq_config.py
files: """ AppScale TaskQueue configuration class. It deals with the configuration
  file given with an application 'queue.yaml' or 'queue.xml'. """
  
  import os
  import sys
  
  from queue import InvalidQueueConfiguration
  from queue import PullQueue
  from queue import PushQueue
  from unpackaged import APPSCALE_LIB_DIR
  from unpackaged import APPSCALE_PYTHON_APPSERVER
  from .utils import logger
  
  sys.path.append(APPSCALE_PYTHON_APPSERVER)
  from google.appengine.api import queueinfo
  from google.appengine.runtime import apiproxy_errors
  
  sys.path.append(APPSCALE_LIB_DIR)
  import appscale_info
  import file_io
  import xmltodict
  
  class TaskQueueConfig():
    """ Contains configuration of the TaskQueue system. """
  
    # Min concurrency per worker.
    MIN_CELERY_CONCURRENCY = 2
  
    # Max concurrency per worker.
    MAX_CELERY_CONCURRENCY = 20
  
    # The default YAML used if a queue.yaml or queue.xml is not supplied.
    DEFAULT_QUEUE_YAML = """
  queue:
  - name: default
    rate: 5/s
  """
  
    # Location of all celery configuration files.
    CELERY_CONFIG_DIR = '/etc/appscale/celery/configuration/'
  
    # Location of all celery workers scripts.
    CELERY_WORKER_DIR = '/etc/appscale/celery/workers/'
  
    # Location where celery workers back up state to.
    CELERY_STATE_DIR = '/opt/appscale/celery/'
  
    # Directory with the task templates.
    TEMPLATE_DIR = os.path.join(
      os.path.dirname(sys.modules['appscale.taskqueue'].__file__), 'templates')
  
    # The location of a header of a queue worker script.
    HEADER_LOC = os.path.join(TEMPLATE_DIR, 'header.py')
    
    # The location of the task template code.
    TASK_LOC = os.path.join(TEMPLATE_DIR, 'task.py')
  
    # XML configs use "-" while yaml uses "_". These are the tags which 
    # need to be converted to match the yaml tags.
    YAML_TO_XML_TAGS_TO_CONVERT = ['bucket-size', 'max-concurrent-request', 
                                   'retry-parameters', 'task-age-limit', 
                                   'total-storage-limit', 'user-email',
                                   'write-email', 'min-backoff-seconds',
                                   'max-backoff-seconds', 'max-doublings',
                                   'task-retry-limit', 'max-concurrent-requests']
  
    # Tags from queue.xml that are ignored.
    TAGS_TO_IGNORE = ['#text']
  
    def __init__(self, app_id, db_access=None):
      """ Configuration constructor. 
  
      Args:
        app_id: The application ID.
        db_access: A DatastoreProxy object.
      """
      self._app_id = app_id
      file_io.mkdir(self.CELERY_CONFIG_DIR)
      file_io.mkdir(self.CELERY_WORKER_DIR)
      self.db_access = db_access
      self.queues = self.load_queues_from_file()
  
    def get_queue_file_location(self, app_id):
      """ Gets the location of the queue.yaml or queue.xml file of a given
      application.
  
      Args:
        app_id: The application ID.
      Returns:
        The location of the queue.yaml or queue.xml file, and 
        an empty string if it could not be found.
      Raises:
        apiproxy_errors.ApplicationError if multiple invalid files are found.
      """
      app_dir = appscale_info.get_app_path(app_id)
  
      queue_yaml = 'queue.yaml'
      queue_xml = 'queue.xml'
      
      paths = []
      queue_yaml_detected = False
      queue_xml_detected = False
      for root, sub_dirs, files in os.walk(app_dir):
        for file in files:
          if file == queue_yaml:
            queue_yaml_detected = True
            paths.append(os.path.abspath(os.path.join(root, file)))
          if file == queue_xml:
            queue_xml_detected = True
            paths.append(os.path.abspath(os.path.join(root, file)))
  
      if not paths:
        return ""
  
      paths = sorted(paths, key = lambda k : len(k.split(os.sep)))
      if len(paths) == 1:
        return paths[0]
  
      # If multiple files were detected and it's Python, return
      # the shortest path.
      if queue_yaml_detected:
        return paths[0]
      
      # If multiple files were detected and it's Java, return
      # the first path that contains WEB-INF.
      for path in paths:
        if queue_xml in path and "WEB-INF" in path and queue_xml_detected:
          return path
      
      raise apiproxy_errors.\
        ApplicationError("Multiple unusable queue configuration files detected")
  
    def load_queues_from_file(self):
      """ Translates an application's queue configuration file to queue objects.
     
      Returns:
        A dictionary mapping queue names to Queue objects.
      Raises:
        ValueError: If queue_file is unable to get loaded.
      """
      using_default = False
      queue_file = ''
  
      try:
        queue_file = self.get_queue_file_location(self._app_id)
        try:
          info = file_io.read(queue_file)
          logger.info('Found queue file for {} in: {}'.
            format(self._app_id, queue_file))
        except IOError:
          logger.error(
            'No queue file found for {}, using default queue'.format(self._app_id))
          info = self.DEFAULT_QUEUE_YAML
          using_default = True
      except apiproxy_errors.ApplicationError as application_error:
        logger.error(application_error.message)
        info = self.DEFAULT_QUEUE_YAML
        using_default = True
  
      #TODO handle bad xml/yaml files.
      if queue_file.endswith('yaml') or using_default:
        queue_info = queueinfo.LoadSingleQueue(info).ToDict()
      elif queue_file.endswith('xml'):
       
  
  [... the rest is omitted ...]
files: AppTaskQueue/appscale/taskqueue/tq_config.py
files: import logging
  import sys
  
  from .unpackaged import APPSCALE_LIB_DIR
  
  sys.path.append(APPSCALE_LIB_DIR)
  from constants import LOG_FORMAT
  
  
  logging.basicConfig(format=LOG_FORMAT)
  logger = logging.getLogger(__name__)
  logger.setLevel(logging.INFO)
  
  
  def get_celery_worker_module_name(app_id):
    """ Returns the python module name of the queue worker script.
  
    Args:
      app_id: The application ID.
    Returns:
      A string of the module name.
    """
    return 'app___' + app_id
  
  
  def get_celery_queue_name(app_id, queue_name):
    """ Gets a usable queue name for celery to prevent collisions where
    mulitple apps have the same name for a queue.
  
    Args:
      app_id: The application ID.
      queue_name: String name of the queue.
    Returns:
      A string to reference the queue name in celery.
    """
    return app_id + "___" + queue_name
  
  
  def get_queue_function_name(queue_name):
    """ Returns the function name of a queue which is not the queue name for
    namespacing and collision reasons.
  
    Args:
      queue_name: The name of a queue.
    Returns:
      The string representing the function name.
    """
    # Remove '-' because that character is not valid for a function name.
    queue_name = queue_name.replace('-', '_')
    return "queue___%s" % queue_name
  
  
  def get_celery_annotation_name(app_id, queue_name):
    """ Returns the annotation name for a celery configuration of a queue
    for a given application id.
  
    Args:
      app_id: The application ID.
      queue_name: The application queue name.
    Returns:
      A string for the annotation tag.
    """
    module_name = get_celery_worker_module_name(app_id)
    function_name = get_queue_function_name(queue_name)
    return "%s.%s" % (module_name, function_name)
  
  
  [... the rest is omitted ...]
files: AppTaskQueue/appscale/taskqueue/utils.py
files: import logging
  import sys
  
  from .unpackaged import APPSCALE_LIB_DIR
  
  sys.path.append(APPSCALE_LIB_DIR)
  from constants import LOG_FORMAT
  
  
  logging.basicConfig(format=LOG_FORMAT)
  logger = logging.getLogger(__name__)
  logger.setLevel(logging.INFO)
  
  
  [... the rest is omitted ...]
files: AppTaskQueue/appscale/taskqueue/utils.py