This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 958737c6128 Rename Google Stackdriver task log handler to Cloud
Logging (#73575)
958737c6128 is described below
commit 958737c6128d72ca42dfa7756777cadcabef3b08
Author: Shahar Epstein <[email protected]>
AuthorDate: Wed Sep 23 17:46:33 2026 +0300
Rename Google Stackdriver task log handler to Cloud Logging (#73575)
Google retired the "Stackdriver" brand years ago. This handler only calls
the Cloud Logging API v2 (google.cloud.logging, LoggingServiceV2Client),
distinct from the Cloud Monitoring API v3 that the hook, operators and
links call, which were already renamed to CloudMonitoring* in a prior
change. Rename this handler to match, for the same reason.
StackdriverRemoteLogIO and StackdriverTaskHandler remain importable as
deprecated aliases of CloudLoggingRemoteLogIO and CloudLoggingTaskHandler
until March 31, 2027, per the provider's deprecation policy. The
stackdriver:// remote-logging scheme is unchanged, so existing
remote_base_log_folder configuration keeps working without edits.
Core's airflow_local_settings.py still imports and instantiates the
now-deprecated StackdriverRemoteLogIO for that scheme, so a released core
paired with this provider emits one deprecation warning at startup until a
follow-up core change switches to CloudLoggingRemoteLogIO.
Part of the rename tracked in #61009.
---
.../logging/{stackdriver.rst => cloud_logging.rst} | 37 +-
providers/google/docs/redirects.txt | 1 +
providers/google/provider.yaml | 4 +-
...sk_handler.py => cloud_logging_task_handler.py} | 38 +-
.../google/cloud/log/stackdriver_task_handler.py | 579 +-------------
.../airflow/providers/google/get_provider_info.py | 4 +-
...ndler.py => test_cloud_logging_task_handler.py} | 290 ++++---
.../cloud/log/test_stackdriver_task_handler.py | 890 +--------------------
8 files changed, 290 insertions(+), 1553 deletions(-)
diff --git a/providers/google/docs/logging/stackdriver.rst
b/providers/google/docs/logging/cloud_logging.rst
similarity index 73%
rename from providers/google/docs/logging/stackdriver.rst
rename to providers/google/docs/logging/cloud_logging.rst
index 5a7634fc008..ada32adaef5 100644
--- a/providers/google/docs/logging/stackdriver.rst
+++ b/providers/google/docs/logging/cloud_logging.rst
@@ -17,10 +17,19 @@
.. _write-logs-stackdriver:
-Writing logs to Google Stackdriver
-----------------------------------
+.. _write-logs-cloud-logging:
-Airflow can be configured to read and write task logs in `Google Stackdriver
Logging <https://cloud.google.com/logging/>`__.
+Writing logs to Google Cloud Logging
+------------------------------------
+
+.. note::
+
+ The ``Stackdriver*`` classes in
``airflow.providers.google.cloud.log.stackdriver_task_handler`` are
+ deprecated aliases for the ``CloudLogging*`` classes documented below, and
will be removed after
+ March 31, 2027.
+
+Airflow can be configured to read and write task logs in `Google Cloud Logging
<https://cloud.google.com/logging/>`__
+(formerly Stackdriver).
To enable this feature, ``airflow.cfg`` must be configured as in this
example:
@@ -39,19 +48,19 @@ All configuration options are in the ``[logging]`` section.
#. By default Application Default Credentials are used to obtain credentials.
You can also
set ``google_key_path`` option in ``[logging]`` section, if you want to use
your own service account.
-#. Make sure with those credentials, you can read/write to/from stackdriver.
+#. Make sure with those credentials, you can read/write to/from Cloud Logging.
#. Install the ``google`` package, like so: ``pip install
'apache-airflow[google]'``.
#. Restart the Airflow webserver and scheduler, and trigger (or wait for) a
new task execution.
#. Verify that logs are showing up for newly executed tasks in the bucket you
have defined.
-#. Verify that the Google Cloud Storage viewer is working in the UI. With
Stackdriver you should see the logs pulled in the real time
+#. Verify that the Google Cloud Storage viewer is working in the UI. With
Cloud Logging you should see the logs pulled in the real time
The value of field ``remote_logging`` must always be set to ``True`` for this
feature to work.
-Turning this option off will result in data not being sent to Stackdriver.
+Turning this option off will result in data not being sent to Cloud Logging.
The ``remote_base_log_folder`` option contains the URL that specifies the type
of handler to be used.
-For integration with Stackdriver, this option should start with
``stackdriver://``.
-The path section of the URL specifies the name of the log e.g.
``stackdriver:///airflow-tasks`` writes
-logs under the name ``airflow-tasks``.
+For integration with Cloud Logging, this option should start with
``stackdriver://`` — the URL scheme
+name is unchanged by this rename. The path section of the URL specifies the
name of the log e.g.
+``stackdriver:///airflow-tasks`` writes logs under the name ``airflow-tasks``.
You can set ``google_key_path`` option in the ``[logging]`` section to specify
the path to `the service
account key file <https://cloud.google.com/iam/docs/service-accounts>`__.
@@ -69,12 +78,14 @@ be used. Make sure that with those credentials, you can
read and write the logs.
By using the ``logging_config_class`` option you can get
:ref:`advanced features <write-logs-advanced>` of
this handler. Details are available in the handler's documentation -
-:class:`~airflow.providers.google.cloud.log.stackdriver_task_handler.StackdriverTaskHandler`.
+:class:`~airflow.providers.google.cloud.log.cloud_logging_task_handler.CloudLoggingTaskHandler`.
.. _log-link-stackdriver:
-Google Stackdriver External Link
-''''''''''''''''''''''''''''''''
+.. _log-link-cloud-logging:
+
+Google Cloud Logging External Link
+''''''''''''''''''''''''''''''''''
-Airflow automatically shows a link to Google Stackdriver when configured to
use it as the remote logging system.
+Airflow automatically shows a link to Google Cloud Logging when configured to
use it as the remote logging system.
diff --git a/providers/google/docs/redirects.txt
b/providers/google/docs/redirects.txt
index 1e5b1be42a8..86b148b8899 100644
--- a/providers/google/docs/redirects.txt
+++ b/providers/google/docs/redirects.txt
@@ -18,3 +18,4 @@
api-auth-backend/index.rst api-auth-backend/google-openid.rst
secrets-backends/index.rst
secrets-backends/google-cloud-secret-manager-backend.rst
operators/cloud/stackdriver.rst operators/cloud/cloud_monitoring.rst
+logging/stackdriver.rst logging/cloud_logging.rst
diff --git a/providers/google/provider.yaml b/providers/google/provider.yaml
index 93f9621f7ac..9dcdd3be9f3 100644
--- a/providers/google/provider.yaml
+++ b/providers/google/provider.yaml
@@ -1516,12 +1516,12 @@ auth-backends:
logging:
- airflow.providers.google.cloud.log.gcs_task_handler.GCSTaskHandler
- -
airflow.providers.google.cloud.log.stackdriver_task_handler.StackdriverTaskHandler
+ -
airflow.providers.google.cloud.log.cloud_logging_task_handler.CloudLoggingTaskHandler
remote-logging:
- classpath:
airflow.providers.google.cloud.log.gcs_task_handler.GCSRemoteLogIO
scheme: gs
- - classpath:
airflow.providers.google.cloud.log.stackdriver_task_handler.StackdriverRemoteLogIO
+ - classpath:
airflow.providers.google.cloud.log.cloud_logging_task_handler.CloudLoggingRemoteLogIO
scheme: stackdriver
queues:
diff --git
a/providers/google/src/airflow/providers/google/cloud/log/stackdriver_task_handler.py
b/providers/google/src/airflow/providers/google/cloud/log/cloud_logging_task_handler.py
similarity index 94%
copy from
providers/google/src/airflow/providers/google/cloud/log/stackdriver_task_handler.py
copy to
providers/google/src/airflow/providers/google/cloud/log/cloud_logging_task_handler.py
index e6cb6c7e2c2..3bc9be0274b 100644
---
a/providers/google/src/airflow/providers/google/cloud/log/stackdriver_task_handler.py
+++
b/providers/google/src/airflow/providers/google/cloud/log/cloud_logging_task_handler.py
@@ -14,7 +14,7 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-"""Handler that integrates with Stackdriver."""
+"""Handler that integrates with Google Cloud Logging."""
from __future__ import annotations
@@ -70,7 +70,7 @@ DEFAULT_LOGGER_NAME = "airflow"
_GLOBAL_RESOURCE = Resource(type="global", labels={})
# Dedicated logger for handler-internal failures (Cloud Logging unavailable,
gRPC errors).
-# Routed to the same
``airflow.providers.google.cloud.log.stackdriver_task_handler`` namespace
+# Routed to the same
``airflow.providers.google.cloud.log.cloud_logging_task_handler`` namespace
# so operators see these alongside the rest of the handler's logs.
_logger = logging.getLogger(__name__)
@@ -85,8 +85,8 @@ LABEL_TRY_NUMBER = "try_number"
@attrs.define(kw_only=True)
-class StackdriverRemoteLogIO(LoggingMixin):
- """Remote log IO that streams logs to and reads from Google Cloud
Stackdriver Logging."""
+class CloudLoggingRemoteLogIO(LoggingMixin):
+ """Remote log IO that streams logs to and reads from Google Cloud
Logging."""
base_log_folder: Path = attrs.field(converter=Path)
delete_local_copy: bool = True
@@ -99,7 +99,7 @@ class StackdriverRemoteLogIO(LoggingMixin):
labels: dict[str, str] | None = None
@classmethod
- def from_config(cls) -> StackdriverRemoteLogIO:
+ def from_config(cls) -> CloudLoggingRemoteLogIO:
"""Build the remote log IO from Airflow logging configuration."""
remote_task_handler_kwargs = conf.getjson("logging",
"remote_task_handler_kwargs", fallback={})
if not isinstance(remote_task_handler_kwargs, dict):
@@ -118,7 +118,7 @@ class StackdriverRemoteLogIO(LoggingMixin):
log_name = urlsplit(remote_base_log_folder).path[1:]
if not log_name:
raise ValueError(
- "Cannot derive a Stackdriver log name from "
+ "Cannot derive a Cloud Logging log name from "
f"logging/remote_base_log_folder: {remote_base_log_folder!r}"
)
return cls(
@@ -159,7 +159,7 @@ class StackdriverRemoteLogIO(LoggingMixin):
@cached_property
def transport(self) -> Transport:
- """Object responsible for sending data to Stackdriver."""
+ """Object responsible for sending data to Cloud Logging."""
return self.transport_type(self._client, self.gcp_log_name)
@cached_property
@@ -246,11 +246,11 @@ class StackdriverRemoteLogIO(LoggingMixin):
shutil.rmtree(parent, ignore_errors=True)
def read(self, relative_path: str, ti: RuntimeTI) -> LogResponse:
- """Read logs from Stackdriver Logging using task instance labels."""
+ """Read logs from Cloud Logging using task instance labels."""
ti_labels = _task_instance_to_labels(ti)
log_filter = self.prepare_log_filter(ti_labels)
messages, end_of_log, _ = self.read_logs(log_filter,
next_page_token=None, all_pages=True)
- return [f"Reading remote log from Stackdriver for {relative_path}"],
[messages] if messages else []
+ return [f"Reading remote log from Cloud Logging for {relative_path}"],
[messages] if messages else []
def prepare_log_filter(self, ti_labels: dict[str, str]) -> str:
def escape_label_key(key: str) -> str:
@@ -316,7 +316,7 @@ class StackdriverRemoteLogIO(LoggingMixin):
def _task_instance_to_labels(ti) -> dict[str, str]:
- """Convert a task instance to Stackdriver labels."""
+ """Convert a task instance to Cloud Logging labels."""
return {
LABEL_TASK_ID: ti.task_id,
LABEL_DAG_ID: ti.dag_id,
@@ -327,13 +327,13 @@ def _task_instance_to_labels(ti) -> dict[str, str]:
}
-class StackdriverTaskHandler(logging.Handler):
+class CloudLoggingTaskHandler(logging.Handler):
"""
- Handler that directly makes Stackdriver logging API calls.
+ Handler that directly makes Cloud Logging API calls.
This is a Python standard ``logging`` handler using that can be used to
- route Python standard logging messages directly to the Stackdriver
- Logging API.
+ route Python standard logging messages directly to the Cloud Logging
+ API.
It can also be used to save logs for executing tasks. To do this, you
should set as a handler with
the name "tasks". In this case, it will also be used to read the log for
display in Web UI.
@@ -345,7 +345,7 @@ class StackdriverTaskHandler(logging.Handler):
<https://cloud.google.com/docs/authentication/production#finding_credentials_automatically>`__
will
be used.
:param scopes: OAuth scopes for the credentials,
- :param name: the name of the custom log in Stackdriver Logging. Defaults
+ :param name: the name of the custom log in Cloud Logging. Defaults
to 'airflow'. The name of the Python logger will be represented
in the ``python_logger`` field.
:param transport: Class for creating new transport objects. It should
@@ -364,7 +364,7 @@ class StackdriverTaskHandler(logging.Handler):
LABEL_LOGICAL_DATE = LABEL_LOGICAL_DATE
LABEL_TRY_NUMBER = LABEL_TRY_NUMBER
LOG_VIEWER_BASE_URL = "https://console.cloud.{domain}/logs/viewer";
- LOG_NAME = "Google Stackdriver"
+ LOG_NAME = "Google Cloud Logging"
trigger_supported = True
trigger_should_queue = False
@@ -391,7 +391,7 @@ class StackdriverTaskHandler(logging.Handler):
gcp_log_name = str(name)
super().__init__()
- self.io = StackdriverRemoteLogIO(
+ self.io = CloudLoggingRemoteLogIO(
base_log_folder=Path("."),
gcp_key_path=gcp_key_path,
scopes=scopes,
@@ -449,7 +449,7 @@ class StackdriverTaskHandler(logging.Handler):
self, task_instance: TaskInstance, try_number: int | None = None,
metadata: dict | None = None
) -> tuple[list[tuple[tuple[str, str]]], list[dict[str, str | bool]]]:
"""
- Read logs of given task instance from Stackdriver logging.
+ Read logs of given task instance from Cloud Logging.
:param task_instance: task instance object
:param try_number: task instance try_number to read logs from. If None
@@ -560,6 +560,6 @@ class StackdriverTaskHandler(logging.Handler):
self.io.transport.flush()
except Exception as exc:
print(
- f"StackdriverTaskHandler.close: transport flush failed:
{type(exc).__name__}: {exc}",
+ f"CloudLoggingTaskHandler.close: transport flush failed:
{type(exc).__name__}: {exc}",
file=sys.stderr,
)
diff --git
a/providers/google/src/airflow/providers/google/cloud/log/stackdriver_task_handler.py
b/providers/google/src/airflow/providers/google/cloud/log/stackdriver_task_handler.py
index e6cb6c7e2c2..3ff177e06e4 100644
---
a/providers/google/src/airflow/providers/google/cloud/log/stackdriver_task_handler.py
+++
b/providers/google/src/airflow/providers/google/cloud/log/stackdriver_task_handler.py
@@ -1,3 +1,4 @@
+#
# 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
@@ -14,552 +15,46 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-"""Handler that integrates with Stackdriver."""
+"""Deprecated aliases for
:mod:`airflow.providers.google.cloud.log.cloud_logging_task_handler`."""
from __future__ import annotations
-import contextlib
-import copy
-import inspect
-import logging
-import os
-import shutil
-import sys
-import warnings
-from collections.abc import Collection
-from datetime import datetime
-from functools import cached_property
-from logging import getLogRecordFactory
-from pathlib import Path
-from typing import TYPE_CHECKING
-from urllib.parse import urlencode, urlsplit
-
-import attrs
-from google.cloud import logging as gcp_logging
-from google.cloud.logging import Resource
-from google.cloud.logging.handlers.transports import
BackgroundThreadTransport, Transport
-from google.cloud.logging_v2.services.logging_service_v2 import
LoggingServiceV2Client
-from google.cloud.logging_v2.types import ListLogEntriesRequest,
ListLogEntriesResponse
-
from airflow.exceptions import AirflowProviderDeprecationWarning
-from airflow.providers.common.compat.sdk import conf
-from airflow.providers.google.cloud.utils.credentials_provider import
get_credentials_and_project_id
-from airflow.providers.google.common.consts import CLIENT_INFO
-from airflow.providers.google.common.hooks.base_google import GoogleBaseHook
-from airflow.providers.google.version_compat import AIRFLOW_V_3_0_PLUS
-from airflow.utils.log.file_task_handler import FileTaskHandler
-from airflow.utils.log.logging_mixin import LoggingMixin
-
-try:
- from airflow.sdk.definitions._internal.types import NOTSET, ArgNotSet
-except ImportError:
- from airflow.utils.types import NOTSET, ArgNotSet # type:
ignore[attr-defined,no-redef]
-
-if not AIRFLOW_V_3_0_PLUS:
- from airflow.utils.log.trigger_handler import ctx_indiv_trigger
-
-if TYPE_CHECKING:
- import structlog.typing
- from google.auth.credentials import Credentials
-
- from airflow.models import TaskInstance
- from airflow.sdk.types import RuntimeTaskInstanceProtocol as RuntimeTI
- from airflow.utils.log.file_task_handler import LogResponse
-
-DEFAULT_LOGGER_NAME = "airflow"
-_GLOBAL_RESOURCE = Resource(type="global", labels={})
-
-# Dedicated logger for handler-internal failures (Cloud Logging unavailable,
gRPC errors).
-# Routed to the same
``airflow.providers.google.cloud.log.stackdriver_task_handler`` namespace
-# so operators see these alongside the rest of the handler's logs.
-_logger = logging.getLogger(__name__)
-
-_DEFAULT_SCOPESS = frozenset(
- ["https://www.googleapis.com/auth/logging.read";,
"https://www.googleapis.com/auth/logging.write";]
+from airflow.providers.google.cloud.log.cloud_logging_task_handler import (
+ DEFAULT_LOGGER_NAME,
+ LABEL_DAG_ID,
+ LABEL_LOGICAL_DATE,
+ LABEL_TASK_ID,
+ LABEL_TRY_NUMBER,
+ CloudLoggingRemoteLogIO,
+ CloudLoggingTaskHandler,
)
+from airflow.providers.google.common.deprecated import deprecated
+
+__all__ = [
+ "DEFAULT_LOGGER_NAME",
+ "LABEL_DAG_ID",
+ "LABEL_LOGICAL_DATE",
+ "LABEL_TASK_ID",
+ "LABEL_TRY_NUMBER",
+ "StackdriverRemoteLogIO",
+ "StackdriverTaskHandler",
+]
+
+
+@deprecated(
+ planned_removal_date="March 31, 2027",
+
use_instead="airflow.providers.google.cloud.log.cloud_logging_task_handler.CloudLoggingRemoteLogIO",
+ category=AirflowProviderDeprecationWarning,
+)
+class StackdriverRemoteLogIO(CloudLoggingRemoteLogIO):
+ """Deprecated. Use :class:`CloudLoggingRemoteLogIO`."""
-LABEL_TASK_ID = "task_id"
-LABEL_DAG_ID = "dag_id"
-LABEL_LOGICAL_DATE = "logical_date" if AIRFLOW_V_3_0_PLUS else "execution_date"
-LABEL_TRY_NUMBER = "try_number"
-
-
[email protected](kw_only=True)
-class StackdriverRemoteLogIO(LoggingMixin):
- """Remote log IO that streams logs to and reads from Google Cloud
Stackdriver Logging."""
-
- base_log_folder: Path = attrs.field(converter=Path)
- delete_local_copy: bool = True
-
- gcp_key_path: str | None = None
- scopes: Collection[str] | None = _DEFAULT_SCOPESS
- gcp_log_name: str = DEFAULT_LOGGER_NAME
- transport_type: type[Transport] = BackgroundThreadTransport
- resource: Resource = _GLOBAL_RESOURCE
- labels: dict[str, str] | None = None
-
- @classmethod
- def from_config(cls) -> StackdriverRemoteLogIO:
- """Build the remote log IO from Airflow logging configuration."""
- remote_task_handler_kwargs = conf.getjson("logging",
"remote_task_handler_kwargs", fallback={})
- if not isinstance(remote_task_handler_kwargs, dict):
- raise ValueError(
- "logging/remote_task_handler_kwargs must be a JSON object (a
python dict), we got "
- f"{type(remote_task_handler_kwargs)}"
- )
- # remote_task_handler_kwargs mixes FileTaskHandler kwargs with IO
kwargs; only the
- # latter belong to this class (same split as
airflow_local_settings.py).
- fth_params =
frozenset(inspect.signature(FileTaskHandler.__init__).parameters) - {
- "self",
- "base_log_folder",
- }
- io_kwargs = {k: v for k, v in remote_task_handler_kwargs.items() if k
not in fth_params}
- remote_base_log_folder = conf.get_mandatory_value("logging",
"remote_base_log_folder")
- log_name = urlsplit(remote_base_log_folder).path[1:]
- if not log_name:
- raise ValueError(
- "Cannot derive a Stackdriver log name from "
- f"logging/remote_base_log_folder: {remote_base_log_folder!r}"
- )
- return cls(
- **{
- "base_log_folder":
os.path.expanduser(conf.get_mandatory_value("logging", "base_log_folder")),
- "gcp_log_name": log_name,
- "gcp_key_path": conf.get_mandatory_value("logging",
"GOOGLE_KEY_PATH", fallback=None),
- "delete_local_copy": conf.getboolean("logging",
"delete_local_logs"),
- }
- | io_kwargs,
- )
-
- @cached_property
- def credentials_and_project(self) -> tuple[Credentials, str]:
- credentials, project = get_credentials_and_project_id(
- key_path=self.gcp_key_path, scopes=self.scopes,
disable_logging=True
- )
- return credentials, project
-
- @cached_property
- def _client(self) -> gcp_logging.Client:
- """The Cloud Library API client."""
- credentials, project = self.credentials_and_project
- return gcp_logging.Client(
- credentials=credentials,
- project=project,
- client_info=CLIENT_INFO,
- )
-
- @cached_property
- def _logging_service_client(self) -> LoggingServiceV2Client:
- """The Cloud logging service v2 client."""
- credentials, _ = self.credentials_and_project
- return LoggingServiceV2Client(
- credentials=credentials,
- client_info=CLIENT_INFO,
- )
-
- @cached_property
- def transport(self) -> Transport:
- """Object responsible for sending data to Stackdriver."""
- return self.transport_type(self._client, self.gcp_log_name)
-
- @cached_property
- def processors(self) -> tuple[structlog.typing.Processor, ...]:
- import structlog.stdlib
-
- from airflow.sdk.log import relative_path_from_logger
-
- log_record_factory = getLogRecordFactory()
- _transport = self.transport
-
- def proc(
- logger: structlog.typing.WrappedLogger,
- method_name: str,
- event: structlog.typing.EventDict,
- ):
- if not logger or not relative_path_from_logger(logger):
- return event
-
- name = event.get("logger_name") or event.get("logger", "")
- level = structlog.stdlib.NAME_TO_LEVEL.get(method_name.lower(),
logging.INFO)
- msg = copy.copy(event)
- created = None
- if ts := msg.pop("timestamp", None):
- with contextlib.suppress(Exception):
- created = datetime.fromisoformat(ts)
- record = log_record_factory(
- name,
- level,
- pathname="",
- lineno=0,
- msg=msg,
- args=(),
- exc_info=None,
- func=None,
- sinfo=None,
- )
- if created is not None:
- ct = created.timestamp()
- record.created = ct
- record.msecs = int((ct - int(ct)) * 1000) + 0.0
-
- ti = getattr(record, "task_instance", None)
- labels: dict[str, str] = {}
- if self.labels:
- labels.update(self.labels)
- if ti:
- labels.update(_task_instance_to_labels(ti))
- else:
- if dag_id := event.get("dag_id"):
- labels[LABEL_DAG_ID] = str(dag_id)
- if task_id := event.get("task_id"):
- labels[LABEL_TASK_ID] = str(task_id)
- if run_id := event.get("run_id"):
- labels["run_id"] = str(run_id)
- if try_number := event.get("try_number"):
- labels[LABEL_TRY_NUMBER] = str(try_number)
- if map_index := event.get("map_index"):
- labels["map_index"] = str(map_index)
-
- _transport.send(record, str(msg.get("event", "")),
resource=self.resource, labels=labels)
- return event
-
- return (proc,)
-
- def upload(self, path: os.PathLike | str, ti: RuntimeTI | None = None) ->
None:
- """Flush the transport and optionally delete local log files."""
- self.transport.flush()
- if self.delete_local_copy:
- base = self.base_log_folder.resolve()
- raw = Path(path)
- local_path = (raw if raw.is_absolute() else base / raw).resolve()
- try:
- local_path.relative_to(base)
- except ValueError:
- self.log.warning(
- "Skipping deletion: path %s is outside base_log_folder %s",
- local_path,
- base,
- )
- return
- parent = local_path.parent
- if parent.exists():
- shutil.rmtree(parent, ignore_errors=True)
-
- def read(self, relative_path: str, ti: RuntimeTI) -> LogResponse:
- """Read logs from Stackdriver Logging using task instance labels."""
- ti_labels = _task_instance_to_labels(ti)
- log_filter = self.prepare_log_filter(ti_labels)
- messages, end_of_log, _ = self.read_logs(log_filter,
next_page_token=None, all_pages=True)
- return [f"Reading remote log from Stackdriver for {relative_path}"],
[messages] if messages else []
-
- def prepare_log_filter(self, ti_labels: dict[str, str]) -> str:
- def escape_label_key(key: str) -> str:
- return f'"{key}"' if "." in key else key
-
- def escape_label_value(value: str) -> str:
- escaped_value = value.replace("\\", "\\\\").replace('"', '\\"')
- return f'"{escaped_value}"'
-
- _, project = self.credentials_and_project
- log_filters = [
- f"resource.type={escape_label_value(self.resource.type)}",
- f'logName="projects/{project}/logs/{self.gcp_log_name}"',
- ]
-
- for key, value in self.resource.labels.items():
-
log_filters.append(f"resource.labels.{escape_label_key(key)}={escape_label_value(value)}")
-
- for key, value in ti_labels.items():
-
log_filters.append(f"labels.{escape_label_key(key)}={escape_label_value(value)}")
- return "\n".join(log_filters)
-
- def read_logs(
- self, log_filter: str, next_page_token: str | None, all_pages: bool
- ) -> tuple[str, bool, str | None]:
- messages = []
- new_messages, next_page_token = self._read_single_logs_page(
- log_filter=log_filter,
- page_token=next_page_token,
- )
- messages.append(new_messages)
- if all_pages:
- while next_page_token:
- new_messages, next_page_token = self._read_single_logs_page(
- log_filter=log_filter, page_token=next_page_token
- )
- messages.append(new_messages)
-
- end_of_log = True
- next_page_token = None
- else:
- end_of_log = not bool(next_page_token)
- return "\n".join(messages), end_of_log, next_page_token
-
- def _read_single_logs_page(self, log_filter: str, page_token: str | None =
None) -> tuple[str, str]:
- _, project = self.credentials_and_project
- request = ListLogEntriesRequest(
- resource_names=[f"projects/{project}"],
- filter=log_filter,
- page_token=page_token,
- order_by="timestamp asc",
- page_size=1000,
- )
- response =
self._logging_service_client.list_log_entries(request=request)
- page: ListLogEntriesResponse = next(response.pages)
- messages: list[str] = []
- for entry in page.entries:
- if "message" in (entry.json_payload or {}):
- messages.append(entry.json_payload["message"]) # type: ignore
- elif entry.text_payload:
- messages.append(entry.text_payload)
- return "\n".join(messages), page.next_page_token
-
-
-def _task_instance_to_labels(ti) -> dict[str, str]:
- """Convert a task instance to Stackdriver labels."""
- return {
- LABEL_TASK_ID: ti.task_id,
- LABEL_DAG_ID: ti.dag_id,
- LABEL_LOGICAL_DATE: str(ti.logical_date.isoformat())
- if AIRFLOW_V_3_0_PLUS
- else str(ti.execution_date.isoformat()),
- LABEL_TRY_NUMBER: str(ti.try_number),
- }
-
-
-class StackdriverTaskHandler(logging.Handler):
- """
- Handler that directly makes Stackdriver logging API calls.
-
- This is a Python standard ``logging`` handler using that can be used to
- route Python standard logging messages directly to the Stackdriver
- Logging API.
-
- It can also be used to save logs for executing tasks. To do this, you
should set as a handler with
- the name "tasks". In this case, it will also be used to read the log for
display in Web UI.
-
- This handler supports both an asynchronous and synchronous transport.
-
- :param gcp_key_path: Path to Google Cloud Credential JSON file.
- If omitted, authorization based on `the Application Default Credentials
-
<https://cloud.google.com/docs/authentication/production#finding_credentials_automatically>`__
will
- be used.
- :param scopes: OAuth scopes for the credentials,
- :param name: the name of the custom log in Stackdriver Logging. Defaults
- to 'airflow'. The name of the Python logger will be represented
- in the ``python_logger`` field.
- :param transport: Class for creating new transport objects. It should
- extend from the base :class:`google.cloud.logging.handlers.Transport`
type and
- implement :meth`google.cloud.logging.handlers.Transport.send`.
Defaults to
- :class:`google.cloud.logging.handlers.BackgroundThreadTransport`. The
other
- option is :class:`google.cloud.logging.handlers.SyncTransport`.
- :param resource: (Optional) Monitored resource of the entry, defaults
- to the global resource type.
- :param labels: (Optional) Mapping of labels for the entry.
- """
-
- # Re-export module-level constants for back-compat with external code
reading them off the class
- LABEL_TASK_ID = LABEL_TASK_ID
- LABEL_DAG_ID = LABEL_DAG_ID
- LABEL_LOGICAL_DATE = LABEL_LOGICAL_DATE
- LABEL_TRY_NUMBER = LABEL_TRY_NUMBER
- LOG_VIEWER_BASE_URL = "https://console.cloud.{domain}/logs/viewer";
- LOG_NAME = "Google Stackdriver"
-
- trigger_supported = True
- trigger_should_queue = False
- trigger_should_wrap = False
- trigger_send_end_marker = False
-
- def __init__(
- self,
- gcp_key_path: str | None = None,
- scopes: Collection[str] | None = _DEFAULT_SCOPESS,
- name: str | ArgNotSet = NOTSET,
- transport: type[Transport] = BackgroundThreadTransport,
- resource: Resource = _GLOBAL_RESOURCE,
- labels: dict[str, str] | None = None,
- gcp_log_name: str = DEFAULT_LOGGER_NAME,
- ):
- if name is not NOTSET:
- warnings.warn(
- "Param `name` is deprecated and will be removed in a future
release. "
- "Please use `gcp_log_name` instead. Planned removal date:
October 5, 2026.",
- AirflowProviderDeprecationWarning,
- stacklevel=2,
- )
- gcp_log_name = str(name)
-
- super().__init__()
- self.io = StackdriverRemoteLogIO(
- base_log_folder=Path("."),
- gcp_key_path=gcp_key_path,
- scopes=scopes,
- gcp_log_name=gcp_log_name,
- transport_type=transport,
- resource=resource,
- labels=labels,
- )
- self.labels: dict[str, str] | None = labels
- self.resource: Resource = resource
- self.task_instance_labels: dict[str, str] | None = {}
- self.task_instance_hostname = "default-hostname"
-
- def _get_labels(self, task_instance=None):
- if task_instance:
- ti_labels = _task_instance_to_labels(task_instance)
- else:
- ti_labels = self.task_instance_labels
- labels: dict[str, str] | None
- if self.labels and ti_labels:
- labels = {}
- labels.update(self.labels)
- labels.update(ti_labels)
- elif self.labels:
- labels = self.labels
- elif ti_labels:
- labels = ti_labels
- else:
- labels = None
- return labels or {}
-
- def emit(self, record: logging.LogRecord) -> None:
- """
- Actually log the specified logging record.
-
- :param record: The record to be logged.
- """
- message = self.format(record)
- ti = None
- if not AIRFLOW_V_3_0_PLUS and getattr(record, ctx_indiv_trigger.name,
None):
- ti = getattr(record, "task_instance", None) # trigger context
- labels = self._get_labels(ti)
- self.io.transport.send(record, message, resource=self.resource,
labels=labels)
-
- def set_context(self, task_instance: TaskInstance) -> None:
- """
- Configure the logger to add information with information about the
current task.
-
- :param task_instance: Currently executed task
- """
- self.task_instance_labels = _task_instance_to_labels(task_instance)
- self.task_instance_hostname = task_instance.hostname or
"default-hostname"
-
- def read(
- self, task_instance: TaskInstance, try_number: int | None = None,
metadata: dict | None = None
- ) -> tuple[list[tuple[tuple[str, str]]], list[dict[str, str | bool]]]:
- """
- Read logs of given task instance from Stackdriver logging.
-
- :param task_instance: task instance object
- :param try_number: task instance try_number to read logs from. If None
- it returns all logs
- :param metadata: log metadata. It is used for steaming log reading and
auto-tailing.
- :return: a tuple of (
- list of (one element tuple with two element tuple - hostname and
logs)
- and list of metadata)
- """
- if try_number is not None and try_number < 1:
- logs = f"Error fetching the logs. Try number {try_number} is
invalid."
- return [((self.task_instance_hostname, logs),)], [{"end_of_log":
"true"}]
-
- if not metadata:
- metadata = {}
-
- ti_labels = _task_instance_to_labels(task_instance)
-
- if try_number is not None:
- ti_labels[LABEL_TRY_NUMBER] = str(try_number)
- else:
- del ti_labels[LABEL_TRY_NUMBER]
-
- log_filter = self.io.prepare_log_filter(ti_labels)
- next_page_token = metadata.get("next_page_token", None)
- all_pages = "download_logs" in metadata and metadata["download_logs"]
-
- try:
- messages, end_of_log, next_page_token =
self.io.read_logs(log_filter, next_page_token, all_pages)
- except Exception:
- # Cloud Logging unavailable / IAM glitch / gRPC error. Without a
guard, the
- # exception used to propagate up as HTTP 500 from the log viewer.
Degrade
- # gracefully instead: surface a short user-facing message, mark
the read
- # complete (no spinning retry), and log the full traceback to the
handler's
- # own logger for the operator.
- _logger.exception("Failed to read logs from Cloud Logging for
filter %s", log_filter)
- return (
- [((self.task_instance_hostname, "Cloud Logging is currently
unavailable."),)],
- [{"end_of_log": True}],
- )
-
- new_metadata: dict[str, str | bool] = {"end_of_log": end_of_log}
-
- if next_page_token:
- new_metadata["next_page_token"] = next_page_token
-
- return [((self.task_instance_hostname, messages),)], [new_metadata]
-
- @classmethod
- def _task_instance_to_labels(cls, ti: TaskInstance) -> dict[str, str]:
- return {
- cls.LABEL_TASK_ID: ti.task_id,
- cls.LABEL_DAG_ID: ti.dag_id,
- cls.LABEL_LOGICAL_DATE: str(ti.logical_date.isoformat())
- if AIRFLOW_V_3_0_PLUS
- else str(ti.execution_date.isoformat()),
- cls.LABEL_TRY_NUMBER: str(ti.try_number),
- }
-
- @property
- def log_name(self):
- """Return log name."""
- return self.LOG_NAME
-
- @cached_property
- def _resource_path(self):
- segments = [self.resource.type]
-
- for key, value in self.resource.labels:
- segments += [key]
- segments += [value]
-
- return "/".join(segments)
-
- def get_external_log_url(self, task_instance: TaskInstance, try_number:
int) -> str:
- """
- Create an address for an external log collecting service.
-
- :param task_instance: task instance object
- :param try_number: task instance try_number to read logs from
- :return: URL to the external log collection service
- """
- _, project_id = self.io.credentials_and_project
-
- ti_labels = _task_instance_to_labels(task_instance)
- ti_labels[LABEL_TRY_NUMBER] = str(try_number)
-
- log_filter = self.io.prepare_log_filter(ti_labels)
-
- url_query_string = {
- "project": project_id,
- "interval": "NO_LIMIT",
- "resource": self._resource_path,
- "advancedFilter": log_filter,
- }
-
- log_viewer_url =
self.LOG_VIEWER_BASE_URL.format(domain=GoogleBaseHook.get_high_value_cookie_domain())
- url = f"{log_viewer_url}?{urlencode(url_query_string)}"
- return url
- def close(self) -> None:
- # ``flush()`` is best-effort during shutdown — if Cloud Logging is
unavailable or
- # the transport raises, that's not a reason to break the rest of the
handler's
- # shutdown chain (and the stdlib logging machinery does not handle
exceptions
- # from ``Handler.close()`` gracefully). Print to stderr as last resort
since
- # logging itself may be shutting down.
- try:
- self.io.transport.flush()
- except Exception as exc:
- print(
- f"StackdriverTaskHandler.close: transport flush failed:
{type(exc).__name__}: {exc}",
- file=sys.stderr,
- )
+@deprecated(
+ planned_removal_date="March 31, 2027",
+
use_instead="airflow.providers.google.cloud.log.cloud_logging_task_handler.CloudLoggingTaskHandler",
+ category=AirflowProviderDeprecationWarning,
+)
+class StackdriverTaskHandler(CloudLoggingTaskHandler):
+ """Deprecated. Use :class:`CloudLoggingTaskHandler`."""
diff --git a/providers/google/src/airflow/providers/google/get_provider_info.py
b/providers/google/src/airflow/providers/google/get_provider_info.py
index 790d37f0d9a..6b61ccf7adf 100644
--- a/providers/google/src/airflow/providers/google/get_provider_info.py
+++ b/providers/google/src/airflow/providers/google/get_provider_info.py
@@ -1725,7 +1725,7 @@ def get_provider_info():
"auth-backends":
["airflow.providers.google.common.auth_backend.google_openid"],
"logging": [
"airflow.providers.google.cloud.log.gcs_task_handler.GCSTaskHandler",
-
"airflow.providers.google.cloud.log.stackdriver_task_handler.StackdriverTaskHandler",
+
"airflow.providers.google.cloud.log.cloud_logging_task_handler.CloudLoggingTaskHandler",
],
"remote-logging": [
{
@@ -1733,7 +1733,7 @@ def get_provider_info():
"scheme": "gs",
},
{
- "classpath":
"airflow.providers.google.cloud.log.stackdriver_task_handler.StackdriverRemoteLogIO",
+ "classpath":
"airflow.providers.google.cloud.log.cloud_logging_task_handler.CloudLoggingRemoteLogIO",
"scheme": "stackdriver",
},
],
diff --git
a/providers/google/tests/unit/google/cloud/log/test_stackdriver_task_handler.py
b/providers/google/tests/unit/google/cloud/log/test_cloud_logging_task_handler.py
similarity index 75%
copy from
providers/google/tests/unit/google/cloud/log/test_stackdriver_task_handler.py
copy to
providers/google/tests/unit/google/cloud/log/test_cloud_logging_task_handler.py
index d93030bfcaf..b360f9c2525 100644
---
a/providers/google/tests/unit/google/cloud/log/test_stackdriver_task_handler.py
+++
b/providers/google/tests/unit/google/cloud/log/test_cloud_logging_task_handler.py
@@ -17,6 +17,7 @@
from __future__ import annotations
import logging
+import warnings
from contextlib import nullcontext
from pathlib import Path
from unittest import mock
@@ -27,10 +28,11 @@ import pytest
from google.cloud.logging import Resource
from google.cloud.logging_v2.types import ListLogEntriesRequest,
ListLogEntriesResponse, LogEntry
+from airflow.exceptions import AirflowProviderDeprecationWarning
from airflow.providers.common.compat.sdk import timezone
-from airflow.providers.google.cloud.log.stackdriver_task_handler import (
- StackdriverRemoteLogIO,
- StackdriverTaskHandler,
+from airflow.providers.google.cloud.log.cloud_logging_task_handler import (
+ CloudLoggingRemoteLogIO,
+ CloudLoggingTaskHandler,
)
from airflow.utils.state import TaskInstanceState
@@ -46,16 +48,16 @@ def _create_list_log_entries_response_mock(messages, token):
@pytest.fixture
-def clean_stackdriver_handlers():
+def clean_cloud_logging_handlers():
yield
for handler_ref in reversed(logging._handlerList[:]):
handler = handler_ref()
- if isinstance(handler, StackdriverTaskHandler):
+ if isinstance(handler, CloudLoggingTaskHandler):
logging._removeHandlerRef(handler_ref)
del handler
-class TestStackdriverRemoteLogIOFromConfig:
+class TestCloudLoggingRemoteLogIOFromConfig:
@conf_vars(
{
("logging", "base_log_folder"): "~/airflow/logs",
@@ -65,7 +67,7 @@ class TestStackdriverRemoteLogIOFromConfig:
}
)
def test_from_config(self):
- subject = StackdriverRemoteLogIO.from_config()
+ subject = CloudLoggingRemoteLogIO.from_config()
assert subject.base_log_folder == Path("~/airflow/logs").expanduser()
assert subject.gcp_log_name == "airflow-tasks"
@@ -81,7 +83,7 @@ class TestStackdriverRemoteLogIOFromConfig:
}
)
def
test_from_config_applies_io_kwargs_and_filters_file_handler_kwargs(self):
- subject = StackdriverRemoteLogIO.from_config()
+ subject = CloudLoggingRemoteLogIO.from_config()
assert subject.delete_local_copy is True
assert not hasattr(subject, "max_bytes")
@@ -89,7 +91,7 @@ class TestStackdriverRemoteLogIOFromConfig:
@conf_vars({("logging", "remote_task_handler_kwargs"): '["not", "a",
"dict"]'})
def test_from_config_rejects_non_dict_remote_task_handler_kwargs(self):
with pytest.raises(ValueError, match="remote_task_handler_kwargs"):
- StackdriverRemoteLogIO.from_config()
+ CloudLoggingRemoteLogIO.from_config()
@pytest.mark.parametrize(
"remote_base_log_folder",
@@ -100,8 +102,8 @@ class TestStackdriverRemoteLogIOFromConfig:
)
def test_from_config_rejects_remote_base_without_log_name(self,
remote_base_log_folder):
with conf_vars({("logging", "remote_base_log_folder"):
remote_base_log_folder}):
- with pytest.raises(ValueError, match="Stackdriver log name"):
- StackdriverRemoteLogIO.from_config()
+ with pytest.raises(ValueError, match="Cloud Logging log name"):
+ CloudLoggingRemoteLogIO.from_config()
def test_provider_registers_stackdriver_scheme(self):
from airflow.providers_manager import ProvidersManager
@@ -114,23 +116,25 @@ class TestStackdriverRemoteLogIOFromConfig:
assert info is not None
assert info.classpath == (
-
"airflow.providers.google.cloud.log.stackdriver_task_handler.StackdriverRemoteLogIO"
+
"airflow.providers.google.cloud.log.cloud_logging_task_handler.CloudLoggingRemoteLogIO"
)
-class TestStackdriverRemoteLogIO:
+class TestCloudLoggingRemoteLogIO:
@pytest.fixture(autouse=True)
def _setup(self, tmp_path):
self.local_log_location = str(tmp_path / "local/stackdriver/logs")
- self.io = StackdriverRemoteLogIO(
+ self.io = CloudLoggingRemoteLogIO(
base_log_folder=self.local_log_location,
gcp_key_path="KEY_PATH",
gcp_log_name="airflow",
delete_local_copy=True,
)
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.LoggingServiceV2Client")
+ @mock.patch(
+
"airflow.providers.google.cloud.log.cloud_logging_task_handler.get_credentials_and_project_id"
+ )
+
@mock.patch("airflow.providers.google.cloud.log.cloud_logging_task_handler.LoggingServiceV2Client")
def test_read_logs(self, mock_client, mock_get_creds_and_project_id):
mock_client.return_value.list_log_entries.return_value.pages = iter(
[_create_list_log_entries_response_mock(["MSG1", "MSG2"], None)]
@@ -149,11 +153,13 @@ class TestStackdriverRemoteLogIO:
messages, logs =
self.io.read("dag_id=test_dag/run_id=run1/task_id=test_task/attempt=1.log", ti)
assert len(messages) == 1
- assert "Stackdriver" in messages[0]
+ assert "Cloud Logging" in messages[0]
assert logs == ["MSG1\nMSG2"]
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.LoggingServiceV2Client")
+ @mock.patch(
+
"airflow.providers.google.cloud.log.cloud_logging_task_handler.get_credentials_and_project_id"
+ )
+
@mock.patch("airflow.providers.google.cloud.log.cloud_logging_task_handler.LoggingServiceV2Client")
def test_read_logs_empty(self, mock_client, mock_get_creds_and_project_id):
mock_client.return_value.list_log_entries.return_value.pages = iter(
[_create_list_log_entries_response_mock([], None)]
@@ -174,8 +180,10 @@ class TestStackdriverRemoteLogIO:
assert len(messages) == 1
assert logs == []
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.gcp_logging.Client")
+ @mock.patch(
+
"airflow.providers.google.cloud.log.cloud_logging_task_handler.get_credentials_and_project_id"
+ )
+
@mock.patch("airflow.providers.google.cloud.log.cloud_logging_task_handler.gcp_logging.Client")
def test_credentials(self, mock_client, mock_get_creds_and_project_id):
mock_get_creds_and_project_id.return_value = ("creds", "project_id")
@@ -193,13 +201,15 @@ class TestStackdriverRemoteLogIO:
)
mock_client.assert_called_once_with(credentials="creds",
client_info=mock.ANY, project="project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.gcp_logging.Client")
+ @mock.patch(
+
"airflow.providers.google.cloud.log.cloud_logging_task_handler.get_credentials_and_project_id"
+ )
+
@mock.patch("airflow.providers.google.cloud.log.cloud_logging_task_handler.gcp_logging.Client")
def test_transport_init(self, mock_client, mock_get_creds_and_project_id):
mock_get_creds_and_project_id.return_value = ("creds", "project_id")
transport_type = mock.MagicMock()
- io = StackdriverRemoteLogIO(
+ io = CloudLoggingRemoteLogIO(
base_log_folder=self.local_log_location,
gcp_log_name="test-log",
transport_type=transport_type,
@@ -209,11 +219,11 @@ class TestStackdriverRemoteLogIO:
@mock.patch("shutil.rmtree")
@mock.patch(
-
"airflow.providers.google.cloud.log.stackdriver_task_handler.StackdriverRemoteLogIO.transport",
+
"airflow.providers.google.cloud.log.cloud_logging_task_handler.CloudLoggingRemoteLogIO.transport",
new_callable=PropertyMock,
)
def test_upload_flushes_transport_and_deletes_local(self,
mock_transport_prop, mock_rmtree):
- io = StackdriverRemoteLogIO(
+ io = CloudLoggingRemoteLogIO(
base_log_folder=self.local_log_location,
gcp_log_name="airflow",
delete_local_copy=True,
@@ -235,11 +245,11 @@ class TestStackdriverRemoteLogIO:
mock_rmtree.assert_called_once_with(log_dir.resolve(),
ignore_errors=True)
@mock.patch(
-
"airflow.providers.google.cloud.log.stackdriver_task_handler.StackdriverRemoteLogIO.transport",
+
"airflow.providers.google.cloud.log.cloud_logging_task_handler.CloudLoggingRemoteLogIO.transport",
new_callable=PropertyMock,
)
def test_upload_no_delete(self, mock_transport_prop):
- io = StackdriverRemoteLogIO(
+ io = CloudLoggingRemoteLogIO(
base_log_folder=self.local_log_location,
gcp_log_name="airflow",
delete_local_copy=False,
@@ -254,14 +264,14 @@ class TestStackdriverRemoteLogIO:
@pytest.mark.skipif(not AIRFLOW_V_3_0_PLUS, reason="airflow.sdk.log only
exists in Airflow 3+")
@mock.patch(
-
"airflow.providers.google.cloud.log.stackdriver_task_handler.StackdriverRemoteLogIO.transport",
+
"airflow.providers.google.cloud.log.cloud_logging_task_handler.CloudLoggingRemoteLogIO.transport",
new_callable=PropertyMock,
)
def test_processors_fallback_to_event_labels(self, mock_transport_prop):
mock_transport = mock.MagicMock()
mock_transport_prop.return_value = mock_transport
- io = StackdriverRemoteLogIO(
+ io = CloudLoggingRemoteLogIO(
base_log_folder=self.local_log_location,
gcp_log_name="airflow",
)
@@ -297,7 +307,9 @@ class TestStackdriverRemoteLogIO:
"map_index": "-1",
}
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
+ @mock.patch(
+
"airflow.providers.google.cloud.log.cloud_logging_task_handler.get_credentials_and_project_id"
+ )
def test_prepare_log_filter(self, mock_get_creds_and_project_id):
mock_get_creds_and_project_id.return_value = ("creds", "project_id")
@@ -313,11 +325,13 @@ class TestStackdriverRemoteLogIO:
assert 'labels.task_id="test_task"' in log_filter
assert 'labels.dag_id="test_dag"' in log_filter
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
+ @mock.patch(
+
"airflow.providers.google.cloud.log.cloud_logging_task_handler.get_credentials_and_project_id"
+ )
def test_prepare_log_filter_with_custom_resource(self,
mock_get_creds_and_project_id):
mock_get_creds_and_project_id.return_value = ("creds", "project_id")
- io = StackdriverRemoteLogIO(
+ io = CloudLoggingRemoteLogIO(
base_log_folder=self.local_log_location,
gcp_log_name="airflow",
resource=Resource(
@@ -335,14 +349,16 @@ class TestStackdriverRemoteLogIO:
assert 'resource.labels.location="europe-west-3"' in log_filter
@pytest.mark.skipif(not AIRFLOW_V_3_0_PLUS, reason="airflow.sdk.log only
exists in Airflow 3+")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.gcp_logging.Client")
+ @mock.patch(
+
"airflow.providers.google.cloud.log.cloud_logging_task_handler.get_credentials_and_project_id"
+ )
+
@mock.patch("airflow.providers.google.cloud.log.cloud_logging_task_handler.gcp_logging.Client")
def test_processors_sends_to_transport(self, mock_client,
mock_get_creds_and_project_id):
mock_get_creds_and_project_id.return_value = ("creds", "project_id")
mock_transport_type = mock.MagicMock()
with mock.patch("airflow.sdk.log.relative_path_from_logger",
return_value="dag/task/1.log"):
- io = StackdriverRemoteLogIO(
+ io = CloudLoggingRemoteLogIO(
base_log_folder=self.local_log_location,
gcp_log_name="airflow",
labels={"env": "test"},
@@ -368,14 +384,16 @@ class TestStackdriverRemoteLogIO:
assert record.levelno == logging.INFO
@pytest.mark.skipif(not AIRFLOW_V_3_0_PLUS, reason="airflow.sdk.log only
exists in Airflow 3+")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.gcp_logging.Client")
+ @mock.patch(
+
"airflow.providers.google.cloud.log.cloud_logging_task_handler.get_credentials_and_project_id"
+ )
+
@mock.patch("airflow.providers.google.cloud.log.cloud_logging_task_handler.gcp_logging.Client")
def test_processors_skips_non_task_logger(self, mock_client,
mock_get_creds_and_project_id):
mock_get_creds_and_project_id.return_value = ("creds", "project_id")
mock_transport_type = mock.MagicMock()
with mock.patch("airflow.sdk.log.relative_path_from_logger",
return_value=None):
- io = StackdriverRemoteLogIO(
+ io = CloudLoggingRemoteLogIO(
base_log_folder=self.local_log_location,
gcp_log_name="airflow",
transport_type=mock_transport_type,
@@ -389,20 +407,20 @@ class TestStackdriverRemoteLogIO:
mock_transport_type.return_value.send.assert_not_called()
[email protected]("clean_stackdriver_handlers")
[email protected]("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
[email protected]("airflow.providers.google.cloud.log.stackdriver_task_handler.gcp_logging.Client")
[email protected]("clean_cloud_logging_handlers")
[email protected]("airflow.providers.google.cloud.log.cloud_logging_task_handler.get_credentials_and_project_id")
[email protected]("airflow.providers.google.cloud.log.cloud_logging_task_handler.gcp_logging.Client")
def test_should_pass_message_to_client(mock_client,
mock_get_creds_and_project_id):
mock_get_creds_and_project_id.return_value = ("creds", "project_id")
transport_type = mock.MagicMock()
- stackdriver_task_handler =
StackdriverTaskHandler(transport=transport_type, labels={"key": "value"})
+ cloud_logging_task_handler =
CloudLoggingTaskHandler(transport=transport_type, labels={"key": "value"})
logger = logging.getLogger("logger")
logger.setLevel(logging.INFO)
- logger.addHandler(stackdriver_task_handler)
+ logger.addHandler(cloud_logging_task_handler)
logger.info("test-message")
- stackdriver_task_handler.flush()
+ cloud_logging_task_handler.flush()
transport_type.assert_called_once_with(mock_client.return_value, "airflow")
transport_type.return_value.send.assert_called_once_with(
@@ -411,9 +429,9 @@ def test_should_pass_message_to_client(mock_client,
mock_get_creds_and_project_i
mock_client.assert_called_once_with(credentials="creds",
client_info=mock.ANY, project="project_id")
[email protected]("clean_stackdriver_handlers")
[email protected]("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
[email protected]("airflow.providers.google.cloud.log.stackdriver_task_handler.gcp_logging.Client")
[email protected]("clean_cloud_logging_handlers")
[email protected]("airflow.providers.google.cloud.log.cloud_logging_task_handler.get_credentials_and_project_id")
[email protected]("airflow.providers.google.cloud.log.cloud_logging_task_handler.gcp_logging.Client")
def test_should_use_configured_log_name(mock_client,
mock_get_creds_and_project_id):
import importlib
@@ -431,32 +449,36 @@ def test_should_use_configured_log_name(mock_client,
mock_get_creds_and_project_
("logging", "remote_base_log_folder"):
"stackdriver://host/path",
}
):
- importlib.reload(airflow_local_settings)
- settings.configure_logging()
+ # Core still instantiates the deprecated classes for
stackdriver://
+ # (StackdriverRemoteLogIO on main, StackdriverTaskHandler on
released cores).
+ with warnings.catch_warnings():
+ warnings.simplefilter("ignore",
AirflowProviderDeprecationWarning)
+ importlib.reload(airflow_local_settings)
+ settings.configure_logging()
task_log = getattr(airflow_local_settings, "REMOTE_TASK_LOG",
None)
if task_log is not None:
# Airflow 3+ uses REMOTE_TASK_LOG instead of handler-based
config
- assert isinstance(task_log, StackdriverRemoteLogIO)
+ assert isinstance(task_log, CloudLoggingRemoteLogIO)
assert task_log.gcp_log_name == "path"
return
# Older Airflow: stackdriver is wired as a logging handler
logger = logging.getLogger("airflow.task")
handler = logger.handlers[0]
- assert isinstance(handler, StackdriverTaskHandler)
+ assert isinstance(handler, CloudLoggingTaskHandler)
finally:
importlib.reload(airflow_local_settings)
settings.configure_logging()
@pytest.mark.db_test
-class TestStackdriverLoggingHandlerTask:
- DAG_ID = "dag_for_testing_stackdriver_file_task_handler"
- TASK_ID = "task_for_testing_stackdriver_task_handler"
+class TestCloudLoggingHandlerTask:
+ DAG_ID = "dag_for_testing_cloud_logging_task_handler"
+ TASK_ID = "task_for_testing_cloud_logging_task_handler"
@pytest.fixture(autouse=True)
- def task_instance(self, create_task_instance, clean_stackdriver_handlers):
+ def task_instance(self, create_task_instance,
clean_cloud_logging_handlers):
self.ti = create_task_instance(
dag_id=self.DAG_ID,
task_id=self.TASK_ID,
@@ -472,21 +494,23 @@ class TestStackdriverLoggingHandlerTask:
def _setup_handler(self, **handler_kwargs):
self.transport_mock = mock.MagicMock()
handler_kwargs = {"transport": self.transport_mock, **handler_kwargs}
- stackdriver_task_handler = StackdriverTaskHandler(**handler_kwargs)
+ cloud_logging_task_handler = CloudLoggingTaskHandler(**handler_kwargs)
self.logger = logging.getLogger("logger")
- self.logger.addHandler(stackdriver_task_handler)
- return stackdriver_task_handler
+ self.logger.addHandler(cloud_logging_task_handler)
+ return cloud_logging_task_handler
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.gcp_logging.Client")
+ @mock.patch(
+
"airflow.providers.google.cloud.log.cloud_logging_task_handler.get_credentials_and_project_id"
+ )
+
@mock.patch("airflow.providers.google.cloud.log.cloud_logging_task_handler.gcp_logging.Client")
def test_should_set_labels(self, mock_client,
mock_get_creds_and_project_id):
mock_get_creds_and_project_id.return_value = ("creds", "project_id")
- stackdriver_task_handler = self._setup_handler()
- stackdriver_task_handler.set_context(self.ti)
+ cloud_logging_task_handler = self._setup_handler()
+ cloud_logging_task_handler.set_context(self.ti)
self.logger.info("test-message")
- stackdriver_task_handler.flush()
+ cloud_logging_task_handler.flush()
date_key = "logical_date" if AIRFLOW_V_3_0_PLUS else "execution_date"
labels = {
@@ -500,18 +524,20 @@ class TestStackdriverLoggingHandlerTask:
mock.ANY, "test-message", labels=labels, resource=resource
)
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.gcp_logging.Client")
+ @mock.patch(
+
"airflow.providers.google.cloud.log.cloud_logging_task_handler.get_credentials_and_project_id"
+ )
+
@mock.patch("airflow.providers.google.cloud.log.cloud_logging_task_handler.gcp_logging.Client")
def test_should_append_labels(self, mock_client,
mock_get_creds_and_project_id):
mock_get_creds_and_project_id.return_value = ("creds", "project_id")
- stackdriver_task_handler = self._setup_handler(
+ cloud_logging_task_handler = self._setup_handler(
labels={"product.googleapis.com/task_id": "test-value"},
)
- stackdriver_task_handler.set_context(self.ti)
+ cloud_logging_task_handler.set_context(self.ti)
self.logger.info("test-message")
- stackdriver_task_handler.flush()
+ cloud_logging_task_handler.flush()
date_key = "logical_date" if AIRFLOW_V_3_0_PLUS else "execution_date"
labels = {
@@ -526,24 +552,26 @@ class TestStackdriverLoggingHandlerTask:
mock.ANY, "test-message", labels=labels, resource=resource
)
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.LoggingServiceV2Client")
+ @mock.patch(
+
"airflow.providers.google.cloud.log.cloud_logging_task_handler.get_credentials_and_project_id"
+ )
+
@mock.patch("airflow.providers.google.cloud.log.cloud_logging_task_handler.LoggingServiceV2Client")
def test_should_read_logs_for_all_try(self, mock_client,
mock_get_creds_and_project_id):
mock_client.return_value.list_log_entries.return_value.pages = iter(
[_create_list_log_entries_response_mock(["MSG1", "MSG2"], None)]
)
mock_get_creds_and_project_id.return_value = ("creds", "project_id")
- stackdriver_task_handler = self._setup_handler()
- logs, metadata = stackdriver_task_handler.read(self.ti)
+ cloud_logging_task_handler = self._setup_handler()
+ logs, metadata = cloud_logging_task_handler.read(self.ti)
date_label = "logical_date" if AIRFLOW_V_3_0_PLUS else "execution_date"
filter_str = (
'resource.type="global"\n'
'logName="projects/project_id/logs/airflow"\n'
- 'labels.task_id="task_for_testing_stackdriver_task_handler"\n'
- 'labels.dag_id="dag_for_testing_stackdriver_file_task_handler"\n'
+ 'labels.task_id="task_for_testing_cloud_logging_task_handler"\n'
+ 'labels.dag_id="dag_for_testing_cloud_logging_task_handler"\n'
f'labels.{date_label}="2016-01-01T00:00:00+00:00"'
)
mock_client.return_value.list_log_entries.assert_called_once_with(
@@ -558,8 +586,10 @@ class TestStackdriverLoggingHandlerTask:
assert logs == [(("default-hostname", "MSG1\nMSG2"),)]
assert metadata == [{"end_of_log": True}]
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.LoggingServiceV2Client")
+ @mock.patch(
+
"airflow.providers.google.cloud.log.cloud_logging_task_handler.get_credentials_and_project_id"
+ )
+
@mock.patch("airflow.providers.google.cloud.log.cloud_logging_task_handler.LoggingServiceV2Client")
def test_should_read_logs_for_task_with_quote(self, mock_client,
mock_get_creds_and_project_id):
mock_client.return_value.list_log_entries.return_value.pages = iter(
[_create_list_log_entries_response_mock(["MSG1", "MSG2"], None)]
@@ -567,15 +597,15 @@ class TestStackdriverLoggingHandlerTask:
mock_get_creds_and_project_id.return_value = ("creds", "project_id")
self.ti.task_id = 'K"OT'
- stackdriver_task_handler = self._setup_handler()
+ cloud_logging_task_handler = self._setup_handler()
- logs, metadata = stackdriver_task_handler.read(self.ti)
+ logs, metadata = cloud_logging_task_handler.read(self.ti)
date_label = "logical_date" if AIRFLOW_V_3_0_PLUS else "execution_date"
filter_str = (
'resource.type="global"\n'
'logName="projects/project_id/logs/airflow"\n'
'labels.task_id="K\\"OT"\n'
- 'labels.dag_id="dag_for_testing_stackdriver_file_task_handler"\n'
+ 'labels.dag_id="dag_for_testing_cloud_logging_task_handler"\n'
f'labels.{date_label}="2016-01-01T00:00:00+00:00"'
)
mock_client.return_value.list_log_entries.assert_called_once_with(
@@ -590,22 +620,24 @@ class TestStackdriverLoggingHandlerTask:
assert logs == [(("default-hostname", "MSG1\nMSG2"),)]
assert metadata == [{"end_of_log": True}]
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.LoggingServiceV2Client")
+ @mock.patch(
+
"airflow.providers.google.cloud.log.cloud_logging_task_handler.get_credentials_and_project_id"
+ )
+
@mock.patch("airflow.providers.google.cloud.log.cloud_logging_task_handler.LoggingServiceV2Client")
def test_should_read_logs_for_single_try(self, mock_client,
mock_get_creds_and_project_id):
mock_client.return_value.list_log_entries.return_value.pages = iter(
[_create_list_log_entries_response_mock(["MSG1", "MSG2"], None)]
)
mock_get_creds_and_project_id.return_value = ("creds", "project_id")
- stackdriver_task_handler = self._setup_handler()
+ cloud_logging_task_handler = self._setup_handler()
- logs, metadata = stackdriver_task_handler.read(self.ti, 3)
+ logs, metadata = cloud_logging_task_handler.read(self.ti, 3)
date_label = "logical_date" if AIRFLOW_V_3_0_PLUS else "execution_date"
filter_str = (
'resource.type="global"\n'
'logName="projects/project_id/logs/airflow"\n'
- 'labels.task_id="task_for_testing_stackdriver_task_handler"\n'
- 'labels.dag_id="dag_for_testing_stackdriver_file_task_handler"\n'
+ 'labels.task_id="task_for_testing_cloud_logging_task_handler"\n'
+ 'labels.dag_id="dag_for_testing_cloud_logging_task_handler"\n'
f'labels.{date_label}="2016-01-01T00:00:00+00:00"\n'
'labels.try_number="3"'
)
@@ -621,23 +653,25 @@ class TestStackdriverLoggingHandlerTask:
assert logs == [(("default-hostname", "MSG1\nMSG2"),)]
assert metadata == [{"end_of_log": True}]
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.LoggingServiceV2Client")
+ @mock.patch(
+
"airflow.providers.google.cloud.log.cloud_logging_task_handler.get_credentials_and_project_id"
+ )
+
@mock.patch("airflow.providers.google.cloud.log.cloud_logging_task_handler.LoggingServiceV2Client")
def test_should_read_logs_with_pagination(self, mock_client,
mock_get_creds_and_project_id):
mock_client.return_value.list_log_entries.side_effect = [
mock.MagicMock(pages=iter([_create_list_log_entries_response_mock(["MSG1",
"MSG2"], "TOKEN1")])),
mock.MagicMock(pages=iter([_create_list_log_entries_response_mock(["MSG3",
"MSG4"], None)])),
]
mock_get_creds_and_project_id.return_value = ("creds", "project_id")
- stackdriver_task_handler = self._setup_handler()
+ cloud_logging_task_handler = self._setup_handler()
- logs, metadata1 = stackdriver_task_handler.read(self.ti, 3)
+ logs, metadata1 = cloud_logging_task_handler.read(self.ti, 3)
date_label = "logical_date" if AIRFLOW_V_3_0_PLUS else "execution_date"
filter_str = (
'resource.type="global"\n'
'logName="projects/project_id/logs/airflow"\n'
- 'labels.task_id="task_for_testing_stackdriver_task_handler"\n'
- 'labels.dag_id="dag_for_testing_stackdriver_file_task_handler"\n'
+ 'labels.task_id="task_for_testing_cloud_logging_task_handler"\n'
+ 'labels.dag_id="dag_for_testing_cloud_logging_task_handler"\n'
f'labels.{date_label}="2016-01-01T00:00:00+00:00"\n'
'labels.try_number="3"'
)
@@ -654,7 +688,7 @@ class TestStackdriverLoggingHandlerTask:
assert metadata1 == [{"end_of_log": False, "next_page_token":
"TOKEN1"}]
mock_client.return_value.list_log_entries.return_value.next_page_token
= None
- logs, metadata2 = stackdriver_task_handler.read(self.ti, 3,
metadata1[0])
+ logs, metadata2 = cloud_logging_task_handler.read(self.ti, 3,
metadata1[0])
mock_client.return_value.list_log_entries.assert_called_with(
request=ListLogEntriesRequest(
@@ -662,8 +696,8 @@ class TestStackdriverLoggingHandlerTask:
filter=(
'resource.type="global"\n'
'logName="projects/project_id/logs/airflow"\n'
-
'labels.task_id="task_for_testing_stackdriver_task_handler"\n'
-
'labels.dag_id="dag_for_testing_stackdriver_file_task_handler"\n'
+
'labels.task_id="task_for_testing_cloud_logging_task_handler"\n'
+
'labels.dag_id="dag_for_testing_cloud_logging_task_handler"\n'
f'labels.{date_label}="2016-01-01T00:00:00+00:00"\n'
'labels.try_number="3"'
),
@@ -675,8 +709,10 @@ class TestStackdriverLoggingHandlerTask:
assert logs == [(("default-hostname", "MSG3\nMSG4"),)]
assert metadata2 == [{"end_of_log": True}]
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.LoggingServiceV2Client")
+ @mock.patch(
+
"airflow.providers.google.cloud.log.cloud_logging_task_handler.get_credentials_and_project_id"
+ )
+
@mock.patch("airflow.providers.google.cloud.log.cloud_logging_task_handler.LoggingServiceV2Client")
def test_should_read_logs_with_download(self, mock_client,
mock_get_creds_and_project_id):
mock_client.return_value.list_log_entries.side_effect = [
mock.MagicMock(pages=iter([_create_list_log_entries_response_mock(["MSG1",
"MSG2"], "TOKEN1")])),
@@ -684,14 +720,16 @@ class TestStackdriverLoggingHandlerTask:
]
mock_get_creds_and_project_id.return_value = ("creds", "project_id")
- stackdriver_task_handler = self._setup_handler()
- logs, metadata1 = stackdriver_task_handler.read(self.ti, 3,
{"download_logs": True})
+ cloud_logging_task_handler = self._setup_handler()
+ logs, metadata1 = cloud_logging_task_handler.read(self.ti, 3,
{"download_logs": True})
assert logs == [(("default-hostname", "MSG1\nMSG2\nMSG3\nMSG4"),)]
assert metadata1 == [{"end_of_log": True}]
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.LoggingServiceV2Client")
+ @mock.patch(
+
"airflow.providers.google.cloud.log.cloud_logging_task_handler.get_credentials_and_project_id"
+ )
+
@mock.patch("airflow.providers.google.cloud.log.cloud_logging_task_handler.LoggingServiceV2Client")
def test_should_read_logs_with_custom_resources(self, mock_client,
mock_get_creds_and_project_id):
mock_get_creds_and_project_id.return_value = ("creds", "project_id")
resource = Resource(
@@ -702,13 +740,13 @@ class TestStackdriverLoggingHandlerTask:
"project_id": "project_id",
},
)
- stackdriver_task_handler = self._setup_handler(resource=resource)
+ cloud_logging_task_handler = self._setup_handler(resource=resource)
entry = mock.MagicMock(json_payload={"message": "TEXT"})
page = mock.MagicMock(entries=[entry, entry], next_page_token=None)
mock_client.return_value.list_log_entries.return_value.pages =
iter([page])
- logs, metadata = stackdriver_task_handler.read(self.ti)
+ logs, metadata = cloud_logging_task_handler.read(self.ti)
date_label = "logical_date" if AIRFLOW_V_3_0_PLUS else "execution_date"
filter_str = (
'resource.type="cloud_composer_environment"\n'
@@ -716,8 +754,8 @@ class TestStackdriverLoggingHandlerTask:
'resource.labels."environment.name"="test-instance"\n'
'resource.labels.location="europe-west-3"\n'
'resource.labels.project_id="project_id"\n'
- 'labels.task_id="task_for_testing_stackdriver_task_handler"\n'
- 'labels.dag_id="dag_for_testing_stackdriver_file_task_handler"\n'
+ 'labels.task_id="task_for_testing_cloud_logging_task_handler"\n'
+ 'labels.dag_id="dag_for_testing_cloud_logging_task_handler"\n'
f'labels.{date_label}="2016-01-01T00:00:00+00:00"'
)
mock_client.return_value.list_log_entries.assert_called_once_with(
@@ -732,13 +770,15 @@ class TestStackdriverLoggingHandlerTask:
assert logs == [(("default-hostname", "TEXT\nTEXT"),)]
assert metadata == [{"end_of_log": True}]
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.gcp_logging.Client")
+ @mock.patch(
+
"airflow.providers.google.cloud.log.cloud_logging_task_handler.get_credentials_and_project_id"
+ )
+
@mock.patch("airflow.providers.google.cloud.log.cloud_logging_task_handler.gcp_logging.Client")
def test_should_use_credentials(self, mock_client,
mock_get_creds_and_project_id):
mock_get_creds_and_project_id.return_value = ("creds", "project_id")
- stackdriver_task_handler =
StackdriverTaskHandler(gcp_key_path="KEY_PATH")
- client = stackdriver_task_handler.io._client
+ cloud_logging_task_handler =
CloudLoggingTaskHandler(gcp_key_path="KEY_PATH")
+ client = cloud_logging_task_handler.io._client
mock_get_creds_and_project_id.assert_called_once_with(
disable_logging=True,
@@ -753,14 +793,16 @@ class TestStackdriverLoggingHandlerTask:
mock_client.assert_called_once_with(credentials="creds",
client_info=mock.ANY, project="project_id")
assert mock_client.return_value == client
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.LoggingServiceV2Client")
+ @mock.patch(
+
"airflow.providers.google.cloud.log.cloud_logging_task_handler.get_credentials_and_project_id"
+ )
+
@mock.patch("airflow.providers.google.cloud.log.cloud_logging_task_handler.LoggingServiceV2Client")
def test_should_return_valid_external_url(self, mock_client,
mock_get_creds_and_project_id, monkeypatch):
mock_get_creds_and_project_id.return_value = ("creds", "project_id")
monkeypatch.setenv("GOOGLE_CLOUD_HIGH_VALUE_COOKIE_DOMAIN",
"googleapis.cn")
- stackdriver_task_handler =
StackdriverTaskHandler(gcp_key_path="KEY_PATH")
- url = stackdriver_task_handler.get_external_log_url(self.ti,
self.ti.try_number)
+ cloud_logging_task_handler =
CloudLoggingTaskHandler(gcp_key_path="KEY_PATH")
+ url = cloud_logging_task_handler.get_external_log_url(self.ti,
self.ti.try_number)
parsed_url = urlsplit(url)
parsed_qs = parse_qs(parsed_url.query)
@@ -783,11 +825,13 @@ class TestStackdriverLoggingHandlerTask:
assert set(expected_filter) == set(filter_params)
-class TestStackdriverTaskHandlerExceptionHandling:
+class TestCloudLoggingTaskHandlerExceptionHandling:
"""Cloud Logging failures must degrade gracefully, not leak internals."""
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.LoggingServiceV2Client")
+ @mock.patch(
+
"airflow.providers.google.cloud.log.cloud_logging_task_handler.get_credentials_and_project_id"
+ )
+
@mock.patch("airflow.providers.google.cloud.log.cloud_logging_task_handler.LoggingServiceV2Client")
def test_read_falls_back_when_cloud_logging_unavailable(
self, mock_client, mock_get_creds_and_project_id, caplog
):
@@ -804,7 +848,7 @@ class TestStackdriverTaskHandlerExceptionHandling:
"Stackdriver returned an internal error for project
secret-project-id"
)
- handler = StackdriverTaskHandler()
+ handler = CloudLoggingTaskHandler()
ti = mock.MagicMock()
ti.task_id = "t"
ti.dag_id = "d"
@@ -821,8 +865,10 @@ class TestStackdriverTaskHandlerExceptionHandling:
assert "secret-project-id" not in message
assert metadata == [{"end_of_log": True}]
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.LoggingServiceV2Client")
+ @mock.patch(
+
"airflow.providers.google.cloud.log.cloud_logging_task_handler.get_credentials_and_project_id"
+ )
+
@mock.patch("airflow.providers.google.cloud.log.cloud_logging_task_handler.LoggingServiceV2Client")
def test_read_does_not_leak_internals_in_user_facing_message(
self, mock_client, mock_get_creds_and_project_id
):
@@ -840,7 +886,7 @@ class TestStackdriverTaskHandlerExceptionHandling:
"service account '[email protected]' lacks
logging.logEntries.list"
)
- handler = StackdriverTaskHandler()
+ handler = CloudLoggingTaskHandler()
ti = mock.MagicMock()
ti.task_id = "t"
ti.dag_id = "d"
@@ -857,11 +903,11 @@ class TestStackdriverTaskHandlerExceptionHandling:
def test_close_swallows_transport_flush_errors(self, capsys):
"""``close()`` must never raise — even when transport ``flush()``
fails."""
- handler = StackdriverTaskHandler()
+ handler = CloudLoggingTaskHandler()
broken_transport = mock.MagicMock()
broken_transport.flush.side_effect = RuntimeError("flush failed during
shutdown")
# ``transport`` is a cached_property on the slotted attrs class
- # ``StackdriverRemoteLogIO``; its value lives in a slot, not
``__dict__``, so
+ # ``CloudLoggingRemoteLogIO``; its value lives in a slot, not
``__dict__``, so
# assign the attribute directly to pre-seed it without building a real
transport.
handler.io.transport = broken_transport
diff --git
a/providers/google/tests/unit/google/cloud/log/test_stackdriver_task_handler.py
b/providers/google/tests/unit/google/cloud/log/test_stackdriver_task_handler.py
index d93030bfcaf..860969fb076 100644
---
a/providers/google/tests/unit/google/cloud/log/test_stackdriver_task_handler.py
+++
b/providers/google/tests/unit/google/cloud/log/test_stackdriver_task_handler.py
@@ -1,3 +1,4 @@
+#
# 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
@@ -16,859 +17,42 @@
# under the License.
from __future__ import annotations
-import logging
-from contextlib import nullcontext
-from pathlib import Path
-from unittest import mock
-from unittest.mock import PropertyMock
-from urllib.parse import parse_qs, urlsplit
-
import pytest
-from google.cloud.logging import Resource
-from google.cloud.logging_v2.types import ListLogEntriesRequest,
ListLogEntriesResponse, LogEntry
-from airflow.providers.common.compat.sdk import timezone
-from airflow.providers.google.cloud.log.stackdriver_task_handler import (
- StackdriverRemoteLogIO,
- StackdriverTaskHandler,
+from airflow.exceptions import AirflowProviderDeprecationWarning
+from airflow.providers.google.cloud.log import cloud_logging_task_handler,
stackdriver_task_handler
+
+
[email protected](
+ ("old_class", "new_class", "kwargs"),
+ [
+ (
+ stackdriver_task_handler.StackdriverRemoteLogIO,
+ cloud_logging_task_handler.CloudLoggingRemoteLogIO,
+ {"base_log_folder": "/tmp/x"},
+ ),
+ (
+ stackdriver_task_handler.StackdriverTaskHandler,
+ cloud_logging_task_handler.CloudLoggingTaskHandler,
+ {},
+ ),
+ ],
)
-from airflow.utils.state import TaskInstanceState
-
-from tests_common.test_utils.config import conf_vars
-from tests_common.test_utils.db import clear_db_dags, clear_db_runs
-from tests_common.test_utils.version_compat import AIRFLOW_V_3_0_PLUS
-
-
-def _create_list_log_entries_response_mock(messages, token):
- return ListLogEntriesResponse(
- entries=[LogEntry(json_payload={"message": message}) for message in
messages], next_page_token=token
- )
-
-
[email protected]
-def clean_stackdriver_handlers():
- yield
- for handler_ref in reversed(logging._handlerList[:]):
- handler = handler_ref()
- if isinstance(handler, StackdriverTaskHandler):
- logging._removeHandlerRef(handler_ref)
- del handler
-
-
-class TestStackdriverRemoteLogIOFromConfig:
- @conf_vars(
- {
- ("logging", "base_log_folder"): "~/airflow/logs",
- ("logging", "remote_base_log_folder"):
"stackdriver:///airflow-tasks",
- ("logging", "delete_local_logs"): "True",
- ("logging", "google_key_path"): "/tmp/google-key.json",
- }
- )
- def test_from_config(self):
- subject = StackdriverRemoteLogIO.from_config()
-
- assert subject.base_log_folder == Path("~/airflow/logs").expanduser()
- assert subject.gcp_log_name == "airflow-tasks"
- assert subject.gcp_key_path == "/tmp/google-key.json"
- assert subject.delete_local_copy is True
-
- @conf_vars(
- {
- ("logging", "base_log_folder"): "/tmp/airflow/logs",
- ("logging", "remote_base_log_folder"):
"stackdriver:///airflow-tasks",
- ("logging", "delete_local_logs"): "False",
- ("logging", "remote_task_handler_kwargs"): '{"delete_local_copy":
true, "max_bytes": 1024}',
- }
- )
- def
test_from_config_applies_io_kwargs_and_filters_file_handler_kwargs(self):
- subject = StackdriverRemoteLogIO.from_config()
-
- assert subject.delete_local_copy is True
- assert not hasattr(subject, "max_bytes")
-
- @conf_vars({("logging", "remote_task_handler_kwargs"): '["not", "a",
"dict"]'})
- def test_from_config_rejects_non_dict_remote_task_handler_kwargs(self):
- with pytest.raises(ValueError, match="remote_task_handler_kwargs"):
- StackdriverRemoteLogIO.from_config()
-
- @pytest.mark.parametrize(
- "remote_base_log_folder",
- [
- pytest.param("stackdriver://", id="scheme-only"),
- pytest.param("stackdriver://host", id="no-path"),
- ],
- )
- def test_from_config_rejects_remote_base_without_log_name(self,
remote_base_log_folder):
- with conf_vars({("logging", "remote_base_log_folder"):
remote_base_log_folder}):
- with pytest.raises(ValueError, match="Stackdriver log name"):
- StackdriverRemoteLogIO.from_config()
-
- def test_provider_registers_stackdriver_scheme(self):
- from airflow.providers_manager import ProvidersManager
-
- manager = ProvidersManager()
- if not hasattr(manager, "remote_logging_handler_by_scheme"):
- pytest.skip("Airflow core does not support remote logging provider
dispatch")
-
- info = manager.remote_logging_handler_by_scheme("stackdriver")
-
- assert info is not None
- assert info.classpath == (
-
"airflow.providers.google.cloud.log.stackdriver_task_handler.StackdriverRemoteLogIO"
- )
-
-
-class TestStackdriverRemoteLogIO:
- @pytest.fixture(autouse=True)
- def _setup(self, tmp_path):
- self.local_log_location = str(tmp_path / "local/stackdriver/logs")
- self.io = StackdriverRemoteLogIO(
- base_log_folder=self.local_log_location,
- gcp_key_path="KEY_PATH",
- gcp_log_name="airflow",
- delete_local_copy=True,
- )
-
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.LoggingServiceV2Client")
- def test_read_logs(self, mock_client, mock_get_creds_and_project_id):
- mock_client.return_value.list_log_entries.return_value.pages = iter(
- [_create_list_log_entries_response_mock(["MSG1", "MSG2"], None)]
- )
- mock_get_creds_and_project_id.return_value = ("creds", "project_id")
-
- ti = mock.MagicMock()
- ti.task_id = "test_task"
- ti.dag_id = "test_dag"
- ti.try_number = 1
- if AIRFLOW_V_3_0_PLUS:
- ti.logical_date = timezone.datetime(2016, 1, 1)
- else:
- ti.execution_date = timezone.datetime(2016, 1, 1)
-
- messages, logs =
self.io.read("dag_id=test_dag/run_id=run1/task_id=test_task/attempt=1.log", ti)
-
- assert len(messages) == 1
- assert "Stackdriver" in messages[0]
- assert logs == ["MSG1\nMSG2"]
-
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.LoggingServiceV2Client")
- def test_read_logs_empty(self, mock_client, mock_get_creds_and_project_id):
- mock_client.return_value.list_log_entries.return_value.pages = iter(
- [_create_list_log_entries_response_mock([], None)]
- )
- mock_get_creds_and_project_id.return_value = ("creds", "project_id")
-
- ti = mock.MagicMock()
- ti.task_id = "test_task"
- ti.dag_id = "test_dag"
- ti.try_number = 1
- if AIRFLOW_V_3_0_PLUS:
- ti.logical_date = timezone.datetime(2016, 1, 1)
- else:
- ti.execution_date = timezone.datetime(2016, 1, 1)
-
- messages, logs = self.io.read("test/path", ti)
-
- assert len(messages) == 1
- assert logs == []
-
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.gcp_logging.Client")
- def test_credentials(self, mock_client, mock_get_creds_and_project_id):
- mock_get_creds_and_project_id.return_value = ("creds", "project_id")
-
- _ = self.io._client
-
- mock_get_creds_and_project_id.assert_called_once_with(
- disable_logging=True,
- key_path="KEY_PATH",
- scopes=frozenset(
- {
- "https://www.googleapis.com/auth/logging.write";,
- "https://www.googleapis.com/auth/logging.read";,
- }
- ),
- )
- mock_client.assert_called_once_with(credentials="creds",
client_info=mock.ANY, project="project_id")
-
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.gcp_logging.Client")
- def test_transport_init(self, mock_client, mock_get_creds_and_project_id):
- mock_get_creds_and_project_id.return_value = ("creds", "project_id")
-
- transport_type = mock.MagicMock()
- io = StackdriverRemoteLogIO(
- base_log_folder=self.local_log_location,
- gcp_log_name="test-log",
- transport_type=transport_type,
- )
- _ = io.transport
- transport_type.assert_called_once_with(mock_client.return_value,
"test-log")
-
- @mock.patch("shutil.rmtree")
- @mock.patch(
-
"airflow.providers.google.cloud.log.stackdriver_task_handler.StackdriverRemoteLogIO.transport",
- new_callable=PropertyMock,
- )
- def test_upload_flushes_transport_and_deletes_local(self,
mock_transport_prop, mock_rmtree):
- io = StackdriverRemoteLogIO(
- base_log_folder=self.local_log_location,
- gcp_log_name="airflow",
- delete_local_copy=True,
- )
- mock_transport = mock.MagicMock()
- mock_transport_prop.return_value = mock_transport
-
- base = Path(self.local_log_location)
- base.mkdir(parents=True, exist_ok=True)
- log_dir = base / "subdir"
- log_dir.mkdir(parents=True, exist_ok=True)
- log_file = log_dir / "test.log"
- log_file.write_text("log content")
-
- ti = mock.MagicMock()
- io.upload(str(log_file), ti)
-
- mock_transport.flush.assert_called_once()
- mock_rmtree.assert_called_once_with(log_dir.resolve(),
ignore_errors=True)
-
- @mock.patch(
-
"airflow.providers.google.cloud.log.stackdriver_task_handler.StackdriverRemoteLogIO.transport",
- new_callable=PropertyMock,
- )
- def test_upload_no_delete(self, mock_transport_prop):
- io = StackdriverRemoteLogIO(
- base_log_folder=self.local_log_location,
- gcp_log_name="airflow",
- delete_local_copy=False,
- )
- mock_transport = mock.MagicMock()
- mock_transport_prop.return_value = mock_transport
-
- ti = mock.MagicMock()
- io.upload("some/path.log", ti)
-
- mock_transport.flush.assert_called_once()
-
- @pytest.mark.skipif(not AIRFLOW_V_3_0_PLUS, reason="airflow.sdk.log only
exists in Airflow 3+")
- @mock.patch(
-
"airflow.providers.google.cloud.log.stackdriver_task_handler.StackdriverRemoteLogIO.transport",
- new_callable=PropertyMock,
- )
- def test_processors_fallback_to_event_labels(self, mock_transport_prop):
- mock_transport = mock.MagicMock()
- mock_transport_prop.return_value = mock_transport
-
- io = StackdriverRemoteLogIO(
- base_log_folder=self.local_log_location,
- gcp_log_name="airflow",
- )
- logger = mock.MagicMock()
- # Mock relative_path_from_logger to return something truthy
- with mock.patch(
- "airflow.sdk.log.relative_path_from_logger",
- return_value="some/path.py",
- ):
- proc = io.processors[0]
- event = {
- "event": "Test message",
- "dag_id": "test_dag_id",
- "task_id": "test_task_id",
- "run_id": "test_run_id",
- "try_number": 2,
- "map_index": -1,
- }
-
- result = proc(logger, "info", event)
-
- assert result == event
-
- mock_transport.send.assert_called_once()
- _, kwargs = mock_transport.send.call_args
-
- labels = kwargs.get("labels", {})
- assert labels == {
- "dag_id": "test_dag_id",
- "task_id": "test_task_id",
- "run_id": "test_run_id",
- "try_number": "2",
- "map_index": "-1",
- }
-
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
- def test_prepare_log_filter(self, mock_get_creds_and_project_id):
- mock_get_creds_and_project_id.return_value = ("creds", "project_id")
-
- ti_labels = {
- "task_id": "test_task",
- "dag_id": "test_dag",
- "try_number": "1",
- }
- log_filter = self.io.prepare_log_filter(ti_labels)
-
- assert 'resource.type="global"' in log_filter
- assert 'logName="projects/project_id/logs/airflow"' in log_filter
- assert 'labels.task_id="test_task"' in log_filter
- assert 'labels.dag_id="test_dag"' in log_filter
-
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
- def test_prepare_log_filter_with_custom_resource(self,
mock_get_creds_and_project_id):
- mock_get_creds_and_project_id.return_value = ("creds", "project_id")
-
- io = StackdriverRemoteLogIO(
- base_log_folder=self.local_log_location,
- gcp_log_name="airflow",
- resource=Resource(
- type="cloud_composer_environment",
- labels={
- "environment.name": "test-instance",
- "location": "europe-west-3",
- },
- ),
- )
- log_filter = io.prepare_log_filter({"task_id": "test"})
-
- assert 'resource.type="cloud_composer_environment"' in log_filter
- assert 'resource.labels."environment.name"="test-instance"' in
log_filter
- assert 'resource.labels.location="europe-west-3"' in log_filter
-
- @pytest.mark.skipif(not AIRFLOW_V_3_0_PLUS, reason="airflow.sdk.log only
exists in Airflow 3+")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.gcp_logging.Client")
- def test_processors_sends_to_transport(self, mock_client,
mock_get_creds_and_project_id):
- mock_get_creds_and_project_id.return_value = ("creds", "project_id")
-
- mock_transport_type = mock.MagicMock()
- with mock.patch("airflow.sdk.log.relative_path_from_logger",
return_value="dag/task/1.log"):
- io = StackdriverRemoteLogIO(
- base_log_folder=self.local_log_location,
- gcp_log_name="airflow",
- labels={"env": "test"},
- transport_type=mock_transport_type,
- )
- processors = io.processors
- assert len(processors) == 1
-
- proc = processors[0]
- mock_logger = mock.MagicMock()
-
- event = {
- "event": "hello world",
- "logger_name": "airflow.task",
- "timestamp": "2026-01-15T10:30:00+00:00",
- }
- result = proc(mock_logger, "info", event)
-
- assert result is event
- mock_transport = mock_transport_type.return_value
- mock_transport.send.assert_called_once()
- record = mock_transport.send.call_args[0][0]
- assert record.levelno == logging.INFO
-
- @pytest.mark.skipif(not AIRFLOW_V_3_0_PLUS, reason="airflow.sdk.log only
exists in Airflow 3+")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.gcp_logging.Client")
- def test_processors_skips_non_task_logger(self, mock_client,
mock_get_creds_and_project_id):
- mock_get_creds_and_project_id.return_value = ("creds", "project_id")
-
- mock_transport_type = mock.MagicMock()
- with mock.patch("airflow.sdk.log.relative_path_from_logger",
return_value=None):
- io = StackdriverRemoteLogIO(
- base_log_folder=self.local_log_location,
- gcp_log_name="airflow",
- transport_type=mock_transport_type,
- )
- proc = io.processors[0]
-
- event = {"event": "should not be sent"}
- result = proc(mock.MagicMock(), "info", event)
-
- assert result is event
- mock_transport_type.return_value.send.assert_not_called()
-
-
[email protected]("clean_stackdriver_handlers")
[email protected]("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
[email protected]("airflow.providers.google.cloud.log.stackdriver_task_handler.gcp_logging.Client")
-def test_should_pass_message_to_client(mock_client,
mock_get_creds_and_project_id):
- mock_get_creds_and_project_id.return_value = ("creds", "project_id")
-
- transport_type = mock.MagicMock()
- stackdriver_task_handler =
StackdriverTaskHandler(transport=transport_type, labels={"key": "value"})
- logger = logging.getLogger("logger")
- logger.setLevel(logging.INFO)
- logger.addHandler(stackdriver_task_handler)
-
- logger.info("test-message")
- stackdriver_task_handler.flush()
-
- transport_type.assert_called_once_with(mock_client.return_value, "airflow")
- transport_type.return_value.send.assert_called_once_with(
- mock.ANY, "test-message", labels={"key": "value"},
resource=Resource(type="global", labels={})
- )
- mock_client.assert_called_once_with(credentials="creds",
client_info=mock.ANY, project="project_id")
-
-
[email protected]("clean_stackdriver_handlers")
[email protected]("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
[email protected]("airflow.providers.google.cloud.log.stackdriver_task_handler.gcp_logging.Client")
-def test_should_use_configured_log_name(mock_client,
mock_get_creds_and_project_id):
- import importlib
-
- from airflow import settings
- from airflow.config_templates import airflow_local_settings
-
- mock_get_creds_and_project_id.return_value = ("creds", "project_id")
-
- try:
- context_manager = nullcontext()
- with context_manager:
- with conf_vars(
- {
- ("logging", "remote_logging"): "True",
- ("logging", "remote_base_log_folder"):
"stackdriver://host/path",
- }
- ):
- importlib.reload(airflow_local_settings)
- settings.configure_logging()
-
- task_log = getattr(airflow_local_settings, "REMOTE_TASK_LOG",
None)
- if task_log is not None:
- # Airflow 3+ uses REMOTE_TASK_LOG instead of handler-based
config
- assert isinstance(task_log, StackdriverRemoteLogIO)
- assert task_log.gcp_log_name == "path"
- return
-
- # Older Airflow: stackdriver is wired as a logging handler
- logger = logging.getLogger("airflow.task")
- handler = logger.handlers[0]
- assert isinstance(handler, StackdriverTaskHandler)
- finally:
- importlib.reload(airflow_local_settings)
- settings.configure_logging()
-
-
[email protected]_test
-class TestStackdriverLoggingHandlerTask:
- DAG_ID = "dag_for_testing_stackdriver_file_task_handler"
- TASK_ID = "task_for_testing_stackdriver_task_handler"
-
- @pytest.fixture(autouse=True)
- def task_instance(self, create_task_instance, clean_stackdriver_handlers):
- self.ti = create_task_instance(
- dag_id=self.DAG_ID,
- task_id=self.TASK_ID,
- logical_date=timezone.datetime(2016, 1, 1),
- state=TaskInstanceState.RUNNING,
- )
- self.ti.try_number = 1
- self.ti.raw = False
- yield
- clear_db_runs()
- clear_db_dags()
-
- def _setup_handler(self, **handler_kwargs):
- self.transport_mock = mock.MagicMock()
- handler_kwargs = {"transport": self.transport_mock, **handler_kwargs}
- stackdriver_task_handler = StackdriverTaskHandler(**handler_kwargs)
- self.logger = logging.getLogger("logger")
- self.logger.addHandler(stackdriver_task_handler)
- return stackdriver_task_handler
-
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.gcp_logging.Client")
- def test_should_set_labels(self, mock_client,
mock_get_creds_and_project_id):
- mock_get_creds_and_project_id.return_value = ("creds", "project_id")
-
- stackdriver_task_handler = self._setup_handler()
- stackdriver_task_handler.set_context(self.ti)
-
- self.logger.info("test-message")
- stackdriver_task_handler.flush()
-
- date_key = "logical_date" if AIRFLOW_V_3_0_PLUS else "execution_date"
- labels = {
- "task_id": self.TASK_ID,
- "dag_id": self.DAG_ID,
- date_key: "2016-01-01T00:00:00+00:00",
- "try_number": "1",
- }
- resource = Resource(type="global", labels={})
- self.transport_mock.return_value.send.assert_called_once_with(
- mock.ANY, "test-message", labels=labels, resource=resource
- )
-
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.gcp_logging.Client")
- def test_should_append_labels(self, mock_client,
mock_get_creds_and_project_id):
- mock_get_creds_and_project_id.return_value = ("creds", "project_id")
-
- stackdriver_task_handler = self._setup_handler(
- labels={"product.googleapis.com/task_id": "test-value"},
- )
- stackdriver_task_handler.set_context(self.ti)
-
- self.logger.info("test-message")
- stackdriver_task_handler.flush()
-
- date_key = "logical_date" if AIRFLOW_V_3_0_PLUS else "execution_date"
- labels = {
- "task_id": self.TASK_ID,
- "dag_id": self.DAG_ID,
- date_key: "2016-01-01T00:00:00+00:00",
- "try_number": "1",
- "product.googleapis.com/task_id": "test-value",
- }
- resource = Resource(type="global", labels={})
- self.transport_mock.return_value.send.assert_called_once_with(
- mock.ANY, "test-message", labels=labels, resource=resource
- )
-
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.LoggingServiceV2Client")
- def test_should_read_logs_for_all_try(self, mock_client,
mock_get_creds_and_project_id):
- mock_client.return_value.list_log_entries.return_value.pages = iter(
- [_create_list_log_entries_response_mock(["MSG1", "MSG2"], None)]
- )
- mock_get_creds_and_project_id.return_value = ("creds", "project_id")
-
- stackdriver_task_handler = self._setup_handler()
- logs, metadata = stackdriver_task_handler.read(self.ti)
-
- date_label = "logical_date" if AIRFLOW_V_3_0_PLUS else "execution_date"
-
- filter_str = (
- 'resource.type="global"\n'
- 'logName="projects/project_id/logs/airflow"\n'
- 'labels.task_id="task_for_testing_stackdriver_task_handler"\n'
- 'labels.dag_id="dag_for_testing_stackdriver_file_task_handler"\n'
- f'labels.{date_label}="2016-01-01T00:00:00+00:00"'
- )
- mock_client.return_value.list_log_entries.assert_called_once_with(
- request=ListLogEntriesRequest(
- resource_names=["projects/project_id"],
- filter=filter_str,
- order_by="timestamp asc",
- page_size=1000,
- page_token=None,
- )
- )
- assert logs == [(("default-hostname", "MSG1\nMSG2"),)]
- assert metadata == [{"end_of_log": True}]
-
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.LoggingServiceV2Client")
- def test_should_read_logs_for_task_with_quote(self, mock_client,
mock_get_creds_and_project_id):
- mock_client.return_value.list_log_entries.return_value.pages = iter(
- [_create_list_log_entries_response_mock(["MSG1", "MSG2"], None)]
- )
- mock_get_creds_and_project_id.return_value = ("creds", "project_id")
-
- self.ti.task_id = 'K"OT'
- stackdriver_task_handler = self._setup_handler()
-
- logs, metadata = stackdriver_task_handler.read(self.ti)
- date_label = "logical_date" if AIRFLOW_V_3_0_PLUS else "execution_date"
- filter_str = (
- 'resource.type="global"\n'
- 'logName="projects/project_id/logs/airflow"\n'
- 'labels.task_id="K\\"OT"\n'
- 'labels.dag_id="dag_for_testing_stackdriver_file_task_handler"\n'
- f'labels.{date_label}="2016-01-01T00:00:00+00:00"'
- )
- mock_client.return_value.list_log_entries.assert_called_once_with(
- request=ListLogEntriesRequest(
- resource_names=["projects/project_id"],
- filter=filter_str,
- order_by="timestamp asc",
- page_size=1000,
- page_token=None,
- )
- )
- assert logs == [(("default-hostname", "MSG1\nMSG2"),)]
- assert metadata == [{"end_of_log": True}]
-
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.LoggingServiceV2Client")
- def test_should_read_logs_for_single_try(self, mock_client,
mock_get_creds_and_project_id):
- mock_client.return_value.list_log_entries.return_value.pages = iter(
- [_create_list_log_entries_response_mock(["MSG1", "MSG2"], None)]
- )
- mock_get_creds_and_project_id.return_value = ("creds", "project_id")
- stackdriver_task_handler = self._setup_handler()
-
- logs, metadata = stackdriver_task_handler.read(self.ti, 3)
- date_label = "logical_date" if AIRFLOW_V_3_0_PLUS else "execution_date"
- filter_str = (
- 'resource.type="global"\n'
- 'logName="projects/project_id/logs/airflow"\n'
- 'labels.task_id="task_for_testing_stackdriver_task_handler"\n'
- 'labels.dag_id="dag_for_testing_stackdriver_file_task_handler"\n'
- f'labels.{date_label}="2016-01-01T00:00:00+00:00"\n'
- 'labels.try_number="3"'
- )
- mock_client.return_value.list_log_entries.assert_called_once_with(
- request=ListLogEntriesRequest(
- resource_names=["projects/project_id"],
- filter=filter_str,
- order_by="timestamp asc",
- page_size=1000,
- page_token=None,
- )
- )
- assert logs == [(("default-hostname", "MSG1\nMSG2"),)]
- assert metadata == [{"end_of_log": True}]
-
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.LoggingServiceV2Client")
- def test_should_read_logs_with_pagination(self, mock_client,
mock_get_creds_and_project_id):
- mock_client.return_value.list_log_entries.side_effect = [
-
mock.MagicMock(pages=iter([_create_list_log_entries_response_mock(["MSG1",
"MSG2"], "TOKEN1")])),
-
mock.MagicMock(pages=iter([_create_list_log_entries_response_mock(["MSG3",
"MSG4"], None)])),
- ]
- mock_get_creds_and_project_id.return_value = ("creds", "project_id")
- stackdriver_task_handler = self._setup_handler()
-
- logs, metadata1 = stackdriver_task_handler.read(self.ti, 3)
- date_label = "logical_date" if AIRFLOW_V_3_0_PLUS else "execution_date"
- filter_str = (
- 'resource.type="global"\n'
- 'logName="projects/project_id/logs/airflow"\n'
- 'labels.task_id="task_for_testing_stackdriver_task_handler"\n'
- 'labels.dag_id="dag_for_testing_stackdriver_file_task_handler"\n'
- f'labels.{date_label}="2016-01-01T00:00:00+00:00"\n'
- 'labels.try_number="3"'
- )
- mock_client.return_value.list_log_entries.assert_called_once_with(
- request=ListLogEntriesRequest(
- resource_names=["projects/project_id"],
- filter=filter_str,
- order_by="timestamp asc",
- page_size=1000,
- page_token=None,
- )
- )
- assert logs == [(("default-hostname", "MSG1\nMSG2"),)]
- assert metadata1 == [{"end_of_log": False, "next_page_token":
"TOKEN1"}]
-
- mock_client.return_value.list_log_entries.return_value.next_page_token
= None
- logs, metadata2 = stackdriver_task_handler.read(self.ti, 3,
metadata1[0])
-
- mock_client.return_value.list_log_entries.assert_called_with(
- request=ListLogEntriesRequest(
- resource_names=["projects/project_id"],
- filter=(
- 'resource.type="global"\n'
- 'logName="projects/project_id/logs/airflow"\n'
-
'labels.task_id="task_for_testing_stackdriver_task_handler"\n'
-
'labels.dag_id="dag_for_testing_stackdriver_file_task_handler"\n'
- f'labels.{date_label}="2016-01-01T00:00:00+00:00"\n'
- 'labels.try_number="3"'
- ),
- order_by="timestamp asc",
- page_size=1000,
- page_token="TOKEN1",
- )
- )
- assert logs == [(("default-hostname", "MSG3\nMSG4"),)]
- assert metadata2 == [{"end_of_log": True}]
-
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.LoggingServiceV2Client")
- def test_should_read_logs_with_download(self, mock_client,
mock_get_creds_and_project_id):
- mock_client.return_value.list_log_entries.side_effect = [
-
mock.MagicMock(pages=iter([_create_list_log_entries_response_mock(["MSG1",
"MSG2"], "TOKEN1")])),
-
mock.MagicMock(pages=iter([_create_list_log_entries_response_mock(["MSG3",
"MSG4"], None)])),
- ]
- mock_get_creds_and_project_id.return_value = ("creds", "project_id")
-
- stackdriver_task_handler = self._setup_handler()
- logs, metadata1 = stackdriver_task_handler.read(self.ti, 3,
{"download_logs": True})
-
- assert logs == [(("default-hostname", "MSG1\nMSG2\nMSG3\nMSG4"),)]
- assert metadata1 == [{"end_of_log": True}]
-
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.LoggingServiceV2Client")
- def test_should_read_logs_with_custom_resources(self, mock_client,
mock_get_creds_and_project_id):
- mock_get_creds_and_project_id.return_value = ("creds", "project_id")
- resource = Resource(
- type="cloud_composer_environment",
- labels={
- "environment.name": "test-instance",
- "location": "europe-west-3",
- "project_id": "project_id",
- },
- )
- stackdriver_task_handler = self._setup_handler(resource=resource)
-
- entry = mock.MagicMock(json_payload={"message": "TEXT"})
- page = mock.MagicMock(entries=[entry, entry], next_page_token=None)
- mock_client.return_value.list_log_entries.return_value.pages =
iter([page])
-
- logs, metadata = stackdriver_task_handler.read(self.ti)
- date_label = "logical_date" if AIRFLOW_V_3_0_PLUS else "execution_date"
- filter_str = (
- 'resource.type="cloud_composer_environment"\n'
- 'logName="projects/project_id/logs/airflow"\n'
- 'resource.labels."environment.name"="test-instance"\n'
- 'resource.labels.location="europe-west-3"\n'
- 'resource.labels.project_id="project_id"\n'
- 'labels.task_id="task_for_testing_stackdriver_task_handler"\n'
- 'labels.dag_id="dag_for_testing_stackdriver_file_task_handler"\n'
- f'labels.{date_label}="2016-01-01T00:00:00+00:00"'
- )
- mock_client.return_value.list_log_entries.assert_called_once_with(
- request=ListLogEntriesRequest(
- resource_names=["projects/project_id"],
- filter=filter_str,
- order_by="timestamp asc",
- page_size=1000,
- page_token=None,
- )
- )
- assert logs == [(("default-hostname", "TEXT\nTEXT"),)]
- assert metadata == [{"end_of_log": True}]
-
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.gcp_logging.Client")
- def test_should_use_credentials(self, mock_client,
mock_get_creds_and_project_id):
- mock_get_creds_and_project_id.return_value = ("creds", "project_id")
-
- stackdriver_task_handler =
StackdriverTaskHandler(gcp_key_path="KEY_PATH")
- client = stackdriver_task_handler.io._client
-
- mock_get_creds_and_project_id.assert_called_once_with(
- disable_logging=True,
- key_path="KEY_PATH",
- scopes=frozenset(
- {
- "https://www.googleapis.com/auth/logging.write";,
- "https://www.googleapis.com/auth/logging.read";,
- }
- ),
- )
- mock_client.assert_called_once_with(credentials="creds",
client_info=mock.ANY, project="project_id")
- assert mock_client.return_value == client
-
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.LoggingServiceV2Client")
- def test_should_return_valid_external_url(self, mock_client,
mock_get_creds_and_project_id, monkeypatch):
- mock_get_creds_and_project_id.return_value = ("creds", "project_id")
- monkeypatch.setenv("GOOGLE_CLOUD_HIGH_VALUE_COOKIE_DOMAIN",
"googleapis.cn")
-
- stackdriver_task_handler =
StackdriverTaskHandler(gcp_key_path="KEY_PATH")
- url = stackdriver_task_handler.get_external_log_url(self.ti,
self.ti.try_number)
-
- parsed_url = urlsplit(url)
- parsed_qs = parse_qs(parsed_url.query)
- assert parsed_url.scheme == "https"
- assert parsed_url.netloc == "console.cloud.googleapis.cn"
- assert parsed_url.path == "/logs/viewer"
- assert {"project", "interval", "resource", "advancedFilter"} ==
set(parsed_qs.keys())
- assert "global" in parsed_qs["resource"]
-
- filter_params = parsed_qs["advancedFilter"][0].splitlines()
- date_label = "logical_date" if AIRFLOW_V_3_0_PLUS else "execution_date"
- expected_filter = [
- 'resource.type="global"',
- 'logName="projects/project_id/logs/airflow"',
- f'labels.task_id="{self.ti.task_id}"',
- f'labels.dag_id="{self.DAG_ID}"',
- f'labels.{date_label}="{self.ti.logical_date.isoformat() if
AIRFLOW_V_3_0_PLUS else self.ti.execution_date.isoformat()}"',
- f'labels.try_number="{self.ti.try_number}"',
- ]
- assert set(expected_filter) == set(filter_params)
-
-
-class TestStackdriverTaskHandlerExceptionHandling:
- """Cloud Logging failures must degrade gracefully, not leak internals."""
-
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.LoggingServiceV2Client")
- def test_read_falls_back_when_cloud_logging_unavailable(
- self, mock_client, mock_get_creds_and_project_id, caplog
- ):
- """``read()`` must surface a user-facing message when Cloud Logging
raises.
-
- Without a guard, a gRPC error from ``list_log_entries`` propagates as
HTTP 500
- on the log viewer. The fix degrades gracefully and logs the full
traceback for
- the operator.
- """
- from google.api_core import exceptions as gapi_exceptions
-
- mock_get_creds_and_project_id.return_value = ("creds", "project_id")
- mock_client.return_value.list_log_entries.side_effect =
gapi_exceptions.ServiceUnavailable(
- "Stackdriver returned an internal error for project
secret-project-id"
- )
-
- handler = StackdriverTaskHandler()
- ti = mock.MagicMock()
- ti.task_id = "t"
- ti.dag_id = "d"
- ti.try_number = 1
- ti.logical_date = mock.MagicMock(isoformat=lambda:
"2020-01-01T00:00:00+00:00")
- ti.execution_date = ti.logical_date
-
- with caplog.at_level(logging.ERROR):
- logs, metadata = handler.read(ti, try_number=1)
-
- # The user-facing message must NOT include the project id / internal
details.
- message = logs[0][0][1]
- assert "Cloud Logging is currently unavailable" in message
- assert "secret-project-id" not in message
- assert metadata == [{"end_of_log": True}]
-
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
-
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.LoggingServiceV2Client")
- def test_read_does_not_leak_internals_in_user_facing_message(
- self, mock_client, mock_get_creds_and_project_id
- ):
- """``read()`` must not propagate gRPC error details into user-visible
messages.
-
- A ``PermissionDenied`` from ``list_log_entries`` typically carries the
service
- account email + the missing IAM permission. The outer guard in
``read()`` must
- replace it with a generic message so an authenticated user sees no
internal
- identifiers.
- """
- from google.api_core import exceptions as gapi_exceptions
-
- mock_get_creds_and_project_id.return_value = ("creds", "project_id")
- mock_client.return_value.list_log_entries.side_effect =
gapi_exceptions.PermissionDenied(
- "service account '[email protected]' lacks
logging.logEntries.list"
- )
-
- handler = StackdriverTaskHandler()
- ti = mock.MagicMock()
- ti.task_id = "t"
- ti.dag_id = "d"
- ti.try_number = 1
- ti.logical_date = mock.MagicMock(isoformat=lambda:
"2020-01-01T00:00:00+00:00")
- ti.execution_date = ti.logical_date
-
- logs, _ = handler.read(ti, try_number=1)
-
- message = logs[0][0][1]
- assert "Cloud Logging is currently unavailable" in message
- assert "[email protected]" not in message
- assert "logging.logEntries.list" not in message
-
- def test_close_swallows_transport_flush_errors(self, capsys):
- """``close()`` must never raise — even when transport ``flush()``
fails."""
- handler = StackdriverTaskHandler()
- broken_transport = mock.MagicMock()
- broken_transport.flush.side_effect = RuntimeError("flush failed during
shutdown")
- # ``transport`` is a cached_property on the slotted attrs class
- # ``StackdriverRemoteLogIO``; its value lives in a slot, not
``__dict__``, so
- # assign the attribute directly to pre-seed it without building a real
transport.
- handler.io.transport = broken_transport
-
- # Must not raise.
- handler.close()
-
- # The failure is surfaced on stderr because the logging machinery may
be shutting down.
- captured = capsys.readouterr()
- assert "transport flush failed" in captured.err
- assert "flush failed during shutdown" in captured.err
+def test_deprecated_class_warns_and_subclasses_new_class(old_class, new_class,
kwargs):
+ with pytest.warns(AirflowProviderDeprecationWarning, match="CloudLogging"):
+ obj = old_class(**kwargs)
+ assert isinstance(obj, new_class)
+
+
[email protected](
+ ("old_constant", "new_constant"),
+ [
+ (stackdriver_task_handler.DEFAULT_LOGGER_NAME,
cloud_logging_task_handler.DEFAULT_LOGGER_NAME),
+ (stackdriver_task_handler.LABEL_DAG_ID,
cloud_logging_task_handler.LABEL_DAG_ID),
+ (stackdriver_task_handler.LABEL_LOGICAL_DATE,
cloud_logging_task_handler.LABEL_LOGICAL_DATE),
+ (stackdriver_task_handler.LABEL_TASK_ID,
cloud_logging_task_handler.LABEL_TASK_ID),
+ (stackdriver_task_handler.LABEL_TRY_NUMBER,
cloud_logging_task_handler.LABEL_TRY_NUMBER),
+ ],
+)
+def test_deprecated_constants_are_importable(old_constant, new_constant):
+ assert old_constant == new_constant
