Does anyone know why Airflow create repeating log on the scheduler log? #33714

adjinugroho · 2023-08-25T03:59:53Z

adjinugroho
Aug 25, 2023

Hi,

As the discussion title say, does anyone here know why that happen?

Here is snapshot of my scheduler log:

[2023-08-24T00:00:01.469+0000] {processor.py:836} INFO - DAG(s) dict_keys(['RKPMSBI_Update_Cube']) retrieved from /opt/airflow/dags/rkpmsbi_update_cube.py
[2023-08-24T00:00:01.493+0000] {logging_mixin.py:149} INFO - [2023-08-24T00:00:01.493+0000] {dag.py:2726} INFO - Sync 1 DAGs
[2023-08-24T00:00:01.518+0000] {logging_mixin.py:149} INFO - [2023-08-24T00:00:01.518+0000] {dag.py:3486} INFO - Setting next_dagrun for RKPMSBI_Update_Cube to 2023-08-23T00:00:01.518198+00:00, run_after=2023-08-24T00:00:01.518198+00:00
[2023-08-24T00:00:01.545+0000] {processor.py:179} INFO - Processing /opt/airflow/dags/rkpmsbi_update_cube.py took 0.965 seconds
[2023-08-24T00:00:01.610+0000] {processor.py:157} INFO - Started process (PID=43667) to work on /opt/airflow/dags/rkpmsbi_update_cube.py
[2023-08-24T00:00:01.612+0000] {processor.py:826} INFO - Processing file /opt/airflow/dags/rkpmsbi_update_cube.py for tasks to queue
[2023-08-24T00:00:01.614+0000] {logging_mixin.py:149} INFO - [2023-08-24T00:00:01.613+0000] {dagbag.py:541} INFO - Filling up the DagBag from /opt/airflow/dags/rkpmsbi_update_cube.py
[2023-08-24T00:00:01.843+0000] {logging_mixin.py:149} INFO - [2023-08-24T00:00:01.843+0000] {base.py:73} INFO - Using connection ID 'mssql_rkpmsbi' for task execution.

[2023-08-24T00:00:02.440+0000] {processor.py:836} INFO - DAG(s) dict_keys(['RKPMSBI_Update_Cube']) retrieved from /opt/airflow/dags/rkpmsbi_update_cube.py
[2023-08-24T00:00:02.463+0000] {logging_mixin.py:149} INFO - [2023-08-24T00:00:02.462+0000] {dag.py:2726} INFO - Sync 1 DAGs
[2023-08-24T00:00:02.486+0000] {logging_mixin.py:149} INFO - [2023-08-24T00:00:02.486+0000] {dag.py:3486} INFO - Setting next_dagrun for RKPMSBI_Update_Cube to 2023-08-23T00:00:02.486019+00:00, run_after=2023-08-24T00:00:02.486019+00:00
[2023-08-24T00:00:02.510+0000] {processor.py:179} INFO - Processing /opt/airflow/dags/rkpmsbi_update_cube.py took 0.903 seconds
[2023-08-24T00:00:03.151+0000] {processor.py:157} INFO - Started process (PID=43691) to work on /opt/airflow/dags/rkpmsbi_update_cube.py
[2023-08-24T00:00:03.153+0000] {processor.py:826} INFO - Processing file /opt/airflow/dags/rkpmsbi_update_cube.py for tasks to queue
[2023-08-24T00:00:03.154+0000] {logging_mixin.py:149} INFO - [2023-08-24T00:00:03.154+0000] {dagbag.py:541} INFO - Filling up the DagBag from /opt/airflow/dags/rkpmsbi_update_cube.py
[2023-08-24T00:00:03.372+0000] {logging_mixin.py:149} INFO - [2023-08-24T00:00:03.371+0000] {base.py:73} INFO - Using connection ID 'mssql_rkpmsbi' for task execution.

