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

arvindp25 · 2025-06-21T17:51:06Z

Issues/feature request link: #51929
This pull request introduces new operator in google provider.
With help of new operator airflow user now can create google cloud log sinks via airflow.

### Checklist

- [x] Unit tests added
- [x] System test added
- [x] Documentation updated (docstrings + relevant .rst files if applicable)
- [x] PR description includes motivation and usage example
- [x] Follows Airflow’s [contribution guidelines](https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst)

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

boring-cyborg · 2025-06-21T17:51:09Z

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide (https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
Here are some useful points:

Pay attention to the quality of your code (ruff, mypy and type annotations). Our pre-commits will help you with that.
In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
Be sure to read the Airflow Coding style.
Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
Apache Airflow is a community-driven project and together we are making it better 🚀.
In case of doubts contact the developers at:
Mailing List: [email protected]
Slack: https://s.apache.org/airflow-slack

shahar1

Looks promising overall! Small comment regarding validation of templated fields.
@VladaZakharova - I'll be happy for additional feedback :)

providers/google/src/airflow/providers/google/cloud/operators/cloud_logging_sink.py

VladaZakharova · 2025-06-23T09:27:20Z

Hi
can you please show system tests results with screenshots? thanks

arvindp25 · 2025-06-23T10:21:49Z

Hi can you please show system tests results with screenshots? thanks
@VladaZakharova
for system test, I have recorded video please fast forward as per your need.

video.mp4

further below I have added some additional screenshot from manual dag testing.

creating multiple log sink with different destination

updating log sink

list sink

delete sinks

Please let me know if i have to do something else as Proof of work.
thanks,

shahar1

LGTM, well done!
Waiting for @VladaZakharova 's approval to merge :)

VladaZakharova · 2025-06-24T10:39:09Z

providers/google/src/airflow/providers/google/cloud/operators/cloud_logging_sink.py

+        if self.filter_:
+            sink_config["filter"] = self.filter_
+        if self.description:
+            sink_config["description"] = self.description
+        if self.exclusion_filter:
+            sink_config["exclusions"] = _handle_excluison_filter(self.exclusion_filter)
+        if self.bigquery_options:
+            if isinstance(self.bigquery_options, dict):
+                bigquery_options = logging_v2.types.BigQueryOptions(**self.bigquery_options)
+            sink_config["bigquery_options"] = bigquery_options


Thank you for the changes
Here I think it will be better to ask user to pass the configuration itself in dict format, instead of constructing this config in the operator - in case if some filed will be deprecated in API side we will need to make changes to the operator every time.

Interesting, One doubt if operator accept sink_cofig::dict then it is possible to render it as native object?
or I should not allow it on template field.
`
def test_template_rendering(self):
operator = CloudLoggingCreateSinkOperator(
task_id=TASK_ID,
sink_config = "{{ var.value.sink_config }}",
project_id="{{ var.value.project_id }}"
)

context = { "var": {"value": {"project_id": PROJECT_ID, "sink_config": sink_config_bq}}, } operator.render_template_fields(context) assert operator.sink_config == sink_config_bq

`

I was trying to run this test case
but it fail due to mismatch in type str!=dict
and we cannot pass render_as_native_object=True at operator level.

I think we have example in example_kubernetes.py from cncf provider.
Still the user is passing object itself, not every field separately (because in my experience it adds more problems in the future :) )

VladaZakharova · 2025-06-24T10:42:45Z

providers/google/src/airflow/providers/google/cloud/operators/cloud_logging_sink.py

