diff --git a/airtime_mvc/application/common/TaskManager.php b/airtime_mvc/application/common/TaskManager.php index 415afa55f..aa250f294 100644 --- a/airtime_mvc/application/common/TaskManager.php +++ b/airtime_mvc/application/common/TaskManager.php @@ -67,10 +67,8 @@ final class TaskManager { try { $lock = $this->_getLock(); if ($lock && microtime(true) < $lock['valstr'] + self::TASK_INTERVAL_SECONDS) { - // Fun fact: Propel caches the database connection and uses it persistently - // (thus why calling Propel::getConnection explicitly and passing a connection - // parameter is often not necessary when making Propel queries). Long story short, - // if we don't use commit() here, we end up blocking other queries made within this request + // Propel caches the database connection and uses it persistently, so if we don't + // use commit() here, we end up blocking other queries made within this request $this->_con->commit(); return; } diff --git a/airtime_mvc/application/services/CeleryService.php b/airtime_mvc/application/services/CeleryService.php index 9f21af3e4..ad18a56b1 100644 --- a/airtime_mvc/application/services/CeleryService.php +++ b/airtime_mvc/application/services/CeleryService.php @@ -50,8 +50,6 @@ class CeleryService { * * @return string the task identifier for the started Celery task so we can fetch the * results asynchronously later - * - * @throws CeleryException when no message is found */ public static function sendCeleryMessage($task, $exchange, $data) { $config = parse_ini_file(Application_Model_RabbitMq::getRmqConfigPath(), true); @@ -67,8 +65,9 @@ class CeleryService { * * @param $task CeleryTasks the Celery task object * - * @return object the message object + * @return array the message response array * + * @throws CeleryException when no message is found * @throws CeleryTimeoutException when no message is found and more than * $_CELERY_MESSAGE_TIMEOUT milliseconds have passed */ @@ -80,13 +79,19 @@ class CeleryService { // If the message isn't ready yet (Celery hasn't finished the task), // only throw an exception if the message has timed out. - if ($message == FALSE && self::_checkMessageTimeout($task)) { - // If the task times out, mark it as failed. We don't want to remove the - // track reference here in case it was a deletion that failed, for example. - $task->setDbStatus(CELERY_FAILED_STATUS); - $task->save(); - throw new CeleryTimeoutException("Celery task " . $task->getDbName() - . " with ID " . $task->getDbId() . " timed out"); + if ($message == FALSE) { + if (self::_checkMessageTimeout($task)) { + // If the task times out, mark it as failed. We don't want to remove the + // track reference here in case it was a deletion that failed, for example. + $task->setDbStatus(CELERY_FAILED_STATUS)->save(); + throw new CeleryTimeoutException("Celery task " . $task->getDbName() + . " with ID " . $task->getDbId() . " timed out"); + } else { + // The message hasn't timed out, but it's still false, which means it hasn't been + // sent back from Celery yet. + throw new CeleryException("Waiting on Celery task " . $task->getDbName() + . " with ID " . $task->getDbId()); + } } return $message; } @@ -121,7 +126,7 @@ class CeleryService { $message = self::_getTaskMessage($task); self::_processTaskMessage($task, $message); } catch (CeleryTimeoutException $e) { - Logging::info($e->getMessage()); + Logging::warn($e->getMessage()); } catch (Exception $e) { // Because $message->result can be either an object or a string, sometimes // we get a json_decode error and end up here @@ -161,10 +166,11 @@ class CeleryService { * * @return object the task message object * - * @throws CeleryException when the result message for this task no longer exists + * @throws CeleryException when the result message for this task is still pending + * @throws CeleryTimeoutException when the result message for this task no longer exists */ protected static function _getTaskMessage($task) { - $message = self::getAsyncResultMessage($task); + $message = self::getAsyncResultMessage($task); return json_decode($message['body']); } diff --git a/airtime_mvc/application/views/scripts/form/preferences_soundcloud.phtml b/airtime_mvc/application/views/scripts/form/preferences_soundcloud.phtml index 594468474..2697a6508 100644 --- a/airtime_mvc/application/views/scripts/form/preferences_soundcloud.phtml +++ b/airtime_mvc/application/views/scripts/form/preferences_soundcloud.phtml @@ -1,19 +1,5 @@
- -element->getElement('SoundCloudTrackType')->getLabel() ?> - - -element->getElement('SoundCloudTrackType') ?> -element->getElement('SoundCloudTrackType')->hasErrors()) : ?> - -element->getElement('SoundCloudTrackType')->getMessages() as $error): ?> - - - - - - hasAccessToken()) { diff --git a/python_apps/airtime-celery/README.rst b/python_apps/airtime-celery/README.rst index f079f78f6..cd5e2b5f2 100644 --- a/python_apps/airtime-celery/README.rst +++ b/python_apps/airtime-celery/README.rst @@ -25,8 +25,16 @@ This program must be run with sudo: Developers ========== +To debug, you can run celery directly from the command line: + + $ cd /my/airtime/root/python_apps/airtime-celery + $ RMQ_CONFIG_FILE=/etc/airtime/airtime.conf celery -A airtime-celery.tasks worker --loglevel=info + +This worker can be run alongside the service without issue. + You may want to use the setuptools develop target to install: + $ cd /my/airtime/root/python_apps/airtime-celery $ sudo python setup.py develop You will need to allow the "airtime" RabbitMQ user to access all exchanges and queues within the /airtime vhost: @@ -39,3 +47,19 @@ Logging By default, logs are saved to: /var/log/airtime/airtime-celery[-DEV_ENV].log + +Troubleshooting +=============== + +If you run into issues getting Celery to accept tasks from Airtime: + + 1) Make sure Celery is running ($ sudo service airtime-celery status). + + 2) Check the log file (/var/log/airtime/airtime-celery[-DEV_ENV].log) to make sure Celery started correctly. + + 3) Check your /etc/airtime/airtime.conf rabbitmq settings. Make sure the settings here align with + /etc/airtime-saas/production/rabbitmq.ini. + + 4) Check RabbitMQ to make sure the celeryresults and task queues were created in the correct vhost. + + 5) Make sure the RabbitMQ user (the default is airtime) has permissions on all vhosts being used. \ No newline at end of file diff --git a/python_apps/airtime-celery/setup.py b/python_apps/airtime-celery/setup.py index bfac57fba..8832ac2a8 100644 --- a/python_apps/airtime-celery/setup.py +++ b/python_apps/airtime-celery/setup.py @@ -4,13 +4,16 @@ import os import sys install_args = ['install', 'install_data', 'develop'] +run_postinst = False # XXX Definitely not the best way of doing this... if sys.argv[1] in install_args and "--no-init-script" not in sys.argv: + run_postinst = True data_files = [('/etc/default', ['install/conf/airtime-celery']), ('/etc/init.d', ['install/initd/airtime-celery'])] else: if "--no-init-script" in sys.argv: + run_postinst = True # We still want to run the postinst here sys.argv.remove("--no-init-script") data_files = [] @@ -20,6 +23,14 @@ def postinst(): # permissions for the defaults config file os.chmod('/etc/init.d/airtime-celery', 0755) os.chmod('/etc/default/airtime-celery', 0640) + # Make the airtime log directory group-writable + os.chmod('/var/log/airtime', 0775) + + # Create the Celery user + call(['adduser', '--no-create-home', '--home', '/var/lib/celery', '--gecos', '""', '--disabled-login', 'celery']) + # Add celery to the www-data group + call(['usermod', '-G', 'www-data', '-a', 'celery']) + print "Reloading initctl configuration" call(['initctl', 'reload-configuration']) print "Setting Celery to start on boot" @@ -43,5 +54,5 @@ setup(name='airtime-celery', zip_safe=False, data_files=data_files) -if data_files: +if run_postinst: postinst()