[AIRFLOW-1385] Create abstraction for Airflow task logging

[allisonwang]

Dear Airflow maintainers,

Please accept this PR. I understand that it will not be reviewed until I have checked off all the steps below!

JIRA

• My PR addresses the following Airflow 1385 issues and references them in the PR title.

Description

• Here are some details about my PR, including screenshots of any UI changes:

This PR adds abilities to provide customized implementations of airflow task logging. It creates an abstraction for setting up, cleaning up and get task instance logs.

Tests

• My PR adds the following unit tests OR does not need testing for this extremely good reason:
  This change is primarily a refactor of logging logic. It is tested locally with custom logging implementations.

Commits

• My commits all reference JIRA issues in their subject lines, and I have squashed multiple commits if they address the same issue. In addition, my commits follow the guidelines from "How to write a good git commit message":
  1. Subject is separated from body by a blank line
  2. Subject is limited to 50 characters
  3. Subject does not end with a period
  4. Subject uses the imperative mood ("add", not "adding")
  5. Body wraps at 72 characters
  6. Body explains "what" and "why", not "how"

@aoen @saguziel

[codecov-io]

Codecov Report

Merging #2422 into master will decrease coverage by <.01%.
The diff coverage is 53.6%.

@@            Coverage Diff             @@
##           master    #2422      +/-   ##
==========================================
- Coverage   69.29%   69.29%   -0.01%     
==========================================
  Files         146      148       +2     
  Lines       11240    11277      +37     
==========================================
+ Hits         7789     7814      +25     
- Misses       3451     3463      +12

Impacted Files	Coverage Δ
airflow/settings.py	89.41% <100%> (+0.52%)	⬆️
airflow/utils/log/airflow_task_logging.py	45.26% <45.26%> (ø)
airflow/utils/log/base_airflow_task_logging.py	63.63% <63.63%> (ø)
airflow/www/views.py	69.42% <80%> (+1.79%)	⬆️
airflow/bin/cli.py	51.13% <88.88%> (-0.74%)	⬇️
airflow/jobs.py	75.05% <0%> (-0.41%)	⬇️

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 4f20f60...f6ec201. Read the comment docs.

[saguziel]

maybe a better name is airflow.utils.logging

[allisonwang]

I was about to use logging but there is another file under airflow.utils that's named logging.py. Should we change the name of that file?

[bolkedebruin]

Other projects (django) use airflow.utils.log

[allisonwang]

We can't assume the logger will log into a file anymore so this line is removed.

[aoen]

If the default Airflow FileHandler has some kind of init call that is run before the other log lines we could put that there.

[allisonwang]

Is BaseAirflowTaskLogging better?

[aoen]

I like BaseAirflowTaskLogging more, it's more explicit.

[aoen]

s/custom/User-defined also AIRFLOW_HOME should be PYTHONPATH . Let's also update the README (airflow/docs/concepts.rst) with this, similar to how "Cluster Policy" is written.

[aoen]

Any reason we can't reuse the existing "airflow_local_settings.py" logic in settings.py? We could basically copy what is done for the policy function in that file. That way we can centralize the logic for loading custom modules.

[aoen]

Nit: grammar
"and optionally uploads to S3/GCS on task completion."

[aoen]

I like BaseAirflowTaskLogging more, it's more explicit.

[aoen]

Same comment as above, i.e. why are we doing this differently from the way airflow_local_settings works for the policy file. Also AIRFLOW_HOME should be PYTHONPATH (in general the comment should be similar to the one in settings.py for "def policy")

[aoen]

Nice, so much cleaner!

[bolkedebruin]

Why do you set this up in cli.py? and not in airflow.logging / airflow.utils.logging?

[allisonwang]

The comment above indicates it needs to be output to STDOUT for parent to read the log. Does it make sense to pass this logic to other places?

[bolkedebruin]

