Can't read S3 remote logs when using gevent/eventlent webserver workers.

[webster-chainalysis]

Hey everyone. I’ve upgraded to 1.10.9. It appears that the logging changes broke the functionality for reading S3 remote logs in the Web UI (writing is ok). In the change log it mentions that Airflow's logging mechanism has been refactored to uses Python’s builtin logging module:

[AIRFLOW-1611] Customize logging

I followed the directions in the changelog and created the following log config:

import os
import six

from airflow import AirflowException
from airflow.configuration import conf
from airflow.utils.file import mkdirs
from typing import Dict, Any

LOG_LEVEL = conf.get('core', 'LOGGING_LEVEL').upper()

FAB_LOG_LEVEL = conf.get('core', 'FAB_LOGGING_LEVEL').upper()

LOG_FORMAT = conf.get('core', 'LOG_FORMAT')

COLORED_LOG_FORMAT = conf.get('core', 'COLORED_LOG_FORMAT')

COLORED_LOG = conf.getboolean('core', 'COLORED_CONSOLE_LOG')

COLORED_FORMATTER_CLASS = conf.get('core', 'COLORED_FORMATTER_CLASS')

BASE_LOG_FOLDER = conf.get('core', 'BASE_LOG_FOLDER')

PROCESSOR_LOG_FOLDER = conf.get('scheduler', 'CHILD_PROCESS_LOG_DIRECTORY')

DAG_PROCESSOR_MANAGER_LOG_LOCATION = \
    conf.get('core', 'DAG_PROCESSOR_MANAGER_LOG_LOCATION')

FILENAME_TEMPLATE = conf.get('core', 'LOG_FILENAME_TEMPLATE')

PROCESSOR_FILENAME_TEMPLATE = conf.get('core', 'LOG_PROCESSOR_FILENAME_TEMPLATE')

FORMATTER_CLASS_KEY = '()' if six.PY2 else 'class'


#
# Getting this from environment because the changelog for 1.10.9 says to set
# the path of `REMOTE_BASE_LOG_FOLDER` explicitly in the config. The
# `REMOTE_BASE_LOG_FOLDER` key is not used anymore.
#
REMOTE_BASE_LOG_FOLDER = os.environ.get('AIRFLOW__CORE__REMOTE_BASE_LOG_FOLDER')

LOGGING_CONFIG = {
    'version': 1,
    'disable_existing_loggers': False,
    'formatters': {
        'airflow.task': {
            'format': LOG_FORMAT,
        },
        'airflow.processor': {
            'format': LOG_FORMAT,
        },
        'airflow_coloured': {
            'format': COLORED_LOG_FORMAT if COLORED_LOG else LOG_FORMAT,
            FORMATTER_CLASS_KEY: COLORED_FORMATTER_CLASS if COLORED_LOG else 'logging.Formatter'
        },
    },
    'handlers': {
        'console': {
            'class': 'logging.StreamHandler',
            'formatter': 'airflow.task',
            'stream': 'ext://sys.stdout'
        },
        'file.task': {
            'class': 'airflow.utils.log.file_task_handler.FileTaskHandler',
            'formatter': 'airflow.task',
            'base_log_folder': os.path.expanduser(BASE_LOG_FOLDER),
            'filename_template': FILENAME_TEMPLATE,
        },
        'file.processor': {
            'class': 'airflow.utils.log.file_processor_handler.FileProcessorHandler',
            'formatter': 'airflow.processor',
            'base_log_folder': os.path.expanduser(PROCESSOR_LOG_FOLDER),
            'filename_template': PROCESSOR_FILENAME_TEMPLATE,
        },
        's3.task': {
            'class': 'airflow.utils.log.s3_task_handler.S3TaskHandler',
            'formatter': 'airflow.task',
            'base_log_folder': os.path.expanduser(BASE_LOG_FOLDER),
            's3_log_folder': REMOTE_BASE_LOG_FOLDER,
            'filename_template': FILENAME_TEMPLATE,
        },
    },
    'loggers': {
        '': {
            'handlers': ['console'],
            'level': LOG_LEVEL
        },
        'airflow': {
            'handlers': ['console'],
            'level': LOG_LEVEL,
            'propagate': False,
        },
        'airflow.processor': {
            'handlers': ['file.processor'],
            'level': LOG_LEVEL,
            'propagate': True,
        },
        'airflow.task': {
            'handlers': ['s3.task'],
            'level': LOG_LEVEL,
            'propagate': False,
        },
        'airflow.task_runner': {
            'handlers': ['s3.task'],
            'level': LOG_LEVEL,
            'propagate': True,
        },
    }
}  # type: Dict[str, Any]

However, the task log reader is always defaulting to using the FileTaskHandler. This should not occur because I have the following settings in airflow.cfg:

remote_logging = True
remote_base_log_folder = s3://my-bucket-name
remote_log_conn_id = aws_default
task_log_reader = s3.task

The s3.task handler passed to the task_log_reader setting should be creating an instance of the S3TaskHandler class to read the task logs to from S3. This occurs when rendering the get_logs_with_metadata view in www/views.py.

---

Apache Airflow version: 1.10.9
Kubernetes version: 1.15

Environment:

