Skip to content

[WIP] queue_job: extract controller's runjob as JobExecutor #945

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
guewen wants to merge 1 commit into
base: 18.0
Choose a base branch
from
+267 −207
Draft

[WIP] queue_job: extract controller's runjob as JobExecutor #945

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
213 changes: 16 additions & 197 deletions queue_job/controllers/main.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,208 +4,25 @@

import logging
import random
import time
import traceback
from contextlib import contextmanager
from io import StringIO

from psycopg2 import OperationalError, errorcodes
from werkzeug.exceptions import BadRequest, Forbidden

from odoo import SUPERUSER_ID, _, api, http
from odoo.service.model import PG_CONCURRENCY_ERRORS_TO_RETRY
from odoo.tools import config
from odoo import SUPERUSER_ID, _, http

from ..delay import chain, group
from ..exception import FailedJobError, RetryableJobError
from ..job import ENQUEUED, Job

_logger = logging.getLogger(__name__)

PG_RETRY = 5 # seconds

DEPENDS_MAX_TRIES_ON_CONCURRENCY_FAILURE = 5


@contextmanager
def _prevent_commit(cr):
"""Context manager to prevent commits on a cursor.

Commiting while the job is not finished would release the job lock, causing
it to be started again by the dead jobs requeuer.
"""

def forbidden_commit(*args, **kwargs):
raise RuntimeError(
"Commit is forbidden in queue jobs. "
'You may want to enable the "Allow Commit" option on the Job '
"Function. Alternatively, if the current job is a cron running as "
"queue job, you can modify it to run as a normal cron. More details on: "
"https://github.com/OCA/queue/wiki/Upgrade-warning:-commits-inside-jobs"
)
# unused imports are kept for backward compatibility
from ..executor import (
DEPENDS_MAX_TRIES_ON_CONCURRENCY_FAILURE, # noqa: F401
PG_RETRY, # noqa: F401
JobExecutor,
_prevent_commit, # noqa: F401
)

original_commit = cr.commit
cr.commit = forbidden_commit
try:
yield
finally:
cr.commit = original_commit
_logger = logging.getLogger(__name__)


class RunJobController(http.Controller):
@classmethod
def _acquire_job(cls, env: api.Environment, job_uuid: str) -> Job | None:
"""Acquire a job for execution.

- make sure it is in ENQUEUED state
- mark it as STARTED and commit the state change
- acquire the job lock

If successful, return the Job instance, otherwise return None. This
function may fail to acquire the job is not in the expected state or is
already locked by another worker.
"""
env.cr.execute(
"SELECT uuid FROM queue_job WHERE uuid=%s AND state=%s "
"FOR NO KEY UPDATE SKIP LOCKED",
(job_uuid, ENQUEUED),
)
if not env.cr.fetchone():
_logger.warning(
"was requested to run job %s, but it does not exist, "
"or is not in state %s, or is being handled by another worker",
job_uuid,
ENQUEUED,
)
return None
job = Job.load(env, job_uuid)
assert job and job.state == ENQUEUED
job.set_started()
job.store()
env.cr.commit()
if not job.lock():
_logger.warning(
"was requested to run job %s, but it could not be locked",
job_uuid,
)
return None
return job

@classmethod
def _try_perform_job(cls, env, job):
"""Try to perform the job, mark it done and commit if successful."""
_logger.debug("%s started", job)
# TODO refactor, the relation between env and job.env is not clear
assert env.cr is job.env.cr
with _prevent_commit(env.cr):
job.perform()
# Triggers any stored computed fields before calling 'set_done'
# so that will be part of the 'exec_time'
env.flush_all()
job.set_done()
job.store()
env.flush_all()
if not config["test_enable"]:
env.cr.commit()
_logger.debug("%s done", job)

@classmethod
def _enqueue_dependent_jobs(cls, env, job):
if not job.should_check_dependents():
return

_logger.debug("%s enqueue depends started", job)
tries = 0
while True:
try:
with job.env.cr.savepoint():
job.enqueue_waiting()
except OperationalError as err:
# Automatically retry the typical transaction serialization
# errors
if err.pgcode not in PG_CONCURRENCY_ERRORS_TO_RETRY:
raise
if tries >= DEPENDS_MAX_TRIES_ON_CONCURRENCY_FAILURE:
_logger.error(
"%s, maximum number of tries reached to update dependencies",
errorcodes.lookup(err.pgcode),
)
raise
wait_time = random.uniform(0.0, 2**tries)
tries += 1
_logger.info(
"%s, retry %d/%d in %.04f sec...",
errorcodes.lookup(err.pgcode),
tries,
DEPENDS_MAX_TRIES_ON_CONCURRENCY_FAILURE,
wait_time,
)
time.sleep(wait_time)
else:
break
_logger.debug("%s enqueue depends done", job)

@classmethod
def _runjob(cls, env: api.Environment, job: Job) -> None:
def retry_postpone(job, message, seconds=None):
job.env.clear()
with job.in_temporary_env():
job.postpone(result=message, seconds=seconds)
job.set_pending(reset_retry=False)
job.store()

try:
try:
cls._try_perform_job(env, job)
except OperationalError as err:
# Automatically retry the typical transaction serialization
# errors
if err.pgcode not in PG_CONCURRENCY_ERRORS_TO_RETRY:
raise

_logger.debug("%s OperationalError, postponed", job)
raise RetryableJobError(err.pgerror, seconds=PG_RETRY) from err

except RetryableJobError as err:
# delay the job later, requeue
retry_postpone(job, str(err), seconds=err.seconds)
_logger.debug("%s postponed", job)
# Do not trigger the error up because we don't want an exception
# traceback in the logs we should have the traceback when all
# retries are exhausted
env.cr.rollback()
return

except (FailedJobError, Exception) as orig_exception:
buff = StringIO()
traceback.print_exc(file=buff)
traceback_txt = buff.getvalue()
_logger.error(traceback_txt)
job.env.clear()
with job.in_temporary_env():
vals = cls._get_failure_values(job, traceback_txt, orig_exception)
job.set_failed(**vals)
job.store()
buff.close()
raise

cls._enqueue_dependent_jobs(env, job)

@classmethod
def _get_failure_values(cls, job, traceback_txt, orig_exception):
"""Collect relevant data from exception."""
exception_name = orig_exception.__class__.__name__
if hasattr(orig_exception, "__module__"):
exception_name = orig_exception.__module__ + "." + exception_name
exc_message = (
orig_exception.args[0] if orig_exception.args else str(orig_exception)
)
return {
"exc_info": traceback_txt,
"exc_name": exception_name,
"exc_message": exc_message,
}

@http.route(
"/queue_job/runjob",
type="http",
Expand All @@ -215,11 +32,13 @@ def _get_failure_values(cls, job, traceback_txt, orig_exception):
)
def runjob(self, db, job_uuid, **kw):
http.request.session.db = db
env = http.request.env(user=SUPERUSER_ID)
job = self._acquire_job(env, job_uuid)
if not job:
return ""
self._runjob(env, job)
# update_env (in contrast to a local request.env(user=...)) replaces
# the uid=None environment installed by auth="none" on the request
# itself. On Odoo >= 19 it additionally repoints
# transaction.default_env, which otherwise makes flushes recompute
# stored fields with uid=None (see OCA/queue issue #922)
http.request.update_env(user=SUPERUSER_ID)
JobExecutor(http.request.env, job_uuid).run()
return ""

# flake8: noqa: C901
Expand Down
Loading
Loading