From 344369d610c391cb5406ca50beea8ecfec725c3a Mon Sep 17 00:00:00 2001 From: Vitalii Lazuka <94494788+vlazuka@users.noreply.github.com> Date: Tue, 2 May 2023 12:50:49 -0700 Subject: [PATCH 1/9] Update default_celery.py to include rediss to the list of supported url connections --- airflow/config_templates/default_celery.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/airflow/config_templates/default_celery.py b/airflow/config_templates/default_celery.py index d3d5a4adf11e2..cb4b171828dce 100644 --- a/airflow/config_templates/default_celery.py +++ b/airflow/config_templates/default_celery.py @@ -26,7 +26,7 @@ def _broker_supports_visibility_timeout(url): - return url.startswith("redis://") or url.startswith("sqs://") + return url.startswith("redis://") or url.startswith("rediss://") or url.startswith("sqs://") log = logging.getLogger(__name__) From bafbe2e21aac1f16f48492b9d7c38c90a4fbb476 Mon Sep 17 00:00:00 2001 From: Vitalii Lazuka <94494788+vlazuka@users.noreply.github.com> Date: Wed, 3 May 2023 02:15:24 -0700 Subject: [PATCH 2/9] Update default_celery.py --- airflow/config_templates/default_celery.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/airflow/config_templates/default_celery.py b/airflow/config_templates/default_celery.py index cb4b171828dce..a8e6e58c40843 100644 --- a/airflow/config_templates/default_celery.py +++ b/airflow/config_templates/default_celery.py @@ -20,13 +20,14 @@ import logging import ssl +import re from airflow.configuration import conf from airflow.exceptions import AirflowConfigException, AirflowException def _broker_supports_visibility_timeout(url): - return url.startswith("redis://") or url.startswith("rediss://") or url.startswith("sqs://") + return url.startswith(("redis://", "rediss://", "sqs://")) log = logging.getLogger(__name__) @@ -67,14 +68,14 @@ def _broker_supports_visibility_timeout(url): try: if celery_ssl_active: - if broker_url and "amqp://" in broker_url: + if broker_url and re.search("amqp://", broker_url): broker_use_ssl = { "keyfile": conf.get("celery", "SSL_KEY"), "certfile": conf.get("celery", "SSL_CERT"), "ca_certs": conf.get("celery", "SSL_CACERT"), "cert_reqs": ssl.CERT_REQUIRED, } - elif broker_url and "redis://" in broker_url: + elif broker_url and re.search("rediss?://", broker_url): broker_use_ssl = { "ssl_keyfile": conf.get("celery", "SSL_KEY"), "ssl_certfile": conf.get("celery", "SSL_CERT"), @@ -100,7 +101,7 @@ def _broker_supports_visibility_timeout(url): f"all necessary certs and key ({e})." ) -if "amqp://" in result_backend or "redis://" in result_backend or "rpc://" in result_backend: +if re.search("rediss?://|amqp://|rpc://", str):: log.warning( "You have configured a result_backend of %s, it is highly recommended " "to use an alternative result_backend (i.e. a database).", From b76565a7249b0cb4eae196100eb6a8a743cc4ff5 Mon Sep 17 00:00:00 2001 From: Vitalii Lazuka <94494788+vlazuka@users.noreply.github.com> Date: Wed, 3 May 2023 09:02:31 -0700 Subject: [PATCH 3/9] Update default_celery.py --- airflow/config_templates/default_celery.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/airflow/config_templates/default_celery.py b/airflow/config_templates/default_celery.py index a8e6e58c40843..e2c656201d318 100644 --- a/airflow/config_templates/default_celery.py +++ b/airflow/config_templates/default_celery.py @@ -101,7 +101,7 @@ def _broker_supports_visibility_timeout(url): f"all necessary certs and key ({e})." ) -if re.search("rediss?://|amqp://|rpc://", str):: +if re.search("rediss?://|amqp://|rpc://", str): log.warning( "You have configured a result_backend of %s, it is highly recommended " "to use an alternative result_backend (i.e. a database).", From 13541de1016dca5c8dd4094e0f917a453dca2acf Mon Sep 17 00:00:00 2001 From: Vitalii Lazuka <94494788+vlazuka@users.noreply.github.com> Date: Wed, 3 May 2023 12:56:29 -0700 Subject: [PATCH 4/9] Update default_celery.py --- airflow/config_templates/default_celery.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/airflow/config_templates/default_celery.py b/airflow/config_templates/default_celery.py index e2c656201d318..989710376a7d9 100644 --- a/airflow/config_templates/default_celery.py +++ b/airflow/config_templates/default_celery.py @@ -68,7 +68,7 @@ def _broker_supports_visibility_timeout(url): try: if celery_ssl_active: - if broker_url and re.search("amqp://", broker_url): + if broker_url and "amqp://" in broker_url: broker_use_ssl = { "keyfile": conf.get("celery", "SSL_KEY"), "certfile": conf.get("celery", "SSL_CERT"), From 6e81217fbad09e641e0bce48f2c7e7ef9dafd25a Mon Sep 17 00:00:00 2001 From: Vitalii Lazuka <94494788+vlazuka@users.noreply.github.com> Date: Fri, 12 May 2023 10:34:07 -0700 Subject: [PATCH 5/9] typo fixed --- airflow/config_templates/default_celery.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/airflow/config_templates/default_celery.py b/airflow/config_templates/default_celery.py index 551abb220ba56..68847ec9a44b6 100644 --- a/airflow/config_templates/default_celery.py +++ b/airflow/config_templates/default_celery.py @@ -111,7 +111,7 @@ def _broker_supports_visibility_timeout(url): f"all necessary certs and key ({e})." ) -if re.search("rediss?://|amqp://|rpc://", str): +if re.search("rediss?://|amqp://|rpc://", result_backend): log.warning( "You have configured a result_backend of %s, it is highly recommended " "to use an alternative result_backend (i.e. a database).", From c812664df486de40376dae8a1edff5a14edcaa3b Mon Sep 17 00:00:00 2001 From: Vitalii Lazuka Date: Thu, 25 May 2023 16:48:49 -0700 Subject: [PATCH 6/9] code from sentinel changes is fixed + typo fixes --- airflow/config_templates/default_celery.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/airflow/config_templates/default_celery.py b/airflow/config_templates/default_celery.py index 68847ec9a44b6..87d6d6a863123 100644 --- a/airflow/config_templates/default_celery.py +++ b/airflow/config_templates/default_celery.py @@ -19,8 +19,8 @@ from __future__ import annotations import logging -import ssl import re +import ssl from airflow.configuration import conf from airflow.exceptions import AirflowConfigException, AirflowException @@ -39,7 +39,7 @@ def _broker_supports_visibility_timeout(url): if _broker_supports_visibility_timeout(broker_url): broker_transport_options["visibility_timeout"] = 21600 -broker_transport_options_for_celery: dict = dict.copy(broker_transport_options) +broker_transport_options_for_celery = broker_transport_options.copy() if "sentinel_kwargs" in broker_transport_options: try: sentinel_kwargs = conf.getjson("celery_broker_transport_options", "sentinel_kwargs") @@ -53,7 +53,7 @@ def _broker_supports_visibility_timeout(url): result_backend = conf.get_mandatory_value("celery", "RESULT_BACKEND") else: 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")}' + result_backend = "db+{}".format(conf.get("database", "SQL_ALCHEMY_CONN")) DEFAULT_CELERY_CONFIG = { "accept_content": ["json"], @@ -107,8 +107,8 @@ def _broker_supports_visibility_timeout(url): ) except Exception as e: raise AirflowException( - f"Exception: There was an unknown Celery SSL Error. Please ensure you want to use SSL and/or have " - f"all necessary certs and key ({e})." + "Exception: There was an unknown Celery SSL Error. Please ensure you " + "want to use SSL and/or have all necessary certs and key ({}).".format(e) ) if re.search("rediss?://|amqp://|rpc://", result_backend): From b8c4426497a41786617ac954c8416b844cb00623 Mon Sep 17 00:00:00 2001 From: Vitalii Lazuka Date: Fri, 26 May 2023 13:28:37 -0700 Subject: [PATCH 7/9] bringing back f-string and fixing sentinel --- airflow/config_templates/default_celery.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/airflow/config_templates/default_celery.py b/airflow/config_templates/default_celery.py index 87d6d6a863123..7acc9a0a9f8a8 100644 --- a/airflow/config_templates/default_celery.py +++ b/airflow/config_templates/default_celery.py @@ -42,18 +42,19 @@ def _broker_supports_visibility_timeout(url): broker_transport_options_for_celery = broker_transport_options.copy() if "sentinel_kwargs" in broker_transport_options: try: - sentinel_kwargs = conf.getjson("celery_broker_transport_options", "sentinel_kwargs") + sentinel_kwargs = broker_transport_options.get("sentinel_kwargs") if not isinstance(sentinel_kwargs, dict): raise ValueError broker_transport_options_for_celery["sentinel_kwargs"] = sentinel_kwargs except Exception: raise AirflowException("sentinel_kwargs should be written in the correct dictionary format.") + if conf.has_option("celery", "RESULT_BACKEND"): result_backend = conf.get_mandatory_value("celery", "RESULT_BACKEND") else: log.debug("Value for celery result_backend not found. Using sql_alchemy_conn with db+ prefix.") - result_backend = "db+{}".format(conf.get("database", "SQL_ALCHEMY_CONN")) + result_backend = f'db+{conf.get("database", "SQL_ALCHEMY_CONN")}' DEFAULT_CELERY_CONFIG = { "accept_content": ["json"], @@ -107,8 +108,8 @@ def _broker_supports_visibility_timeout(url): ) except Exception as e: raise AirflowException( - "Exception: There was an unknown Celery SSL Error. Please ensure you " - "want to use SSL and/or have all necessary certs and key ({}).".format(e) + f"Exception: There was an unknown Celery SSL Error. Please ensure you want to use SSL and/or have " + f"all necessary certs and key ({e})." ) if re.search("rediss?://|amqp://|rpc://", result_backend): From 662689d3918cad159b85d9b620028eb55029b0db Mon Sep 17 00:00:00 2001 From: Tzu-ping Chung Date: Mon, 29 May 2023 15:30:45 +0800 Subject: [PATCH 8/9] Add back annotation to resolve Mypy error --- airflow/config_templates/default_celery.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/airflow/config_templates/default_celery.py b/airflow/config_templates/default_celery.py index 7acc9a0a9f8a8..bd5363dc347b3 100644 --- a/airflow/config_templates/default_celery.py +++ b/airflow/config_templates/default_celery.py @@ -39,7 +39,7 @@ def _broker_supports_visibility_timeout(url): if _broker_supports_visibility_timeout(broker_url): broker_transport_options["visibility_timeout"] = 21600 -broker_transport_options_for_celery = broker_transport_options.copy() +broker_transport_options_for_celery: dict = broker_transport_options.copy() if "sentinel_kwargs" in broker_transport_options: try: sentinel_kwargs = broker_transport_options.get("sentinel_kwargs") From 08164b624ccb2bd4dfd09c06c48c27c798218166 Mon Sep 17 00:00:00 2001 From: Tzu-ping Chung Date: Mon, 29 May 2023 15:31:23 +0800 Subject: [PATCH 9/9] MInimize diff --- airflow/config_templates/default_celery.py | 1 - 1 file changed, 1 deletion(-) diff --git a/airflow/config_templates/default_celery.py b/airflow/config_templates/default_celery.py index bd5363dc347b3..78dea869e368f 100644 --- a/airflow/config_templates/default_celery.py +++ b/airflow/config_templates/default_celery.py @@ -49,7 +49,6 @@ def _broker_supports_visibility_timeout(url): except Exception: raise AirflowException("sentinel_kwargs should be written in the correct dictionary format.") - if conf.has_option("celery", "RESULT_BACKEND"): result_backend = conf.get_mandatory_value("celery", "RESULT_BACKEND") else: