Блог

Я получаю следующую ошибку, когда запускаю свой конвейер и записываю журнал в S3.

Версия Airflow: 1.10.14

Я определил s3_conn в веб-интерфейсе Airflow вместе с идентификатором ключа и секретным кодом.

В файле airflow.cfg у меня есть следующие записи:

     base_log_folder = /root/airflow/logs
    
    # Airflow can store logs remotely in AWS S3, Google Cloud Storage or Elastic Search.
    # Set this to True if you want to enable remote logging.
    remote_logging = True
    
    # Users must supply an Airflow connection id that provides access to the storage
    # location.
    remote_log_conn_id = s3_conn
    remote_base_log_folder = s3://airflow-s3-buck2/test/
    encrypt_s3_logs = False
    logging_config_class = log_config.LOGGING_CONFIG
 

Вот мой log_config.py:

 import os

from airflow import configuration as conf

# TODO: Logging format and level should be configured
# in this file instead of from airflow.cfg. Currently
# there are other log format and level configurations in
# settings.py and cli.py. Please see AIRFLOW-1455.

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

BASE_LOG_FOLDER = conf.get('core', 'remote_base_log_folder')
PROCESSOR_LOG_FOLDER = conf.get('scheduler', 'child_process_log_directory')

FILENAME_TEMPLATE = '{{ ti.dag_id }}/{{ ti.task_id }}/{{ ts }}/{{ try_number }}.log'
PROCESSOR_FILENAME_TEMPLATE = '{{ filename }}.log'

LOGGING_CONFIG = {
    'version': 1,
    'disable_existing_loggers': False,
    'formatters': {
        'airflow.task': {
            'format': LOG_FORMAT,
        },
        'airflow.processor': {
            'format': LOG_FORMAT,
        },
    },
    'handlers': {
        'console': {
            'class': 'logging.StreamHandler',
            'formatter': 'airflow.task',
            'stream': 'ext://sys.stdout'
        },
        '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': BASE_LOG_FOLDER,
            'filename_template': FILENAME_TEMPLATE,
        },
        '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,
        }
        # When using s3 or gcs, provide a customized LOGGING_CONFIG
        # in airflow_local_settings within your PYTHONPATH, see UPDATING.md
        # for details
        # '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': S3_LOG_FOLDER,
        #     'filename_template': FILENAME_TEMPLATE,
        # },
        # 'gcs.task': {
        #     'class': 'airflow.utils.log.gcs_task_handler.GCSTaskHandler',
        #     'formatter': 'airflow.task',
        #     'base_log_folder': os.path.expanduser(BASE_LOG_FOLDER),
        #     'gcs_log_folder': GCS_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,
        },
    }
}
 

Ошибка (в журнале планировщика)

 botocore.exceptions.SSLError: SSL validation failed for https://airflow-s3-buck2.s3.amazonaws.com/test/s3_dag_test/bash_test/2016-11-01T00:00:00+00:00/1.log *[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: self signed certificate in certificate chain* (_ssl.c:1124)
 

Есть какие-либо рекомендации по устранению проблемы с сертификацией SSL?

Проблема, с которой вы столкнулись, связана с тем, что ваш сервер не доверяет сертификату, поскольку вы, скорее всего, используете пользовательский поставщик S3. Вам нужно будет добавить этот сертификат в свои сертификаты CA. Если у вас есть надлежащий доступ для этого, пожалуйста, прочтите это:

https://manpages .debian.org/buster/ca-certificates/update-ca-certificates.8.en.html

Вы просто поместите файл в /usr/share/ca-certificates и запустите update-ca-certificates .

Предполагая, что вы используете официальный образ docker / debian.

Однако, если вы действительно используете docker image в Kubernetes. Тогда я не смог найти правильного решения за пределами встраивания сертификата в изображение, поскольку контейнер запускается не root, а код Airflow не предоставляет никаких «дополнительных» настроек, чтобы пометить его как доверенный центр сертификации. Кроме того, Boto3 также не предоставляет никаких настроек, которые можно было бы использовать без фактического изменения кода в Airflow, поскольку параметр конечной точки не является частью переменных ENV или частью файлов конфигурации.