Allow global postgres_dsn
import json
-from kazoo.exceptions import ZookeeperError, NoNodeError
+from kazoo.exceptions import ZookeeperError
from tornado.ioloop import IOLoop, PeriodicCallback
+from appscale.taskqueue.pg_connection_wrapper import PostgresConnectionWrapper
from appscale.taskqueue.queue import PostgresPullQueue
from appscale.taskqueue.utils import create_celery_for_app
from .queue import PullQueue
@@ -28,23 +29,32 @@ class ProjectQueueManager(dict):
super(ProjectQueueManager, self).__init__()
self.zk_client = zk_client
self.project_id = project_id
- pg_dns_node = '/appscale/projects/{}/postgres_dsn'.format(project_id)
- try:
- pg_dsn = self.zk_client.get(pg_dns_node)
- logger.info('Using PostgreSQL as a backend for Pull Queues of "{}"'
+
+ project_dsn_node = '/appscale/projects/{}/postgres_dsn'.format(project_id)
+ global_dsn_node = '/appscale/tasks/postgres_dsn'
+ if self.zk_client.exists(project_dsn_node):
+ pg_dsn = self.zk_client.get(project_dsn_node)
+ logger.info('Using project-specific PostgreSQL as a backend for '
+ 'Pull Queues of project "{}" '.format(project_id))
+ elif self.zk_client.exists(global_dsn_node):
+ pg_dsn = self.zk_client.get(global_dsn_node)
+ logger.info('Using deployment-wide PostgreSQL as a backend for '
+ 'Pull Queues"'.format(project_id))
+ else:
+ pg_dsn = None
+ logger.info('Using Cassandra as a backend for Pull Queues of "{}"'
.format(project_id))
- # Import pg_connection_wrapper (and psycopg2) lazily
- from appscale.taskqueue import pg_connection_wrapper
+
+ if pg_dsn:
# TODO: PostgresConnectionWrapper may need an update when
# TaskQueue becomes concurrent
- self.pg_connection_wrapper = (
- pg_connection_wrapper.PostgresConnectionWrapper(dsn=pg_dsn[0].decode('utf-8'))
+ self.pg_connection_wrapper = PostgresConnectionWrapper(
+ dsn=pg_dsn[0].decode('utf-8')
)
self._configure_periodical_flush()
- except NoNodeError:
- logger.info('Using Cassandra as a backend for Pull Queues of "{}"'
- .format(project_id))
+ else:
self.pg_connection_wrapper = None
+
self.queues_node = '/appscale/projects/{}/queues'.format(project_id)
self.watch = zk_client.DataWatch(self.queues_node,
self._update_queues_watch)
