Issue 1:
Description

Add the option for the celery worker to create a new virtual env, install some packages, and run airflow run command inside it (based on executor_config params).

Really nice to have - have reusable virtual env that can be shared between tasks with the same param (based on user configuration).

Use case / motivation

Once getting to a point when you want to create cluster for different types of python tasks and you've multiple teams working on the same cluster, you need to start splitting into different python packages the business login code to allow better versioning control and avoid the need of restarting the workers when deploying new util code.

I think it would be amazing if we can allow creating new virtual envs as part of Airflow and control the package versions.

I know that PythonVirtualenvOperator exists, but:

Creating env related thing feels like an executor job to me, the coder should not use specific operators for it.

The big downside to it is that if I want to use ShortCircuitOperator or BranchPythonOperator or any kind of new python based operator, I've to create a new operator that will inherit from PythonVirtualenvOperator and duplicate the desired functionality.

Are you willing to submit a PR?

Yes, would love to.

Related Issues

Not that I can find.

Issue 2:
Apache Airflow version

2.3.0 (latest released)

What happened

After upgrading Airflow from 2.2.4 to 2.3.0, the Airflow webserver encounters the following errors:

[2022-05-10 05:04:43,530] {manager.py:543} INFO - Removed Permission View: can_create on Users
[2022-05-10 05:04:43,639] {manager.py:585} INFO - Removed Permission menu access on Permissions to role Admin
[2022-05-10 05:04:43,645] {manager.py:585} INFO - Removed Permission menu access on Permissions to role Admin
[2022-05-10 05:04:43,650] {manager.py:543} INFO - Removed Permission View: menu_access on Permissions
[2022-05-10 05:04:43,656] {manager.py:543} INFO - Removed Permission View: menu_access on Permissions
[2022-05-10 05:04:43,718] {manager.py:508} INFO - Created Permission View: menu access on Permissions
[2022-05-10 05:04:43,727] {manager.py:568} INFO - Added Permission menu access on Permissions to role Admin
[2022-05-10 05:04:43,728] {manager.py:570} ERROR - Add Permission to Role Error: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "ab_permission_view_role_permission_view_id_role_id_key"
DETAIL: Key (permission_view_id, role_id)=(410, 1) already exists.

