OCA / queue / 130

pending completion

Build # 130

Build Type

Pull #51

travis-ci

Committed by

web-flow

Commit Message

[ADD] configurable scheme, host and http auth

Initial work by Nils Hamerlinck <nils@hamerlinck.fr> on the connector
module for Odoo v9.0 (commit 383959afc)

Pull Request Pull Request #51: [FIX] Allow to run jobs when Odoo is not listening on localhost

Run Details

12 of 12 new or added lines in 1 file covered. (100.0%)

922 of 1187 relevant lines covered (77.67%)

1.55 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

21.59

/queue_job/jobrunner/runner.py

# -*- coding: utf-8 -*-
# Copyright (c) 2015-2016 ACSONE SA/NV (<http://acsone.eu>)
# Copyright 2015-2016 Camptocamp SA
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html)
"""
What is the job runner?
-----------------------
The job runner is the main process managing the dispatch of delayed jobs to
available Odoo workers

How does it work?
-----------------

* It starts as a thread in the Odoo main process
* It receives postgres NOTIFY messages each time jobs are
  added or updated in the queue_job table.
* It maintains an in-memory priority queue of jobs that
  is populated from the queue_job tables in all databases.
* It does not run jobs itself, but asks Odoo to run them through an
  anonymous ``/queue_job/runjob`` HTTP request. [1]_

How to use it?
--------------

* Optionally adjust your configuration through environment variables:

  - ``ODOO_QUEUE_JOB_CHANNELS=root:4`` (or any other channels
    configuration), default ``root:1``.
  - ``ODOO_QUEUE_JOB_SCHEME=https``, default ``http``.
  - ``ODOO_QUEUE_JOB_HOST=load-balancer``, default ``localhost``.
  - ``ODOO_QUEUE_JOB_PORT=443``, default ``xmlrpc-port`` or 8069.
  - ``ODOO_QUEUE_JOB_HTTP_AUTH_USER=jobrunner``, default empty.
  - ``ODOO_QUEUE_JOB_HTTP_AUTH_PASSWORD=s3cr3t``, default empty.

* Alternatively, configure the channels through the Odoo configuration
  file, like:

.. code-block:: ini

  [queue_job]
  channels = root:4
  scheme = https
  host = load-balancer
  port = 443
  http_auth_user = jobrunner
  http_auth_password = s3cr3t

* Or, if using ``anybox.recipe.odoo``, add this to your buildout configuration:

.. code-block:: ini

  [odoo]
  recipe = anybox.recipe.odoo
  (...)
  queue_job.channels = root:4
  queue_job.scheme = https
  queue_job.host = load-balancer
  queue_job.port = 443
  queue_job.http_auth_user = jobrunner
  queue_job.http_auth_password = s3cr3t

* Start Odoo with ``--load=web,web_kanban,queue_job``
  and ``--workers`` greater than 1 [2]_, or set the ``server_wide_modules``
  option in The Odoo configuration file:

.. code-block:: ini

  [options]
  (...)
  workers = 4
  server_wide_modules = web,web_kanban,queue_job
  (...)

* Or, if using ``anybox.recipe.odoo``:

.. code-block:: ini

  [odoo]
  recipe = anybox.recipe.odoo
  (...)
  options.workers = 4
  options.server_wide_modules = web,web_kanban,queue_job

* Confirm the runner is starting correctly by checking the odoo log file:

.. code-block:: none

  ...INFO...queue_job.jobrunner.runner: starting
  ...INFO...queue_job.jobrunner.runner: initializing database connections
  ...INFO...queue_job.jobrunner.runner: queue job runner ready for db <dbname>
  ...INFO...queue_job.jobrunner.runner: database connections ready

* Create jobs (eg using base_import_async) and observe they
  start immediately and in parallel.

* Tip: to enable debug logging for the queue job, use
  ``--log-handler=odoo.addons.queue_job:DEBUG``

Caveat
------

* After creating a new database or installing queue_job on an
  existing database, Odoo must be restarted for the runner to detect it.

* When Odoo shuts down normally, it waits for running jobs to finish.
  However, when the Odoo server crashes or is otherwise force-stopped,
  running jobs are interrupted while the runner has no chance to know
  they have been aborted. In such situations, jobs may remain in
  ``started`` or ``enqueued`` state after the Odoo server is halted.
  Since the runner has no way to know if they are actually running or
  not, and does not know for sure if it is safe to restart the jobs,
  it does not attempt to restart them automatically. Such stale jobs
  therefore fill the running queue and prevent other jobs to start.
  You must therefore requeue them manually, either from the Jobs view,
  or by running the following SQL statement *before starting Odoo*:

.. code-block:: sql

  update queue_job set state='pending' where state in ('started', 'enqueued')

.. rubric:: Footnotes

.. [1] From a security standpoint, it is safe to have an anonymous HTTP
       request because this request only accepts to run jobs that are
       enqueued.
.. [2] It works with the threaded Odoo server too, although this way
       of running Odoo is obviously not for production purposes.
"""

