Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions composer/tools/README.md
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
# Composer Tools

* [Composer DAGs Pausing/Unpausing script](composer_dags.md)
* [google-api-core Python 3.11 Deprecation Workaround](google_api_core_workaround/README.md)

86 changes: 86 additions & 0 deletions composer/tools/google_api_core_workaround/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,86 @@
# Workaround for google-api-core Python 3.11 Deprecation Warning

This tool provides a workaround for Cloud Composer environments running Python 3.11 where `google-api-core` triggers high DAG parsing times, elevated worker CPU usage, or deprecation log spam.

## When to Use This

Use this workaround if:
- Your Cloud Composer environment is running on Python 3.11.
- You see `FutureWarning` logs from `google.api_core._python_version_support` indicating that Python 3.11 will reach end of life.
- You notice increased DAG parsing latency or worker CPU usage after entering the 365-day Python 3.11 EOL window (October 2026).

### Root Cause
`google-api-core` emits a deprecation warning starting 365 days before a Python version reaches community EOL. While emitting this warning, it calls `importlib.metadata.packages_distributions()`, which scans all installed packages across `site-packages` on disk. In Cloud Composer, where DAG processors and workers frequently spawn fresh Python subprocesses, repeated filesystem metadata scans cause measurable parse latency and CPU overhead.

## What the Workaround Does

`sitecustomize.py` is an import hook that CPython runs automatically on startup before user code executes. This script:
1. Changes `DEPRECATION_WARNING_PERIOD` from 365 days to 60 days, deferring the warning until August 2027.
2. Stubs `_get_pypi_package_name` to return `None`, preventing `packages_distributions()` from scanning disk.
3. Cleans up its own folder from `sys.path` immediately after load so subsequent imports do not re-scan the patch directory.

Because the file contains no DAG objects and avoids DAG-related keywords, Airflow's default `dag_discovery_safe_mode = True` ignores it during DAG parsing.

## How to Apply

### Step 1: Upload `sitecustomize.py` to GCS

Copy `sitecustomize.py` to a `patches/` directory in your environment's DAGs bucket:

```bash
gcloud storage cp sitecustomize.py gs://<YOUR_DAGS_BUCKET>/dags/patches/sitecustomize.py
```

You can find your DAGs bucket name by running:
```bash
gcloud composer environments describe <ENVIRONMENT_NAME> \
--location <LOCATION> \
--format="value(config.dagGcsPrefix)"
```

### Step 2: Set the `PYTHONPATH` Environment Variable

Add the patch folder to `PYTHONPATH` on your Composer environment:

```bash
gcloud composer environments update <ENVIRONMENT_NAME> \
--location <LOCATION> \
--update-env-variables PYTHONPATH=/home/airflow/gcs/dags/patches
```

> **Important:** If your environment already defines a custom `PYTHONPATH`, preserve your existing paths by separating them with a colon (`:`):
> ```bash
> --update-env-variables PYTHONPATH=/home/airflow/gcs/dags/patches:<EXISTING_PATH>
> ```

This update triggers a rolling restart of Composer workloads. Once restarted, all Python subprocesses (scheduler, workers, DAG processor) will load `sitecustomize.py` on startup.

### Note on `dag_discovery_safe_mode`

If you explicitly disabled safe mode (`[core] dag_discovery_safe_mode = False`), add `patches/*` to your `dags/.airflowignore` file to prevent the DAG file processor from attempting to parse the patch directory:

```text
patches/*
```

## Verification

After the environment finishes updating:
1. Check Cloud Logging or task logs in the Airflow UI. The `_python_version_support` `FutureWarning` messages should stop appearing.
2. In the Cloud Monitoring dashboard for Composer, check the **DAG Processing: Total Parse Time** metric. Parse times should return to baseline levels.

## How to Remove

Once your environment is upgraded to a Composer image or `google-api-core` version with the upstream fix, remove the workaround:

1. Remove `PYTHONPATH` (or restore your original value):
```bash
gcloud composer environments update <ENVIRONMENT_NAME> \
--location <LOCATION> \
--remove-env-variables PYTHONPATH
```

2. Delete the patch file from the bucket:
```bash
gcloud storage rm gs://<YOUR_DAGS_BUCKET>/dags/patches/sitecustomize.py
```
91 changes: 91 additions & 0 deletions composer/tools/google_api_core_workaround/sitecustomize.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,91 @@
# Copyright 2026 Google LLC
#
# Licensed 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.

