New RotateAirflowLogs DAG for deleting empty log directories

[jabortell]

Links:

• Design: STAD-198 Add log rotation to Airflow logs
• Jira: STAD-198 Add log rotation to Airflow logs

Testing instructions:

In a stadium or stadium-like project, import the new DAG and instantiate it like:

fromea_airflow_utilimportRotateLogsDAGglobals()[rotate_logs_dag.dag.dag_id] =rotate_logs_dag.dag

This DAG support configuration via airflow_config.yml under the key rotate_logs_dag. For example, if your logs root directory is located someplace besides $AIRFLOW_HOME/logs, you can configure it like:

rotate_logs_dag:
schedule_interval: nulldefault_args: *default_task_argslogs_dir: /some/other/airflow/logs

then update your instantiation code to load the config:

fromutilimportio_helpers# Or wherever your config is:airflow_configs_file="/home/airflow/airflow/configs/airflow_config.yml"AIRFLOW_CONFIGS=io_helpers.safe_load_yaml(configs_dir, airflow_configs_file)
dag_args=AIRFLOW_CONFIGS.get("rotate_logs_dag", {})
rotate_logs_dag=RotateLogsDAG(dag_id="rotate_airflow_logs", **dag_args)
globals()[rotate_logs_dag.dag.dag_id] =rotate_logs_dag.dag

Finally, when you run the DAG, confirm that it runs successfully, that empty sub-directories in the logging directory are deleted, and non-empty directories are not deleted. The DAG uses os.rmdir(), which will throw an exception if called on a non-empty directory, so there should not be a risk of deleting non-empty directories.

[jabortell]

@rlittle08, Fyi @ea-mtenhoor commented in the Jira about wanting to discuss the scope of the design further.

[jabortell]

Pushed a change to scan for empty files, via the meeting/notes from the Jira. I also updated the directory scanner to be a loop, so that it handles nested empty directories.

[tomreitz]

I tested this in TX – some observations and feedback:

1. A naming question: this dag is called RotateAirflowLogs but that's not actually what it does. Might a better name be PruneEmptyAirflowLogs?

2. To avoid import errors, I had to add the following line to ea_airflow_util/ea_airflow_util/__init__.py:

   fromea_airflow_util.dags.rotate_airflow_logsimportRotateLogsDAG

3. This was my final DAG code:

   fromea_airflow_utilimportRotateLogsDAGfromutilimportio_helpersconfigs_dir='/home/airflow/airflow/configs'airflow_configs_file="airflow_config.yml"AIRFLOW_CONFIGS=io_helpers.safe_load_yaml(configs_dir, airflow_configs_file)
   dag_args=AIRFLOW_CONFIGS.get("rotate_logs_dag", {})
   rotate_logs_dag=RotateLogsDAG(dag_id="rotate_airflow_logs", **dag_args)
   globals()[rotate_logs_dag.dag.dag_id] =rotate_logs_dag.dag

   It could be helpful to add a full example like this to the docs.