[2023-08-24T00:00:03.970+0000] {processor.py:836} INFO - DAG(s) dict_keys(['RKPMSBI_Update_Cube']) retrieved from /opt/airflow/dags/rkpmsbi_update_cube.py
[2023-08-24T00:00:03.991+0000] {logging_mixin.py:149} INFO - [2023-08-24T00:00:03.991+0000] {dag.py:2726} INFO - Sync 1 DAGs
[2023-08-24T00:00:04.015+0000] {logging_mixin.py:149} INFO - [2023-08-24T00:00:04.014+0000] {dag.py:3486} INFO - Setting next_dagrun for RKPMSBI_Update_Cube to 2023-08-23T00:00:04.014620+00:00, run_after=2023-08-24T00:00:04.014620+00:00
[2023-08-24T00:00:04.040+0000] {processor.py:179} INFO - Processing /opt/airflow/dags/rkpmsbi_update_cube.py took 0.892 seconds
[2023-08-24T00:00:04.098+0000] {processor.py:157} INFO - Started process (PID=43709) to work on /opt/airflow/dags/rkpmsbi_update_cube.py
[2023-08-24T00:00:04.099+0000] {processor.py:826} INFO - Processing file /opt/airflow/dags/rkpmsbi_update_cube.py for tasks to queue
[2023-08-24T00:00:04.101+0000] {logging_mixin.py:149} INFO - [2023-08-24T00:00:04.101+0000] {dagbag.py:541} INFO - Filling up the DagBag from /opt/airflow/dags/rkpmsbi_update_cube.py
[2023-08-24T00:00:04.268+0000] {logging_mixin.py:149} INFO - [2023-08-24T00:00:04.268+0000] {base.py:73} INFO - Using connection ID 'mssql_rkpmsbi' for task execution.

[2023-08-24T00:00:04.871+0000] {processor.py:836} INFO - DAG(s) dict_keys(['RKPMSBI_Update_Cube']) retrieved from /opt/airflow/dags/rkpmsbi_update_cube.py
[2023-08-24T00:00:04.893+0000] {logging_mixin.py:149} INFO - [2023-08-24T00:00:04.892+0000] {dag.py:2726} INFO - Sync 1 DAGs
[2023-08-24T00:00:04.914+0000] {logging_mixin.py:149} INFO - [2023-08-24T00:00:04.914+0000] {dag.py:3486} INFO - Setting next_dagrun for RKPMSBI_Update_Cube to 2023-08-23T00:00:04.914111+00:00, run_after=2023-08-24T00:00:04.914111+00:00
[2023-08-24T00:00:04.940+0000] {processor.py:179} INFO - Processing /opt/airflow/dags/rkpmsbi_update_cube.py took 0.848 seconds
[2023-08-24T00:00:04.991+0000] {processor.py:157} INFO - Started process (PID=43724) to work on /opt/airflow/dags/rkpmsbi_update_cube.py
[2023-08-24T00:00:04.992+0000] {processor.py:826} INFO - Processing file /opt/airflow/dags/rkpmsbi_update_cube.py for tasks to queue
[2023-08-24T00:00:04.993+0000] {logging_mixin.py:149} INFO - [2023-08-24T00:00:04.993+0000] {dagbag.py:541} INFO - Filling up the DagBag from /opt/airflow/dags/rkpmsbi_update_cube.py
[2023-08-24T00:00:05.178+0000] {logging_mixin.py:149} INFO - [2023-08-24T00:00:05.178+0000] {base.py:73} INFO - Using connection ID 'mssql_rkpmsbi' for task execution.
[2023-08-24T00:00:05.779+0000] {processor.py:836} INFO - DAG(s) dict_keys(['RKPMSBI_Update_Cube']) retrieved from /opt/airflow/dags/rkpmsbi_update_cube.py

As you can see on the snapshot above, Airflow keeps repeating 8 lines logs. This is happening on scheduler log, for each of DAG I have.

Here is the example the DAG I have:

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

from rea.common.airflow import AirflowSQLConn
from rea.service.rkpmsbi_update_cube.db_logship import LogShipping
from rea.service.rkpmsbi_update_cube.job_rainydays import RainyDays

default_args = {
    "depends_on_past": True,
    "owner": "REA - Dm Team",
    "start_date": datetime(2023, 1, 1),
    "retries": 1,
    "retry_delay": timedelta(minutes=5),
}

with DAG(
    dag_id="RKPMSBI_Update_Cube",
    default_args=default_args,
    schedule_interval=timedelta(days=1),
    catchup=False,
) as dag:
    rkpmsConn = AirflowSQLConn.RKPMSBI()
    logShip = LogShipping(rkpmsConn)
    rainDays = RainyDays(rkpmsConn)

    task1 = PythonOperator(
        task_id="DB_StopLogShip",
        python_callable=logShip.Disable,
    )

    task2 = PythonOperator(
        task_id="DB_RestoreLog",
        python_callable=logShip.RestoreLog,
    )
   
   ... And so on with task similar like the task above. Just PythonOperator calling another python methods.

