Skip to content
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

Add support for custom celery configs #45038

Open
wants to merge 5 commits into
base: main
Choose a base branch
from
+23 −0
Open

Add support for custom celery configs #45038

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
6084c54
#45037: Support for additional celery config directly from airflow.cf…
sachin-arora-cashfree Dec 18, 2024
03431e6
Added unit test for the additional config and addressed comments
sachin-arora-cashfree Dec 18, 2024
bfaf924
Added unit test for the additional config and addressed comments
sachin-arora-cashfree Dec 18, 2024
303af23
Addressed comments: Added sample config in the docs as well as refere…
sachin-arora-cashfree Dec 22, 2024
382d45e
Addressed comments: Added sample config in the docs as well as refere…
sachin-arora-cashfree Dec 22, 2024
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
3 changes: 3 additions & 0 deletions providers/src/airflow/providers/celery/executors/default_celery.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,8 @@ def _broker_supports_visibility_timeout(url):
log.debug("Value for celery result_backend not found. Using sql_alchemy_conn with db+ prefix.")
result_backend = f'db+{conf.get("database", "SQL_ALCHEMY_CONN")}'

extra_celery_config: dict = conf.getjson("celery", "extra_celery_config", fallback={})

DEFAULT_CELERY_CONFIG = {
"accept_content": ["json"],
"event_serializer": "json",
Expand All @@ -85,6 +87,7 @@ def _broker_supports_visibility_timeout(url):
),
"worker_concurrency": conf.getint("celery", "WORKER_CONCURRENCY", fallback=16),
"worker_enable_remote_control": conf.getboolean("celery", "worker_enable_remote_control", fallback=True),
**extra_celery_config,
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As CI did not get to this stage (yet) as failing earlier, there will be a problem reported by mypy for providers like:

providers/src/airflow/providers/celery/executors/default_celery.py:72: error:
Incompatible types in assignment (expression has type
"Union[dict[Any, Any], list[Any], str, int, float, None]", variable has type
"dict[Any, Any]")  [assignment]
    extra_celery_config: dict = conf.getjson("celery", "extra_celery_confi...
                                ^~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~...
Found 1 error in 1 file (checked 1 source file)
Error 1 returned

As the JSON result can be other values than a dict (irrespective of the type declaration above, you need to use:

Suggested change
**extra_celery_config,
**(extra_celery_config if isinstance(extra_celery_config, dict) else {}),

}


Expand Down
11 changes: 11 additions & 0 deletions providers/src/airflow/providers/celery/provider.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -330,6 +330,17 @@ config:
type: string
example: ~
default: "False"
extra_celery_config:
description: |
Extra celery configs to include in the celery worker.
Any of the celery config can be added to this config and it
will be applied while starting the celery worker. e.g. {"worker_max_tasks_per_child": 10}
See also:
https://docs.celeryq.dev/en/stable/userguide/configuration.html#configuration-and-defaults
version_added: ~
type: string
example: ~
default: "{}"
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Okay, I also did not know but defaults seem to be Format-String formatted and the string "{}" is an invalid sequence. So to effectively have an {} as default we need to use {{}} here to quote the brackets.

Suggested change
default: "{}"
default: "{{}}"

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Okay ... interesting, this also resolves the CI problem - it seems the constrains also can not be built as the templating fails there. I did a PR with the fixes applied as test and this is turning green: #45160
As with the other proposal a check is being made that the settings are a dict, you can even lave an empty string as default...

Suggested change
default: "{}"
default: ""

celery_broker_transport_options:
description: |
This section is for specifying options which can be passed to the
Expand Down
9 changes: 9 additions & 0 deletions providers/tests/celery/executors/test_celery_executor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -399,3 +399,12 @@ def test_celery_task_acks_late_loaded_from_string():
# reload celery conf to apply the new config
importlib.reload(default_celery)
assert default_celery.DEFAULT_CELERY_CONFIG["task_acks_late"] is False


@conf_vars({("celery", "extra_celery_config"): '{"worker_max_tasks_per_child": 10}'})
def test_celery_extra_celery_config_loaded_from_string():
import importlib

# reload celery conf to apply the new config
importlib.reload(default_celery)
assert default_celery.DEFAULT_CELERY_CONFIG["extra_celery_config"] == {"worker_max_tasks_per_child": 10}
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As you did not hit it (yet) as CI was broken, in the test baloon PR #45160 the check fail.

Loading