Catch arbitrary exception from run_job to prevent zombie scheduler

[xuganyu96]

This PR attempts to address issue #32706

When run_job raises unhandled exception, the scheduler process will become zombie. Placing run_job inside a try-catch block will allow serve_logs and health_check to properly terminate the sub-processes.

[uranusjr]

This needs a test case.

[xuganyu96]

@uranusjr

Thank you for the feedback. I have added a test case that mocked airflow.jobs.job.run_job to simulate the situation where run_job throws some Exception, then test that the exception is caught and sub_proc.terminate is still called.

[pankajkoti]

Suggested change

	log = logging.getLogger(__name__)
	logger = logging.getLogger(__name__)

Maybe this? Usually I see it's called logger

[xuganyu96]

Could you please point to where the logger is named logger? The modules I've touched so far all named logger as log example

[pankajkoti]

yes. The example you shared is using self.log and not log. I believe this comes from extending the LoggingMixin class which has the log property.

What I meant is in general when using logging.getLogger(__name__), python modules call it as logger across the projects I have seen in so far. One example in our project I could find is here

airflow/airflow/utils/log/file_task_handler.py

Line 46 in 73b90c4

logger = logging.getLogger(__name__)

But I think I have another suggestion here. Since this is a CLI command I guess we could maybe use the AirflowConsole here to print the message? e.g. https://github.com/apache/airflow/blob/main/airflow/cli/commands/connection_command.py#L360

[xuganyu96]

Naming variables one way or another seems ultimately trivial, so I am perfectly okay with changing the variable name to logger.

On the other hand, I am not sure about using AirflowConsole for logging purposes. The code changes are located under the cli module, but they are ultimately meant to catch exceptions that came up within the inner logic of the scheduler, so I think it should be treated like a normal logger.

Another situation where I think using a logger is more appropriate is when the scheduler is run as a daemon (with airflow scheduler -D), in which case using the AirflowConsole will not log the exception.

[uranusjr]

Naming on this is unfortunately pretty inconsistent in Airflow. Personally I much prefer logger but 🤷

[pankajkoti]

Should we also log what the exception message was?

[xuganyu96]

https://docs.python.org/3/library/logging.html#logging.exception
Calling log.exception will automatically include the exception information so there is no need to include it separately.

[pankajkoti]

yes, sorry missed reading the logging level.