"Failed to load task run error" occur in "on_failure_callback" after upgrading from 2.0.0 to 2.1.2

[kaojunsong]

Apache Airflow version: 2.1.2

Kubernetes version (if you are using kubernetes) (use kubectl version): "v1.19.6"

Environment:

• Cloud provider or hardware configuration:
• OS (e.g. from /etc/os-release):

NAME="CentOS Linux"
VERSION="7 (Core)"
ID="centos"
ID_LIKE="rhel fedora"
VERSION_ID="7"
PRETTY_NAME="CentOS Linux 7 (Core)"
ANSI_COLOR="0;31"
CPE_NAME="cpe:/o:centos:centos:7"
HOME_URL="https://www.centos.org/"
BUG_REPORT_URL="https://bugs.centos.org/"

CENTOS_MANTISBT_PROJECT="CentOS-7"
CENTOS_MANTISBT_PROJECT_VERSION="7"
REDHAT_SUPPORT_PRODUCT="centos"

• Kernel (e.g. uname -a):
• Install tools:
• Others:

What happened: "Failed to load task run error" occur in "on_failure_callback" after task failed

What you expected to happen: The actually exception should be caught

How to reproduce it:

def validate_parameters(**kwargs):
    conf = kwargs.get('dag_run').conf
    kwargs['ti'].xcom_push(key='conf', value=conf)
    job_identifier = conf.get('job_identifier')
    s3_connection_id = conf.get('s3_connection_id')
    if not job_identifier:
        raise ParameterValidationException(reason='job_identifier should not be empty')
    if not s3_connection_id:
        raise ParameterValidationException(reason='s3_connection_id should not be empty')
  

def handle_failure(context):
    exception = context.get('exception')
    // the ParameterValidationException should be caught here, but I got "Failed to load task run error"
    // other business
    

DEFAULT_ARGS = {
    'owner': 'boss_admin',
    'start_date': datetime(2021, 1, 2),
    "retries": 0,
    'on_failure_callback': handle_failure
}

Anything else we need to know:

[eladkal]

Can you please add a minimal working reproduce example and also format your code?
Also please add the traceback

[kaojunsong]

Hi @eladkal, thanks a lot for your reply.
The minimal working reproduce example is in https://github.com/kaojunsong/airflow-test/blob/main/dags/test.py

[potiuk]

I run your example using local executor and it all works fine (see below).

The problem you most likely experienced was problem with retrieving task logs. You need to make sure you have
AIRFLOW__WEBSERVER__SECRET_KEY the same for workers and webserver - then your logs will be retrieved by the webserver https://airflow.apache.org/docs/apache-airflow/stable/configurations-ref.html#secret-key

I am closing it now - please re-open if the issue is not fixed (but I am quite sure it will be).

*** Reading local file: /root/airflow/logs/monitor/validate_parameters/2021-07-25T17:02:55.522373+00:00/1.log
[2021-07-25, 17:02:57 UTC] {taskinstance.py:913} INFO - Dependencies all met for <TaskInstance: monitor.validate_parameters 2021-07-25T17:02:55.522373+00:00 [queued]>
[2021-07-25, 17:02:57 UTC] {taskinstance.py:913} INFO - Dependencies all met for <TaskInstance: monitor.validate_parameters 2021-07-25T17:02:55.522373+00:00 [queued]>
[2021-07-25, 17:02:57 UTC] {taskinstance.py:1104} INFO - 
--------------------------------------------------------------------------------
[2021-07-25, 17:02:57 UTC] {taskinstance.py:1105} INFO - Starting attempt 1 of 1
[2021-07-25, 17:02:57 UTC] {taskinstance.py:1106} INFO - 
--------------------------------------------------------------------------------
[2021-07-25, 17:02:57 UTC] {taskinstance.py:1124} INFO - Executing <Task(PythonOperator): validate_parameters> on 2021-07-25T17:02:55.522373+00:00
[2021-07-25, 17:02:57 UTC] {standard_task_runner.py:52} INFO - Started process 1353 to run task
[2021-07-25, 17:02:57 UTC] {standard_task_runner.py:76} INFO - Running: ['airflow', 'tasks', 'run', 'monitor', 'validate_parameters', '2021-07-25T17:02:55.522373+00:00', '--job-id', '2', '--pool', 'default_pool', '--raw', '--subdir', 'DAGS_FOLDER/test_dag_failure.py', '--cfg-path', '/tmp/tmpook1fgsn', '--error-file', '/tmp/tmp45ahny1g']
[2021-07-25, 17:02:57 UTC] {standard_task_runner.py:77} INFO - Job 2: Subtask validate_parameters
[2021-07-25, 17:02:57 UTC] {logging_mixin.py:109} INFO - Running <TaskInstance: monitor.validate_parameters 2021-07-25T17:02:55.522373+00:00 [running]> on host 19f13fbfde00
[2021-07-25, 17:02:57 UTC] {taskinstance.py:1257} INFO - Exporting the following env vars:
AIRFLOW_CTX_DAG_OWNER=boss_admin
AIRFLOW_CTX_DAG_ID=monitor
AIRFLOW_CTX_TASK_ID=validate_parameters
AIRFLOW_CTX_EXECUTION_DATE=2021-07-25T17:02:55.522373+00:00
AIRFLOW_CTX_DAG_RUN_ID=manual__2021-07-25T17:02:55.522373+00:00
[2021-07-25, 17:02:57 UTC] {taskinstance.py:1458} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/opt/airflow/airflow/models/taskinstance.py", line 1185, in _run_raw_task
    self._prepare_and_execute_task_with_callbacks(context, task)
  File "/opt/airflow/airflow/models/taskinstance.py", line 1288, in _prepare_and_execute_task_with_callbacks
    result = self._execute_task(context, task_copy)
  File "/opt/airflow/airflow/models/taskinstance.py", line 1318, in _execute_task
    result = task_copy.execute(context=context)
  File "/opt/airflow/airflow/operators/python.py", line 150, in execute
    return_value = self.execute_callable()
  File "/opt/airflow/airflow/operators/python.py", line 161, in execute_callable
    return self.python_callable(*self.op_args, **self.op_kwargs)
  File "/files/dags/test_dag_failure.py", line 24, in validate_parameters
    raise ParameterValidationException(reason='job_identifier should not be empty')