from contextlib import closing
import datetime
import logging
import os
import select
import threading
import time

import psycopg2
from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT
import requests

import odoo
from odoo.tools import config

from .channels import ChannelManager, PENDING, ENQUEUED, NOT_DONE

SELECT_TIMEOUT = 60
ERROR_RECOVERY_DELAY = 5

_logger = logging.getLogger(__name__)


# Unfortunately, it is not possible to extend the Odoo
# server command line arguments, so we resort to environment variables
# to configure the runner (channels mostly).
#
# On the other hand, the odoo configuration file can be extended at will,
# so we check it in addition to the environment variables.


def _channels():
    return (
        os.environ.get('ODOO_QUEUE_JOB_CHANNELS') or
        config.misc.get("queue_job", {}).get("channels") or
        "root:1"
    )


def _datetime_to_epoch(dt):
    # important: this must return the same as postgresql
    # EXTRACT(EPOCH FROM TIMESTAMP dt)
    return (dt - datetime.datetime(1970, 1, 1)).total_seconds()


def _odoo_now():
    dt = datetime.datetime.utcnow()
    return _datetime_to_epoch(dt)


def _async_http_get(scheme, host, port, user, password, db_name, job_uuid):
    # Method to set failed job (due to timeout, etc) as pending,
    # to avoid keeping it as enqueued.
    def set_job_pending():
        connection_info = odoo.sql_db.connection_info_for(db_name)[1]
        conn = psycopg2.connect(**connection_info)
        conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)
        with closing(conn.cursor()) as cr:
            cr.execute(
                "UPDATE queue_job SET state=%s, "
                "date_enqueued=NULL, date_started=NULL "
                "WHERE uuid=%s and state=%s", (PENDING, job_uuid, ENQUEUED)
            )

    # TODO: better way to HTTP GET asynchronously (grequest, ...)?
    #       if this was python3 I would be doing this with
    #       asyncio, aiohttp and aiopg
    def urlopen():
        url = ('%s://%s:%s/connector/runjob?db=%s&job_uuid=%s' %
               (scheme, host, port, db_name, job_uuid))
        try:
            auth = None
            if user:
                auth = (user, password)
            # we are not interested in the result, so we set a short timeout
            # but not too short so we trap and log hard configuration errors
            response = requests.get(url, timeout=1, auth=auth)

            # raise_for_status will result in either nothing, a Client Error
            # for HTTP Response codes between 400 and 500 or a Server Error
            # for codes between 500 and 600
            response.raise_for_status()
        except requests.Timeout:
            set_job_pending()
        except:
            _logger.exception("exception in GET %s", url)
            set_job_pending()
    thread = threading.Thread(target=urlopen)
    thread.daemon = True
    thread.start()


