JSON logging support

[pirackr]

Description
Make logging in Airlfow more extensible or support for json

Motivation/Usecase
Currenty, we're working on using json formatter for our logging (for several reasons: k8s deployment, better search/filter).

When we're moving away, we found some issue, one of them is that when the task runner try to run a task, it would try to capture the output of the task and redirect to the log

# capture the log
log_reader = threading.Thread(
            target=self._read_task_logs,
            args=(proc.stdout,),
        )
log_reader.daemon = True
log_reader.start()

and

# and log it out
self.log.info('Job %s: Subtask %s %s',
                      self._task_instance.job_id, self._task_instance.task_id,
                      line.rstrip('\n'))

Anyway, when we do log it out, we're trying to use string interpolation, this is not extensible, especially when we do json, it could produce something like this:

{"asctime": 123, "message": "Job 1: Subtask 1 {\"asctime\": 123, \"message\": \"Task 2\"}}

So, do we have a way to handle this? I have a few ideas in mind:

• Override the task runner: I know we have the ability to specific a task runner, but is there a usual way to add a new task runner? (rather than patching airflow?)
• We change the log and use extra (this is supported by python logging) instead:

self.log.info(line,  extra={"job_id": self._task_instance.job_id, "task_id": self._task_instance.task_id}_

Can you kindly let me know what you're thinking about this?

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template!