unusual_prefix_f60d67b5f19a88c1f508e2153f05719e51032699_test_dag_failure.ParameterValidationException
[2021-07-25, 17:02:57 UTC] {taskinstance.py:1132} INFO - Marking task as FAILED. dag_id=monitor, task_id=validate_parameters, execution_date=20210725T170255, start_date=20210725T170257, end_date=20210725T170257
[2021-07-25, 17:02:57 UTC] {local_task_job.py:152} INFO - Task exited with return code 1
[2021-07-25, 17:02:57 UTC] {test_dag_failure.py:31} INFO - Failed to load task run error
[2021-07-25, 17:02:57 UTC] {logging_mixin.py:109} INFO - NoneType: None
[2021-07-25, 17:02:57 UTC] {local_task_job.py:255} INFO - 0 downstream tasks scheduled from follow-on schedule check

[kaojunsong]

@potiuk Thanks a lot for your reply.
I'm using Celery Executor. And the webserver and worker are using the same airflow.cfg: https://github.com/kaojunsong/airflow-test/blob/main/config/airflow.cfg, they are using the same secret_key, am I right? If not, how can I configure the secret_key for the workers?

And the deployment of kubenetes is in: https://github.com/kaojunsong/airflow-test/tree/main/kubenetes

[kaojunsong]

@potiuk Hi, I still can't solve this issue, could you please kindly have a look?

[potiuk]

I suggest you to switch to Airflow's official Helm rather than running your own deployment. It is proven and works and it's the easiest way to get Airflow Up-and-running on Kubernetes, I think no-one in the community has time to debug hundreds of different manual deployments like yours, but the Official Helm Chart from the community is very versatile, supported and it has a lot of problems solved that you might not realise by running your own deployment. I think this is the best route you can take.

See https://airflow.apache.org/docs/helm-chart/stable/index.html

[Calder-Ty]

Bottom Line Up Front: I had a similar issue, and found that the cause was a third party packages code that did not implement exceptions that could be loaded from a picklefile. This is the most likely cause of the problem.

For me, switching to Airflows official helm wasn't an option (not running Kubernetes). I did some digging around and found that this message is spawned when trying to load an original exception from a pickle file.

airflow/airflow/models/taskinstance.py

Lines 136 to 147 in 9922287

	def load_error_file(fd: IO[bytes]) -> Optional[Union[str, Exception]]:
	"""Load and return error from error file"""
	if fd.closed:
	return None
	fd.seek(0, os.SEEK_SET)
	data = fd.read()
	if not data:
	return None
	try:
	return pickle.loads(data)
	except Exception:
	return "Failed to load task run error"

Unfortunately the causing error message is not logged. I applied this patch to my instance of airflow to see what error was causing the pickle.loads to fail.

@@ -115,7 +115,10 @@
         return None
     try:
         return pickle.loads(data)
-    except Exception:  # pylint: disable=broad-except
+    except Exception as e:  # pylint: disable=broad-except
+        log.exception("Failed to load the exception! Oh NO!!! Here is why")
+        log.exception(e.msg)
+        log.exception(e)
         return "Failed to load task run error"

I found that the error was being caused because the third party tooling I was using had an error that could be pickled, but then not loaded back. This is a common issue with custom errors in python. I'm guessing that something similar is happening in your case.

[kaojunsong]

@Calder-Ty Thanks a lot, I have fixed my issue by adding
def __reduce__(self): return ParameterValidationException, (self.reason, self.component)
to my custom Exception, as the pickle library needs to unpicle data from log files.
https://www.synopsys.com/blogs/software-security/python-pickling/#:~:text=Whenever%20an%20object%20is%20pickled,reconstruct%20this%20object%20when%20unpickling.