class Database(object):

    def __init__(self, db_name):
        self.db_name = db_name
        connection_info = odoo.sql_db.connection_info_for(db_name)[1]
        self.conn = psycopg2.connect(**connection_info)
        self.conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)
        self.has_queue_job = self._has_queue_job()
        if self.has_queue_job:
            self._initialize()

    def close(self):
        try:
            self.conn.close()
        except:
            pass
        self.conn = None

    def _has_queue_job(self):
        with closing(self.conn.cursor()) as cr:
            cr.execute("SELECT 1 FROM pg_tables WHERE tablename=%s",
                       ('ir_module_module',))
            if not cr.fetchone():
                return False
            cr.execute(
                "SELECT 1 FROM ir_module_module WHERE name=%s AND state=%s",
                ('queue_job', 'installed')
            )
            return cr.fetchone()

    def _initialize(self):
        with closing(self.conn.cursor()) as cr:
            # this is the trigger that sends notifications when jobs change
            cr.execute("""
                DROP TRIGGER IF EXISTS queue_job_notify ON queue_job;

                CREATE OR REPLACE
                    FUNCTION queue_job_notify() RETURNS trigger AS $$
                BEGIN
                    IF TG_OP = 'DELETE' THEN
                        IF OLD.state != 'done' THEN
                            PERFORM pg_notify('queue_job', OLD.uuid);
                        END IF;
                    ELSE
                        PERFORM pg_notify('queue_job', NEW.uuid);
                    END IF;
                    RETURN NULL;
                END;
                $$ LANGUAGE plpgsql;

                CREATE TRIGGER queue_job_notify
                    AFTER INSERT OR UPDATE OR DELETE
                    ON queue_job
                    FOR EACH ROW EXECUTE PROCEDURE queue_job_notify();
            """)
            cr.execute("LISTEN queue_job")

    def select_jobs(self, where, args):
        query = ("SELECT channel, uuid, id as seq, date_created, "
                 "priority, EXTRACT(EPOCH FROM eta), state "
                 "FROM queue_job WHERE %s" %
                 (where, ))
        with closing(self.conn.cursor()) as cr:
            cr.execute(query, args)
            return list(cr.fetchall())

    def set_job_enqueued(self, uuid):
        with closing(self.conn.cursor()) as cr:
            cr.execute("UPDATE queue_job SET state=%s, "
                       "date_enqueued=date_trunc('seconds', "
                       "                         now() at time zone 'utc') "
                       "WHERE uuid=%s",
                       (ENQUEUED, uuid))


