diff --git a/composer/tools/README.md b/composer/tools/README.md index ad686ca7c8b..2ee8aecf589 100644 --- a/composer/tools/README.md +++ b/composer/tools/README.md @@ -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) + diff --git a/composer/tools/google_api_core_workaround/README.md b/composer/tools/google_api_core_workaround/README.md new file mode 100644 index 00000000000..7083e15e865 --- /dev/null +++ b/composer/tools/google_api_core_workaround/README.md @@ -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:///dags/patches/sitecustomize.py +``` + +You can find your DAGs bucket name by running: +```bash +gcloud composer environments describe \ + --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 \ + --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: +> ``` + +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 \ + --location \ + --remove-env-variables PYTHONPATH + ``` + +2. Delete the patch file from the bucket: + ```bash + gcloud storage rm gs:///dags/patches/sitecustomize.py + ``` diff --git a/composer/tools/google_api_core_workaround/sitecustomize.py b/composer/tools/google_api_core_workaround/sitecustomize.py new file mode 100644 index 00000000000..ad4dfbc2a6a --- /dev/null +++ b/composer/tools/google_api_core_workaround/sitecustomize.py @@ -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 + ) + ] + 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 + + +if not any(type(h).__name__ == "_GoogleApiCorePatcher" for h in sys.meta_path): + sys.meta_path.insert(0, _GoogleApiCorePatcher())