factory.py 文件源码

def make_celery(app=None):
    if app is None:
        app = create_app('kubedock', os.path.dirname(__file__))
    if SENTRY_ENABLE:
        import socket
        import celery
        import raven
        from raven.contrib.celery import register_signal
        from raven.contrib.celery import register_logger_signal
        from kubedock.settings import MASTER_IP
        from kubedock.settings import SENTRY_DSN, SENTRY_EXCLUDE_PATHS
        from kubedock.settings import SENTRY_PROCESSORS
        from kubedock.utils import get_version
        from kubedock.kapi.licensing import get_license_info
        authkey = get_license_info().get('auth_key', 'no installation id')
        from celery.utils import log

        class Celery(celery.Celery):

            def on_configure(self):
                hostname = "{}({})".format(socket.gethostname(), MASTER_IP)
                tags = {'installation_id': authkey}
                client = raven.Client(SENTRY_DSN, name=hostname,
                                      release=get_version('kuberdock'),
                                      tags=tags, processors=SENTRY_PROCESSORS,
                                      exclude_paths=SENTRY_EXCLUDE_PATHS)

                # register a custom filter to filter out duplicate logs
                register_logger_signal(client)

                # hook into the Celery error handler
                register_signal(client)
    else:
        from celery import Celery
    celery = Celery(app.import_name, broker=app.config['CELERY_BROKER_URL'])
    celery.conf.update(app.config)
    TaskBase = celery.Task

    class ContextTask(TaskBase):
        abstract = True
        flask_app = app

        def __call__(self, *args, **kwargs):
            with app.app_context():
                env.user = 'root'
                env.key_filename = SSH_KEY_FILENAME
                return TaskBase.__call__(self, *args, **kwargs)
    celery.Task = ContextTask
    return celery