class QueueJobRunner(object):

    def __init__(self,
                 scheme='http',
                 host='localhost',
                 port=8069,
                 user=None,
                 password=None,
                 channel_config_string=None):
        self.scheme = scheme
        self.host = host
        self.port = port
        self.user = user
        self.password = password
        self.channel_manager = ChannelManager()
        if channel_config_string is None:
            channel_config_string = _channels()
        self.channel_manager.simple_configure(channel_config_string)
        self.db_by_name = {}
        self._stop = False
        self._stop_pipe = os.pipe()

    def get_db_names(self):
        if odoo.tools.config['db_name']:
            db_names = odoo.tools.config['db_name'].split(',')
        else:
            db_names = odoo.service.db.exp_list(True)
        return db_names

    def close_databases(self, remove_jobs=True):
        for db_name, db in self.db_by_name.items():
            try:
                if remove_jobs:
                    self.channel_manager.remove_db(db_name)
                db.close()
            except:
                _logger.warning('error closing database %s',
                                db_name, exc_info=True)
        self.db_by_name = {}

    def initialize_databases(self):
        for db_name in self.get_db_names():
            db = Database(db_name)
            if not db.has_queue_job:
                _logger.debug('queue_job is not installed for db %s', db_name)
            else:
                self.db_by_name[db_name] = db
                for job_data in db.select_jobs('state in %s', (NOT_DONE,)):
                    self.channel_manager.notify(db_name, *job_data)
                _logger.info('queue job runner ready for db %s', db_name)

    def run_jobs(self):
        now = _odoo_now()
        for job in self.channel_manager.get_jobs_to_run(now):
            if self._stop:
                break
            _logger.info("asking Odoo to run job %s on db %s",
                         job.uuid, job.db_name)
            self.db_by_name[job.db_name].set_job_enqueued(job.uuid)
            _async_http_get(self.scheme,
                            self.host,
                            self.port,
                            self.user,
                            self.password,
                            job.db_name,
                            job.uuid)

    def process_notifications(self):
        for db in self.db_by_name.values():
            while db.conn.notifies:
                if self._stop:
                    break
                notification = db.conn.notifies.pop()
                uuid = notification.payload
                job_datas = db.select_jobs('uuid = %s', (uuid,))
                if job_datas:
                    self.channel_manager.notify(db.db_name, *job_datas[0])
                else:
                    self.channel_manager.remove_job(uuid)

    def wait_notification(self):
        for db in self.db_by_name.values():
            if db.conn.notifies:
                # something is going on in the queue, no need to wait
                return
        # wait for something to happen in the queue_job tables
        # we'll select() on database connections and the stop pipe
        conns = [db.conn for db in self.db_by_name.values()]
        conns.append(self._stop_pipe[0])
        # look if the channels specify a wakeup time
        wakeup_time = self.channel_manager.get_wakeup_time()
        if not wakeup_time:
            # this could very well be no timeout at all, because
            # any activity in the job queue will wake us up, but
            # let's have a timeout anyway, just to be safe
            timeout = SELECT_TIMEOUT
        else:
            timeout = wakeup_time - _odoo_now()
        # wait for a notification or a timeout;
        # if timeout is negative (ie wakeup time in the past),
        # do not wait; this should rarely happen
        # because of how get_wakeup_time is designed; actually
        # if timeout remains a large negative number, it is most
        # probably a bug
        _logger.debug("select() timeout: %.2f sec", timeout)
        if timeout > 0:
            conns, _, _ = select.select(conns, [], [], timeout)
            if conns and not self._stop:
                for conn in conns:
                    conn.poll()

    def stop(self):
        _logger.info("graceful stop requested")
        self._stop = True
        # wakeup the select() in wait_notification
        os.write(self._stop_pipe[1], b'.')

    def run(self):
        _logger.info("starting")
        while not self._stop:
            # outer loop does exception recovery
            try:
                _logger.info("initializing database connections")
                # TODO: how to detect new databases or databases
                #       on which queue_job is installed after server start?
                self.initialize_databases()
                _logger.info("database connections ready")
                # inner loop does the normal processing
                while not self._stop:
                    self.process_notifications()
                    self.run_jobs()
                    self.wait_notification()
            except KeyboardInterrupt:
                self.stop()
            except:
                _logger.exception("exception: sleeping %ds and retrying",
                                  ERROR_RECOVERY_DELAY)
                self.close_databases()
                time.sleep(ERROR_RECOVERY_DELAY)
        self.close_databases(remove_jobs=False)
        _logger.info("stopped")

