Skip to content

climate_ref_celery.routing #

Queue routing for Celery task submission

A deployment may wish to place executions on different queues depending on their size. The routing table maps diagnostics to queue names, so that each execution lands on a queue such as esmvaltool-large. Differently sized worker pools can then consume the queues independently.

The table is a TOML file whose path is given by the REF_CELERY_ROUTES environment variable. When the variable is unset, no table is loaded and every execution uses the bare provider queue.

Example:

.. code-block:: toml

default = "{provider}"

[esmvaltool]
default = "esmvaltool-medium"
rules = [
  { match = "portrait-*", queue = "esmvaltool-large" },
  { match = "sea-ice-basic", queue = "esmvaltool-small" },
]

[ilamb]
default = "ilamb-small"

Rules are matched against the diagnostic slug in order, first match wins. Patterns use :func:fnmatch.fnmatchcase semantics, so exact strings and glob wildcards both work. Queue names are templates in which {provider} expands to the provider slug.

A provider default applies when no rule matches. The top-level default applies when the provider has no entry. With no default and no match, the queue is the bare provider slug, equivalent to a default of "{provider}".

ROUTES_ENV_VAR = 'REF_CELERY_ROUTES' module-attribute #

Environment variable holding the path to the routing table file

ProviderRoutes #

The ordered rules and optional default queue for one provider

Source code in packages/climate-ref-celery/src/climate_ref_celery/routing.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
@frozen
class ProviderRoutes:
    """
    The ordered rules and optional default queue for one provider
    """

    rules: tuple[RoutingRule, ...] = ()
    default: str | None = None

    def queue_template_for(self, diagnostic_slug: str) -> str | None:
        """
        Resolve the queue template for a diagnostic, or the provider default if no rule matches
        """
        for rule in self.rules:
            if fnmatch.fnmatchcase(diagnostic_slug, rule.match):
                return rule.queue
        return self.default

queue_template_for(diagnostic_slug) #

Resolve the queue template for a diagnostic, or the provider default if no rule matches

Source code in packages/climate-ref-celery/src/climate_ref_celery/routing.py 
113
114
115
116
117
118
119
120
def queue_template_for(self, diagnostic_slug: str) -> str | None:
    """
    Resolve the queue template for a diagnostic, or the provider default if no rule matches
    """
    for rule in self.rules:
        if fnmatch.fnmatchcase(diagnostic_slug, rule.match):
            return rule.queue
    return self.default

RoutingRule #

A single pattern to queue rule

Source code in packages/climate-ref-celery/src/climate_ref_celery/routing.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
@frozen
class RoutingRule:
    """
    A single pattern to queue rule
    """

    match: str
    """Pattern matched against the diagnostic slug, with ``fnmatch`` semantics"""

    queue: str
    """Queue name template used when the pattern matches"""

match instance-attribute #

Pattern matched against the diagnostic slug, with fnmatch semantics

queue instance-attribute #

Queue name template used when the pattern matches

RoutingTable #

Deployment-supplied mapping of diagnostics to queue names

An empty table routes everything to the bare provider queue, which matches the behaviour when no table is configured, i.e. default = "{provider}".

Source code in packages/climate-ref-celery/src/climate_ref_celery/routing.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
@frozen
class RoutingTable:
    """
    Deployment-supplied mapping of diagnostics to queue names

    An empty table routes everything to the bare provider queue,
    which matches the behaviour when no table is configured,
    i.e. `default = "{provider}"`.
    """

    providers: Mapping[str, ProviderRoutes] = field(factory=dict)
    default: str | None = None

    def queue_for(self, provider_slug: str, diagnostic_slug: str) -> str:
        """
        Compute the queue name for an execution

        Returns
        -------
        :
            The matched queue template with ``{provider}`` expanded to the provider slug,
            or the bare provider slug when neither a rule nor a default applies
        """
        provider = self.providers.get(provider_slug)
        if provider is None:
            template = self.default
        else:
            template = provider.queue_template_for(diagnostic_slug)
        if template is None:
            return provider_slug
        return template.format(provider=provider_slug)

    @classmethod
    def from_file(cls, path: Path, known_providers: Collection[str] | None = None) -> "RoutingTable":
        """
        Load and validate a routing table from a TOML file

        Parameters
        ----------
        path
            Path to the TOML file
        known_providers
            Slugs of the currently registered providers.
            An entry for a provider not in this collection logs a warning, not an error,
            because deployments may share one table across environments with different provider sets.
            ``None`` skips the check.

        Raises
        ------
        RoutingTableError
            The file is missing, is not valid TOML, or contains a malformed entry
        """
        try:
            raw = tomllib.loads(path.read_text())
        except OSError as exc:
            raise RoutingTableError(f"Routing table {path}: cannot read file ({exc})") from exc
        except tomllib.TOMLDecodeError as exc:
            raise RoutingTableError(f"Routing table {path}: invalid TOML ({exc})") from exc

        default: str | None = None
        providers: dict[str, ProviderRoutes] = {}

        for key, value in raw.items():
            if key == "default":
                default = _require_queue_template(value, path, "top-level 'default'")
                continue
            if not isinstance(value, dict):
                raise RoutingTableError(
                    f"Routing table {path}: '{key}' must be a provider table, got {value!r}"
                )
            providers[key] = cls._parse_provider(value, path, key)

        if known_providers is not None:
            for slug in providers.keys() - set(known_providers):
                logger.warning(f"Routing table {path}: provider '{slug}' is not currently registered")

        return cls(providers=providers, default=default)

    @classmethod
    def _parse_provider(cls, raw: dict[str, Any], path: Path, provider: str) -> ProviderRoutes:
        unknown = raw.keys() - {"default", "rules"}
        if unknown:
            raise RoutingTableError(f"Routing table {path}: [{provider}] has unknown keys {sorted(unknown)}")

        default: str | None = None
        if "default" in raw:
            default = _require_queue_template(raw["default"], path, f"[{provider}] 'default'")

        raw_rules = raw.get("rules", [])
        if not isinstance(raw_rules, list):
            raise RoutingTableError(f"Routing table {path}: [{provider}] 'rules' must be a list")

        rules = []
        for index, raw_rule in enumerate(raw_rules):
            entry = f"[{provider}] rule {index}"
            if not isinstance(raw_rule, dict) or raw_rule.keys() != {"match", "queue"}:
                raise RoutingTableError(
                    f"Routing table {path}: {entry} must have exactly 'match' and 'queue' keys, "
                    f"got {raw_rule!r}"
                )
            rules.append(
                RoutingRule(
                    match=_require_string(raw_rule["match"], path, f"{entry} 'match'"),
                    queue=_require_queue_template(raw_rule["queue"], path, f"{entry} 'queue'"),
                )
            )

        patterns = [rule.match for rule in rules]
        for pattern in sorted({p for p in patterns if patterns.count(p) > 1}):
            logger.warning(f"Routing table {path}: [{provider}] has duplicate pattern '{pattern}'")

        return ProviderRoutes(rules=tuple(rules), default=default)

