utils.py

import logging
import os

from appscale.common.constants import LOG_FORMAT
from celery import Celery
from kombu import (
  Exchange,
  Queue as KombuQueue
)
from .brokers import rabbitmq


logging.basicConfig(format=LOG_FORMAT)
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

# The directory where Celery configuration files are stored.
CELERY_CONFIG_DIR = os.path.join('/etc', 'appscale', 'celery', 'configuration')

# The working directory for Celery workers.
CELERY_WORKER_DIR = os.path.join('/etc', 'appscale', 'celery', 'workers')


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)


def get_celery_configuration_path(app_id):
  """ Returns the full path of the configuration used for Celery.

  Args:
    app_id: The application ID.
  Returns:
    A string of the full file name of the configuration file.
  """
  return os.path.join(CELERY_CONFIG_DIR, '{}.json'.format(app_id))


def create_celery_for_app(app, rates):
  """ Create Celery interface for a given app.

  Args:
    app: A string specifying an application ID.
    rates: A dictionary mapping queue names to their rates.
  Returns:
    A Celery application object.
  """
  module_name = get_celery_worker_module_name(app)
  celery = Celery(module_name, broker=rabbitmq.get_connection_string(),
                  backend='amqp://')

  kombu_queues = []
  annotations = []
  for queue_name, rate in rates.items():
    celery_name = get_celery_queue_name(app, queue_name)
    kombu_queue = KombuQueue(celery_name, Exchange(app),
                             routing_key=celery_name)

    kombu_queues.append(kombu_queue)
    annotation_name = get_celery_annotation_name(app, queue_name)
    annotations.append({annotation_name: {'rate_limit': rate}})

  celery.conf.CELERY_QUEUES = kombu_queues
  celery.conf.CELERY_ANNOTATIONS = annotations

  # Everytime a task is enqueued, a temporary queue is created to store
  # results into rabbitmq. This can be bad in a high enqueue environment
  # We use the following to make sure these temp queues are not created.
  # See stackoverflow.com/questions/7144025/temporary-queue-made-in-celery.
  celery.conf.CELERY_IGNORE_RESULT = True
  celery.conf.CELERY_STORE_ERRORS_EVEN_IF_IGNORED = False

  # One month expiration date because a task can be deferred that long.
  celery.conf.CELERY_AMQP_TASK_RESULT_EXPIRES = 2678400

  # Disable prefetching for celery workers. If tasks are small in duration this
  # should be set to a higher value (64-128) for increased performance. See
  # celery.readthedocs.org/en/latest/userguide/optimizing.html#worker-settings.
  celery.conf.CELERYD_PREFETCH_MULTIPLIER = 1

  return celery