1	# -- coding: utf-8 --
2	# Copyright (c) 2015-2016 ACSONE SA/NV (<http://acsone.eu>)
3	# Copyright 2015-2016 Camptocamp SA
4	# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html)
5	"""	2✔
6	What is the job runner?
7	-----------------------
8	The job runner is the main process managing the dispatch of delayed jobs to
9	available Odoo workers
10
11	How does it work?
12	-----------------
13
14	* It starts as a thread in the Odoo main process
15	* It receives postgres NOTIFY messages each time jobs are
16	added or updated in the queue_job table.
17	* It maintains an in-memory priority queue of jobs that
18	is populated from the queue_job tables in all databases.
19	* It does not run jobs itself, but asks Odoo to run them through an
20	anonymous ``/queue_job/runjob`` HTTP request. [1]_
21
22	How to use it?
23	--------------
24
25	* Optionally adjust your configuration through environment variables:
26
27	- ``ODOO_QUEUE_JOB_CHANNELS=root:4`` (or any other channels
28	configuration), default ``root:1``.
29	- ``ODOO_QUEUE_JOB_SCHEME=https``, default ``http``.
30	- ``ODOO_QUEUE_JOB_HOST=load-balancer``, default ``localhost``.
31	- ``ODOO_QUEUE_JOB_PORT=443``, default ``xmlrpc-port`` or 8069.
32	- ``ODOO_QUEUE_JOB_HTTP_AUTH_USER=jobrunner``, default empty.
33	- ``ODOO_QUEUE_JOB_HTTP_AUTH_PASSWORD=s3cr3t``, default empty.
34
35	* Alternatively, configure the channels through the Odoo configuration
36	file, like:
37
38	.. code-block:: ini
39
40	[queue_job]
41	channels = root:4
42	scheme = https
43	host = load-balancer
44	port = 443
45	http_auth_user = jobrunner
46	http_auth_password = s3cr3t
47
48	* Or, if using ``anybox.recipe.odoo``, add this to your buildout configuration:
49
50	.. code-block:: ini
51
52	[odoo]
53	recipe = anybox.recipe.odoo
54	(...)
55	queue_job.channels = root:4
56	queue_job.scheme = https
57	queue_job.host = load-balancer
58	queue_job.port = 443
59	queue_job.http_auth_user = jobrunner
60	queue_job.http_auth_password = s3cr3t
61
62	* Start Odoo with ``--load=web,web_kanban,queue_job``
63	and ``--workers`` greater than 1 [2]_, or set the ``server_wide_modules``
64	option in The Odoo configuration file:
65
66	.. code-block:: ini
67
68	[options]
69	(...)
70	workers = 4
71	server_wide_modules = web,web_kanban,queue_job
72	(...)
73
74	* Or, if using ``anybox.recipe.odoo``:
75
76	.. code-block:: ini
77
78	[odoo]
79	recipe = anybox.recipe.odoo
80	(...)
81	options.workers = 4
82	options.server_wide_modules = web,web_kanban,queue_job
83
84	* Confirm the runner is starting correctly by checking the odoo log file:
85
86	.. code-block:: none
87
88	...INFO...queue_job.jobrunner.runner: starting
89	...INFO...queue_job.jobrunner.runner: initializing database connections
90	...INFO...queue_job.jobrunner.runner: queue job runner ready for db <dbname>
91	...INFO...queue_job.jobrunner.runner: database connections ready
92
93	* Create jobs (eg using base_import_async) and observe they
94	start immediately and in parallel.
95
96	* Tip: to enable debug logging for the queue job, use
97	``--log-handler=odoo.addons.queue_job:DEBUG``
98
99	Caveat
100	------
101
102	* After creating a new database or installing queue_job on an
103	existing database, Odoo must be restarted for the runner to detect it.
104
105	* When Odoo shuts down normally, it waits for running jobs to finish.
106	However, when the Odoo server crashes or is otherwise force-stopped,
107	running jobs are interrupted while the runner has no chance to know
108	they have been aborted. In such situations, jobs may remain in
109	``started`` or ``enqueued`` state after the Odoo server is halted.
110	Since the runner has no way to know if they are actually running or
111	not, and does not know for sure if it is safe to restart the jobs,
112	it does not attempt to restart them automatically. Such stale jobs
113	therefore fill the running queue and prevent other jobs to start.
114	You must therefore requeue them manually, either from the Jobs view,
115	or by running the following SQL statement before starting Odoo:
116
117	.. code-block:: sql
118
119	update queue_job set state='pending' where state in ('started', 'enqueued')
120
121	.. rubric:: Footnotes
122
123	.. [1] From a security standpoint, it is safe to have an anonymous HTTP
124	request because this request only accepts to run jobs that are
125	enqueued.
126	.. [2] It works with the threaded Odoo server too, although this way
127	of running Odoo is obviously not for production purposes.
128	"""
129
130	from contextlib import closing	2✔
131	import datetime	2✔
132	import logging	2✔
133	import os	2✔
134	import select	2✔
135	import threading	2✔
136	import time	2✔
137
138	import psycopg2	2✔
139	from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT	2✔
140	import requests	2✔
141
142	import odoo	2✔
143	from odoo.tools import config	2✔
144
145	from .channels import ChannelManager, PENDING, ENQUEUED, NOT_DONE	2✔
146
147	SELECT_TIMEOUT = 60	2✔
148	ERROR_RECOVERY_DELAY = 5	2✔
149
150	_logger = logging.getLogger(__name__)	2✔
151
152
153	# Unfortunately, it is not possible to extend the Odoo
154	# server command line arguments, so we resort to environment variables
155	# to configure the runner (channels mostly).
156	#
157	# On the other hand, the odoo configuration file can be extended at will,
158	# so we check it in addition to the environment variables.
159
160
161	def _channels():	2✔
162	return (	×
163	os.environ.get('ODOO_QUEUE_JOB_CHANNELS') or
164	config.misc.get("queue_job", {}).get("channels") or
165	"root:1"
166	)
167
168
169	def _datetime_to_epoch(dt):	2✔
170	# important: this must return the same as postgresql
171	# EXTRACT(EPOCH FROM TIMESTAMP dt)
172	return (dt - datetime.datetime(1970, 1, 1)).total_seconds()	×
173
174
175	def _odoo_now():	2✔
176	dt = datetime.datetime.utcnow()	×
177	return _datetime_to_epoch(dt)	×
178
179
180	def _async_http_get(scheme, host, port, user, password, db_name, job_uuid):	2✔
181	# Method to set failed job (due to timeout, etc) as pending,
182	# to avoid keeping it as enqueued.
183	def set_job_pending():	×
184	connection_info = odoo.sql_db.connection_info_for(db_name)[1]	×
185	conn = psycopg2.connect(**connection_info)	×
186	conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)	×
187	with closing(conn.cursor()) as cr:	×
188	cr.execute(	×
189	"UPDATE queue_job SET state=%s, "
190	"date_enqueued=NULL, date_started=NULL "
191	"WHERE uuid=%s and state=%s", (PENDING, job_uuid, ENQUEUED)
192	)
193
194	# TODO: better way to HTTP GET asynchronously (grequest, ...)?
195	# if this was python3 I would be doing this with
196	# asyncio, aiohttp and aiopg
197	def urlopen():	×
198	url = ('%s://%s:%s/connector/runjob?db=%s&job_uuid=%s' %	×
199	(scheme, host, port, db_name, job_uuid))
200	try:	×
201	auth = None	×
202	if user:	×
203	auth = (user, password)	×
204	# we are not interested in the result, so we set a short timeout
205	# but not too short so we trap and log hard configuration errors
206	response = requests.get(url, timeout=1, auth=auth)	×
207
208	# raise_for_status will result in either nothing, a Client Error
209	# for HTTP Response codes between 400 and 500 or a Server Error
210	# for codes between 500 and 600
211	response.raise_for_status()	×
212	except requests.Timeout:	×
213	set_job_pending()	×
214	except:	×
215	_logger.exception("exception in GET %s", url)	×
216	set_job_pending()	×
217	thread = threading.Thread(target=urlopen)	×
218	thread.daemon = True	×
219	thread.start()	×
220
221
222	class Database(object):	2✔
223
224	def __init__(self, db_name):	2✔
225	self.db_name = db_name	×
226	connection_info = odoo.sql_db.connection_info_for(db_name)[1]	×
227	self.conn = psycopg2.connect(**connection_info)	×
228	self.conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)	×
229	self.has_queue_job = self._has_queue_job()	×
230	if self.has_queue_job:	×
231	self._initialize()	×
232
233	def close(self):	2✔
234	try:	×
235	self.conn.close()	×
236	except:	×
237	pass	×
238	self.conn = None	×
239
240	def _has_queue_job(self):	2✔
241	with closing(self.conn.cursor()) as cr:	×
242	cr.execute("SELECT 1 FROM pg_tables WHERE tablename=%s",	×
243	('ir_module_module',))
244	if not cr.fetchone():	×
245	return False	×
246	cr.execute(	×
247	"SELECT 1 FROM ir_module_module WHERE name=%s AND state=%s",
248	('queue_job', 'installed')
249	)
250	return cr.fetchone()	×
251
252	def _initialize(self):	2✔
253	with closing(self.conn.cursor()) as cr:	×
254	# this is the trigger that sends notifications when jobs change
255	cr.execute("""	×
256	DROP TRIGGER IF EXISTS queue_job_notify ON queue_job;
257
258	CREATE OR REPLACE
259	FUNCTION queue_job_notify() RETURNS trigger AS $$
260	BEGIN
261	IF TG_OP = 'DELETE' THEN
262	IF OLD.state != 'done' THEN
263	PERFORM pg_notify('queue_job', OLD.uuid);
264	END IF;
265	ELSE
266	PERFORM pg_notify('queue_job', NEW.uuid);
267	END IF;
268	RETURN NULL;
269	END;
270	$$ LANGUAGE plpgsql;
271
272	CREATE TRIGGER queue_job_notify
273	AFTER INSERT OR UPDATE OR DELETE
274	ON queue_job
275	FOR EACH ROW EXECUTE PROCEDURE queue_job_notify();
276	""")
277	cr.execute("LISTEN queue_job")	×
278
279	def select_jobs(self, where, args):	2✔
280	query = ("SELECT channel, uuid, id as seq, date_created, "	×
281	"priority, EXTRACT(EPOCH FROM eta), state "
282	"FROM queue_job WHERE %s" %
283	(where, ))
284	with closing(self.conn.cursor()) as cr:	×
285	cr.execute(query, args)	×
286	return list(cr.fetchall())	×
287
288	def set_job_enqueued(self, uuid):	2✔
289	with closing(self.conn.cursor()) as cr:	×
290	cr.execute("UPDATE queue_job SET state=%s, "	×
291	"date_enqueued=date_trunc('seconds', "
292	" now() at time zone 'utc') "
293	"WHERE uuid=%s",
294	(ENQUEUED, uuid))
295
296
297	class QueueJobRunner(object):	2✔
298
299	def __init__(self,	2✔
300	scheme='http',
301	host='localhost',
302	port=8069,
303	user=None,
304	password=None,
305	channel_config_string=None):
306	self.scheme = scheme	×
307	self.host = host	×
308	self.port = port	×
309	self.user = user	×
310	self.password = password	×
311	self.channel_manager = ChannelManager()	×
312	if channel_config_string is None:	×
313	channel_config_string = _channels()	×
314	self.channel_manager.simple_configure(channel_config_string)	×
315	self.db_by_name = {}	×
316	self._stop = False	×
317	self._stop_pipe = os.pipe()	×
318
319	def get_db_names(self):	2✔
320	if odoo.tools.config['db_name']:	×
321	db_names = odoo.tools.config['db_name'].split(',')	×
322	else:
323	db_names = odoo.service.db.exp_list(True)	×
324	return db_names	×
325
326	def close_databases(self, remove_jobs=True):	2✔
327	for db_name, db in self.db_by_name.items():	×
328	try:	×
329	if remove_jobs:	×
330	self.channel_manager.remove_db(db_name)	×
331	db.close()	×
332	except:	×
333	_logger.warning('error closing database %s',	×
334	db_name, exc_info=True)
335	self.db_by_name = {}	×
336
337	def initialize_databases(self):	2✔
338	for db_name in self.get_db_names():	×
339	db = Database(db_name)	×
340	if not db.has_queue_job:	×
341	_logger.debug('queue_job is not installed for db %s', db_name)	×
342	else:
343	self.db_by_name[db_name] = db	×
344	for job_data in db.select_jobs('state in %s', (NOT_DONE,)):	×
345	self.channel_manager.notify(db_name, *job_data)	×
346	_logger.info('queue job runner ready for db %s', db_name)	×
347
348	def run_jobs(self):	2✔
349	now = _odoo_now()	×
350	for job in self.channel_manager.get_jobs_to_run(now):	×
351	if self._stop:	×
352	break	×
353	_logger.info("asking Odoo to run job %s on db %s",	×
354	job.uuid, job.db_name)
355	self.db_by_name[job.db_name].set_job_enqueued(job.uuid)	×
356	_async_http_get(self.scheme,	×
357	self.host,
358	self.port,
359	self.user,
360	self.password,
361	job.db_name,
362	job.uuid)
363
364	def process_notifications(self):	2✔
365	for db in self.db_by_name.values():	×
366	while db.conn.notifies:	×
367	if self._stop:	×
368	break	×
369	notification = db.conn.notifies.pop()	×
370	uuid = notification.payload	×
371	job_datas = db.select_jobs('uuid = %s', (uuid,))	×
372	if job_datas:	×
373	self.channel_manager.notify(db.db_name, *job_datas[0])	×
374	else:
375	self.channel_manager.remove_job(uuid)	×
376
377	def wait_notification(self):	2✔
378	for db in self.db_by_name.values():	×
379	if db.conn.notifies:	×
380	# something is going on in the queue, no need to wait
381	return	×
382	# wait for something to happen in the queue_job tables
383	# we'll select() on database connections and the stop pipe
384	conns = [db.conn for db in self.db_by_name.values()]	×
385	conns.append(self._stop_pipe[0])	×
386	# look if the channels specify a wakeup time
387	wakeup_time = self.channel_manager.get_wakeup_time()	×
388	if not wakeup_time:	×
389	# this could very well be no timeout at all, because
390	# any activity in the job queue will wake us up, but
391	# let's have a timeout anyway, just to be safe
392	timeout = SELECT_TIMEOUT	×
393	else:
394	timeout = wakeup_time - _odoo_now()	×
395	# wait for a notification or a timeout;
396	# if timeout is negative (ie wakeup time in the past),
397	# do not wait; this should rarely happen
398	# because of how get_wakeup_time is designed; actually
399	# if timeout remains a large negative number, it is most
400	# probably a bug
401	_logger.debug("select() timeout: %.2f sec", timeout)	×
402	if timeout > 0:	×
403	conns, _, _ = select.select(conns, [], [], timeout)	×
404	if conns and not self._stop:	×
405	for conn in conns:	×
406	conn.poll()	×
407
408	def stop(self):	2✔
409	_logger.info("graceful stop requested")	×
410	self._stop = True	×
411	# wakeup the select() in wait_notification
412	os.write(self._stop_pipe[1], b'.')	×
413
414	def run(self):	2✔
415	_logger.info("starting")	×
416	while not self._stop:	×
417	# outer loop does exception recovery
418	try:	×
419	_logger.info("initializing database connections")	×
420	# TODO: how to detect new databases or databases
421	# on which queue_job is installed after server start?
422	self.initialize_databases()	×
423	_logger.info("database connections ready")	×
424	# inner loop does the normal processing
425	while not self._stop:	×
426	self.process_notifications()	×
427	self.run_jobs()	×
428	self.wait_notification()	×
429	except KeyboardInterrupt:	×
430	self.stop()	×
431	except:	×
432	_logger.exception("exception: sleeping %ds and retrying",	×
433	ERROR_RECOVERY_DELAY)
434	self.close_databases()	×
435	time.sleep(ERROR_RECOVERY_DELAY)	×
436	self.close_databases(remove_jobs=False)	×
437	_logger.info("stopped")	×

OCA / queue / 130

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous