fix(google,log): Avoid log name overriding

Avoid the use of the generic `name` attribute which is overrode by the
dict configurator.

airflow/config_templates/airflow_local_settings.py

@@ -286,7 +286,7 @@
             "task": {
                 "class": "airflow.providers.google.cloud.log.stackdriver_task_handler.StackdriverTaskHandler",
                 "formatter": "airflow",
-                "name": log_name,
+                "gcp_log_name": log_name,
                 "gcp_key_path": key_path,
             }
         }

airflow/providers/google/cloud/log/stackdriver_task_handler.py

@@ -22,16 +22,19 @@
 from functools import cached_property
 from typing import TYPE_CHECKING, Collection
 from urllib.parse import urlencode
+import warnings
 
 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 RemovedInAirflow3Warning
 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.utils.log.trigger_handler import ctx_indiv_trigger
+from airflow.utils.types import NOTSET, ArgNotSet
 
 if TYPE_CHECKING:
     from google.auth.credentials import Credentials
@@ -92,15 +95,25 @@ def __init__(
         self,
         gcp_key_path: str | None = None,
         scopes: Collection[str] | None = _DEFAULT_SCOPESS,
-        name: str = DEFAULT_LOGGER_NAME,
+        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. ",
+                RemovedInAirflow3Warning,
+                stacklevel=2,
+            )
+            gcp_log_name = str(name)
+
         super().__init__()
         self.gcp_key_path: str | None = gcp_key_path
         self.scopes: Collection[str] | None = scopes
-        self.name: str = name
+        self.gcp_log_name: str = gcp_log_name
         self.transport_type: type[Transport] = transport
         self.resource: Resource = resource
         self.labels: dict[str, str] | None = labels
@@ -140,7 +153,7 @@ def _transport(self) -> Transport:
         """Object responsible for sending data to Stackdriver."""
         # The Transport object is badly defined (no init) but in the docs client/name as constructor
         # arguments are a requirement for any class that derives from Transport class, hence ignore:
-        return self.transport_type(self._client, self.name)  # type: ignore[call-arg]
+        return self.transport_type(self._client, self.gcp_log_name)  # type: ignore[call-arg]
 
     def _get_labels(self, task_instance=None):
         if task_instance:
@@ -245,7 +258,7 @@ def escale_label_value(value: str) -> str:
         _, project = self._credentials_and_project
         log_filters = [
             f"resource.type={escale_label_value(self.resource.type)}",
-            f'logName="projects/{project}/logs/{self.name}"',
+            f'logName="projects/{project}/logs/{self.gcp_log_name}"',
         ]
 
         for key, value in self.resource.labels.items():

docs/apache-airflow-providers-google/logging/stackdriver.rst

@@ -33,7 +33,7 @@ example:
     # location. If remote_logging is set to true, see UPDATING.md for additional
     # configuration requirements.
     remote_logging = True
-    remote_base_log_folder = stackdriver://logs-name
+    remote_base_log_folder = stackdriver:///logs-name
 
 All configuration options are in the ``[logging]`` section.
 
@@ -50,7 +50,7 @@ Turning this option off will result in data not being sent to Stackdriver.
 
 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
+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

newsfragments/38071.bugfix.rst

@@ -0,0 +1 @@
+Rename the ``name`` attribute of the StackdriverTaskHandler to avoid name overriding by the the ``DictConfigurator``.

tests/providers/google/cloud/log/test_stackdriver_task_handler.py

@@ -382,3 +382,26 @@ def test_should_return_valid_external_url(self, mock_client, mock_get_creds_and_
             f'labels.try_number="{self.ti.try_number}"',
         ]
         assert set(expected_filter) == set(filter_params)
+
+    def test_should_use_configured_log_name(self):
+        with mock.patch.dict(
+            "os.environ",
+            AIRFLOW__LOGGING__REMOTE_LOGGING="true",
+            AIRFLOW__LOGGING__REMOTE_BASE_LOG_FOLDER="stackdriver://host/path",
+        ):
+            import importlib
+            import logging
+
+            from airflow import settings
+            from airflow.config_templates import airflow_local_settings
+
+            importlib.reload(airflow_local_settings)
+            settings.configure_logging()
+
+            logger = logging.getLogger("airflow.task")
+            handler = logger.handlers[0]
+            assert isinstance(handler, StackdriverTaskHandler)
+            with mock.patch.object(handler, "transport_type") as transport_type_mock:
+                logger.error("foo")
+                transport_type_mock.assert_called_once()
+                assert transport_type_mock.call_args.args[1] == "path"