Hi guys, Im not sure if I follow the approach. It seems very complex to me and a bit java-esque (ohhh the horror :P ). Should we do a vid conf on this?

[aoen]

@bolkedebruin are you free now? I just messaged you on your gmail.

[aoen]

Is this line needed?

[aoen]

s/loaded/defined

[aoen]

Let's log debug here whether or not the user defined airflow task logging was loaded.

[bolkedebruin]

@aoen yes I am, no message received yet it seems?

[allisonwang]

@bolkedebruin Looking forward to the discussion! Here are some of my thoughts when I was designing the abstraction:
From a high-level perspective, the system needs to provide three major logging operations (use airflow worker as an example):

1. Setup airflow logging before task run such as adding log handlers
2. Clean up / additional operations once worker finished the task, if any, like uploading to cloud storage
3. Query logs and display on Airflow UI

Let's take the current Airflow logging as an example for each of the above steps:

1. Worker configures logging before each task run based on whether it's raw mode. It output logs to stdout if raw mode otherwise it logs to a file named dag_id-task_id-execution_date.
2. Once the task is finished, the log can be optionally uploaded to S3 or GCS.
3. Webserver queries the log by referring to the file named dag_id-task_id-execution_date. It first checks S3/GCS. If not loaded, it directly fetches the log from the remote worker machine.

What's the best way to use conf to represent all those situations without making user implementing the interface? I considered multiple configuration style abstractions but none of them seems to even be able to abstract the current implementations.

[bolkedebruin]

Hi @allisonwang

So from my point of view I would take a leaf out of Django's book:

So the configure logging I would expect something like

DEFAULT_LOGGING = {
    'version': 1,
    'disable_existing_loggers': False,
    },
    'formatters': {
        'airflow.task': {
            '()': 'airflow.utils.log.TaskFormatter',
            'format': '[%(server_time)s] %(message)s',
        }
    },
    'handlers': {
        'console': {
            'level': 'INFO',
            'filters': ['require_debug_true'],
            'class': 'logging.StreamHandler',
        },
        'airflow.task': {
            'level': 'INFO',
            'class': 'airflow.utils.log.TaskHandler',
            'formatter': 'airflow.task',
        },
    },
    'loggers': {
        'django.server': {
            'handlers': ['airflow.task'],
            'level': 'INFO',
            'propagate': False,
        },
    }
}

Here TaskHandler could be S3TaskHandler if needed. The S3TaskHandler could/should overwrite the "flush" / "close" methods in order to do clean ups and if really needed maybe the createLock or acquire lock methods. It could also derive from RotatingFileHandler to move the log to S3 at a certain point in time.

In the TaskHandler you can then use something like:

    def handle(self, record):
        msg = record.msg
        task_try = getattr(record, 'task_try', None)
        execution_date = getattr(record, 'execution_date', None)
        dag_id = getattr(record, 'execution_date', None)

       do_something(xx)

Doing it this way allows us to continue to use import logging and use the standard frameworks. It also gets rid of file opening in airflow itself for logging and all those messy bits. next to that, it is a simpler path for your internship as TaskHandler can be implemented quite independently from the rest, i.e. it is not necessary to overhaul everything cause the importing of the handler will get dependent on what logging.getLogger('XXX') gives back. This will be the default Handler when it is an arbitrary place, but will be logging.getLogger('airflow.task') in case of a task.

To get the location of the log (and to display it in the UI) several options exist, but I could imagine that you would extend the TaskHandler with a read(task_id, dag_id, execution_date) method that returns the contents for all tries or something along those lines.

This approach ensures that you only have to create two new classes in your case (I guess you are using S3) airflow.utils.log.TaskHandler and airflow.utils.log.S3TaskHandler. And some adjustments to configure_logging and to TaskInstance in models.py to obtain the right logger. Maybe also to the some of the operators if they are miss behaving (ie. not using the default handler), but I just checked I do not think works needs to be done there.

[allisonwang]