At the moment, each log produce ~5mb/DAG/daily. And IMO, 5mb for log file is too much. Especially if the cause is repeating line like that.

I know I can stop the log with change the AIRFLOW__LOGGING__LOGGING_LEVEL to WARNING or above level. But I need to keep logging level on INFO.

Does anyone know how to adjust this?

Answered by Taragolis

Aug 25, 2023

As the discussion title say, does anyone here know why that happen?

Because scheduler/dag_processor continuously parse DAG from .py files every min_file_process_interval (30 sec by default)

Does anyone know how to adjust this?

I would start with this one
-Fine-tuning your Scheduler performance
-Advanced Logging

At the moment, each log produce ~5mb/DAG/daily. And IMO, 5mb for log file is too much.

I would start worrying on this point, that might refers to some issues with you're DAG code, so better the check that your DAG follow Best Practices

That how you logs should look like, please attention on: time to process, logs frequencies and next_dagrun

[2023-08-23T00:00:33.588+0000] {pro…

View full answer

Taragolis · 2023-08-25T07:38:12Z

Taragolis
Aug 25, 2023
Collaborator

As the discussion title say, does anyone here know why that happen?

Because scheduler/dag_processor continuously parse DAG from .py files every min_file_process_interval (30 sec by default)

Does anyone know how to adjust this?

I would start with this one
-Fine-tuning your Scheduler performance
-Advanced Logging

At the moment, each log produce ~5mb/DAG/daily. And IMO, 5mb for log file is too much.

I would start worrying on this point, that might refers to some issues with you're DAG code, so better the check that your DAG follow Best Practices

That how you logs should look like, please attention on: time to process, logs frequencies and next_dagrun

[2023-08-23T00:00:33.588+0000] {processor.py:157} INFO - Started process (PID=57137) to work on /files/dags/ecs_sample.py
[2023-08-23T00:00:33.589+0000] {processor.py:820} INFO - Processing file /files/dags/ecs_sample.py for tasks to queue
[2023-08-23T00:00:33.590+0000] {logging_mixin.py:151} INFO - [2023-08-23T00:00:33.589+0000] {dagbag.py:539} INFO - Filling up the DagBag from /files/dags/ecs_sample.py
[2023-08-23T00:00:33.594+0000] {processor.py:830} INFO - DAG(s) dict_keys(['example_ecs']) retrieved from /files/dags/ecs_sample.py
[2023-08-23T00:00:33.608+0000] {logging_mixin.py:151} INFO - [2023-08-23T00:00:33.608+0000] {dag.py:2906} INFO - Sync 1 DAGs
[2023-08-23T00:00:33.619+0000] {logging_mixin.py:151} INFO - [2023-08-23T00:00:33.619+0000] {dag.py:3687} INFO - Setting next_dagrun for example_ecs to 2021-01-01T00:00:00+00:00, run_after=2021-01-01T00:00:00+00:00
[2023-08-23T00:00:33.629+0000] {processor.py:179} INFO - Processing /files/dags/ecs_sample.py took 0.044 seconds

[2023-08-23T00:01:04.029+0000] {processor.py:157} INFO - Started process (PID=57502) to work on /files/dags/ecs_sample.py
[2023-08-23T00:01:04.032+0000] {processor.py:820} INFO - Processing file /files/dags/ecs_sample.py for tasks to queue
[2023-08-23T00:01:04.033+0000] {logging_mixin.py:151} INFO - [2023-08-23T00:01:04.033+0000] {dagbag.py:539} INFO - Filling up the DagBag from /files/dags/ecs_sample.py
[2023-08-23T00:01:04.042+0000] {processor.py:830} INFO - DAG(s) dict_keys(['example_ecs']) retrieved from /files/dags/ecs_sample.py
[2023-08-23T00:01:04.058+0000] {logging_mixin.py:151} INFO - [2023-08-23T00:01:04.058+0000] {dag.py:2906} INFO - Sync 1 DAGs
[2023-08-23T00:01:04.070+0000] {logging_mixin.py:151} INFO - [2023-08-23T00:01:04.069+0000] {dag.py:3687} INFO - Setting next_dagrun for example_ecs to 2021-01-01T00:00:00+00:00, run_after=2021-01-01T00:00:00+00:00
[2023-08-23T00:01:04.080+0000] {processor.py:179} INFO - Processing /files/dags/ecs_sample.py took 0.053 seconds

1 reply

adjinugroho Aug 25, 2023
Author

Thank you for your link about Best Practices for building DAG code. That's help me a lot!

From that link, I think I can guess what's wrong with my code is. The top-level import on the Python code.

I am new on Python (just starting on the last month), and I am coming from C# developer. That's why that's become a habit of mine to use top level statement like using on C#.

I still need to re-apply it first though to see what will change.

potiuk · 2023-08-25T07:43:48Z

potiuk
Aug 25, 2023
Collaborator

Yes. Scheduler is writing logs for every action it performs, and it scans and re-parses (actually dag file processor does it) the files in your DAG folder regularly - you can see all the parameters of scheduler in https://airflow.apache.org/docs/apache-airflow/stable/configurations-ref.html#scheduler but the one that controls the frequency is https://airflow.apache.org/docs/apache-airflow/stable/configurations-ref.html#min-file-process-interval

There multiple reasons why just "scanning once" is not enough:

the DAG gets re-parsed regularly to account for Python code that might dynamically change the result of parsing (for example if you DAG code produced a dynamic DAG, reading "dynamic" data from a file, parsing the DAG might produce a different structure or parameters of the DAG object based on the changes in the file. Or your DAG structure might be time-depending etc.
DAG file Processor also uses the opportunity that the DAG is parsed to run certain callbacks. There are cases where callbacks need to be run (for example when your task was forcefully killed and had no time to run the callback, DAGFile processor will run the "task_failed" callback instead. And it needs to have the parsed DAG in memory for that.

Regarding the log entries - Airflow provides a basic configuration of loggers and handlers following the standar Python logging framework and assumes that it's up to your deployment to do somothing with the logs. For example, in our chart there is a log rotation script that simply cleans up old logs. This is pretty standard approach of modern applications that they just produce the logs and whatever happens with the logs is up to Deployment that does something with it. Generally Airflow does not make opinionated decisiosn on wheter to avoid similar duplicated lines or not it just logs whatever happens. And leaves the decision what to do with it for you.

So you can do whatever you want there and you have two choices:

You can post-process the logs (i.e. rotate, archive etc. etc.) - this is the default assumption
Or you can configure your loggers in the way to make something more complex - you can do whatever you want with your loggers - airflow uses standard Python logging framework and you can for example write or use some handler that keeps the logs in memory and removes duplicates before writing (and replaces it with ... and 123 times similar log lines). That is an example of course - but you are pretty free to configure the loggers as you want: https://airflow.apache.org/docs/apache-airflow/stable/administration-and-deployment/logging-monitoring/advanced-logging-configuration.html#advanced-logging-configuration

1 reply

adjinugroho Aug 25, 2023
Author

I see.

I know the scans progress will generate a log. But I think that log is dag_processor_manager.log. I don't think the scan process will also have impact on the scheduler log.

Yes, I understand why the logs need to be treated based on personal (or team) choice. And I am happy with that because we can treat that aspect like our needs.

But still, the repeating line code on the log file make me curious why that's happening.

potiuk · 2023-08-25T07:44:11Z

potiuk
Aug 25, 2023
Collaborator

I see our posts with @Taragolis crossed - and we gave you similar answers :)

1 reply

potiuk Aug 25, 2023
Collaborator

Mine is a bit longer though :)

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Does anyone know why Airflow create repeating log on the scheduler log? #33714

{{title}}

Replies: 3 comments 3 replies

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

Select a reply

Does anyone know why Airflow create repeating log on the scheduler log? #33714

adjinugroho Aug 25, 2023

Replies: 3 comments · 3 replies

Taragolis Aug 25, 2023 Collaborator

adjinugroho Aug 25, 2023 Author

potiuk Aug 25, 2023 Collaborator

adjinugroho Aug 25, 2023 Author

potiuk Aug 25, 2023 Collaborator

potiuk Aug 25, 2023 Collaborator

adjinugroho
Aug 25, 2023

Replies: 3 comments 3 replies

Taragolis
Aug 25, 2023
Collaborator

adjinugroho Aug 25, 2023
Author

potiuk
Aug 25, 2023
Collaborator

adjinugroho Aug 25, 2023
Author

potiuk
Aug 25, 2023
Collaborator

potiuk Aug 25, 2023
Collaborator