diff --git a/queue_job/README.rst b/queue_job/README.rst index b27a1675ad..f1ee154c52 100644 --- a/queue_job/README.rst +++ b/queue_job/README.rst @@ -666,6 +666,9 @@ Known issues / Roadmap - After creating a new database or installing ``queue_job`` on an existing database, Odoo must be restarted for the runner to detect it. +- Reload runner configuration on SIGHUP +- ``eval_capacity`` ``workers`` should evaluate to sum of all workers + across all containers Changelog ========= diff --git a/queue_job/jobrunner/channels.py b/queue_job/jobrunner/channels.py index c895d9caf3..ab951ca6d9 100644 --- a/queue_job/jobrunner/channels.py +++ b/queue_job/jobrunner/channels.py @@ -7,6 +7,9 @@ from heapq import heappop, heappush from weakref import WeakValueDictionary +from odoo.tools import config as odoo_config +from odoo.tools.safe_eval import safe_eval + from ..exception import ChannelNotFound from ..job import CANCELLED, DONE, ENQUEUED, FAILED, PENDING, STARTED, WAIT_DEPENDENCIES @@ -806,8 +809,38 @@ def __init__(self): self._root_channel = Channel(name="root", parent=None, capacity=1) self._channels_by_name = WeakValueDictionary(root=self._root_channel) - @classmethod - def parse_simple_config(cls, config_string): + def configure(self, config_string): + """Configure the channel manager from a complex configuration string + + >>> from odoo.tools import config + >>> from pprint import pprint as pp + >>> config['workers'] = 10 + + >>> cm = ChannelManager() + >>> pp(cm.configure('root:workers')) + [{'capacity': 10, 'name': 'root'}] + + >>> cm = ChannelManager() + >>> pp(cm.configure('root:workers * 0.8')) + [{'capacity': 8, 'name': 'root'}] + + >>> cm = ChannelManager() + >>> pp(cm.configure('root:workers,child:root * 0.6')) + [{'capacity': 10, 'name': 'root'}, {'capacity': 6, 'name': 'child'}] + + >>> cm = ChannelManager() + >>> pp(cm.configure('root:workers,root.child:parent - 2,root.child.grandchild:parent * 0.5')) + [{'capacity': 10, 'name': 'root'}, + {'capacity': 8, 'name': 'root.child'}, + {'capacity': 4, 'name': 'root.child.grandchild'}] + """ # noqa: E501 + res = [] + for cfg in self.parse_config(config_string): + self.get_channel_from_config(cfg) + res.append(cfg) + return res + + def parse_simple_config(self, config_string): """Parse a simple channels configuration string. The general form is as follow: @@ -821,24 +854,25 @@ def parse_simple_config(cls, config_string): Returns a list of channel configuration dictionaries. >>> from pprint import pprint as pp - >>> pp(ChannelManager.parse_simple_config('root:4')) + >>> cm = ChannelManager() + >>> pp(cm.parse_simple_config('root:4')) [{'capacity': 4, 'name': 'root'}] - >>> pp(ChannelManager.parse_simple_config('root:4,root.sub:2')) + >>> pp(cm.parse_simple_config('root:4,root.sub:2')) [{'capacity': 4, 'name': 'root'}, {'capacity': 2, 'name': 'root.sub'}] - >>> pp(ChannelManager.parse_simple_config('root:4,root.sub:2:' + >>> pp(cm.parse_simple_config('root:4,root.sub:2:' ... 'sequential:k=v')) [{'capacity': 4, 'name': 'root'}, {'capacity': 2, 'k': 'v', 'name': 'root.sub', 'sequential': True}] - >>> pp(ChannelManager.parse_simple_config('root')) + >>> pp(cm.parse_simple_config('root')) [{'capacity': 1, 'name': 'root'}] - >>> pp(ChannelManager.parse_simple_config('sub:2')) + >>> pp(cm.parse_simple_config('sub:2')) [{'capacity': 2, 'name': 'sub'}] It ignores whitespace around values, and drops empty entries which would be generated by trailing commas, or commented lines on the Odoo config file. - >>> pp(ChannelManager.parse_simple_config(''' + >>> pp(cm.parse_simple_config(''' ... root : 4, ... , ... foo bar:1: k=va lue, @@ -849,7 +883,7 @@ def parse_simple_config(cls, config_string): It's also possible to replace commas with line breaks, which is more readable if the channel configuration comes from the odoo config file. - >>> pp(ChannelManager.parse_simple_config(''' + >>> pp(cm.parse_simple_config(''' ... root : 4 ... foo bar:1: k=va lue ... baz @@ -858,50 +892,59 @@ def parse_simple_config(cls, config_string): {'capacity': 1, 'k': 'va lue', 'name': 'foo bar'}, {'capacity': 1, 'name': 'baz'}] """ - res = [] + return list(self.parse_config(config_string, simple=True)) + + def parse_config(self, config_string, simple=False): config_string = config_string.replace("\n", ",") for channel_config_string in split_strip(config_string, ","): if not channel_config_string: # ignore empty entries (commented lines, trailing commas) continue - config = {} - config_items = split_strip(channel_config_string, ":") - name = config_items[0] - if not name: - raise ValueError( - f"Invalid channel config {config_string}: missing channel name" + yield self.parse_config_line(channel_config_string, simple=simple) + + def parse_config_line(self, config_string, simple=False): + config = {} + config_items = split_strip(config_string, ":") + name = config_items[0] + if not name: + raise ValueError( + f"Invalid channel config {config_string}: missing channel name" + ) + if not simple and (name == "parent"): + raise ValueError( + f"Invalid channel config {config_string}: channel name parent is reserved" # noqa: E501 + ) + config["name"] = name + if len(config_items) > 1: + capacity = config_items[1] + try: + config["capacity"] = ( + int(capacity) if simple else self.eval_capacity(name, capacity) ) - config["name"] = name - if len(config_items) > 1: - capacity = config_items[1] - try: - config["capacity"] = int(capacity) - except Exception as ex: + except Exception as ex: + raise ValueError( + f"Invalid channel config {config_string}: " + f"invalid capacity {capacity}" + ) from ex + for config_item in config_items[2:]: + kv = split_strip(config_item, "=") + if len(kv) == 1: + k, v = kv[0], True + elif len(kv) == 2: + k, v = kv + else: raise ValueError( f"Invalid channel config {config_string}: " - f"invalid capacity {capacity}" - ) from ex - for config_item in config_items[2:]: - kv = split_strip(config_item, "=") - if len(kv) == 1: - k, v = kv[0], True - elif len(kv) == 2: - k, v = kv - else: - raise ValueError( - f"Invalid channel config {config_string}: " - f"incorrect config item {config_item}" - ) - if k in config: - raise ValueError( - f"Invalid channel config {config_string}: " - f"duplicate key {k}" - ) - config[k] = v - else: - config["capacity"] = 1 - res.append(config) - return res + f"incorrect config item {config_item}" + ) + if k in config: + raise ValueError( + f"Invalid channel config {config_string}: " f"duplicate key {k}" + ) + config[k] = v + else: + config["capacity"] = 1 + return config def simple_configure(self, config_string): """Configure the channel manager from a simple configuration string @@ -927,9 +970,21 @@ def simple_configure(self, config_string): >>> cm.get_channel_by_name('seq').sequential True """ - for config in ChannelManager.parse_simple_config(config_string): + for config in self.parse_simple_config(config_string): self.get_channel_from_config(config) + def eval_capacity(self, name, expr) -> int: + locals_dict = {"workers": odoo_config["workers"]} + if name != "root": + root = self.get_channel_by_name("root") + # FIXME Can't do math on None, but does zero make sense? + locals_dict["root"] = root.capacity or 0 + # HACK Could this get the named channel by mistake? + parent = self.get_channel_by_name(name, parent_fallback=True) + locals_dict["parent"] = parent.capacity or 0 + capacity = safe_eval(expr, locals_dict=locals_dict) + return int(capacity) + def get_channel_from_config(self, config): """Return a Channel object from a parsed configuration. diff --git a/queue_job/readme/ROADMAP.md b/queue_job/readme/ROADMAP.md index df33142b88..1e3f25a8f2 100644 --- a/queue_job/readme/ROADMAP.md +++ b/queue_job/readme/ROADMAP.md @@ -1,2 +1,4 @@ - After creating a new database or installing `queue_job` on an existing database, Odoo must be restarted for the runner to detect it. +- Reload runner configuration on SIGHUP +- `eval_capacity` `workers` should evaluate to sum of all workers across all containers \ No newline at end of file diff --git a/queue_job/static/description/index.html b/queue_job/static/description/index.html index c5920c7ac7..93da65cad1 100644 --- a/queue_job/static/description/index.html +++ b/queue_job/static/description/index.html @@ -3,7 +3,7 @@ -README.rst +Job Queue