@bolkedebruin
Good point. Here is my understanding of TaskHandler class: it should at least have the following methods:

1. handle(record)
2. Flush / close: operations done once task finishes running
3. Read: read the log given task_id, dag_id and execution_date. We can get the log by using logging.getLogger('airflow.task').handlers[<handler_number>].read(task_instance)

I am still confused about how to address questions that's in current Airflow logging:

1. TaskHandler can handle each log line but how can it be setup before each task run? (like using the task_instance information to create directories if needed and give them correct permissions)
2. How to differentiate between logging to stdout and file given different args.raw value?
3. In handle method, currently airflow record does not have information like task_id, dag_id. This requires additional change to current logger.

From my perspective, TaskHandler with configuration is a lot more complex and less extensible than having one interface.

1. handle + init handler for each task run → setup_logging(ti)
2. flush/close/acquire_lock → post_logging(ti)
3. read → get_task_log(ti)
4. getLogger('airflow.task') → get_task_logger(ti)

It's definitely good to use and extend existing Python logging module if we can solve the above mention problems.

[bolkedebruin]

1. To my understanding the Handler will be opened at first call (ie. logger.getLogger) thus it is automatically per task run. RotatingFileHandler faces the same issue and has solved it.
2. That is just a configuration and you could log to both at the same time if you want. But maybe this is for the discussion in a minute as I don't quite understand why you think this could be an issue. If you don't want to STDOUT and a file at the same time, just use a Filter which is part of the same logging configuration.
3. This is what you pass to the record i.e. log.warning("Hello", dag_id="my_dag") results in a record attribute that you can access by getattr(record, 'dag_id', None). So while you need to override the handle method it does not require a lot of extra changes. See for inspirational purposes the ServerFormatter here: https://github.com/django/django/blob/master/django/utils/log.py

So no I don't think your approach is more flexible, but the contrary:

1. Only one handler needed, no additional call, standard config, just replace logging.getLogger(name) in TaskInstance with logging.getLogger('airflow.task'). It auto configures everything for everyone.
2. post_logging: not required, gets automatically called
3. Yes needs to be implemented
4. You are re-implementing what Python's "logging" was invented for. I don't think that is smart to do.

[allisonwang]

@bolkedebruin Thanks for the discussing the problem with us. As I mentioned, this solution has difficulties in actual implementation. So I spent some time writing the logic out. I have my working code pasted below and some pros and cons for this approach.

Pros:

1. Re-use python logging module config parser (dictConfig)
2. Easy to extend existing Python log handlers by creating subclass

Cons:

1. [Critical] Need to change all operator code to use airflow.task logger instead of root logger
   1. This can be avoid by configing for root logger instead of getLogger(“airflow.task”), but how can webserver read the log as itself also uses root logger?
2. [Design] Dynamically format configuration file defeats the purpose of using static configuration
3. [Design] read method in log handler breaks the handler abstraction
   1. Webserver doesn't know which handler to read the log from, if we have multiple handlers configured.

Code

task_handler.py

import logging

class TaskHandler(logging.FileHandler):

    def __init__(self, dag_id, task_id, execution_date, try_number):
        # Do directory creation here, for simplicity let's just use the filename
        self.filename = "{}-{}-{}-{}.log".format(dag_id, task_id, execution_date, try_number)
        logging.FileHandler.__init__(self, self.filename)

    def close(self):
        # Upload to S3/GCS
        super(logging.FileHandler, self).close()

    def read(self, dag_id, task_id, execution_date, try_number):
        # Parse the airflow.cfg and read from logging backend and worker machine.
        return "log"

cli.py

import logging
import logging.config


