Issue 1:
Apache Airflow version: 2.0.1

Kubernetes version (if you are using kubernetes) (use kubectl version):

Client Version: version.Info{Major:"1", Minor:"18", GitVersion:"v1.18.1", GitCommit:"7879fc12a63337efff607952a323df90cdc7a335", GitTreeState:"clean", BuildDate:"2020-04-10T21:53:58Z", GoVersion:"go1.14.2", Compiler:"gc", Platform:"darwin/amd64"}

Server Version: version.Info{Major:"1", Minor:"16+", GitVersion:"v1.16.15-gke.7800", GitCommit:"cef3156c566a1d1a4b23ee360a760f45bfbaaac1", GitTreeState:"clean", BuildDate:"2020-12-14T09:12:37Z", GoVersion:"go1.13.15b4", Compiler:"gc", Platform:"linux/amd64"}

Environment:

Cloud provider or hardware configuration: GKE

OS (e.g. from /etc/os-release):

Kernel (e.g. uname -a):

Install tools:

Others: We use scheduler HA with 2 instances.

What happened:

We are running Airflow 2.0.1 using KubernetesExecutor and PostgreSQL 9.6.2.

Task is up_for_retry after its worker pod fails to start

It gets stuck in queued state

It runs only after a scheduler restart.

What you expected to happen:

Task gets rescheduled and runs successfully.

How to reproduce it:

Anything else we need to know:

Logs below.

Its upstream task succeeds and schedules it via fast follow.

[2021-03-01 17:08:51,306] {taskinstance.py:1166} INFO - Marking task as SUCCESS. dag_id=datalake_dag_id, task_id=processidlogs, execution_date=20210301T163000, start_date=20210301T170546, end_date=20210301T170851

[2021-03-01 17:08:51,339] {taskinstance.py:1220} INFO - 1 downstream tasks scheduled from follow-on schedule check

[2021-03-01 17:08:51,357] {local_task_job.py:146} INFO - Task exited with return code 0

Task is attempted to be run.

[2021-03-01 17:08:52,229] {scheduler_job.py:1105} INFO - Sending TaskInstanceKey(dag_id='datalake_dag_id', task_id='delta_id_logs', execution_date=datetime.datetime(2021, 3, 1, 16, 30, tzinfo=Timezone('UTC')), try_number=1) to executor with priority 1 and queue default

[2021-03-01 17:08:52,308] {kubernetes_executor.py:306} DEBUG - Kubernetes running for command ['airflow', 'tasks', 'run', 'datalake_dag_id', 'delta_id_logs', '2021-03-01T16:30:00+00:00', '--local', '--pool', 'default_pool', '--subdir', '/opt/airflow/dags/data_lake/some_tasks/some_tasks.py']

[2021-03-01 17:08:52,332] {scheduler_job.py:1206} INFO - Executor reports execution of datalake_dag_id.delta_id_logs execution_date=2021-03-01 16:30:00+00:00 exited with status queued for try_number 1

Pod fails to start.

[2021-03-01 17:12:17,319] {kubernetes_executor.py:197} INFO - Event: Failed to start pod datalakedagiddeltaidlogs.5fa98ae3856f4cb4b6c8810ac13e5c6a, will reschedule

It is put as up_for_reschedule.

[2021-03-01 17:12:23,912] {kubernetes_executor.py:343} DEBUG - Processing task ('datalakedagiddeltaidlogs.5fa98ae3856f4cb4b6c8810ac13e5c6a', 'prod', 'up_for_reschedule', {'dag_id': 'datalake_dag_id', 'task_id': 'delta_id_logs', 'execution_date': '2021-03-01T16:30:00+00:00', 'try_number': '1'}, '1172208829')

[2021-03-01 17:12:23,930] {kubernetes_executor.py:528} INFO - Changing state of (TaskInstanceKey(dag_id='datalake_dag_id', task_id='delta_id_logs', execution_date=datetime.datetime(2021, 3, 1, 16, 30, tzinfo=tzlocal()), try_number=1), 'up_for_reschedule', 'datalakedagiddeltaidlogs.5fa98ae3856f4cb4b6c8810ac13e5c6a', 'prod', '1172208829') to up_for_reschedule

