feature: Added support for Google cloud log sink management via google provider

docs/spelling_wordlist.txt

@@ -771,6 +771,7 @@ golang
 googleads
 GoogleAdsRow
 googleapiclient
+googleapis
 GoogleDisplayVideo
 gpu
 gpus

providers/google/docs/operators/cloud/cloud_logging_sink.rst

@@ -0,0 +1,102 @@
+ .. 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
+    regarding copyright ownership.  The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License.  You may obtain a copy of the License at
+
+ ..   http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied.  See the License for the
+    specific language governing permissions and limitations
+    under the License.
+
+Google Cloud Logging Sink Operators
+===================================
+
+Cloud Logging allows you to export log entries outside of Cloud Logging using sinks.
+
+For more information, visit `Cloud Logging documentation <https://cloud.google.com/logging/docs/export>`__.
+
+Prerequisite Tasks
+^^^^^^^^^^^^^^^^^^
+
+.. include:: /operators/_partials/prerequisite_tasks.rst
+
+
+Create a log sink (Dict-based)
+------------------------------
+
+You can create a Cloud Logging sink using a Python dictionary:
+
+:class:`~airflow.providers.google.cloud.operators.cloud_logging_sink.CloudLoggingCreateSinkOperator`
+
+.. exampleinclude:: /../../google/tests/system/google/cloud/cloud_logging_sink/example_cloud_logging_sink.py
+   :language: python
+   :dedent: 4
+   :start-after: [START howto_operator_cloud_logging_create_sink_native_obj]
+   :end-before: [END howto_operator_cloud_logging_create_sink_native_obj]
+
+***Required fields in ``sink_config`` (dict):***
+
+- ``name``: The name of the sink.
+- ``destination``: The export destination (e.g., ``storage.googleapis.com/...``, ``bigquery.googleapis.com/...``).
+
+Other fields such as ``description``, ``filter``, ``disabled``, and ``exclusions`` are optional.
+
+Update a log Sink (Protobuf)
+----------------------------
+
+You can also provide the ``sink_config`` as a ``google.cloud.logging_v2.types.LogSink`` Protobuf object,
+and the ``update_mask`` as a ``google.protobuf.field_mask_pb2.FieldMask``.
+
+The following import is required when using a Protobuf object:
+
+.. exampleinclude:: /../../google/tests/system/google/cloud/cloud_logging_sink/example_cloud_logging_sink.py
+   :language: python
+   :start-after: [START howto_operator_import_protobuf_obj]
+   :end-before: [END howto_operator_import_protobuf_obj]
+
+
+
+:class:`~airflow.providers.google.cloud.operators.cloud_logging_sink.CloudLoggingUpdateSinkOperator`
+
+.. exampleinclude:: /../../google/tests/system/google/cloud/cloud_logging_sink/example_cloud_logging_sink.py
+   :language: python
+   :dedent: 4
+   :start-after: [START howto_operator_cloud_logging_update_sink_protobuf_obj]
+   :end-before: [END howto_operator_cloud_logging_update_sink_protobuf_obj]
+
+**When updating a sink**, only include the fields you want to change in ``sink_config``, and list those fields in the ``update_mask["paths"]``.
+
+List log sinks
+--------------
+
+To list all sinks in a Google Cloud project:
+
+:class:`~airflow.providers.google.cloud.operators.cloud_logging_sink.CloudLoggingListSinksOperator`
+
+.. exampleinclude:: /../../google/tests/system/google/cloud/cloud_logging_sink/example_cloud_logging_sink.py
+   :language: python
+   :dedent: 4
+   :start-after: [START howto_operator_cloud_logging_list_sinks]
+   :end-before: [END howto_operator_cloud_logging_list_sinks]
+
+This operator returns a list of sink dictionaries from the project.
+
+Delete a log sink
+-----------------
+
+To delete a sink from a Google Cloud project:
+
+:class:`~airflow.providers.google.cloud.operators.cloud_logging_sink.CloudLoggingDeleteSinkOperator`
+
+.. exampleinclude:: /../../google/tests/system/google/cloud/cloud_logging_sink/example_cloud_logging_sink.py
+   :language: python
+   :dedent: 4
+   :start-after: [START howto_operator_cloud_logging_delete_sink]
+   :end-before: [END howto_operator_cloud_logging_delete_sink]

providers/google/provider.yaml

@@ -450,6 +450,12 @@ integrations:
     how-to-guide:
       - /docs/apache-airflow-providers-google/operators/cloud/looker.rst
     tags: [gcp]
+  - integration-name: Google Cloud Logging Sink
+    external-doc-url: https://cloud.google.com/logging
+    logo: /docs/integration-logos/Cloud-Logging-Sink.png
+    how-to-guide:
+      - /docs/apache-airflow-providers-google/operators/cloud/cloud_logging_sink.rst
+    tags: [gcp]
 
 operators:
   - integration-name: Google Ads
