search

aws / aws-mwaa-local-runner

This repository provides a command line interface (CLI) utility that replicates an Amazon Managed Workflows for Apache Airflow (MWAA) environment locally.
MIT No Attribution
692 stars 702 forks source link

conn_id `postgres` isn't defined - f"The conn_id `{conn_id}` isn't defined" #209

Open ck-here opened 1 year ago

ck-here commented 1 year ago

Though I added postgres connection using UI, test run failing with below error, there are no typos in the connection string which I added using UI.

File "/usr/local/airflow/.local/lib/python3.10/site-packages/airflow/hooks/base.py", line 70, in get_connection conn = Connection.get_connection_from_secrets(conn_id) File "/usr/local/airflow/.local/lib/python3.10/site-packages/airflow/models/connection.py", line 432, in get_connection_from_secrets
raise AirflowNotFoundException(f"The conn_id {conn_id} isn't defined") airflow.exceptions.AirflowNotFoundException: The conn_id postgres isn't defined

Entire log is here

[airflow@8ccee5c9c63e ~]$ airflow tasks test second_dag create_table 2022-01-01 /usr/local/airflow/.local/lib/python3.10/site-packages/airflow/configuration.py:545: DeprecationWarning: The sql_alchemy_conn option in [core] has been moved to the sql_alchemy_conn option in [database] - the old setting has been used, but please update your config. option = self._get_environment_variables(deprecated_key, deprecated_section, key, section) /usr/local/airflow/.local/lib/python3.10/site-packages/airflow/configuration.py:443: FutureWarning: The 'dag_default_view' setting in [webserver] has the old default value of 'tree'. This value has been changed to 'grid' in the running config, but please update your config before Apache Airflow 3.0. warnings.warn( /usr/local/airflow/.local/lib/python3.10/site-packages/airflow/configuration.py:443: FutureWarning: The 'log_filename_template' setting in [logging] has the old default value of '{{ ti.dag_id }}/{{ ti.task_id }}/{{ ts }}/{{ try_number }}.log'. This value has been changed to 'dag_id={{ ti.dag_id }}/run_id={{ ti.run_id }}/task_id={{ ti.task_id }}/{% if ti.map_index >= 0 %}map_index={{ ti.map_index }}/{% endif %}attempt={{ try_number }}.log' in the running config, but please update your config before Apache Airflow 3.0. warnings.warn( /usr/local/airflow/.local/lib/python3.10/site-packages/airflow/configuration.py:549: DeprecationWarning: The auth_backend option in [api] has been renamed to auth_backends - the old setting has been used, but please update your config. option = self._get_option_from_config_file(deprecated_key, deprecated_section, key, kwargs, section) /usr/local/airflow/.local/lib/python3.10/site-packages/airflow/configuration.py:443: FutureWarning: The 'auth_backends' setting in [api] has the old default value of 'airflow.api.auth.backend.deny_all'. This value has been changed to 'airflow.api.auth.backend.session' in the running config, but please update your config before Apache Airflow 3.0. warnings.warn( /usr/local/airflow/.local/lib/python3.10/site-packages/airflow/configuration.py:549 DeprecationWarning: The sql_alchemy_pool_enabled option in [core] has been moved to the sql_alchemy_pool_enabled option in [database] - the old setting has been used, but please update your config. /usr/local/airflow/.local/lib/python3.10/site-packages/airflow/configuration.py:549 DeprecationWarning: The sql_engine_encoding option in [core] has been moved to the sql_engine_encoding option in [database] - the old setting has been used, but please update your config.
/usr/local/airflow/.local/lib/python3.10/site-packages/airflow/configuration.py:549 DeprecationWarning: The sql_alchemy_schema option in [core] has been moved to the sql_alchemy_schema option in [database] - the old setting has been used, but please update your config.
/usr/local/airflow/.local/lib/python3.10/site-packages/airflow/configuration.py:549 DeprecationWarning: The max_db_retries option in [core] has been moved to the max_db_retries option in [database] - the old setting has been used, but please update your config. [2023-02-12 19:48:31,592] {{dagbag.py:537}} INFO - Filling up the DagBag from /usr/local//dags /usr/local/airflow/.local/lib/python3.10/site-packages/airflow/models/dag.py:3405 RemovedInAirflow3Warning: Param schedule_interval is deprecated and will be removed in a future release. Please use schedule instead. [2023-02-12 19:48:33,108] {{example_kubernetes_executor.py:41}} WARNING - The example_kubernetes_executor example DAG requires the kubernetes provider. Please install it with: pip install apache-[cncf.kubernetes] [2023-02-12 19:48:33,130] {{taskmixin.py:205}} WARNING - Dependency <Task(_PythonDecoratedOperator): prepare_email>, send_email already registered for DAG: example_dag_decorator [2023-02-12 19:48:33,131] {{taskmixin.py:205}} WARNING - Dependency <Task(EmailOperator): send_email>, prepare_email already registered for DAG: example_dag_decorator [2023-02-12 19:48:33,177] {{taskmixin.py:205}} WARNING - Dependency <Task(BashOperator): create_entry_group>, delete_entry_group already registered for DAG: example_complex [2023-02-12 19:48:33,177] {{taskmixin.py:205}} WARNING - Dependency <Task(BashOperator): delete_entry_group>, create_entry_group already registered for DAG: example_complex [2023-02-12 19:48:33,178] {{taskmixin.py:205}} WARNING - Dependency <Task(BashOperator): create_entry_gcs>, delete_entry already registered for DAG: example_complex [2023-02-12 19:48:33,178] {{taskmixin.py:205}} WARNING - Dependency <Task(BashOperator): delete_entry>, create_entry_gcs already registered for DAG: example_complex [2023-02-12 19:48:33,179] {{taskmixin.py:205}} WARNING - Dependency <Task(BashOperator): create_tag>, delete_tag already registered for DAG: example_complex [2023-02-12 19:48:33,179] {{taskmixin.py:205}} WARNING - Dependency <Task(BashOperator): delete_tag>, create_tag already registered for DAG: example_complex [2023-02-12 19:48:33,233] {{taskmixin.py:205}} WARNING - Dependency <Task(_PythonDecoratedOperator): print_the_context>, log_sql_query already registered for DAG: example_python_operator [2023-02-12 19:48:33,234] {{taskmixin.py:205}} WARNING - Dependency <Task(_PythonDecoratedOperator): log_sql_query>, print_the_context already registered for DAG: example_python_operator [2023-02-12 19:48:33,235] {{taskmixin.py:205}} WARNING - Dependency <Task(_PythonDecoratedOperator): print_the_context>, log_sql_query already registered for DAG: example_python_operator [2023-02-12 19:48:33,236] {{taskmixin.py:205}} WARNING - Dependency <Task(_PythonDecoratedOperator): log_sql_query>, print_the_context already registered for DAG: example_python_operator [2023-02-12 19:48:33,237] {{taskmixin.py:205}} WARNING - Dependency <Task(_PythonDecoratedOperator): print_the_context>, log_sql_query already registered for DAG: example_python_operator [2023-02-12 19:48:33,237] {{taskmixin.py:205}} WARNING - Dependency <Task(_PythonDecoratedOperator): log_sql_query>, print_the_context already registered for DAG: example_python_operator [2023-02-12 19:48:33,239] {{taskmixin.py:205}} WARNING - Dependency <Task(_PythonDecoratedOperator): print_the_context>, log_sql_query already registered for DAG: example_python_operator [2023-02-12 19:48:33,239] {{taskmixin.py:205}} WARNING - Dependency <Task(_PythonDecoratedOperator): log_sql_query>, print_the_context already registered for DAG: example_python_operator [2023-02-12 19:48:33,301] {{example_local_kubernetes_executor.py:39}} WARNING - Could not import DAGs in example_local_kubernetes_executor.py Traceback (most recent call last): File "/usr/local/airflow/.local/lib/python3.10/site-packages/airflow/example_dags/example_local_kubernetes_executor.py", line 37, in from kubernetes.client import models as k8s ModuleNotFoundError: No module named 'kubernetes' [2023-02-12 19:48:33,301] {{example_local_kubernetes_executor.py:40}} WARNING - Install Kubernetes dependencies with: pip install apache-[cncf.kubernetes] /usr/local/airflow/.local/lib/python3.10/site-packages/airflow/cli/commands/task_command.py:129 RemovedInAirflow3Warning: Calling DAG.create_dagrun() without an explicit data interval is deprecated /usr/local/airflow/.local/lib/python3.10/site-packages/airflow/configuration.py:545 DeprecationWarning: The sql_alchemy_conn option in [core] has been moved to the sql_alchemy_conn option in [database] - the old setting has been used, but please update your config. [2023-02-12 19:48:33,561] {{taskinstance.py:1165}} INFO - Dependencies all met for <TaskInstance: second_dag.create_table ___temporary_run_2023-02-12T19:48:33.493737+00:00 [None]> [2023-02-12 19:48:33,573] {{taskinstance.py:1165}} INFO - Dependencies all met for <TaskInstance: second_dag.create_table ***_temporary_run_2023-02-12T19:48:33.493737+00:00__ [None]> [2023-02-12 19:48:33,574] {{taskinstance.py:1362}} INFO -

[2023-02-12 19:48:33,574] {{taskinstance.py:1363}} INFO - Starting attempt 1 of 1 [2023-02-12 19:48:33,574] {{taskinstance.py:1364}} INFO -

[2023-02-12 19:48:33,582] {{taskinstance.py:1383}} INFO - Executing <Task(PostgresOperator): create_table> on 2022-01-01T00:00:00+00:00 [2023-02-12 19:48:33,767] {{taskinstance.py:1590}} INFO - Exporting the following env vars: AIRFLOW_CTX_DAG_OWNER= AIRFLOW_CTX_DAG_ID=second_dag AIRFLOW_CTX_TASK_ID=create_table AIRFLOW_CTX_EXECUTION_DATE=2022-01-01T00:00:00+00:00 AIRFLOW_CTX_TRY_NUMBER=1 AIRFLOW_CTX_DAG_RUN_ID=___temporary_run_2023-02-12T19:48:33.493737+00:00 [2023-02-12 19:48:33,785] {{taskinstance.py:1851}} ERROR - Task failed with exception Traceback (most recent call last): File "/usr/local//.local/lib/python3.10/site-packages//providers/postgres/operators/postgres.py", line 94, in execute self.hook.run(self.sql, self.autocommit, parameters=self.parameters) File "/usr/local//.local/lib/python3.10/site-packages//providers/common/sql/hooks/sql.py", line 288, in run with closing(self.get_conn()) as conn: File "/usr/local//.local/lib/python3.10/site-packages//providers/postgres/hooks/postgres.py", line 88, in get_conn conn = deepcopy(self.connection or self.get_connection(conn_id)) File "/usr/local//.local/lib/python3.10/site-packages//hooks/base.py", line 70, in get_connection conn = Connection.get_connection_from_secrets(conn_id) File "/usr/local//.local/lib/python3.10/site-packages//models/connection.py", line 432, in get_connection_from_secrets raise AirflowNotFoundException(f"The conn_id {conn_id} isn't defined") ***.exceptions.AirflowNotFoundException: The conn_id postgres isn't defined [2023-02-12 19:48:33,787] {{taskinstance.py:1401}} INFO - Marking task as FAILED. dag_id=second_dag, task_id=create_table, execution_date=20220101T000000, start_date=, end_date=20230212T194833 Traceback (most recent call last): File "/usr/local/airflow/.local/bin/airflow", line 8, in sys.exit(main()) File "/usr/local/airflow/.local/lib/python3.10/site-packages/airflow/main__.py", line 39, in main args.func(args) File "/usr/local/airflow/.local/lib/python3.10/site-packages/airflow/cli/cli_parser.py", line 52, in command return func(*args, kwargs) File "/usr/local/airflow/.local/lib/python3.10/site-packages/airflow/utils/cli.py", line 103, in wrapper return f(*args, *kwargs) File "/usr/local/airflow/.local/lib/python3.10/site-packages/airflow/cli/commands/task_command.py", line 554, in task_test ti.run(ignore_task_deps=True, ignore_ti_state=True, test_mode=True) File "/usr/local/airflow/.local/lib/python3.10/site-packages/airflow/utils/session.py", line 75, in wrapper return func(args, session=session, kwargs) File "/usr/local/airflow/.local/lib/python3.10/site-packages/airflow/models/taskinstance.py", line 1752, in run self._run_raw_task( File "/usr/local/airflow/.local/lib/python3.10/site-packages/airflow/utils/session.py", line 72, in wrapper return func(*args, **kwargs) File "/usr/local/airflow/.local/lib/python3.10/site-packages/airflow/models/taskinstance.py", line 1457, in _run_raw_task self._execute_task_with_callbacks(context, test_mode) File "/usr/local/airflow/.local/lib/python3.10/site-packages/airflow/models/taskinstance.py", line 1603, in _execute_task_with_callbacks result = self._execute_task(context, task_orig) File "/usr/local/airflow/.local/lib/python3.10/site-packages/airflow/models/taskinstance.py", line 1664, in _execute_task result = execute_callable(context=context) File "/usr/local/airflow/.local/lib/python3.10/site-packages/airflow/providers/postgres/operators/postgres.py", line 94, in execute
self.hook.run(self.sql, self.autocommit, parameters=self.parameters) File "/usr/local/airflow/.local/lib/python3.10/site-packages/airflow/providers/common/sql/hooks/sql.py", line 288, in run with closing(self.get_conn()) as conn: File "/usr/local/airflow/.local/lib/python3.10/site-packages/airflow/providers/postgres/hooks/postgres.py", line 88, in get_conn
conn = deepcopy(self.connection or self.get_connection(conn_id)) File "/usr/local/airflow/.local/lib/python3.10/site-packages/airflow/hooks/base.py", line 70, in get_connection conn = Connection.get_connection_from_secrets(conn_id) File "/usr/local/airflow/.local/lib/python3.10/site-packages/airflow/models/connection.py", line 432, in get_connection_from_secrets
raise AirflowNotFoundException(f"The conn_id {conn_id} isn't defined") airflow.exceptions.AirflowNotFoundException: The conn_id postgres isn't defined

trlopes1974 commented 9 months ago

I'm having the exact same issue with HttpOperator.

I believe the issue is when retrieving the connection from a backend secret . In my case we are using Hashicorp Vault entreprise and you can see the connection info that is retrieved from vault,. Yet, just after returning the connection info, the AirflowNotFoundException(f"The conn_id {conn_id} isn't defined") is raised

in base_secrets.py, both get_conn_uri and get_conn_value are not implemented...

def get_conn_value(self, conn_id: str) -> str | None: """ Retrieve from Secrets Backend a string value representing the Connection object.

    If the client your secrets backend uses already returns a python dict, you should override
    ``get_connection`` instead.

    :param conn_id: connection id
    """
    raise NotImplementedError

[2024-02-09, 15:47:04 WET] {h_secrets_backend.py:162} DEBUG - response: {'conn_type': 'http', 'extra': {'Authorization': 'Bearer bcGd5AYrSwwDFDZ6DpkMzbmYHRNA1n', 'content-type': 'application/json', 'proxies': {'http': 'http://1.1.1.1:3128', 'https': 'http://1.1.1.1:3128'}, 'verify': False}, 'host': 'ansible-tower.host', 'port': '443', 'schema': 'https'} [2024-02-09, 15:47:04 WET] {taskinstance.py:2699} ERROR - Task failed with exception Traceback (most recent call last): File "/opt/tkapp/env_airflow/lib64/python3.9/site-packages/airflow/models/taskinstance.py", line 433, in _execute_task result = execute_callable(context=context, **execute_callable_kwargs) File "/opt/tkapp/env_airflow/lib64/python3.9/site-packages/airflow/providers/http/operators/http.py", line 166, in execute return self.execute_sync(context=context) File "/opt/tkapp/env_airflow/lib64/python3.9/site-packages/airflow/providers/http/operators/http.py", line 170, in execute_sync response = self.hook.run(self.endpoint, self.data, self.headers, self.extra_options) File "/opt/tkapp/env_airflow/lib64/python3.9/site-packages/airflow/providers/http/operators/http.py", line 148, in hook conn = BaseHook.get_connection(conn_id) File "/opt/tkapp/env_airflow/lib64/python3.9/site-packages/airflow/hooks/base.py", line 82, in get_connection conn = Connection.get_connection_from_secrets(conn_id) File "/opt/tkapp/env_airflow/lib64/python3.9/site-packages/airflow/models/connection.py", line 479, in get_connection_from_secrets raise AirflowNotFoundException(f"The conn_id {conn_id} isn't defined")