[2021-03-01 17:12:23,941] {scheduler_job.py:1206} INFO - Executor reports execution of datalake_dag_id.delta_id_logs execution_date=2021-03-01 16:30:00+00:00 exited with status up_for_reschedule for try_number 1

A few minutes later, another scheduler finds it in queued state.

[2021-03-01 17:15:39,177] {taskinstance.py:851} DEBUG - Dependencies all met for <TaskInstance: datalake_dag_id.delta_id_logs 2021-03-01 16:30:00+00:00 [queued]>

[2021-03-01 17:15:40,477] {taskinstance.py:866} DEBUG - <TaskInstance: datalake_dag_id.delta_id_logs 2021-03-01 16:30:00+00:00 [queued]> dependency 'Not In Retry Period' PASSED: True, The context specified that being in a retry period was permitted.

[2021-03-01 17:15:40,478] {taskinstance.py:866} DEBUG - <TaskInstance: datalake_dag_id.delta_id_logs 2021-03-01 16:30:00+00:00 [queued]> dependency 'Previous Dagrun State' PASSED: True, The task did not have depends_on_past set.

It stays in that state for another hour and a half until the scheduler is restarted.

Finally, it is rescheduled.

[2021-03-01 18:58:10,475] {kubernetes_executor.py:463} INFO - TaskInstance: <TaskInstance: datalake_dag_id.delta_id_logs 2021-03-01 16:30:00+00:00 [queued]> found in queued state but was not launched, rescheduling

Other tasks run fine while that one is stuck in queued state. We have a cron job to restart the scheduler as a hack to recover from when such cases happen, but we would like to avoid it as much as possible.

We run 60 DAGs with 50-100 tasks each every 30 minutes. We have been seeing this issue at least once daily since we upgraded to Airflow 2.0.1.

I understand there are some open issues about the scheduler or tasks getting stuck. But I could not tell if this is related, since hundreds of other tasks run as expected. Apologies if this turns out to be a duplicate of an existing issue. Thank you.

Issue 2:
Apache Airflow version: 2.1.0

What happened:

When I tried to call logging.exception(), the task instances fails.

def func():
import logging
logging.exception("some error message")

with dag:
p = PythonOperator(
task_id='test',
python_callable=func,
)

task failure stack strace

Traceback (most recent call last):
File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1138, in _run_raw_task
self._prepare_and_execute_task_with_callbacks(context, task)
File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1311, in _prepare_and_execute_task_with_callbacks
result = self._execute_task(context, task_copy)
File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1341, in _execute_task
result = task_copy.execute(context=context)
File "/usr/local/lib/python3.7/site-packages/airflow/operators/python.py", line 117, in execute
return_value = self.execute_callable()
File "/usr/local/lib/python3.7/site-packages/airflow/operators/python.py", line 128, in execute_callable
return self.python_callable(*self.op_args, **self.op_kwargs)
File "/usr/local/airflow/src/__main__.py", line 12, in main
src.file_get_sftp(sftp_site_folder,dest_dir)
File "/usr/local/airflow/src/controller/source.py", line 102, in file_get_sftp
logging.exception("Destination directory doesn't exist")
File "/usr/local/lib/python3.7/logging/__init__.py", line 1967, in exception
error(msg, *args, exc_info=exc_info, **kwargs)
File "/usr/local/lib/python3.7/logging/__init__.py", line 1959, in error
root.error(msg, *args, **kwargs)
File "/usr/local/lib/python3.7/logging/__init__.py", line 1407, in error
self._log(ERROR, msg, args, **kwargs)
File "/usr/local/lib/python3.7/logging/__init__.py", line 1514, in _log
self.handle(record)
File "/usr/local/lib/python3.7/logging/__init__.py", line 1523, in handle
if (not self.disabled) and self.filter(record):
File "/usr/local/lib/python3.7/logging/__init__.py", line 751, in filter
result = f.filter(record)
File "/usr/local/lib/python3.7/site-packages/airflow/utils/log/secrets_masker.py", line 161, in filter
exc.args = (self.redact(v) for v in exc.args)
AttributeError: 'NoneType' object has no attribute 'args'

What you expected to happen:

I expect an exception message to be printed out.

How to reproduce it:

Create a simple DAG that calls logging.exception()

Execute the function

Anything else we need to know:

This issue was created for the PR (#16047).