@@ -614,6 +620,9 @@ operators:
   - integration-name: Google Cloud Managed Service for Apache Kafka
     python-modules:
       - airflow.providers.google.cloud.operators.managed_kafka
+  - integration-name: Google Cloud Logging Sink
+    python-modules:
+      - airflow.providers.google.cloud.operators.cloud_logging_sink
 
 sensors:
   - integration-name: Google BigQuery
@@ -894,6 +903,10 @@ hooks:
   - integration-name: Google Cloud Managed Service for Apache Kafka
     python-modules:
       - airflow.providers.google.cloud.hooks.managed_kafka
+  - integration-name: Google Cloud Logging
+    python-modules:
+      - airflow.providers.google.cloud.hooks.cloud_logging
+
 
 triggers:
   - integration-name: Google BigQuery Data Transfer Service

providers/google/src/airflow/providers/google/cloud/hooks/cloud_logging.py

@@ -0,0 +1,109 @@
+#
+# 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
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from collections.abc import Sequence
+from typing import TYPE_CHECKING
+
+from google.cloud.logging_v2.services.config_service_v2 import ConfigServiceV2Client
+from google.cloud.logging_v2.types import (
+    CreateSinkRequest,
+    DeleteSinkRequest,
+    GetSinkRequest,
+    ListSinksRequest,
+    LogSink,
+    UpdateSinkRequest,
+)
+
+from airflow.providers.google.common.consts import CLIENT_INFO
+from airflow.providers.google.common.hooks.base_google import PROVIDE_PROJECT_ID, GoogleBaseHook
+
+if TYPE_CHECKING:
+    from google.protobuf.field_mask_pb2 import FieldMask
+
+
+class CloudLoggingHook(GoogleBaseHook):
+    """
+    Hook for Google Cloud Logging Log Sinks API.
+
+    :param gcp_conn_id: The connection ID to use when fetching connection info.
+    :param impersonation_chain: Optional service account to impersonate.
+    """
+
+    def __init__(
+        self,
+        gcp_conn_id: str = "google_cloud_default",
+        impersonation_chain: str | Sequence[str] | None = None,
+        **kwargs,
+    ) -> None:
+        super().__init__(gcp_conn_id=gcp_conn_id, impersonation_chain=impersonation_chain, **kwargs)
+        self._client: ConfigServiceV2Client | None = None
+
+    def get_conn(self) -> ConfigServiceV2Client:
+        """Return the Google Cloud Logging Config client."""
+        if not self._client:
+            self._client = ConfigServiceV2Client(credentials=self.get_credentials(), client_info=CLIENT_INFO)
+        return self._client
+
+    def get_parent(self, project_id):
+        return f"projects/{project_id}"
+
+    @GoogleBaseHook.fallback_to_default_project_id
+    def create_sink(
+        self, sink: LogSink | dict, unique_writer_identity: bool = True, project_id: str = PROVIDE_PROJECT_ID
+    ) -> LogSink:
+        if isinstance(sink, dict):
+            sink = LogSink(**sink)
+        request = CreateSinkRequest(
+            parent=self.get_parent(project_id), sink=sink, unique_writer_identity=unique_writer_identity
+        )
+        return self.get_conn().create_sink(request=request)
+
+    @GoogleBaseHook.fallback_to_default_project_id
+    def get_sink(self, sink_name: str, project_id: str = PROVIDE_PROJECT_ID) -> LogSink:
+        request = GetSinkRequest(sink_name=f"projects/{project_id}/sinks/{sink_name}")
+        return self.get_conn().get_sink(request=request)
+
+    @GoogleBaseHook.fallback_to_default_project_id
+    def list_sinks(self, page_size: int | None = None, project_id: str = PROVIDE_PROJECT_ID) -> list[LogSink]:
+        request = ListSinksRequest(parent=self.get_parent(project_id), page_size=page_size)
+        return list(self.get_conn().list_sinks(request=request))
+
+    @GoogleBaseHook.fallback_to_default_project_id
+    def delete_sink(self, sink_name: str, project_id: str = PROVIDE_PROJECT_ID) -> None:
+        request = DeleteSinkRequest(sink_name=f"projects/{project_id}/sinks/{sink_name}")
+        self.get_conn().delete_sink(request=request)
+
+    @GoogleBaseHook.fallback_to_default_project_id
+    def update_sink(
+        self,
+        sink_name: str,
+        sink: LogSink | dict,
+        unique_writer_identity: bool,
+        update_mask: FieldMask | dict,
+        project_id: str = PROVIDE_PROJECT_ID,
+    ) -> LogSink:
+        if isinstance(sink, dict):
+            sink = LogSink(**sink)
+        request = UpdateSinkRequest(
+            sink_name=f"projects/{project_id}/sinks/{sink_name}",
+            sink=sink,
+            unique_writer_identity=unique_writer_identity,
+            update_mask=update_mask,
+        )
+        return self.get_conn().update_sink(request=request)