"""CPython sitecustomize hook for patching google.api_core deprecation check."""

import datetime
import importlib.util
import os
import sys

# Remove this patch directory from sys.path to avoid redundant lookups
# on subsequent module imports when loaded via PYTHONPATH.
_patch_file = globals().get("__file__")
if _patch_file:
_patch_dir = os.path.abspath(os.path.dirname(_patch_file))
_is_site_dir = any(
part in ("site-packages", "dist-packages")
for part in _patch_dir.split(os.sep)
)
if not _is_site_dir:
_patch_dir_slash = _patch_dir + os.sep
sys.path[:] = [
p
for p in sys.path
if not isinstance(p, (str, os.PathLike))
or not os.path.isabs(p)
or (
os.path.abspath(p) != _patch_dir
and os.path.abspath(p) != _patch_dir_slash
)
]
Comment on lines +33 to +42

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

When modifying sys.path to remove the patch directory, we should be defensive against non-string elements (which can occasionally be present in sys.path depending on the environment/test runners) and avoid accidentally removing relative paths (like "" or ".") if the current working directory happens to match the patch directory. Checking os.path.isabs(p) ensures we only remove absolute paths.

Suggested change
sys.path[:] = [
p
for p in sys.path
if os.path.abspath(p) != _patch_dir
and os.path.abspath(p) != _patch_dir_slash
]
sys.path[:] = [
p
for p in sys.path
if not isinstance(p, (str, os.PathLike))
or not os.path.isabs(p)
or (os.path.abspath(p) != _patch_dir and os.path.abspath(p) != _patch_dir_slash)
]

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

addressed.

sys.path_importer_cache.pop(_patch_dir, None)
sys.path_importer_cache.pop(_patch_dir_slash, None)

TARGET_MODULE = "google.api_core._python_version_support"


class _GoogleApiCorePatcher:
"""Import hook to patch DEPRECATION_WARNING_PERIOD on load."""

def find_spec(self, fullname, path=None, target=None):
if fullname == TARGET_MODULE:
try:
sys.meta_path.remove(self)
except ValueError:
pass

try:
spec = importlib.util.find_spec(fullname)
except Exception:
return None

if spec and spec.loader:
orig_exec_module = spec.loader.exec_module

def exec_module(module):
import importlib.metadata

orig_packages_distributions = getattr(
importlib.metadata, "packages_distributions", None
)
if orig_packages_distributions:
importlib.metadata.packages_distributions = lambda: {}
try:
orig_exec_module(module)
finally:
if orig_packages_distributions:
importlib.metadata.packages_distributions = (
orig_packages_distributions
)
module.DEPRECATION_WARNING_PERIOD = datetime.timedelta(days=60)
module._get_pypi_package_name = lambda module_name: None

spec.loader.exec_module = exec_module
return spec
return None
Comment on lines +52 to +87

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

To make the import hook extremely robust, we should handle potential exceptions from importlib.util.find_spec (e.g., if parent packages are missing or there are unexpected import errors) by returning None to let the default import system handle it. Additionally, if the deprecation check in google.api_core._python_version_support is executed at module load time (during orig_exec_module), modifying module.DEPRECATION_WARNING_PERIOD afterwards is too late to prevent the disk scan and log spam. Temporarily patching importlib.metadata.packages_distributions during the execution of the module ensures the disk scan is bypassed in all scenarios.

    def find_spec(self, fullname, path=None, target=None):
        if fullname == TARGET_MODULE:
            try:
                sys.meta_path.remove(self)
            except ValueError:
                pass

            try:
                spec = importlib.util.find_spec(fullname)
            except Exception:
                return None

            if spec and spec.loader:
                orig_exec_module = spec.loader.exec_module

                def exec_module(module):
                    import importlib.metadata
                    orig_packages_distributions = getattr(importlib.metadata, "packages_distributions", None)
                    if orig_packages_distributions:
                        importlib.metadata.packages_distributions = lambda: {}
                    try:
                        orig_exec_module(module)
                    finally:
                        if orig_packages_distributions:
                            importlib.metadata.packages_distributions = orig_packages_distributions
                    module.DEPRECATION_WARNING_PERIOD = datetime.timedelta(days=60)
                    module._get_pypi_package_name = lambda module_name: None

                spec.loader.exec_module = exec_module
            return spec
        return None

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

addressed.



if not any(type(h).__name__ == "_GoogleApiCorePatcher" for h in sys.meta_path):
sys.meta_path.insert(0, _GoogleApiCorePatcher())
Loading