airflow - sesion - bitrix24 netitrix

Eliminar registros de tareas de flujo de aire

Question

Estoy ejecutando 5 DAG que han generado un total de aproximadamente 6 GB de datos de registro en la base_log_folder durante un período de meses. Acabo de agregar una remote_base_log_folder pero parece que no excluye el registro en la base_log_folder .

¿Hay alguna forma de eliminar automáticamente los archivos de registro antiguos, rotarlos o forzar el flujo de aire para que no inicie sesión en el disco (base_log_folder) solo en almacenamiento remoto?

Answer 1

Consulte https://github.com/teamclairvoyant/airflow-maintenance-dags

Este complemento tiene DAG que pueden eliminar tareas detenidas y limpieza de registros. Puede captar los conceptos y crear un nuevo DAG que se pueda limpiar según sus necesidades.

Answer 2

Los encargados del flujo de aire no piensan que truncar los registros es parte de la lógica del núcleo del flujo de aire. Para ver this , y luego, en este problema, los encargados del mantenimiento sugieren cambiar LOG_LEVEL para evitar demasiados datos de registro.

Y en this PR, podemos aprender cómo cambiar el nivel de registro en airflow.cfg .

buena suerte.

Answer 3

No creo que exista un mecanismo de rotación, pero puede almacenarlos en S3 o en el almacenamiento en la nube de Google como se describe aquí: https://airflow.incubator.apache.org/configuration.html#logs

Answer 4

FileTaskHandler los registros de tareas implementando nuestro propio FileTaskHandler , y luego lo airflow.cfg en airflow.cfg . Por lo tanto, sobrescribimos el LogHandler predeterminado para mantener solo N registros de tareas, sin programar DAG adicionales.

Estamos utilizando Airflow==1.10.1 .

[core]
logging_config_class = log_config.LOGGING_CONFIG

log_config.LOGGING_CONFIG

BASE_LOG_FOLDER = conf.get(''core'', ''BASE_LOG_FOLDER'')
FOLDER_TASK_TEMPLATE = ''{{ ti.dag_id }}/{{ ti.task_id }}''
FILENAME_TEMPLATE = ''{{ ti.dag_id }}/{{ ti.task_id }}/{{ ts }}/{{ try_number }}.log''
LOGGING_CONFIG = {
''formatters'': {},
''handlers'': {
''...'': {},
''task'': {
''class'': ''file_task_handler.FileTaskRotationHandler'',
''formatter'': ''airflow.job'',
''base_log_folder'': os.path.expanduser(BASE_LOG_FOLDER),
''filename_template'': FILENAME_TEMPLATE,
''folder_task_template'': FOLDER_TASK_TEMPLATE,
''retention'': 20
},
''...'': {}
},
''loggers'': {
''airflow.task'': {
''handlers'': [''task''],
''level'': JOB_LOG_LEVEL,
''propagate'': False,
},
''airflow.task_runner'': {
''handlers'': [''task''],
''level'': LOG_LEVEL,
''propagate'': True,
},
''...'': {}
}
}

file_task_handler.FileTaskRotationHandler

import os
import shutil
from airflow.utils.helpers import parse_template_string
from airflow.utils.log.file_task_handler import FileTaskHandler
class FileTaskRotationHandler(FileTaskHandler):
def __init__(self, base_log_folder, filename_template, folder_task_template, retention):
"""
:param base_log_folder: Base log folder to place logs.
:param filename_template: template filename string.
:param folder_task_template: template folder task path.
:param retention: Number of folder logs to keep
"""
super(FileTaskRotationHandler, self).__init__(base_log_folder, filename_template)
self.retention = retention
self.folder_task_template, self.folder_task_template_jinja_template = /
parse_template_string(folder_task_template)
@staticmethod
def _get_directories(path=''.''):
return next(os.walk(path))[1]
def _render_folder_task_path(self, ti):
if self.folder_task_template_jinja_template:
jinja_context = ti.get_template_context()
return self.folder_task_template_jinja_template.render(**jinja_context)
return self.folder_task_template.format(dag_id=ti.dag_id, task_id=ti.task_id)
def _init_file(self, ti):
relative_path = self._render_folder_task_path(ti)
folder_task_path = os.path.join(self.local_base, relative_path)
subfolders = self._get_directories(folder_task_path)
to_remove = set(subfolders) - set(subfolders[-self.retention:])
for dir_to_remove in to_remove:
full_dir_to_remove = os.path.join(folder_task_path, dir_to_remove)
print(''Removing'', full_dir_to_remove)
shutil.rmtree(full_dir_to_remove)
return FileTaskHandler._init_file(self, ti)