+
+            response = client.create_sink(
+                request={
+                    "parent": parent,


usually we define and pass parent in the hook itself, as well as constructing request body

VladaZakharova · 2025-06-24T10:43:10Z

providers/google/src/airflow/providers/google/cloud/operators/cloud_logging_sink.py

+        hook = CloudLoggingHook(gcp_conn_id=self.gcp_conn_id, impersonation_chain=self.impersonation_chain)
+
+        client = hook.get_conn()
+        parent = f"projects/{self.project_id}"


it shouldn't be here, can you please move it to hook?

VladaZakharova · 2025-06-24T10:43:29Z

providers/google/src/airflow/providers/google/cloud/operators/cloud_logging_sink.py

+        self._validate_inputs()
+        hook = CloudLoggingHook(gcp_conn_id=self.gcp_conn_id, impersonation_chain=self.impersonation_chain)
+
+        client = hook.get_conn()


this one should be in hooks as well

VladaZakharova · 2025-06-24T10:44:42Z

providers/google/src/airflow/providers/google/cloud/operators/cloud_logging_sink.py

+        hook = CloudLoggingHook(gcp_conn_id=self.gcp_conn_id, impersonation_chain=self.impersonation_chain)
+
+        client = hook.get_conn()
+        sink_path = f"projects/{self.project_id}/sinks/{self.sink_name}"


Can we move it to hook?

Just for my own knowledge; what are you looking to move to the hook?

creating client and constructing sink_path :)
please check other operators we have in google provider for better example

VladaZakharova · 2025-06-24T10:46:45Z

providers/google/src/airflow/providers/google/cloud/operators/cloud_logging_sink.py

+        self._validate_inputs()
+        hook = CloudLoggingHook(gcp_conn_id=self.gcp_conn_id, impersonation_chain=self.impersonation_chain)
+
+        client = hook.get_conn()


same comment, can you please move it to hook?

VladaZakharova · 2025-06-24T10:47:02Z

providers/google/src/airflow/providers/google/cloud/operators/cloud_logging_sink.py

+        hook = CloudLoggingHook(gcp_conn_id=self.gcp_conn_id, impersonation_chain=self.impersonation_chain)
+
+        client = hook.get_conn()
+        sink_path = f"projects/{self.project_id}/sinks/{self.sink_name}"


it should be moved to hook as well

VladaZakharova · 2025-06-24T10:49:34Z

providers/google/src/airflow/providers/google/cloud/operators/cloud_logging_sink.py

+            if self.destination is not None:
+                current_sink.destination = self.destination
+                update_mask.append("destination")
+
+            if self.filter_ is not None:
+                current_sink.filter = self.filter_
+                update_mask.append("filter")
+
+            if self.description is not None:
+                current_sink.description = self.description
+                update_mask.append("description")
+
+            if self.disabled is not None:
+                current_sink.disabled = self.disabled
+                update_mask.append("disabled")
+


here the same comment as for Create operator: we don't usually pass all the parameters in the operator that we want to update. Instead we use update_mask and the body of the NEW object that will be used to update existing one.

so here i can ask user to provide params sink_config and update_mask and pass it to hook, instead of creating mask on my own?

VladaZakharova · 2025-06-24T10:49:49Z

providers/google/src/airflow/providers/google/cloud/operators/cloud_logging_sink.py

+            self.log.info("Updating fields: %s", ", ".join(update_mask))
+
+            response = client.update_sink(
+                request={


the same here about request, can you move it to hook?

VladaZakharova · 2025-06-24T10:51:20Z

providers/google/tests/system/google/cloud/cloud_logging_sink/example_cloud_logging_sink.py

+]
+
+with DAG(
+    dag_id="example_google_cloud_logging_sink",


we removed "example_" from dags names. can you please change it to only "google_cloud_logging_sink"

arvindp25 · 2025-06-28T14:27:02Z

System Test:-
video
https://github.com/user-attachments/assets/01fce6e2-cc4f-4d11-b8c7-b1d06b61a47e

unit test hook:-

unit test operator:-

manual dag testing:-
creating sink ✅

test_bq_sink and test_pubsub_sink got created.

update_sink ✅
my-airflow-test-sink updated successfully.

list sink ✅
total number of sink on project 5 can be matched in log output

delete_sink ✅
all three sink got deleted.

arvindp25 · 2025-06-28T16:37:00Z

@VladaZakharova I have made requested change. Could you please re-review?
change:-
1.moved connection logic to hook
2. used sink_config for configuration support both dict and protobuf obj
3. removed example_ from dag names

Added:-
protobuf support.
system test with protobuf object.

…sink.

VladaZakharova

Looks good, small comments left 😺

VladaZakharova · 2025-06-30T10:39:03Z

providers/google/src/airflow/providers/google/cloud/operators/cloud_logging_sink.py

+        hook = CloudLoggingHook(gcp_conn_id=self.gcp_conn_id, impersonation_chain=self.impersonation_chain)
+
+        try:
+            sink_to_delete = hook.get_sink(sink_name=self.sink_name, project_id=self.project_id)


i am not sure we need this call here to retrieve it, maybe if we call delete() by passing the name only, it's okay to just call delete() alone? Also i hope delete() can throw NotFound if the resource doesn't exist

I can do that but then I will not able to return log sink config
delete api don't response back anything
https://cloud.google.com/logging/docs/reference/v2/rest/v2/projects.sinks/delete
I hope that is OK?

VladaZakharova · 2025-06-30T10:40:44Z

providers/google/src/airflow/providers/google/cloud/operators/cloud_logging_sink.py

+        hook = CloudLoggingHook(gcp_conn_id=self.gcp_conn_id, impersonation_chain=self.impersonation_chain)
+
+        try:
+            current_sink = hook.get_sink(sink_name=self.sink_name, project_id=self.project_id)


Nice way here to show the configuration before updating, good job!

VladaZakharova · 2025-06-30T10:42:47Z

...rs/google/tests/system/google/cloud/cloud_logging_sink/example_cloud_logging_protobuf_obj.py

@@ -0,0 +1,121 @@
+# Licensed to the Apache Software Foundation (ASF) under one


I think we can have only one system test, no need to add 2 almost identical. All the information can be just listed in docs, and that will be enough :)

VladaZakharova · 2025-06-30T10:46:30Z

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

+You can also provide the ``sink_config`` as a native ``google.cloud.logging_v2.types.LogSink`` Protobuf object,
+and the ``update_mask`` as a ``google.protobuf.field_mask_pb2.FieldMask``.
+


this will be enough, tbh, we don't need separate system tests to show that. Maybe you can add to one of the operator in the system tests input in protobuf type

arvindp25 · 2025-06-30T14:04:16Z

@VladaZakharova made requested change. Please check

boring-cyborg bot added area:providers kind:documentation provider:google Google (including GCP) related issues labels Jun 21, 2025

arvindp25 changed the title ~~added support for Google cloud log sink management via google provider~~ Added support for Google cloud log sink management via google provider Jun 21, 2025

arvindp25 changed the title ~~Added support for Google cloud log sink management via google provider~~ feature: Added support for Google cloud log sink management via google provider Jun 21, 2025

shahar1 reviewed Jun 22, 2025

View reviewed changes

providers/google/src/airflow/providers/google/cloud/operators/cloud_logging_sink.py Outdated Show resolved Hide resolved

arvindp25 requested a review from shahar1 June 22, 2025 15:31

arvindp25 force-pushed the feature/add-cloud-logging-export-operator branch from 5889af3 to ee8bc9d Compare June 22, 2025 16:34

shahar1 approved these changes Jun 23, 2025

View reviewed changes

potiuk approved these changes Jun 23, 2025

View reviewed changes

VladaZakharova suggested changes Jun 24, 2025

View reviewed changes

arvindp25 force-pushed the feature/add-cloud-logging-export-operator branch 2 times, most recently from 8455f95 to 83e661b Compare June 28, 2025 12:17

potiuk force-pushed the feature/add-cloud-logging-export-operator branch from 83e661b to 2e67e19 Compare June 28, 2025 14:28

arvindp25 requested a review from VladaZakharova June 28, 2025 16:37

arvindp25 added 4 commits June 29, 2025 10:58

added hook,operator,unit test,system test, docs for google cloud log …

563cff3

…sink.

moved validation in execute and accordingly updated unittest

88e79ed

updated code,docs,test for Google cloud logging sink as requested in PR

ae3b1cb

cloud logging sink:pre-commit check fix

e8e6689

arvindp25 force-pushed the feature/add-cloud-logging-export-operator branch from b88400b to e8e6689 Compare June 29, 2025 10:59

remove db dependency on unittest for clog logging sink

f16f7f5

VladaZakharova suggested changes Jun 30, 2025

View reviewed changes

arvindp25 added 2 commits June 30, 2025 12:25

updated Delete operator code

edb6347

deleted similar system test for cloud logging sink.

08cef5c

arvindp25 requested a review from VladaZakharova June 30, 2025 14:01

		@@ -0,0 +1,121 @@
		# Licensed to the Apache Software Foundation (ASF) under one

		You can also provide the ``sink_config`` as a native ``google.cloud.logging_v2.types.LogSink`` Protobuf object,
		and the ``update_mask`` as a ``google.protobuf.field_mask_pb2.FieldMask``.

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

Are you sure you want to change the base?

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

Uh oh!

Conversation

arvindp25 commented Jun 21, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

boring-cyborg bot commented Jun 21, 2025

Uh oh!

shahar1 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

VladaZakharova commented Jun 23, 2025

Uh oh!

arvindp25 commented Jun 23, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

shahar1 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

arvindp25 Jun 25, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

arvindp25 commented Jun 28, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

System Test:- video https://github.com/user-attachments/assets/01fce6e2-cc4f-4d11-b8c7-b1d06b61a47e

unit test hook:-

unit test operator:-

Uh oh!

arvindp25 commented Jun 28, 2025

Uh oh!

VladaZakharova left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

arvindp25 Jun 30, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

arvindp25 commented Jun 30, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

arvindp25 commented Jun 21, 2025 •

edited

Loading

arvindp25 commented Jun 23, 2025 •

edited

Loading

arvindp25 Jun 25, 2025 •

edited

Loading

arvindp25 commented Jun 28, 2025 •

edited

Loading

System Test:-
video
https://github.com/user-attachments/assets/01fce6e2-cc4f-4d11-b8c7-b1d06b61a47e

arvindp25 Jun 30, 2025 •

edited

Loading

arvindp25 commented Jun 30, 2025 •

edited

Loading