CONFIG = {
    'version': 1,
    'disable_existing_loggers': False,
    'formatters': {
        'airflow.task': {
            # '()': 'airflow.utils.log.TaskFormatter',
            'format': '[%(asctime)s] %(message)s',
        }
    },
    'handlers': {
        'console': {
            'level': 'INFO',
            'class': 'logging.StreamHandler',
        },
        'airflow.task': {
            'level': 'INFO',
            'class': 'task_handler.TaskHandler',
            'formatter': 'airflow.task',
            'dag_id': '{}',
            'task_id': '{}',
            'execution_date': '{}',
            'try_number': '{}',
        },
    },
    'loggers': {
        'airflow.task': {
            'handlers': ['airflow.task'],
            'level': 'INFO',
            'propagate': False,
        },
    }
}

TASK_LOGGER = 'airflow.task'

if __name__ == "__main__":

    dag_id = "dag_id"
    task_id = "task_id"
    execution_date = "execution_date"
    try_number = 1

    ### We need to include those in CLI run method:

    # One of the two ways to dynamically construct the config based on
    # http://codeinthehole.com/tips/a-deferred-logging-file-handler-for-django/
    for handler in CONFIG.get('handlers').values():
        if handler.get('class') == 'task_handler.TaskHandler':
            handler['dag_id'] = handler['dag_id'].format(dag_id)
            handler['task_id'] = handler['task_id'].format(task_id)
            handler['execution_date'] = handler['execution_date'].format(execution_date)
            handler['try_number'] = handler['try_number'].format(try_number)

    print CONFIG

    logging.config.dictConfig(CONFIG)

    logger = logging.getLogger(TASK_LOGGER)

    logger.info("testing")

    # close all handlers. Maybe we should flush all handlers first?
    for handler in logger.handlers:
        handler.close()

    ##################
    # webserver side #
    ##################
    logger = logging.getLogger(TASK_LOGGER)
    # What if there are more than one handlers? 
    print logger.handlers[0].read(dag_id, task_id, execution_date, try_number)

[bolkedebruin]

1. Why would you need to change all operator code (airflow.operator.*) to change its logger? It will inherit it from TaskInstance (I think), if not you could use "propagate = True" in the logging config and set "airflow.operators". Or add an extra method called something like "setTaskInstance" to TaskHandler, as they operators will inherit the parent logger. Obviously we need to try this out. I wouldn't necessarily call it critical as these kind of changes could be acceptable.

2. I think you would need to grab the template from the configuration and not do it in TaskHandler 'fixed'. Arguments from the config are password as keyword arguments to the constructor of the Handler. https://docs.python.org/2/library/logging.config.html#logging-config-dictschema

3. I don't think that really matters. You can just iterate over logging.handlers and see if one exposes the ReadMixin (or read method) and you are fine: they are logging all the same information. You could also make this a configuration item in airflow.cfg, i.e. Handler to use to read the logs.