4. At least in TX, the environment variable AIRFLOW_HOME is not defined. (Based on here, I don't think it would be defined in any implementation.) This means the rotate_logs_dag.logs_dir YAML config is not optional (though the comments imply it is); I had to specify it as logs_dir: /home/airflow/airflow/logs, and even then there's still a DAG-parse key error on os.environ["AIRFLOW_HOME"]here. I'd recommend changing to something like os.environ.get("AIRFLOW_HOME", "/home/airflow/airflow/logs").

5. When I got the DAG to run, it successfully deleted 74,821 empty files/folders in 61s, freeing up ~300MB of space on disk. It logged every (nested) file/folder that's deleted, so the task's logs were 159k lines and 33MB. Have we considered checking if a folder contains any non-empty files and only recursing in if so? (otherwise, just delete the entire folder) I suspect such an approach could run faster, and produce more concise logs.

   defprune_empty(folder=logs_diroros.environ.get("AIRFLOW_HOME", "/home/airflow/airflow/logs")):
   folder_path=Path(folder)
   forsubfolderinfolder_path.iterdir():
   ifsubfolder.is_dir():
   total_size=sum(f.stat().st_sizeforfinsubfolder.rglob('*') iff.is_file())
   iftotal_size>0: prune_empty(subfolder) # (recurse)else:
   logging.info(f"Deleting folder {subfolder} (contains no non-empty files)")
   shutil.rmtree(subfolder)

[jabortell]

Links:

• Design: STAD-198 Add log rotation to Airflow logs
• Jira: STAD-198 Add log rotation to Airflow logs

Testing instructions:

In a stadium or stadium-like project, import the new DAG and instantiate it like:

fromea_airflow_utilimportRotateLogsDAGglobals()[rotate_logs_dag.dag.dag_id] =rotate_logs_dag.dag

This DAG support configuration via airflow_config.yml under the key rotate_logs_dag. For example, if your logs root directory is located someplace besides $AIRFLOW_HOME/logs, you can configure it like:

rotate_logs_dag:
schedule_interval: nulldefault_args: *default_task_argslogs_dir: /some/other/airflow/logs

then update your instantiation code to load the config:

fromutilimportio_helpers# Or wherever your config is:airflow_configs_file="/home/airflow/airflow/configs/airflow_config.yml"AIRFLOW_CONFIGS=io_helpers.safe_load_yaml(configs_dir, airflow_configs_file)
dag_args=AIRFLOW_CONFIGS.get("rotate_logs_dag", {})
rotate_logs_dag=RotateLogsDAG(dag_id="rotate_airflow_logs", **dag_args)
globals()[rotate_logs_dag.dag.dag_id] =rotate_logs_dag.dag

Finally, when you run the DAG, confirm that it runs successfully, that empty sub-directories in the logging directory are deleted, and non-empty directories are not deleted. The DAG uses os.rmdir(), which will throw an exception if called on a non-empty directory, so there should not be a risk of deleting non-empty directories.

[jabortell]

@rlittle08, Fyi @ea-mtenhoor commented in the Jira about wanting to discuss the scope of the design further.

[jabortell]

Pushed a change to scan for empty files, via the meeting/notes from the Jira. I also updated the directory scanner to be a loop, so that it handles nested empty directories.

[tomreitz]

I tested this in TX – some observations and feedback:

1. A naming question: this dag is called RotateAirflowLogs but that's not actually what it does. Might a better name be PruneEmptyAirflowLogs?

2. To avoid import errors, I had to add the following line to ea_airflow_util/ea_airflow_util/__init__.py:

   fromea_airflow_util.dags.rotate_airflow_logsimportRotateLogsDAG

3. This was my final DAG code:

   fromea_airflow_utilimportRotateLogsDAGfromutilimportio_helpersconfigs_dir='/home/airflow/airflow/configs'airflow_configs_file="airflow_config.yml"AIRFLOW_CONFIGS=io_helpers.safe_load_yaml(configs_dir, airflow_configs_file)
   dag_args=AIRFLOW_CONFIGS.get("rotate_logs_dag", {})
   rotate_logs_dag=RotateLogsDAG(dag_id="rotate_airflow_logs", **dag_args)
   globals()[rotate_logs_dag.dag.dag_id] =rotate_logs_dag.dag

   It could be helpful to add a full example like this to the docs.

4. At least in TX, the environment variable AIRFLOW_HOME is not defined. (Based on here, I don't think it would be defined in any implementation.) This means the rotate_logs_dag.logs_dir YAML config is not optional (though the comments imply it is); I had to specify it as logs_dir: /home/airflow/airflow/logs, and even then there's still a DAG-parse key error on os.environ["AIRFLOW_HOME"]here. I'd recommend changing to something like os.environ.get("AIRFLOW_HOME", "/home/airflow/airflow/logs").

5. When I got the DAG to run, it successfully deleted 74,821 empty files/folders in 61s, freeing up ~300MB of space on disk. It logged every (nested) file/folder that's deleted, so the task's logs were 159k lines and 33MB. Have we considered checking if a folder contains any non-empty files and only recursing in if so? (otherwise, just delete the entire folder) I suspect such an approach could run faster, and produce more concise logs.

   defprune_empty(folder=logs_diroros.environ.get("AIRFLOW_HOME", "/home/airflow/airflow/logs")):
   folder_path=Path(folder)
   forsubfolderinfolder_path.iterdir():
   ifsubfolder.is_dir():
   total_size=sum(f.stat().st_sizeforfinsubfolder.rglob('*') iff.is_file())
   iftotal_size>0: prune_empty(subfolder) # (recurse)else:
   logging.info(f"Deleting folder {subfolder} (contains no non-empty files)")
   shutil.rmtree(subfolder)

[jabortell]

Links:

• Design: STAD-198 Add log rotation to Airflow logs
• Jira: STAD-198 Add log rotation to Airflow logs

Testing instructions:

In a stadium or stadium-like project, import the new DAG and instantiate it like:

fromea_airflow_utilimportRotateLogsDAGglobals()[rotate_logs_dag.dag.dag_id] =rotate_logs_dag.dag

This DAG support configuration via airflow_config.yml under the key rotate_logs_dag. For example, if your logs root directory is located someplace besides $AIRFLOW_HOME/logs, you can configure it like:

rotate_logs_dag:
schedule_interval: nulldefault_args: *default_task_argslogs_dir: /some/other/airflow/logs

then update your instantiation code to load the config:

fromutilimportio_helpers# Or wherever your config is:airflow_configs_file="/home/airflow/airflow/configs/airflow_config.yml"AIRFLOW_CONFIGS=io_helpers.safe_load_yaml(configs_dir, airflow_configs_file)
dag_args=AIRFLOW_CONFIGS.get("rotate_logs_dag", {})
rotate_logs_dag=RotateLogsDAG(dag_id="rotate_airflow_logs", **dag_args)
globals()[rotate_logs_dag.dag.dag_id] =rotate_logs_dag.dag

Finally, when you run the DAG, confirm that it runs successfully, that empty sub-directories in the logging directory are deleted, and non-empty directories are not deleted. The DAG uses os.rmdir(), which will throw an exception if called on a non-empty directory, so there should not be a risk of deleting non-empty directories.

[jabortell]

@rlittle08, Fyi @ea-mtenhoor commented in the Jira about wanting to discuss the scope of the design further.

[jabortell]

Pushed a change to scan for empty files, via the meeting/notes from the Jira. I also updated the directory scanner to be a loop, so that it handles nested empty directories.

[tomreitz]

I tested this in TX – some observations and feedback:

1. A naming question: this dag is called RotateAirflowLogs but that's not actually what it does. Might a better name be PruneEmptyAirflowLogs?

2. To avoid import errors, I had to add the following line to ea_airflow_util/ea_airflow_util/__init__.py:

   fromea_airflow_util.dags.rotate_airflow_logsimportRotateLogsDAG

3. This was my final DAG code:

   fromea_airflow_utilimportRotateLogsDAGfromutilimportio_helpersconfigs_dir='/home/airflow/airflow/configs'airflow_configs_file="airflow_config.yml"AIRFLOW_CONFIGS=io_helpers.safe_load_yaml(configs_dir, airflow_configs_file)
   dag_args=AIRFLOW_CONFIGS.get("rotate_logs_dag", {})
   rotate_logs_dag=RotateLogsDAG(dag_id="rotate_airflow_logs", **dag_args)
   globals()[rotate_logs_dag.dag.dag_id] =rotate_logs_dag.dag

   It could be helpful to add a full example like this to the docs.

4. At least in TX, the environment variable AIRFLOW_HOME is not defined. (Based on here, I don't think it would be defined in any implementation.) This means the rotate_logs_dag.logs_dir YAML config is not optional (though the comments imply it is); I had to specify it as logs_dir: /home/airflow/airflow/logs, and even then there's still a DAG-parse key error on os.environ["AIRFLOW_HOME"]here. I'd recommend changing to something like os.environ.get("AIRFLOW_HOME", "/home/airflow/airflow/logs").

5. When I got the DAG to run, it successfully deleted 74,821 empty files/folders in 61s, freeing up ~300MB of space on disk. It logged every (nested) file/folder that's deleted, so the task's logs were 159k lines and 33MB. Have we considered checking if a folder contains any non-empty files and only recursing in if so? (otherwise, just delete the entire folder) I suspect such an approach could run faster, and produce more concise logs.

   defprune_empty(folder=logs_diroros.environ.get("AIRFLOW_HOME", "/home/airflow/airflow/logs")):
   folder_path=Path(folder)
   forsubfolderinfolder_path.iterdir():
   ifsubfolder.is_dir():
   total_size=sum(f.stat().st_sizeforfinsubfolder.rglob('*') iff.is_file())
   iftotal_size>0: prune_empty(subfolder) # (recurse)else:
   logging.info(f"Deleting folder {subfolder} (contains no non-empty files)")
   shutil.rmtree(subfolder)

[jabortell]

Links:

• Design: STAD-198 Add log rotation to Airflow logs
• Jira: STAD-198 Add log rotation to Airflow logs

Testing instructions:

In a stadium or stadium-like project, import the new DAG and instantiate it like:

fromea_airflow_utilimportRotateLogsDAGglobals()[rotate_logs_dag.dag.dag_id] =rotate_logs_dag.dag

This DAG support configuration via airflow_config.yml under the key rotate_logs_dag. For example, if your logs root directory is located someplace besides $AIRFLOW_HOME/logs, you can configure it like:

rotate_logs_dag:
schedule_interval: nulldefault_args: *default_task_argslogs_dir: /some/other/airflow/logs

then update your instantiation code to load the config:

fromutilimportio_helpers# Or wherever your config is:airflow_configs_file="/home/airflow/airflow/configs/airflow_config.yml"AIRFLOW_CONFIGS=io_helpers.safe_load_yaml(configs_dir, airflow_configs_file)
dag_args=AIRFLOW_CONFIGS.get("rotate_logs_dag", {})
rotate_logs_dag=RotateLogsDAG(dag_id="rotate_airflow_logs", **dag_args)
globals()[rotate_logs_dag.dag.dag_id] =rotate_logs_dag.dag

Finally, when you run the DAG, confirm that it runs successfully, that empty sub-directories in the logging directory are deleted, and non-empty directories are not deleted. The DAG uses os.rmdir(), which will throw an exception if called on a non-empty directory, so there should not be a risk of deleting non-empty directories.

[jabortell]

@rlittle08, Fyi @ea-mtenhoor commented in the Jira about wanting to discuss the scope of the design further.

[jabortell]

Pushed a change to scan for empty files, via the meeting/notes from the Jira. I also updated the directory scanner to be a loop, so that it handles nested empty directories.

[tomreitz]

I tested this in TX – some observations and feedback:

1. A naming question: this dag is called RotateAirflowLogs but that's not actually what it does. Might a better name be PruneEmptyAirflowLogs?

2. To avoid import errors, I had to add the following line to ea_airflow_util/ea_airflow_util/__init__.py:

   fromea_airflow_util.dags.rotate_airflow_logsimportRotateLogsDAG

3. This was my final DAG code:

   fromea_airflow_utilimportRotateLogsDAGfromutilimportio_helpersconfigs_dir='/home/airflow/airflow/configs'airflow_configs_file="airflow_config.yml"AIRFLOW_CONFIGS=io_helpers.safe_load_yaml(configs_dir, airflow_configs_file)
   dag_args=AIRFLOW_CONFIGS.get("rotate_logs_dag", {})
   rotate_logs_dag=RotateLogsDAG(dag_id="rotate_airflow_logs", **dag_args)
   globals()[rotate_logs_dag.dag.dag_id] =rotate_logs_dag.dag

   It could be helpful to add a full example like this to the docs.

4. At least in TX, the environment variable AIRFLOW_HOME is not defined. (Based on here, I don't think it would be defined in any implementation.) This means the rotate_logs_dag.logs_dir YAML config is not optional (though the comments imply it is); I had to specify it as logs_dir: /home/airflow/airflow/logs, and even then there's still a DAG-parse key error on os.environ["AIRFLOW_HOME"]here. I'd recommend changing to something like os.environ.get("AIRFLOW_HOME", "/home/airflow/airflow/logs").

5. When I got the DAG to run, it successfully deleted 74,821 empty files/folders in 61s, freeing up ~300MB of space on disk. It logged every (nested) file/folder that's deleted, so the task's logs were 159k lines and 33MB. Have we considered checking if a folder contains any non-empty files and only recursing in if so? (otherwise, just delete the entire folder) I suspect such an approach could run faster, and produce more concise logs.

   defprune_empty(folder=logs_diroros.environ.get("AIRFLOW_HOME", "/home/airflow/airflow/logs")):
   folder_path=Path(folder)
   forsubfolderinfolder_path.iterdir():
   ifsubfolder.is_dir():
   total_size=sum(f.stat().st_sizeforfinsubfolder.rglob('*') iff.is_file())
   iftotal_size>0: prune_empty(subfolder) # (recurse)else:
   logging.info(f"Deleting folder {subfolder} (contains no non-empty files)")
   shutil.rmtree(subfolder)

[jabortell]

Links:

• Design: STAD-198 Add log rotation to Airflow logs
• Jira: STAD-198 Add log rotation to Airflow logs

Testing instructions:

In a stadium or stadium-like project, import the new DAG and instantiate it like:

fromea_airflow_utilimportRotateLogsDAGglobals()[rotate_logs_dag.dag.dag_id] =rotate_logs_dag.dag

This DAG support configuration via airflow_config.yml under the key rotate_logs_dag. For example, if your logs root directory is located someplace besides $AIRFLOW_HOME/logs, you can configure it like:

rotate_logs_dag:
schedule_interval: nulldefault_args: *default_task_argslogs_dir: /some/other/airflow/logs

then update your instantiation code to load the config:

fromutilimportio_helpers# Or wherever your config is:airflow_configs_file="/home/airflow/airflow/configs/airflow_config.yml"AIRFLOW_CONFIGS=io_helpers.safe_load_yaml(configs_dir, airflow_configs_file)
dag_args=AIRFLOW_CONFIGS.get("rotate_logs_dag", {})
rotate_logs_dag=RotateLogsDAG(dag_id="rotate_airflow_logs", **dag_args)
globals()[rotate_logs_dag.dag.dag_id] =rotate_logs_dag.dag

Finally, when you run the DAG, confirm that it runs successfully, that empty sub-directories in the logging directory are deleted, and non-empty directories are not deleted. The DAG uses os.rmdir(), which will throw an exception if called on a non-empty directory, so there should not be a risk of deleting non-empty directories.

[jabortell]

@rlittle08, Fyi @ea-mtenhoor commented in the Jira about wanting to discuss the scope of the design further.

[jabortell]

Pushed a change to scan for empty files, via the meeting/notes from the Jira. I also updated the directory scanner to be a loop, so that it handles nested empty directories.

[tomreitz]

I tested this in TX – some observations and feedback:

1. A naming question: this dag is called RotateAirflowLogs but that's not actually what it does. Might a better name be PruneEmptyAirflowLogs?

2. To avoid import errors, I had to add the following line to ea_airflow_util/ea_airflow_util/__init__.py:

   fromea_airflow_util.dags.rotate_airflow_logsimportRotateLogsDAG

3. This was my final DAG code:

   fromea_airflow_utilimportRotateLogsDAGfromutilimportio_helpersconfigs_dir='/home/airflow/airflow/configs'airflow_configs_file="airflow_config.yml"AIRFLOW_CONFIGS=io_helpers.safe_load_yaml(configs_dir, airflow_configs_file)
   dag_args=AIRFLOW_CONFIGS.get("rotate_logs_dag", {})
   rotate_logs_dag=RotateLogsDAG(dag_id="rotate_airflow_logs", **dag_args)
   globals()[rotate_logs_dag.dag.dag_id] =rotate_logs_dag.dag

   It could be helpful to add a full example like this to the docs.

4. At least in TX, the environment variable AIRFLOW_HOME is not defined. (Based on here, I don't think it would be defined in any implementation.) This means the rotate_logs_dag.logs_dir YAML config is not optional (though the comments imply it is); I had to specify it as logs_dir: /home/airflow/airflow/logs, and even then there's still a DAG-parse key error on os.environ["AIRFLOW_HOME"]here. I'd recommend changing to something like os.environ.get("AIRFLOW_HOME", "/home/airflow/airflow/logs").

5. When I got the DAG to run, it successfully deleted 74,821 empty files/folders in 61s, freeing up ~300MB of space on disk. It logged every (nested) file/folder that's deleted, so the task's logs were 159k lines and 33MB. Have we considered checking if a folder contains any non-empty files and only recursing in if so? (otherwise, just delete the entire folder) I suspect such an approach could run faster, and produce more concise logs.

   defprune_empty(folder=logs_diroros.environ.get("AIRFLOW_HOME", "/home/airflow/airflow/logs")):
   folder_path=Path(folder)
   forsubfolderinfolder_path.iterdir():
   ifsubfolder.is_dir():
   total_size=sum(f.stat().st_sizeforfinsubfolder.rglob('*') iff.is_file())
   iftotal_size>0: prune_empty(subfolder) # (recurse)else:
   logging.info(f"Deleting folder {subfolder} (contains no non-empty files)")
   shutil.rmtree(subfolder)

[jabortell]

Links:

• Design: STAD-198 Add log rotation to Airflow logs
• Jira: STAD-198 Add log rotation to Airflow logs

Testing instructions:

In a stadium or stadium-like project, import the new DAG and instantiate it like:

fromea_airflow_utilimportRotateLogsDAGglobals()[rotate_logs_dag.dag.dag_id] =rotate_logs_dag.dag

This DAG support configuration via airflow_config.yml under the key rotate_logs_dag. For example, if your logs root directory is located someplace besides $AIRFLOW_HOME/logs, you can configure it like:

rotate_logs_dag:
schedule_interval: nulldefault_args: *default_task_argslogs_dir: /some/other/airflow/logs

then update your instantiation code to load the config:

fromutilimportio_helpers# Or wherever your config is:airflow_configs_file="/home/airflow/airflow/configs/airflow_config.yml"AIRFLOW_CONFIGS=io_helpers.safe_load_yaml(configs_dir, airflow_configs_file)
dag_args=AIRFLOW_CONFIGS.get("rotate_logs_dag", {})
rotate_logs_dag=RotateLogsDAG(dag_id="rotate_airflow_logs", **dag_args)
globals()[rotate_logs_dag.dag.dag_id] =rotate_logs_dag.dag

Finally, when you run the DAG, confirm that it runs successfully, that empty sub-directories in the logging directory are deleted, and non-empty directories are not deleted. The DAG uses os.rmdir(), which will throw an exception if called on a non-empty directory, so there should not be a risk of deleting non-empty directories.

[jabortell]

@rlittle08, Fyi @ea-mtenhoor commented in the Jira about wanting to discuss the scope of the design further.

[jabortell]

Pushed a change to scan for empty files, via the meeting/notes from the Jira. I also updated the directory scanner to be a loop, so that it handles nested empty directories.

[tomreitz]

I tested this in TX – some observations and feedback:

1. A naming question: this dag is called RotateAirflowLogs but that's not actually what it does. Might a better name be PruneEmptyAirflowLogs?

2. To avoid import errors, I had to add the following line to ea_airflow_util/ea_airflow_util/__init__.py:

   fromea_airflow_util.dags.rotate_airflow_logsimportRotateLogsDAG

3. This was my final DAG code:

   fromea_airflow_utilimportRotateLogsDAGfromutilimportio_helpersconfigs_dir='/home/airflow/airflow/configs'airflow_configs_file="airflow_config.yml"AIRFLOW_CONFIGS=io_helpers.safe_load_yaml(configs_dir, airflow_configs_file)
   dag_args=AIRFLOW_CONFIGS.get("rotate_logs_dag", {})
   rotate_logs_dag=RotateLogsDAG(dag_id="rotate_airflow_logs", **dag_args)
   globals()[rotate_logs_dag.dag.dag_id] =rotate_logs_dag.dag

   It could be helpful to add a full example like this to the docs.

4. At least in TX, the environment variable AIRFLOW_HOME is not defined. (Based on here, I don't think it would be defined in any implementation.) This means the rotate_logs_dag.logs_dir YAML config is not optional (though the comments imply it is); I had to specify it as logs_dir: /home/airflow/airflow/logs, and even then there's still a DAG-parse key error on os.environ["AIRFLOW_HOME"]here. I'd recommend changing to something like os.environ.get("AIRFLOW_HOME", "/home/airflow/airflow/logs").

5. When I got the DAG to run, it successfully deleted 74,821 empty files/folders in 61s, freeing up ~300MB of space on disk. It logged every (nested) file/folder that's deleted, so the task's logs were 159k lines and 33MB. Have we considered checking if a folder contains any non-empty files and only recursing in if so? (otherwise, just delete the entire folder) I suspect such an approach could run faster, and produce more concise logs.

   defprune_empty(folder=logs_diroros.environ.get("AIRFLOW_HOME", "/home/airflow/airflow/logs")):
   folder_path=Path(folder)
   forsubfolderinfolder_path.iterdir():
   ifsubfolder.is_dir():
   total_size=sum(f.stat().st_sizeforfinsubfolder.rglob('*') iff.is_file())
   iftotal_size>0: prune_empty(subfolder) # (recurse)else:
   logging.info(f"Deleting folder {subfolder} (contains no non-empty files)")
   shutil.rmtree(subfolder)

[jabortell]

Links:

• Design: STAD-198 Add log rotation to Airflow logs
• Jira: STAD-198 Add log rotation to Airflow logs

Testing instructions:

In a stadium or stadium-like project, import the new DAG and instantiate it like:

fromea_airflow_utilimportRotateLogsDAGglobals()[rotate_logs_dag.dag.dag_id] =rotate_logs_dag.dag

This DAG support configuration via airflow_config.yml under the key rotate_logs_dag. For example, if your logs root directory is located someplace besides $AIRFLOW_HOME/logs, you can configure it like:

rotate_logs_dag:
schedule_interval: nulldefault_args: *default_task_argslogs_dir: /some/other/airflow/logs

then update your instantiation code to load the config:

fromutilimportio_helpers# Or wherever your config is:airflow_configs_file="/home/airflow/airflow/configs/airflow_config.yml"AIRFLOW_CONFIGS=io_helpers.safe_load_yaml(configs_dir, airflow_configs_file)
dag_args=AIRFLOW_CONFIGS.get("rotate_logs_dag", {})
rotate_logs_dag=RotateLogsDAG(dag_id="rotate_airflow_logs", **dag_args)
globals()[rotate_logs_dag.dag.dag_id] =rotate_logs_dag.dag

Finally, when you run the DAG, confirm that it runs successfully, that empty sub-directories in the logging directory are deleted, and non-empty directories are not deleted. The DAG uses os.rmdir(), which will throw an exception if called on a non-empty directory, so there should not be a risk of deleting non-empty directories.

[jabortell]

@rlittle08, Fyi @ea-mtenhoor commented in the Jira about wanting to discuss the scope of the design further.

[jabortell]

Pushed a change to scan for empty files, via the meeting/notes from the Jira. I also updated the directory scanner to be a loop, so that it handles nested empty directories.

[tomreitz]

I tested this in TX – some observations and feedback:

1. A naming question: this dag is called RotateAirflowLogs but that's not actually what it does. Might a better name be PruneEmptyAirflowLogs?

2. To avoid import errors, I had to add the following line to ea_airflow_util/ea_airflow_util/__init__.py:

   fromea_airflow_util.dags.rotate_airflow_logsimportRotateLogsDAG

3. This was my final DAG code:

   fromea_airflow_utilimportRotateLogsDAGfromutilimportio_helpersconfigs_dir='/home/airflow/airflow/configs'airflow_configs_file="airflow_config.yml"AIRFLOW_CONFIGS=io_helpers.safe_load_yaml(configs_dir, airflow_configs_file)
   dag_args=AIRFLOW_CONFIGS.get("rotate_logs_dag", {})
   rotate_logs_dag=RotateLogsDAG(dag_id="rotate_airflow_logs", **dag_args)
   globals()[rotate_logs_dag.dag.dag_id] =rotate_logs_dag.dag

   It could be helpful to add a full example like this to the docs.

4. At least in TX, the environment variable AIRFLOW_HOME is not defined. (Based on here, I don't think it would be defined in any implementation.) This means the rotate_logs_dag.logs_dir YAML config is not optional (though the comments imply it is); I had to specify it as logs_dir: /home/airflow/airflow/logs, and even then there's still a DAG-parse key error on os.environ["AIRFLOW_HOME"]here. I'd recommend changing to something like os.environ.get("AIRFLOW_HOME", "/home/airflow/airflow/logs").

5. When I got the DAG to run, it successfully deleted 74,821 empty files/folders in 61s, freeing up ~300MB of space on disk. It logged every (nested) file/folder that's deleted, so the task's logs were 159k lines and 33MB. Have we considered checking if a folder contains any non-empty files and only recursing in if so? (otherwise, just delete the entire folder) I suspect such an approach could run faster, and produce more concise logs.

   defprune_empty(folder=logs_diroros.environ.get("AIRFLOW_HOME", "/home/airflow/airflow/logs")):
   folder_path=Path(folder)
   forsubfolderinfolder_path.iterdir():
   ifsubfolder.is_dir():
   total_size=sum(f.stat().st_sizeforfinsubfolder.rglob('*') iff.is_file())
   iftotal_size>0: prune_empty(subfolder) # (recurse)else:
   logging.info(f"Deleting folder {subfolder} (contains no non-empty files)")
   shutil.rmtree(subfolder)

[jabortell]

Links:

• Design: STAD-198 Add log rotation to Airflow logs
• Jira: STAD-198 Add log rotation to Airflow logs

Testing instructions:

In a stadium or stadium-like project, import the new DAG and instantiate it like:

fromea_airflow_utilimportRotateLogsDAGglobals()[rotate_logs_dag.dag.dag_id] =rotate_logs_dag.dag

This DAG support configuration via airflow_config.yml under the key rotate_logs_dag. For example, if your logs root directory is located someplace besides $AIRFLOW_HOME/logs, you can configure it like:

rotate_logs_dag:
schedule_interval: nulldefault_args: *default_task_argslogs_dir: /some/other/airflow/logs

then update your instantiation code to load the config:

fromutilimportio_helpers# Or wherever your config is:airflow_configs_file="/home/airflow/airflow/configs/airflow_config.yml"AIRFLOW_CONFIGS=io_helpers.safe_load_yaml(configs_dir, airflow_configs_file)
dag_args=AIRFLOW_CONFIGS.get("rotate_logs_dag", {})
rotate_logs_dag=RotateLogsDAG(dag_id="rotate_airflow_logs", **dag_args)
globals()[rotate_logs_dag.dag.dag_id] =rotate_logs_dag.dag

Finally, when you run the DAG, confirm that it runs successfully, that empty sub-directories in the logging directory are deleted, and non-empty directories are not deleted. The DAG uses os.rmdir(), which will throw an exception if called on a non-empty directory, so there should not be a risk of deleting non-empty directories.

[jabortell]

@rlittle08, Fyi @ea-mtenhoor commented in the Jira about wanting to discuss the scope of the design further.

[jabortell]

Pushed a change to scan for empty files, via the meeting/notes from the Jira. I also updated the directory scanner to be a loop, so that it handles nested empty directories.

[tomreitz]

I tested this in TX – some observations and feedback:

1. A naming question: this dag is called RotateAirflowLogs but that's not actually what it does. Might a better name be PruneEmptyAirflowLogs?

2. To avoid import errors, I had to add the following line to ea_airflow_util/ea_airflow_util/__init__.py:

   fromea_airflow_util.dags.rotate_airflow_logsimportRotateLogsDAG

3. This was my final DAG code:

   fromea_airflow_utilimportRotateLogsDAGfromutilimportio_helpersconfigs_dir='/home/airflow/airflow/configs'airflow_configs_file="airflow_config.yml"AIRFLOW_CONFIGS=io_helpers.safe_load_yaml(configs_dir, airflow_configs_file)
   dag_args=AIRFLOW_CONFIGS.get("rotate_logs_dag", {})
   rotate_logs_dag=RotateLogsDAG(dag_id="rotate_airflow_logs", **dag_args)
   globals()[rotate_logs_dag.dag.dag_id] =rotate_logs_dag.dag

   It could be helpful to add a full example like this to the docs.

4. At least in TX, the environment variable AIRFLOW_HOME is not defined. (Based on here, I don't think it would be defined in any implementation.) This means the rotate_logs_dag.logs_dir YAML config is not optional (though the comments imply it is); I had to specify it as logs_dir: /home/airflow/airflow/logs, and even then there's still a DAG-parse key error on os.environ["AIRFLOW_HOME"]here. I'd recommend changing to something like os.environ.get("AIRFLOW_HOME", "/home/airflow/airflow/logs").

5. When I got the DAG to run, it successfully deleted 74,821 empty files/folders in 61s, freeing up ~300MB of space on disk. It logged every (nested) file/folder that's deleted, so the task's logs were 159k lines and 33MB. Have we considered checking if a folder contains any non-empty files and only recursing in if so? (otherwise, just delete the entire folder) I suspect such an approach could run faster, and produce more concise logs.

   defprune_empty(folder=logs_diroros.environ.get("AIRFLOW_HOME", "/home/airflow/airflow/logs")):
   folder_path=Path(folder)
   forsubfolderinfolder_path.iterdir():
   ifsubfolder.is_dir():
   total_size=sum(f.stat().st_sizeforfinsubfolder.rglob('*') iff.is_file())
   iftotal_size>0: prune_empty(subfolder) # (recurse)else:
   logging.info(f"Deleting folder {subfolder} (contains no non-empty files)")
   shutil.rmtree(subfolder)