• Cloud provider or hardware configuration: AWS
• OS (e.g. from /etc/os-release): python:3.7.6-buster docker image

What happened: Logs did not appear in the Airflow Web UI. The FileTaskHandler tries to fetch the file locally or from the worker on port 8793. However, the logs do not exist in either location since we are using the Kubernetes Executor. This produces the following errors messages:

*** Log file does not exist: /usr/local/airflow/logs/MY_DAG_NAME/MY_TASK_NAME/2020-04-07T20:59:19.312402+00:00/6.log
*** Fetching from: http://MY_DAG_NAME-0dde5ff5a786437cb14234:8793/log/MY_DAG_NAME/MY_TASK_NAME/2020-04-07T20:59:19.312402+00:00/6.log
*** Failed to fetch log file from worker. HTTPConnectionPool(host='MY_DAG_NAME-0dde5ff5a786437cb14234', port=8793): Max retries exceeded with url: /log/MY_DAG_NAME/MY_TASK_NAME/2020-04-07T20:59:19.312402+00:00/6.log (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x7f708332fc90>: Failed to establish a new connection: [Errno -2] Name or service not known'))

What you expected to happen:

The logs should be rendered in the Web UI using the S3TaskHandler class.

[boring-cyborg]

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

[dimon222]

@webster-chainalysis
I noticed such issue on 1.10.10. Do you also have same behaviour on this version?

[webster-chainalysis]

@dimon222 - Yes.

[dimon222]

@webster-chainalysis
Worker class switch in airflow.cfg from gevent/eventlet to "sync" seem to resolve the issue for me.

[webster-chainalysis]

@dimon222 - That makes sense. I'm currently using gevent behind an Amazon elastic load balancer. When using the sync worker class with the ELB I had unacceptable performance.

[TRReeve]

I have had this issue as well on Airflow 10.10 and been tearing my hear out at it a couple of days now. I'm running using the KubernetesOperator and the logging seems to upload to S3 fine it is only the reading of the log back from the web ui that is always in local mode.

[TRReeve]

Having gotten remote logging working on 1.10.10 I've noticed there seems to be a difference in how the workers handle the log upload versus the webserver component. The Connection you define needs to point to the same folder as the remote_base_logs_folder. E.g if your remote base logs folder is data/airflow/logs then your connection used for remote logging also needs to point to it e.g s3://access_key:secret@data/airflow/logs. You cannot just have a generic "AWS connection" and then have it figure out which folder it needs to point to on its own using the filepath from the worker.

[marclamberti]

@TRReeve I didn’t need to do that. Actually, if you want to have the remote logging working with the kubernetes executor, you have to define additional kubernetes environment variables so that your PODs are in sync with the scheduler/web server. Eg:
AIRFLOW__CORE__REMOTE_LOGGING=True
AIRFLOW__CORE__REMOTE_BASE_LOGS_FOLDER=s3://my-bucket/my-key
AIRFLOW__CORE__REMOTE_LOG_CONN_ID=myawsconn
AIRFLOW__CORE__FERNET_KEY=myfernetkey

AIRFLOW__KUBERNETES_ENVIRONMENT_VARIABALES__AIRFLOW__CORE__REMOTE_LOGGING=True
AIRFLOW__KUBERNETES_ENVIRONMENT_VARIABALES__AIRFLOW__CORE__REMOTE_BASE_LOGS_FOLDER=s3://my-bucket/my-key
AIRFLOW__KUBERNETES_ENVIRONMENT_VARIABALES__AIRFLOW__CORE__REMOTE_LOG_CONN_ID=myawsconn
AIRFLOW__KUBERNETES_ENVIRONMENT_VARIABALES__AIRFLOW__CORE__FERNET_KEY=myfernetkey

The connection myawsconn with type S3 and extra field

• aws_access_key_id
• aws_secret_acess_key_id
• region_name

Also I didn’t need to change the task handler as it is automatically changed when remote logging is set to True, and I didn’t have to define a custom logging class.

Hope it helps :)

[TRReeve]

@marclamberti Your answer is how I understood it would work as well and it half worked for me I would get the logs uploading fine into the S3 bucket but then when i went to "view logs" in the UI it would give the "logs not found" error with no output in the logs to indicate it was using the s3 connection or the read_key function to retrieve anything.

It would be really nice if I could just define AIRFLOW_CONN_S3_URI = s3://user:pass@S3 then have REMOTE_BASE_LOGS_FOLDER=s3://airflow-stuff/logs and the UI would build the path but I could only get logs uploading. My working helm template for airflow on k8s builds the connection s3://access_key:secret_key@{{ mys3path }} and then remote_log_path is s3://{{ mys3path }}. Aside that it's exactly the same as you defined above with the same variables defined under AIRFLOW__KUBERNETES__ENVIRONMENT_VARIABLES.

And yes I can confirm for anyone reading there was no need for any custom logging classes or task handlers.

[marclamberti]

Well, in my case it works for both reading and writing logs from S3 with Airflow 1.10.10 and the Kubernetes executor
I agreed it could be simpler :)

[ashb]

@webster-chainalysis I've just opened #9118 to improve the debugging of S3 log handler.

Could you also try the debug steps I put in that description there -- lets rule out that sort of problem.