[SQL: INSERT INTO ab_permission_view_role (id, permission_view_id, role_id) VALUES (nextval('ab_permission_view_role_id_seq'), %(permission_view_id)s, %(role_id)s) RETURNING ab_permission_view_role.id]
[parameters: {'permission_view_id': 410, 'role_id': 1}]
(Background on this error at: http://sqlalche.me/e/14/gkpj)
[2022-05-10 05:04:45,666] {manager.py:508} INFO - Created Permission View: can create on Users
[2022-05-10 05:04:45,666] {manager.py:511} ERROR - Creation of Permission View Error: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "ab_permission_view_permission_id_view_menu_id_key"
DETAIL: Key (permission_id, view_menu_id)=(5, 51) already exists.

[SQL: INSERT INTO ab_permission_view (id, permission_id, view_menu_id) VALUES (nextval('ab_permission_view_id_seq'), %(permission_id)s, %(view_menu_id)s) RETURNING ab_permission_view.id]
[parameters: {'permission_id': 5, 'view_menu_id': 51}]
(Background on this error at: http://sqlalche.me/e/14/gkpj)
[2022-05-10 05:04:45 -0400] [650935] [ERROR] Exception in worker process
Traceback (most recent call last):
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1705, in _execute_context
self.dialect.do_execute(
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 716, in do_execute
cursor.execute(statement, parameters)
psycopg2.errors.UniqueViolation: duplicate key value violates unique constraint "ab_permission_view_role_permission_view_id_role_id_key"
DETAIL: Key (permission_view_id, role_id)=(412, 1) already exists.

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/gunicorn/arbiter.py", line 589, in spawn_worker
worker.init_process()
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/gunicorn/workers/base.py", line 134, in init_process
self.load_wsgi()
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/gunicorn/workers/base.py", line 146, in load_wsgi
self.wsgi = self.app.wsgi()
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/gunicorn/app/base.py", line 67, in wsgi
self.callable = self.load()
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/gunicorn/app/wsgiapp.py", line 58, in load
return self.load_wsgiapp()
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/gunicorn/app/wsgiapp.py", line 48, in load_wsgiapp
return util.import_app(self.app_uri)
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/gunicorn/util.py", line 412, in import_app
app = app(*args, **kwargs)
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/airflow/www/app.py", line 158, in cached_app
app = create_app(config=config, testing=testing)
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/airflow/www/app.py", line 146, in create_app
sync_appbuilder_roles(flask_app)
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/airflow/www/app.py", line 68, in sync_appbuilder_roles
flask_app.appbuilder.sm.sync_roles()
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/airflow/www/security.py", line 580, in sync_roles
self.update_admin_permission()
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/airflow/www/security.py", line 562, in update_admin_permission
self.get_session.commit()
File "<string>", line 2, in commit
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 1423, in commit
self._transaction.commit(_to_root=self.future)
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 829, in commit
self._prepare_impl()
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 808, in _prepare_impl
self.session.flush()
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 3255, in flush
self._flush(objects)
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 3395, in _flush
transaction.rollback(_capture_exception=True)
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/util/langhelpers.py", line 70, in __exit__
compat.raise_(
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/util/compat.py", line 211, in raise_
raise exception
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 3355, in _flush
flush_context.execute()
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/orm/unitofwork.py", line 453, in execute
rec.execute(self)
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/orm/unitofwork.py", line 576, in execute
self.dependency_processor.process_saves(uow, states)
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/orm/dependency.py", line 1182, in process_saves
self._run_crud(
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/orm/dependency.py", line 1245, in _run_crud
connection.execute(statement, secondary_insert)
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1200, in execute
return meth(self, multiparams, params, _EMPTY_EXECUTION_OPTS)
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/sql/elements.py", line 313, in _execute_on_connection
return connection._execute_clauseelement(
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1389, in _execute_clauseelement
ret = self._execute_context(
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1748, in _execute_context
self._handle_dbapi_exception(
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1929, in _handle_dbapi_exception
util.raise_(
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/util/compat.py", line 211, in raise_
raise exception
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1705, in _execute_context
self.dialect.do_execute(
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 716, in do_execute
cursor.execute(statement, parameters)
sqlalchemy.exc.IntegrityError: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "ab_permission_view_role_permission_view_id_role_id_key"
DETAIL: Key (permission_view_id, role_id)=(412, 1) already exists.

[SQL: INSERT INTO ab_permission_view_role (id, permission_view_id, role_id) VALUES (nextval('ab_permission_view_role_id_seq'), %(permission_view_id)s, %(role_id)s) RETURNING ab_permission_view_role.id]
[parameters: {'permission_view_id': 412, 'role_id': 1}]
(Background on this error at: http://sqlalche.me/e/14/gkpj)
[2022-05-10 05:04:45 -0400] [650935] [INFO] Worker exiting (pid: 650935)
[2022-05-10 05:04:45 -0400] [650934] [ERROR] Exception in worker process
Traceback (most recent call last):
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1705, in _execute_context
self.dialect.do_execute(
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 716, in do_execute
cursor.execute(statement, parameters)
psycopg2.errors.UniqueViolation: duplicate key value violates unique constraint "ab_permission_view_role_permission_view_id_role_id_key"
DETAIL: Key (permission_view_id, role_id)=(412, 1) already exists.

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/gunicorn/arbiter.py", line 589, in spawn_worker
worker.init_process()
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/gunicorn/workers/base.py", line 134, in init_process
self.load_wsgi()
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/gunicorn/workers/base.py", line 146, in load_wsgi
self.wsgi = self.app.wsgi()
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/gunicorn/app/base.py", line 67, in wsgi
self.callable = self.load()
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/gunicorn/app/wsgiapp.py", line 58, in load
return self.load_wsgiapp()
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/gunicorn/app/wsgiapp.py", line 48, in load_wsgiapp
return util.import_app(self.app_uri)
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/gunicorn/util.py", line 412, in import_app
app = app(*args, **kwargs)
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/airflow/www/app.py", line 158, in cached_app
app = create_app(config=config, testing=testing)
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/airflow/www/app.py", line 146, in create_app
sync_appbuilder_roles(flask_app)
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/airflow/www/app.py", line 68, in sync_appbuilder_roles
flask_app.appbuilder.sm.sync_roles()
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/airflow/www/security.py", line 580, in sync_roles
self.update_admin_permission()
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/airflow/www/security.py", line 562, in update_admin_permission
self.get_session.commit()
File "<string>", line 2, in commit
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 1423, in commit
self._transaction.commit(_to_root=self.future)
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 829, in commit
self._prepare_impl()
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 808, in _prepare_impl
self.session.flush()
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 3255, in flush
self._flush(objects)
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 3395, in _flush
transaction.rollback(_capture_exception=True)
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/util/langhelpers.py", line 70, in __exit__
compat.raise_(
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/util/compat.py", line 211, in raise_
raise exception
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 3355, in _flush
flush_context.execute()
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/orm/unitofwork.py", line 453, in execute
rec.execute(self)
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/orm/unitofwork.py", line 576, in execute
self.dependency_processor.process_saves(uow, states)
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/orm/dependency.py", line 1182, in process_saves
self._run_crud(
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/orm/dependency.py", line 1245, in _run_crud
connection.execute(statement, secondary_insert)
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1200, in execute
return meth(self, multiparams, params, _EMPTY_EXECUTION_OPTS)
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/sql/elements.py", line 313, in _execute_on_connection
return connection._execute_clauseelement(
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1389, in _execute_clauseelement
ret = self._execute_context(
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1748, in _execute_context
self._handle_dbapi_exception(
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1929, in _handle_dbapi_exception
util.raise_(
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/util/compat.py", line 211, in raise_
raise exception
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1705, in _execute_context
self.dialect.do_execute(
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 716, in do_execute
cursor.execute(statement, parameters)
sqlalchemy.exc.IntegrityError: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "ab_permission_view_role_permission_view_id_role_id_key"
DETAIL: Key (permission_view_id, role_id)=(412, 1) already exists.

[SQL: INSERT INTO ab_permission_view_role (id, permission_view_id, role_id) VALUES (nextval('ab_permission_view_role_id_seq'), %(permission_view_id)s, %(role_id)s) RETURNING ab_permission_view_role.id]
[parameters: {'permission_view_id': 412, 'role_id': 1}]
(Background on this error at: http://sqlalche.me/e/14/gkpj)
[2022-05-10 05:04:45 -0400] [650934] [INFO] Worker exiting (pid: 650934)
[2022-05-10 05:04:46 -0400] [650932] [INFO] Worker exiting (pid: 650932)
[2022-05-10 05:04:46 -0400] [650933] [INFO] Worker exiting (pid: 650933)
Traceback (most recent call last):
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/gunicorn/arbiter.py", line 209, in run
self.sleep()
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/gunicorn/arbiter.py", line 357, in sleep
ready = select.select([self.PIPE[0]], [], [], 1.0)
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/gunicorn/arbiter.py", line 242, in handle_chld
self.reap_workers()
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/gunicorn/arbiter.py", line 525, in reap_workers
raise HaltServer(reason, self.WORKER_BOOT_ERROR)
gunicorn.errors.HaltServer: <HaltServer 'Worker failed to boot.' 3>

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
File "/usr/lib/python3.8/runpy.py", line 194, in _run_module_as_main
return _run_code(code, main_globals, None,
File "/usr/lib/python3.8/runpy.py", line 87, in _run_code
exec(code, run_globals)
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/gunicorn/__main__.py", line 7, in <module>
run()
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/gunicorn/app/wsgiapp.py", line 67, in run
WSGIApplication("%(prog)s [OPTIONS] [APP_MODULE]").run()
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/gunicorn/app/base.py", line 231, in run
super().run()
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/gunicorn/app/base.py", line 72, in run
Arbiter(self).run()
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/gunicorn/arbiter.py", line 229, in run
self.halt(reason=inst.reason, exit_status=inst.exit_status)
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/gunicorn/arbiter.py", line 342, in halt
self.stop()
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/gunicorn/arbiter.py", line 393, in stop
time.sleep(0.1)
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/gunicorn/arbiter.py", line 242, in handle_chld
self.reap_workers()
File "/home/etlprod/venv/airflow-3141031@1652080547/lib/python3.8/site-packages/gunicorn/arbiter.py", line 525, in reap_workers
raise HaltServer(reason, self.WORKER_BOOT_ERROR)
gunicorn.errors.HaltServer: <HaltServer 'Worker failed to boot.' 3>

What you think should happen instead

Obviously, the Airflow webserver is trying to fix some permissions issues, every time it starts.

Unfortunately, when this process fails, it leaves the webserver hanging without any active gunicorn workers, thus preventing it from serving the UI

How to reproduce

upgrade from 2.2.4 to 2.3.0

Operating System

Ubuntu 20.04.3 LTS

Versions of Apache Airflow Providers

apache-airflow-providers-amazon @ file:///home/airflow/deploy/wheelfreeze/wheels/apache_airflow_providers_amazon-3.3.0-py3-none-any.whl
apache-airflow-providers-ftp @ file:///home/airflow/deploy/wheelfreeze/wheels/apache_airflow_providers_ftp-2.1.2-py3-none-any.whl
apache-airflow-providers-http @ file:///home/airflow/deploy/wheelfreeze/wheels/apache_airflow_providers_http-2.1.2-py3-none-any.whl
apache-airflow-providers-imap @ file:///home/airflow/deploy/wheelfreeze/wheels/apache_airflow_providers_imap-2.2.3-py3-none-any.whl
apache-airflow-providers-mongo @ file:///home/airflow/deploy/wheelfreeze/wheels/apache_airflow_providers_mongo-2.3.3-py3-none-any.whl
apache-airflow-providers-mysql @ file:///home/airflow/deploy/wheelfreeze/wheels/apache_airflow_providers_mysql-2.2.3-py3-none-any.whl
apache-airflow-providers-pagerduty @ file:///home/airflow/deploy/wheelfreeze/wheels/apache_airflow_providers_pagerduty-2.1.3-py3-none-any.whl
apache-airflow-providers-postgres @ file:///home/airflow/deploy/wheelfreeze/wheels/apache_airflow_providers_postgres-4.1.0-py3-none-any.whl
apache-airflow-providers-sendgrid @ file:///home/airflow/deploy/wheelfreeze/wheels/apache_airflow_providers_sendgrid-2.0.4-py3-none-any.whl
apache-airflow-providers-slack @ file:///home/airflow/deploy/wheelfreeze/wheels/apache_airflow_providers_slack-4.2.3-py3-none-any.whl
apache-airflow-providers-sqlite @ file:///home/airflow/deploy/wheelfreeze/wheels/apache_airflow_providers_sqlite-2.1.3-py3-none-any.whl
apache-airflow-providers-ssh @ file:///home/airflow/deploy/wheelfreeze/wheels/apache_airflow_providers_ssh-2.4.3-py3-none-any.whl
apache-airflow-providers-vertica @ file:///home/airflow/deploy/wheelfreeze/wheels/apache_airflow_providers_vertica-2.1.3-py3-none-any.whl

Deployment

Virtualenv installation

Deployment details

python 3.8.10

Anything else

No response

Are you willing to submit PR?

Yes I am willing to submit a PR!

Code of Conduct

I agree to follow this project's Code of Conduct