from_file(path, known_providers=None) classmethod #

Load and validate a routing table from a TOML file

Parameters:

Name Type Description Default
path Path

Path to the TOML file

 required
known_providers Collection[str] | None

Slugs of the currently registered providers. An entry for a provider not in this collection logs a warning, not an error, because deployments may share one table across environments with different provider sets. None skips the check.

 None

Raises:

Type Description
RoutingTableError

The file is missing, is not valid TOML, or contains a malformed entry
Source code in packages/climate-ref-celery/src/climate_ref_celery/routing.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
@classmethod
def from_file(cls, path: Path, known_providers: Collection[str] | None = None) -> "RoutingTable":
    """
    Load and validate a routing table from a TOML file

    Parameters
    ----------
    path
        Path to the TOML file
    known_providers
        Slugs of the currently registered providers.
        An entry for a provider not in this collection logs a warning, not an error,
        because deployments may share one table across environments with different provider sets.
        ``None`` skips the check.

    Raises
    ------
    RoutingTableError
        The file is missing, is not valid TOML, or contains a malformed entry
    """
    try:
        raw = tomllib.loads(path.read_text())
    except OSError as exc:
        raise RoutingTableError(f"Routing table {path}: cannot read file ({exc})") from exc
    except tomllib.TOMLDecodeError as exc:
        raise RoutingTableError(f"Routing table {path}: invalid TOML ({exc})") from exc

    default: str | None = None
    providers: dict[str, ProviderRoutes] = {}

    for key, value in raw.items():
        if key == "default":
            default = _require_queue_template(value, path, "top-level 'default'")
            continue
        if not isinstance(value, dict):
            raise RoutingTableError(
                f"Routing table {path}: '{key}' must be a provider table, got {value!r}"
            )
        providers[key] = cls._parse_provider(value, path, key)

    if known_providers is not None:
        for slug in providers.keys() - set(known_providers):
            logger.warning(f"Routing table {path}: provider '{slug}' is not currently registered")

    return cls(providers=providers, default=default)

queue_for(provider_slug, diagnostic_slug) #

Compute the queue name for an execution

Returns:

Type Description
str

The matched queue template with {provider} expanded to the provider slug, or the bare provider slug when neither a rule nor a default applies
Source code in packages/climate-ref-celery/src/climate_ref_celery/routing.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def queue_for(self, provider_slug: str, diagnostic_slug: str) -> str:
    """
    Compute the queue name for an execution

    Returns
    -------
    :
        The matched queue template with ``{provider}`` expanded to the provider slug,
        or the bare provider slug when neither a rule nor a default applies
    """
    provider = self.providers.get(provider_slug)
    if provider is None:
        template = self.default
    else:
        template = provider.queue_template_for(diagnostic_slug)
    if template is None:
        return provider_slug
    return template.format(provider=provider_slug)

RoutingTableError #

Bases: ValueError

Raised when a routing table file is malformed

A malformed table fails hard rather than falling back to default routing, because silent fallback would misplace large jobs onto small workers.

Source code in packages/climate-ref-celery/src/climate_ref_celery/routing.py 
53
54
55
56
57
58
59
class RoutingTableError(ValueError):
    """
    Raised when a routing table file is malformed

    A malformed table fails hard rather than falling back to default routing,
    because silent fallback would misplace large jobs onto small workers.
    """

load_routing_table(known_providers=None) #

Load the routing table named by REF_CELERY_ROUTES, or an empty table if unset

Parameters:

Name Type Description Default
known_providers Collection[str] | None

Slugs of the currently registered providers, used to warn about stale entries. None skips the check.

 None

Raises:

Type Description
RoutingTableError

The variable is set but the file is missing or malformed
Source code in packages/climate-ref-celery/src/climate_ref_celery/routing.py 
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
def load_routing_table(known_providers: Collection[str] | None = None) -> RoutingTable:
    """
    Load the routing table named by ``REF_CELERY_ROUTES``, or an empty table if unset

    Parameters
    ----------
    known_providers
        Slugs of the currently registered providers, used to warn about stale entries.
        ``None`` skips the check.

    Raises
    ------
    RoutingTableError
        The variable is set but the file is missing or malformed
    """
    path = os.environ.get(ROUTES_ENV_VAR)
    if not path:
        return RoutingTable()
    return RoutingTable.from_file(Path(path), known_providers=known_providers)