I would use the flush method for uploading (https://github.com/python/cpython/blob/2.7/Lib/logging/__init__.py#L770), but that's arguable.

I think you don't you need to close the handlers by hand, I presume these are closed on interpreter shutdown. But that can be tested of course.

[allisonwang]

@bolkedebruin

1. Operators (at least majority) uses root logger like logging.info instead of logging into a specific logger. How can it be changed to use 'airflow.task' logger for all operators?
2. Would you be able to provide a concrete example? Given a task_instnace, how can we provide this task instance context when we use dictConfig(CONFIG) to instantiate handler and get the logger.
3. Why does it not matter? If webserver only grabs a random one and call the read method, it won't return the correct log unless user only configures one handler with the read method. But I guess we could make user specify in airflow.cfg the name of the handler.

Max and I discussed using the configuration style today. This configuration is ultimately static and doesn't provide enough context when instantiating the handler (unless the second point is addressed). Ideally, we can use a log handler generator that returns a callable handler. It can then be instantiated during run time with task instance and other contexts as arguments. Then config dictionary would not be used in this design but might be used in each handler's implementation. I will sketch this new design out soon.

[bolkedebruin]

1. operator.logging.getLogger.setHandler from TaskInstance and do a operator.logging.getLogger.removeHandler at the end? Didn't try and it is kind of hackish. We might need to fix the operators to log properly, because ultimately it is a bug how we do it now.

2. Here is an example: https://stackoverflow.com/questions/9405890/dynamic-filepath-filename-for-filehandler-in-logger-config-file-in-python and just use a ".format" for the string instead. You don't need to open the file at instantiation time, this can be deferred to handle time. The python core libs provide some inspiration: https://github.com/python/cpython/blob/2.7/Lib/logging/__init__.py#L907

What I would do is something like (high level pseudo code):

def __init__(xxx, template=None):
   self.template = template
   self.file_handle = None
 
def _open(dag_id, execution_date, task_id):
  file_name = self.template.format(dag_id, execution_date, task_id)
  self.file_handle = open(file_name)

def emit():
  if not self.file_handle:
    open(...)

  write(self.file_handle, xxx)

def close():
  if self.file_handle:
    self.file_handle.close()

You probably want to add a check if you are a still logging for the same dag_id,execution_date, task_id, but I think you get the point.

3. It isn't random. Just just goes through the handlers and tries to find one that exposes the "read" method. Indeed if that doesn't exist it won't do anything, but we help the user by providing sensible defaults. Buy using airflow.cfg seems more practical (or maybe add it to the LoggingConfig dict which puts all logging config in one place).

[allisonwang]

@bolkedebruin
For the second point if we defer the handler instantiation to handle method, does it mean we have to provide the context to the first log, such as logger.info("msg", dag_id='dag_id', task_id='task_id', execution_date='today', ...) Then we actually rely on this log message to correctly configure the handler, which could be problematic in the future.

[bolkedebruin]

@allisonwang Some further thoughts on your No 1 (redirecting other logs), if logging is properly configured in the operators/task_runners etc (e.g. logging.getLogger(name) ), it is just a matter of correctly configuring the handler in the logging config to point to the TaskHandler. This will work as it will get the reference to the same logger that does have set_context, by python logging design.

[allisonwang]

@bolkedebruin Make sense to me. Both jobs and task runner uses LoggingMixin which returns logger with format module_name.class_name. We can set propagate = True to make logs go through parent log handlers. Setting up context for task_runner is easy but for jobs, we don't have a TI object to pass in the handlers. Actually, do we even want to display the logs from models.py or jobs.py to UI? Airflow users are more concerned about everything happening inside task_runner while other logs are mainly for debugging airflow itself. Can we maybe have a separate handler for those system logs?

[bolkedebruin]

@allisonwang completely agree with not wanting to show models and jobs logging. This is also something from a risk perspective (separation of concerns) we like.

[allisonwang]

@bolkedebruin I refactored the original handler into FileTaskHandler and S3TaskHandler. GCSTaskHandler will closely resemble S3TaskHandler once the general structure is set. Regarding having one handler inside another handler, I think it's okay in this case considering how we need to combine set_context and logging.FileHandler together. But I am open to trying other approaches if any.

[bolkedebruin]

@allisonwang Nice! I don't have a very very strong opinion on using another handler in a handler, but it does feel somewhat unclean. I would prefer one that doesn't do this. As to your struggle on what the FileHandler does, you can just override the init method of course I guess?

Overall I like it much better now, it starts making sense.

[bolkedebruin]

Don't forget to correct this (DEFAULT_LOGGING)

[bolkedebruin]

this can be private

[bolkedebruin]

this can be private

[bolkedebruin]

this can be private

[bolkedebruin]

I don't think we should lose log entries if a handler isn't present. Maybe throw an exception when context isn't present?

[allisonwang]

This PR is closed by accident. I will create a new PR for this change.

[bolkedebruin]

We can just revert (and should)

[allisonwang]

Dan should revert it already but PR doesn't seem to be re-openable.

[bolkedebruin]

Just open a new one and link this one from there. This thread is getting pretty long anyways.