Fix ownership for mounted volume for dag processing logs (#431)

*Description of changes:*
Change in entrypoint.py

Adding the mount point to the Airflow container caused an error when
deploying, which is needed to expose the log files to the FluentBit
container.

This changes resolves this, changing the ownership of the directory to
the 'airflow' user , when the director is mounted.

Added unit tests to match this change.

Edit 03/18:

Moved the fix to execute earlier in entrypoint.py. This is due to AF 2.x
interacting with the mounted folder earlier than in 3.x , causing the
permission error earlier in the deployment

Also added this change to AF 2.X since NCL may be enabled , which means
this fix is required

---------

Co-authored-by: Sergio AnguloIbarra <aglser@amazon.com>

Author: SergioAAI

images/airflow/2.10.1/python/mwaa/entrypoint.py

@@ -6,11 +6,36 @@
 after setting up the necessary configurations.
 """
 
+# Fix shared log volume permissions before any Airflow imports, since Airflow's
+# logging config tries to create subdirectories under /usr/local/airflow/logs
+# at import time.
+# ruff: noqa: E402
+import os
+import subprocess
+
+def _fix_shared_log_volume_permissions():
+    """
+    When /usr/local/airflow/logs is backed by a shared volume (e.g. for Fluent Bit
+    to tail log files), Docker creates it owned by root. Fix ownership so the
+    airflow user can write logs.
+    """
+    log_dir = "/usr/local/airflow/logs"
+    if os.path.ismount(log_dir):
+        try:
+            subprocess.run(
+                ["sudo", "chown", "-R", "airflow:", log_dir],
+                check=True,
+            )
+            print(f"Fixed ownership of shared log volume {log_dir}")
+        except Exception as e:
+            print(f"WARNING: Could not fix ownership of {log_dir}: {e}")
+
+_fix_shared_log_volume_permissions()
+
 # Setup logging first thing to make sure all logs happen under the right setup. The
 # reason for needing this is that typically a `logger` object is defined at the top
 # of the module and is used through out it. So, if we import a module before logging
 # is setup, its `logger` object will not have the right setup.
-# ruff: noqa: E402
 # fmt: off
 from airflow.config_templates.airflow_local_settings import DEFAULT_LOGGING_CONFIG
 import logging.config
@@ -20,7 +45,6 @@
 # Python imports
 import asyncio
 import logging
-import os
 import sys
 import time
 

images/airflow/2.10.3/python/mwaa/entrypoint.py

@@ -6,11 +6,36 @@
 after setting up the necessary configurations.
 """
 
+# Fix shared log volume permissions before any Airflow imports, since Airflow's
+# logging config tries to create subdirectories under /usr/local/airflow/logs
+# at import time.
+# ruff: noqa: E402
+import os
+import subprocess
+
+def _fix_shared_log_volume_permissions():
+    """
+    When /usr/local/airflow/logs is backed by a shared volume (e.g. for Fluent Bit
+    to tail log files), Docker creates it owned by root. Fix ownership so the
+    airflow user can write logs.
+    """
+    log_dir = "/usr/local/airflow/logs"
+    if os.path.ismount(log_dir):
+        try:
+            subprocess.run(
+                ["sudo", "chown", "-R", "airflow:", log_dir],
+                check=True,
+            )
+            print(f"Fixed ownership of shared log volume {log_dir}")
+        except Exception as e:
+            print(f"WARNING: Could not fix ownership of {log_dir}: {e}")
+
+_fix_shared_log_volume_permissions()
+
 # Setup logging first thing to make sure all logs happen under the right setup. The
 # reason for needing this is that typically a `logger` object is defined at the top
 # of the module and is used through out it. So, if we import a module before logging
 # is setup, its `logger` object will not have the right setup.
-# ruff: noqa: E402
 # fmt: off
 from airflow.config_templates.airflow_local_settings import DEFAULT_LOGGING_CONFIG
 import logging.config
@@ -20,7 +45,6 @@
 # Python imports
 import asyncio
 import logging
-import os
 import sys
 import time
 

images/airflow/2.11.0/python/mwaa/entrypoint.py

@@ -6,11 +6,36 @@
 after setting up the necessary configurations.
 """
 
+# Fix shared log volume permissions before any Airflow imports, since Airflow's
+# logging config tries to create subdirectories under /usr/local/airflow/logs
+# at import time.
+# ruff: noqa: E402
+import os
+import subprocess
+
+def _fix_shared_log_volume_permissions():
+    """
+    When /usr/local/airflow/logs is backed by a shared volume (e.g. for Fluent Bit
+    to tail log files), Docker creates it owned by root. Fix ownership so the
+    airflow user can write logs.
+    """
+    log_dir = "/usr/local/airflow/logs"
+    if os.path.ismount(log_dir):
+        try:
+            subprocess.run(
+                ["sudo", "chown", "-R", "airflow:", log_dir],
+                check=True,
+            )
+            print(f"Fixed ownership of shared log volume {log_dir}")
+        except Exception as e:
+            print(f"WARNING: Could not fix ownership of {log_dir}: {e}")
+
+_fix_shared_log_volume_permissions()
+
 # Setup logging first thing to make sure all logs happen under the right setup. The
 # reason for needing this is that typically a `logger` object is defined at the top
 # of the module and is used through out it. So, if we import a module before logging
 # is setup, its `logger` object will not have the right setup.
-# ruff: noqa: E402
 # fmt: off
 from airflow.config_templates.airflow_local_settings import DEFAULT_LOGGING_CONFIG
 import logging.config
@@ -20,7 +45,6 @@
 # Python imports
 import asyncio
 import logging
-import os
 import sys
 import time
 
@@ -244,7 +268,6 @@ async def main() -> None:
 
     execute_command(command, environ, CONTAINER_START_TIME)
 
-
 if __name__ == "__main__":
     try:
         asyncio.run(main())

images/airflow/2.9.2/python/mwaa/entrypoint.py

@@ -6,11 +6,36 @@
 after setting up the necessary configurations.
 """
 
+# Fix shared log volume permissions before any Airflow imports, since Airflow's
+# logging config tries to create subdirectories under /usr/local/airflow/logs
+# at import time.
+# ruff: noqa: E402
+import os
+import subprocess
+
+def _fix_shared_log_volume_permissions():
+    """
+    When /usr/local/airflow/logs is backed by a shared volume (e.g. for Fluent Bit
+    to tail log files), Docker creates it owned by root. Fix ownership so the
+    airflow user can write logs.
+    """
+    log_dir = "/usr/local/airflow/logs"
+    if os.path.ismount(log_dir):
+        try:
+            subprocess.run(
+                ["sudo", "chown", "-R", "airflow:", log_dir],
+                check=True,
+            )
+            print(f"Fixed ownership of shared log volume {log_dir}")
+        except Exception as e:
+            print(f"WARNING: Could not fix ownership of {log_dir}: {e}")
+
+_fix_shared_log_volume_permissions()
+
 # Setup logging first thing to make sure all logs happen under the right setup. The
 # reason for needing this is that typically a `logger` object is defined at the top
 # of the module and is used through out it. So, if we import a module before logging
 # is setup, its `logger` object will not have the right setup.
-# ruff: noqa: E402
 # fmt: off
 from airflow.config_templates.airflow_local_settings import DEFAULT_LOGGING_CONFIG
 import logging.config
@@ -21,7 +46,6 @@
 from datetime import datetime
 import asyncio
 import logging
-import os
 import sys
 import time
 

images/airflow/3.0.6/python/mwaa/entrypoint.py

@@ -6,11 +6,36 @@
 after setting up the necessary configurations.
 """
 
+# Fix shared log volume permissions before any Airflow imports, since Airflow's
+# logging config tries to create subdirectories under /usr/local/airflow/logs
+# at import time.
+# ruff: noqa: E402
+import os
+import subprocess
+
+def _fix_shared_log_volume_permissions():
+    """
+    When /usr/local/airflow/logs is backed by a shared volume (e.g. for Fluent Bit
+    to tail log files), Docker creates it owned by root. Fix ownership so the
+    airflow user can write logs.
+    """
+    log_dir = "/usr/local/airflow/logs"
+    if os.path.ismount(log_dir):
+        try:
+            subprocess.run(
+                ["sudo", "chown", "-R", "airflow:", log_dir],
+                check=True,
+            )
+            print(f"Fixed ownership of shared log volume {log_dir}")
+        except Exception as e:
+            print(f"WARNING: Could not fix ownership of {log_dir}: {e}")
+
+_fix_shared_log_volume_permissions()
+
 # Setup logging first thing to make sure all logs happen under the right setup. The
 # reason for needing this is that typically a `logger` object is defined at the top
 # of the module and is used through out it. So, if we import a module before logging
 # is setup, its `logger` object will not have the right setup.
-# ruff: noqa: E402
 # fmt: off
 from airflow.config_templates.airflow_local_settings import DEFAULT_LOGGING_CONFIG
 import logging.config
@@ -20,7 +45,6 @@
 # Python imports
 import asyncio
 import logging
-import os
 import sys
 import time
 

images/airflow/3.1.6/python/mwaa/entrypoint.py

@@ -6,11 +6,36 @@
 after setting up the necessary configurations.
 """
 
+# Fix shared log volume permissions before any Airflow imports, since Airflow's
+# logging config tries to create subdirectories under /usr/local/airflow/logs
+# at import time.
+# ruff: noqa: E402
+import os
+import subprocess
+
+def _fix_shared_log_volume_permissions():
+    """
+    When /usr/local/airflow/logs is backed by a shared volume (e.g. for Fluent Bit
+    to tail log files), Docker creates it owned by root. Fix ownership so the
+    airflow user can write logs.
+    """
+    log_dir = "/usr/local/airflow/logs"
+    if os.path.ismount(log_dir):
+        try:
+            subprocess.run(
+                ["sudo", "chown", "-R", "airflow:", log_dir],
+                check=True,
+            )
+            print(f"Fixed ownership of shared log volume {log_dir}")
+        except Exception as e:
+            print(f"WARNING: Could not fix ownership of {log_dir}: {e}")
+
+_fix_shared_log_volume_permissions()
+
 # Setup logging first thing to make sure all logs happen under the right setup. The
 # reason for needing this is that typically a `logger` object is defined at the top
 # of the module and is used through out it. So, if we import a module before logging
 # is setup, its `logger` object will not have the right setup.
-# ruff: noqa: E402
 # fmt: off
 from airflow.config_templates.airflow_local_settings import DEFAULT_LOGGING_CONFIG
 import logging.config
@@ -20,7 +45,6 @@
 # Python imports
 import asyncio
 import logging
-import os
 import sys
 import time
 

tests/images/airflow/2.10.1/python/mwaa/test_entrypoint_2_10_1.py

@@ -261,3 +261,34 @@ def test_mark_as_unhealthy_error(mock_environ):
 
         # Verify sleep was still called
         mock_sleep.assert_called_once_with(1100)
+
+
+def test_fix_shared_log_volume_permissions_on_mount(mock_environ):
+    """Test _fix_shared_log_volume_permissions runs chown when path is a mount point."""
+    with patch('os.path.ismount', return_value=True) as mock_ismount, \
+         patch('subprocess.run') as mock_run:
+        entrypoint._fix_shared_log_volume_permissions()
+
+        mock_ismount.assert_called_once_with('/usr/local/airflow/logs')
+        mock_run.assert_called_once_with(
+            ['sudo', 'chown', '-R', 'airflow:', '/usr/local/airflow/logs'],
+            check=True,
+        )
+
+
+def test_fix_shared_log_volume_permissions_not_mount(mock_environ):
+    """Test _fix_shared_log_volume_permissions skips chown when path is not a mount."""
+    with patch('os.path.ismount', return_value=False) as mock_ismount, \
+         patch('subprocess.run') as mock_run:
+        entrypoint._fix_shared_log_volume_permissions()
+
+        mock_ismount.assert_called_once_with('/usr/local/airflow/logs')
+        mock_run.assert_not_called()
+
+
+def test_fix_shared_log_volume_permissions_chown_failure(mock_environ):
+    """Test _fix_shared_log_volume_permissions handles chown failure gracefully."""
+    with patch('os.path.ismount', return_value=True), \
+         patch('subprocess.run', side_effect=Exception('Permission denied')):
+        # Should not raise — just prints a warning
+        entrypoint._fix_shared_log_volume_permissions()

tests/images/airflow/2.10.3/python/mwaa/test_entrypoint_2_10_3.py

@@ -261,3 +261,34 @@ def test_mark_as_unhealthy_error(mock_environ):
 
         # Verify sleep was still called
         mock_sleep.assert_called_once_with(1100)
+
+
+def test_fix_shared_log_volume_permissions_on_mount(mock_environ):
+    """Test _fix_shared_log_volume_permissions runs chown when path is a mount point."""
+    with patch('os.path.ismount', return_value=True) as mock_ismount, \
+         patch('subprocess.run') as mock_run:
+        entrypoint._fix_shared_log_volume_permissions()
+
+        mock_ismount.assert_called_once_with('/usr/local/airflow/logs')
+        mock_run.assert_called_once_with(
+            ['sudo', 'chown', '-R', 'airflow:', '/usr/local/airflow/logs'],
+            check=True,
+        )
+
+
+def test_fix_shared_log_volume_permissions_not_mount(mock_environ):
+    """Test _fix_shared_log_volume_permissions skips chown when path is not a mount."""
+    with patch('os.path.ismount', return_value=False) as mock_ismount, \
+         patch('subprocess.run') as mock_run:
+        entrypoint._fix_shared_log_volume_permissions()
+
+        mock_ismount.assert_called_once_with('/usr/local/airflow/logs')
+        mock_run.assert_not_called()
+
+
+def test_fix_shared_log_volume_permissions_chown_failure(mock_environ):
+    """Test _fix_shared_log_volume_permissions handles chown failure gracefully."""
+    with patch('os.path.ismount', return_value=True), \
+         patch('subprocess.run', side_effect=Exception('Permission denied')):
+        # Should not raise — just prints a warning
+        entrypoint._fix_shared_log_volume_permissions()

tests/images/airflow/2.11.0/python/mwaa/test_entrypoint_2_11_0.py

@@ -261,3 +261,34 @@ def test_mark_as_unhealthy_error(mock_environ):
 
         # Verify sleep was still called
         mock_sleep.assert_called_once_with(1100)
+
+
+def test_fix_shared_log_volume_permissions_on_mount(mock_environ):
+    """Test _fix_shared_log_volume_permissions runs chown when path is a mount point."""
+    with patch('os.path.ismount', return_value=True) as mock_ismount, \
+         patch('subprocess.run') as mock_run:
+        entrypoint._fix_shared_log_volume_permissions()
+
+        mock_ismount.assert_called_once_with('/usr/local/airflow/logs')
+        mock_run.assert_called_once_with(
+            ['sudo', 'chown', '-R', 'airflow:', '/usr/local/airflow/logs'],
+            check=True,
+        )
+
+
+def test_fix_shared_log_volume_permissions_not_mount(mock_environ):
+    """Test _fix_shared_log_volume_permissions skips chown when path is not a mount."""
+    with patch('os.path.ismount', return_value=False) as mock_ismount, \
+         patch('subprocess.run') as mock_run:
+        entrypoint._fix_shared_log_volume_permissions()
+
+        mock_ismount.assert_called_once_with('/usr/local/airflow/logs')
+        mock_run.assert_not_called()
+
+
+def test_fix_shared_log_volume_permissions_chown_failure(mock_environ):
+    """Test _fix_shared_log_volume_permissions handles chown failure gracefully."""
+    with patch('os.path.ismount', return_value=True), \
+         patch('subprocess.run', side_effect=Exception('Permission denied')):
+        # Should not raise — just prints a warning
+        entrypoint._fix_shared_log_volume_permissions()