haseebmalik18 commented on code in PR #65198:
URL: https://github.com/apache/airflow/pull/65198#discussion_r3192604642
##########
providers/google/src/airflow/providers/google/cloud/log/stackdriver_task_handler.py:
##########
@@ -56,6 +66,209 @@
["https://www.googleapis.com/auth/logging.read";,
"https://www.googleapis.com/auth/logging.write";]
)
+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
+
+ @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, ...]:
+ from datetime import datetime
+ from logging import getLogRecordFactory
+
+ import structlog.stdlib
Review Comment:
`structlog` and `airflow.sdk.log` are lazy here to match the
`CloudWatchRemoteLogIO.processors` pattern. `datetime` and
`getLogRecordFactory` can move to the top. Want me to move just those two?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
