Download as ODP, PPTX
Uploaded byMauro Rocco
Europython 2011 - Playing tasks with Django & Celery
This document discusses using Django and Celery for asynchronous task processing. Some key points: - Celery allows defining and processing asynchronous tasks across multiple machines in a distributed, fault-tolerant way. It supports scheduling via cron-like functionality. - Django-celery provides a Django app that integrates Celery with the Django admin for task scheduling and monitoring. It replaces Celery configuration with Django settings. - Unique tasks ensure only one instance of a task runs at a time across all machines. This is useful for accessing sensitive resources atomically. Custom locking and concurrency handling is implemented. - Centralized logging is implemented by connecting signal handlers to Celery's logging setup. This allows sending all worker logs to a syslog
Europython 2011 - Playing tasks with Django & Celery
- 1. Playing tasks withDjango & Celery Mauro Rocco @fireantology
- 2. About me AboutJamendo Jamendo is a community of free, legal and unlimited music published under Creative Commons licenses
- 3. Free Music forusers
- 4. Popularity and earningsfor artists
- 5. Music licensing andbackground music at competitive prices for companies I'm a Web Developer
- 6. Python, Javascript, PHP,Java/Android
- 7. celery contributor (justone of the hundreds )
- 8. Jamendo needs Multi-formatmusic encoding
- 9. Statistics (downloads, listens,reviews, stars, fb likes) on different units
- 10. Music analysis troughexternal services
- 11. Music qualification troughseveral sources
- 12. Integration with thirdpart services
- 13. Common jobs (contractgenerations, certifications, bills, search index update)
- 14. Celery Async &Sync processes
- 15.
- 16.
- 17.
- 18.
- 19. Subtask, Set oftasks
- 20. Web monitoring (django-celeryand others) “ Celery is an asynchronous task queue/job queue based on distributed message passing. It is focused on real-time operation, but supports scheduling as well”
- 21. AMPQ The AdvancedMessage Queuing Protocol (AMQP) is an open standard application layer protocol for Message Oriented Middleware.
- 22.
- 23. Celery worker Isthe celery process that execute the tasks
- 24. Can serve oneor multiple queues
- 25. Have a maxnumber of tasks that can be executed at the same time
- 26. Can be remotelycontrolled
- 27. Have a greatconfiguration option called MAX_TASK_PER_CHILD
- 28. $ $celeryd -l INFO -c 5 -Q queue1 -E
- 29.
- 30. Defining a simpletask from celery.decorators import task @task def make_money (how_much): logger = make_money.get_logger() logger.info( "Congratulation, you earned %s$" % how_much) if how_much>1000000: return "Bora Bora" return "Keep working" >>> result = make_money.delay(200) >>> result.get() “ Keep working”
- 31. Retrying a taskif something fails from celery.decorators import task @task def make_money_real_life (how_much, wife=True): try : logger = make_money.get_logger() if wife: raise Exception( "Incompatibility exception" ) logger.info( "Congratulation, you earned %s$" % how_much) if how_much>1000000: return "Bora Bora" return "Keep working" except Exception,exc: make_money_real_life.retry(exc=exc, countdown=60, args=[how_much,False])
- 32. Task set example def run (self, setid=None, subtasks=None, **kwargs): … if not setid or not subtasks: … tasks = [] for slice in slices: tasks.append(uploadTrackSlice.subtask((slice,folder_name))) job = TaskSet(tasks=tasks) task_set_result = job.apply_async() setid = task_set_result.taskset_id subtasks = [result.task_id for result in task_set_result.subtasks] self.incrementalRetry("Result not ready", args=[setid,subtasks]) #Is a retry than we just have to check the results tasks_result = TaskSetResult(setid, map(AsyncResult,subtasks)) if not tasks_result.ready(): self.incrementalRetry( "Result not ready" , args=[setid,subtasks]) else : if tasks_result.successful(): return tasks_result.join() else : raise Exception( "Some of the tasks was failing" ) Extract from a jamendo task that upload track metadata in xml format to an ftp server for music analysis
- 33. The Jamendo Taskclass class JamTask (Task): def __call__ (self, *args, **kwargs): """This method is in charge of call the run method of the task""" self.max_retries = 30 self.sandbox = SandBox(self.name, self.request.id, settings.PATH_SANDBOX, settings.DEBUG) self.taskLogger = TaskLogger(args, kwargs) self.taskLogger.__enter__() . . return self.run(*args, **kwargs) . . def after_return (self, status, retval, task_id, args, kwargs, einfo): """This method is called when the tasks end, on whatever return state""" self.taskLogger.__exit__(status, retval, args, kwargs, einfo) self.cleanTaskSandBox(status,kwargs) self.closeAllConnections() The way for define common behaviour to all your tasks is to override __call__ and after_return methods of the celery Task class
- 34. Web Monitoring toolsdjango-celery https://github.com/ask/django-celery/
- 35.
- 36.
- 37. django-celery Task schedulingand monitoring trough the Django admin interface The celeryconf.py file is replaced by the django settings
- 38. The CELERY_IMPORTS confvar is replaced by the Django INSTALLED_APPS You run celery trough the manage.py of your project $ python manage.py celeryd -l INFO -E
- 39. django-celery settings.py INSTALLED_APPS+= ("djcelery", ) . . import djcelery djcelery.setup_loader() . . CELERYBEAT_SCHEDULER = "djcelery.schedulers.DatabaseScheduler" . . #standard celery conf vars (Broker settings, concurrency ,...)
- 40.
- 41.
- 42.
- 43. Some little niceextensions Execute tasks directly from the django admin interface
- 44. Some little niceextensions
- 45. Jamendo needs UNIQUEtasks A task is unique when can run only one instance of it at the same time in the whole cloud Rational utilization of shared resources
- 46. Atomic access tosensitive resources
- 47.
- 48. Define a listof UNIQUE tasks in settings.py
- 49. If a lockis found define the behaviour retry or fail
- 50. Allow the possibilityof define a task UNIQUE on arguments (same task type with different arguments can run)
- 51. Our solution : mongodb for write and release locks.
- 52. Best solution: cache,virtual file system ?
- 53. Unique tasks UNIQUE_TASKS= { "searchengines.solr.index": { "retry_on_lock": False, "lock_on_type": True, }, "stats.album.rebuild": { "retry_on_lock": True, "lock_on_type": False, }, } On task start ( method __call__ ) self.taskConcurrency = None if kwargs[ “task_name” ] in settings.UNIQUE_TASKS: self.taskConcurrency = TaskConcurrency(kwargs, args, settings.UNIQUE_TASKS\ [kwargs[ “task_name” ]]) if not self.taskConcurrency.canRun(): if self.taskConcurrency.retry: self.incrementalRetry(Exception( "Concurrency Exception" )) else : raise Exception( "Concurrency Exception" ) On task end ( method after_return ) if self.taskConcurrency: self.taskConcurrency.__exit__()
- 54. Celery logs Thelogger object is not unique, the same handler is added to different logs object
- 55. Main Process logger,PoolWorker logger, TaskLogger
- 56. The command logging.getLogger(“Celery”)give you back only the Main Process logger
- 57. Extend logging featureswas a bit tricky until the last version
- 58. Centralized logging Wegive a very little contribute to celery by adding the signal after_setup_logger and after_setup_task_logger (the name are self explanatory)
- 59. after_setup_logger istriggered after the build of the Main Process logger and after the build of each PoolWorker logger
- 60. The signals giveyou back a log object, in this way you can add additional handler for implement a centralized logging
- 61. In our specificcase we are sending the logs of all workers to a syslog server that store log lines in a separated file.
- 62. Centralized logging import logging from celery.signals import after_setup_logger, after_setup_task_logger def after_setup_logger_handler (sender=None, logger=None, loglevel=None, logfile=None, format=None, colorize=None, **kwds): handler = logging.handlers.SysLogHandler(address=( 'syslogserver' , 514)) handler.setFormatter(logging.Formatter(format)) handler.setLevel(logging.INFO) logger.addHandler(handler) after_setup_logger.connect(after_setup_logger_handler) after_setup_task_logger.connect(after_setup_logger_handler)
- 63.