Repository: apache/airflow
Pull request: Advises the kernel to not cache log files generated by Airflow
Extends the standard python logging.FileHandler with advise to the
Kernel to not cache the file in PageCache when it is written. While
there is nothing wrong with such cache (it will be cleaned when memory
is needed), it causes ever-growing memory usage when scheduler is
running as it keeps on writing new log files and the files are not
rotated later on. This might lead to confusion for our users, who are
monitoring memory usage of Scheduler - without realising that it is
harmless and expected in this case.

Adding the advice to Kernel might help with not generating the cache
memory growth in the first place.

Closes: #14924

<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.

Feel free to ping committers for the review!

In case of existing issue, reference it using one of the following:

closes: #ISSUE
related: #ISSUE

How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->

---
**^ Add meaningful description above**

Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#pull-request-guidelines)** for more information.
In case of fundamental code change, Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvements+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in [UPDATING.md](https://github.com/apache/airflow/blob/main/UPDATING.md).


Problem statement:
Scheduler Memory Leak in Airflow 2.0.1
**Apache Airflow version**: 2.0.1

**Kubernetes version (if you are using kubernetes)** (use `kubectl version`): v1.17.4

**Environment**: Dev
- **OS** (e.g. from /etc/os-release): RHEL7

**What happened**:

After running fine for some time my airflow tasks got stuck in scheduled state with below error in Task Instance Details:
"All dependencies are met but the task instance is not running. In most cases this just means that the task will probably be scheduled soon unless: - The scheduler is down or under heavy load If this task instance does not start soon please contact your Airflow administrator for assistance."


**What you expected to happen**:

I restarted the scheduler then it started working fine. When i checked my metrics i realized the scheduler has a memory leak and over past 4 days it has reached up to 6GB of memory utilization

In version >2.0 we don't even have the run_duration config option to restart scheduler periodically to avoid this issue until it is resolved.

**How to reproduce it**:
I saw this issue in multiple dev instances of mine all running Airflow 2.0.1 on kubernetes with KubernetesExecutor. 
Below are the configs that i changed from the default config.
max_active_dag_runs_per_dag=32
parallelism=64
dag_concurrency=32
sql_Alchemy_pool_size=50
sql_Alchemy_max_overflow=30

**Anything else we need to know**:

The scheduler memory leaks occurs consistently in all instances i have been running. The memory utilization keeps growing for scheduler.

Hints from issue discussion:
Thanks for opening your first issue here! Be sure to follow the issue template!

i have same problem.  i looked for all the channels and methods but did not solve it！
Thanks for your question and let me know what the problem is.
I have to use crontab script to restart scheduler process regularly.  But this is stupid.
If  can't solve it, I can only fall back to 1.10.

This issue is also there in version 1.10.* however in version 2.0.* the issue is more severe and also we don't have option of run_duration hence have to deploy our own cron jobs to refresh scheduler regularly.
https://issues.apache.org/jira/browse/AIRFLOW-4593
Until there is a fix or you find a specific reason, you could handle the OOM -> https://github.com/kubernetes/kubernetes/issues/40157

There is few initiatives :+1: 
Thanks for yours comments.
It seems that the memory leak problem has not been solved well.
I can only call cron script to restart process every hour.
It seems like this is specific to the Kubernetes executor? It’d be awesome if you can confirm.
not completely.I use `CeleryExecutor`  in the project also have this problem
After schedule, VSZ and RSS  keep growing  always.
One observation I have is that the rate of memory leak increases with number of dags (irrespective of whether they are being run). It definitely has something to do with the dag parsing process.
yep.In my production environment, when using a small number of jobs, no problems are found temporarily.
but as the business is connected, the increase in the amount of work will cause the schedule process  die.
guys. i also have some other questions.
In my project, use `BashOperator` task while get  exception occasionally- 'Bash command failed. The command returned a non-zero exit code.'  But it will succeed after retrying. 
 No value in log file.I donot konw what reason is. If you have experience please let me know, thank you very much.

Commit message: Advises the kernel to not cache log files generated by Airflow

Extends the standard python logging.FileHandler with advise to the
Kernel to not cache the file in PageCache when it is written. While
there is nothing wrong with such cache (it will be cleaned when memory
is needed), it causes ever-growing memory usage when scheduler is
running as it keeps on writing new log files and the files are not
rotated later on. This might lead to confusion for our users, who are
monitoring memory usage of Scheduler - without realising that it is
harmless and expected in this case.

Adding the advice to Kernel might help with not generating the cache
memory growth in the first place.

Closes: #14924

Patch to review:
diff --git a/airflow/utils/cli.py b/airflow/utils/cli.py
index acdb2910da78f..64156765930b6 100644
--- a/airflow/utils/cli.py
+++ b/airflow/utils/cli.py
@@ -34,6 +34,7 @@
 from airflow import settings
 from airflow.exceptions import AirflowException
 from airflow.utils import cli_action_loggers
+from airflow.utils.log.non_caching_file_handler import NonCachingFileHandler
 from airflow.utils.platform import getuser, is_terminal_support_colors
 from airflow.utils.session import provide_session
 
@@ -244,7 +245,7 @@ def setup_locations(process, pid=None, stdout=None, stderr=None, log=None):
 def setup_logging(filename):
     """Creates log file handler for daemon process"""
     root = logging.getLogger()
-    handler = logging.FileHandler(filename)
+    handler = NonCachingFileHandler(filename)
     formatter = logging.Formatter(settings.SIMPLE_LOG_FORMAT)
     handler.setFormatter(formatter)
     root.addHandler(handler)
diff --git a/airflow/utils/log/file_processor_handler.py b/airflow/utils/log/file_processor_handler.py
index 7792faf48eed4..e1ce4918b7ea9 100644
--- a/airflow/utils/log/file_processor_handler.py
+++ b/airflow/utils/log/file_processor_handler.py
@@ -23,6 +23,7 @@
 
 from airflow import settings
 from airflow.utils.helpers import parse_template_string
+from airflow.utils.log.non_caching_file_handler import NonCachingFileHandler
 
 
 class FileProcessorHandler(logging.Handler):
@@ -54,7 +55,7 @@ def set_context(self, filename):
         :param filename: filename in which the dag is located
         """
         local_loc = self._init_file(filename)
-        self.handler = logging.FileHandler(local_loc)
+        self.handler = NonCachingFileHandler(local_loc)
         self.handler.setFormatter(self.formatter)
         self.handler.setLevel(self.level)
 
diff --git a/airflow/utils/log/file_task_handler.py b/airflow/utils/log/file_task_handler.py
index 0a1d4334489f0..e4531035d3f15 100644
--- a/airflow/utils/log/file_task_handler.py
+++ b/airflow/utils/log/file_task_handler.py
@@ -26,6 +26,7 @@
 
 from airflow.configuration import AirflowConfigException, conf
 from airflow.utils.helpers import parse_template_string
+from airflow.utils.log.non_caching_file_handler import NonCachingFileHandler
 
 if TYPE_CHECKING:
     from airflow.models import TaskInstance
@@ -55,7 +56,7 @@ def set_context(self, ti: "TaskInstance"):
         :param ti: task instance object
         """
         local_loc = self._init_file(ti)
-        self.handler = logging.FileHandler(local_loc, encoding='utf-8')
+        self.handler = NonCachingFileHandler(local_loc, encoding='utf-8')
         if self.formatter:
             self.handler.setFormatter(self.formatter)
         self.handler.setLevel(self.level)
diff --git a/airflow/utils/log/non_caching_file_handler.py b/airflow/utils/log/non_caching_file_handler.py
new file mode 100644
index 0000000000000..3228ddaeb6c52
--- /dev/null
+++ b/airflow/utils/log/non_caching_file_handler.py
@@ -0,0 +1,46 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import logging
+import os
+
+
+class NonCachingFileHandler(logging.FileHandler):
+    """
+    This is an extension of the python FileHandler that advises the Kernel to not cache the file
+    in PageCache when it is written. While there is nothing wrong with such cache (it will be cleaned
+    when memory is needed), it causes ever-growing memory usage when scheduler is running as it keeps
+    on writing new log files and the files are not rotated later on. This might lead to confusion
+    for our users, who are monitoring memory usage of Scheduler - without realising that it is
+    harmless and expected in this case.
+
+    See https://github.com/apache/airflow/issues/14924
+
+    Adding the advice to Kernel might help with not generating the cache memory growth in the first place.
+    """
+
+    def _open(self):
+        wrapper = logging.FileHandler._open(self)
+        try:
+            fd = wrapper.fileno()
+            os.posix_fadvise(fd, 0, 0, os.POSIX_FADV_DONTNEED)
+        except Exception:
+            # in case either file descriptor cannot be retrieved or fadvise is not available
+            # we should simply return the wrapper retrieved by FileHandler's open method
+            # the advise to the kernel is just an advise and if we cannot give it, we won't
+            pass
+        return wrapper


Code review comments:
diff_hunk: @@ -0,0 +1,46 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import logging
+import os
+
+
+class NonCachingFileHandler(logging.FileHandler):
+    """
+    This is an extension of the python FileHandler that advises the Kernel to not cache the file
+    in PageCache when it is written. While there is nothing wrong with such cache (it will be cleaned
+    when memory is needed), it causes ever-growing memory usage when scheduler is running as it keeps
+    on writing new log files and the files are not rotated later on. This might lead to confusion
+    for our users, who are monitoring memory usage of Scheduler - without realising that it is
+    harmless and expected in this case.
+
+    See https://github.com/apache/airflow/issues/14924
+
+    Adding the advice to Kernel might help with not generating the cache memory growth in the first place.
+    """
+
+    def _open(self):
+        wrapper = logging.FileHandler._open(self)
path: airflow/utils/log/non_caching_file_handler.py
text: @user1:
```suggestion
        wrapper = super()._open()
```