Simplified GCSTaskHandler configuration (#10365)

Author: mik-laj

UPDATING.md

@@ -561,6 +561,20 @@ better handle the case when a DAG file has multiple DAGs.
 Sentry is disabled by default. To enable these integrations, you need set ``sentry_on`` option
 in ``[sentry]`` section to ``"True"``.
 
+#### Simplified GCSTaskHandler configuration
+
+In previous versions, in order to configure the service account key file, you had to create a connection entry.
+In the current version, you can configure ``google_key_path`` option in ``[logging]`` section to set
+the key file path.
+
+Users using Application Default Credentials (ADC) need not take any action.
+
+The change aims to simplify the configuration of logging, to prevent corruption of
+the instance configuration by changing the value controlled by the user - connection entry. If you
+configure a backend secret, it also means the webserver doesn't need to connect to it. This
+simplifies setups with multiple GCP projects, because only one project will require the Secret Manager API
+to be enabled.
+
 ### Changes to the core operators/hooks
 
 We strive to ensure that there are no changes that may affect the end user and your files, but this

airflow/config_templates/airflow_local_settings.py

@@ -196,13 +196,15 @@
 
         DEFAULT_LOGGING_CONFIG['handlers'].update(CLOUDWATCH_REMOTE_HANDLERS)
     elif REMOTE_BASE_LOG_FOLDER.startswith('gs://'):
+        key_path = conf.get('logging', 'GOOGLE_KEY_PATH', fallback=None)
         GCS_REMOTE_HANDLERS: Dict[str, Dict[str, str]] = {
             'task': {
-                'class': 'airflow.utils.log.gcs_task_handler.GCSTaskHandler',
+                'class': 'airflow.providers.google.cloud.log.gcs_task_handler.GCSTaskHandler',
                 'formatter': 'airflow',
                 'base_log_folder': str(os.path.expanduser(BASE_LOG_FOLDER)),
                 'gcs_log_folder': REMOTE_BASE_LOG_FOLDER,
                 'filename_template': FILENAME_TEMPLATE,
+                'gcp_key_path': key_path
             },
         }
 
@@ -222,7 +224,7 @@
 
         DEFAULT_LOGGING_CONFIG['handlers'].update(WASB_REMOTE_HANDLERS)
     elif REMOTE_BASE_LOG_FOLDER.startswith('stackdriver://'):
-        key_path = conf.get('logging', 'STACKDRIVER_KEY_PATH', fallback=None)
+        key_path = conf.get('logging', 'GOOGLE_KEY_PATH', fallback=None)
         # stackdriver:///airflow-tasks => airflow-tasks
         log_name = urlparse(REMOTE_BASE_LOG_FOLDER).path[1:]
         STACKDRIVER_REMOTE_HANDLERS = {

airflow/config_templates/config.yml

@@ -402,9 +402,9 @@
       type: string
       example: ~
       default: ""
-    - name: stackdriver_key_path
+    - name: google_key_path
       description: |
-        Path to GCP Credential JSON file. If omitted, authorization based on `the Application Default
+        Path to Google 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.

airflow/config_templates/default_airflow.cfg

@@ -230,11 +230,11 @@ remote_logging = False
 # location.
 remote_log_conn_id =
 
-# Path to GCP Credential JSON file. If omitted, authorization based on `the Application Default
+# Path to Google 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.
-stackdriver_key_path =
+google_key_path =
 
 # Storage bucket URL for remote logging
 # S3 buckets should start with "s3://"

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

@@ -16,48 +16,86 @@
 # specific language governing permissions and limitations
 # under the License.
 import os
-from urllib.parse import urlparse
+from typing import Collection, Optional
 
 from cached_property import cached_property
+from google.api_core.client_info import ClientInfo
+from google.cloud import storage
 
-from airflow.configuration import conf
-from airflow.exceptions import AirflowException
+from airflow import version
+from airflow.providers.google.cloud.utils.credentials_provider import get_credentials_and_project_id
 from airflow.utils.log.file_task_handler import FileTaskHandler
 from airflow.utils.log.logging_mixin import LoggingMixin
 
+_DEFAULT_SCOPESS = frozenset([
+    "https://www.googleapis.com/auth/devstorage.read_write",
+])
+
 
 class GCSTaskHandler(FileTaskHandler, LoggingMixin):
     """
     GCSTaskHandler is a python log handler that handles and reads
     task instance logs. It extends airflow FileTaskHandler and
     uploads to and reads from GCS remote storage. Upon log reading
     failure, it reads from host machine's local disk.
+
+    :param base_log_folder: Base log folder to place logs.
+    :type base_log_folder: str
+    :param gcs_log_folder: Path to a remote location where logs will be saved. It must have the prefix
+        ``gs://``. For example: ``gs://bucket/remote/log/location``
+    :type gcs_log_folder: str
+    :param filename_template: template filename string
+    :type filename_template: str
+    :param gcp_key_path: Path to GCP Credential JSON file. Mutually exclusive with gcp_keyfile_dict.
+        If omitted, authorization based on `the Application Default Credentials
+        <https://cloud.google.com/docs/authentication/production#finding_credentials_automatically>`__ will
+        be used.
+    :type gcp_key_path: str
+    :param gcp_keyfile_dict: Dictionary of keyfile parameters. Mutually exclusive with gcp_key_path.
+    :type gcp_keyfile_dict: dict
+    :param gcp_scopes: Comma-separated string containing GCP scopes
+    :type gcp_scopes: str
+    :param project_id: Project ID to read the secrets from. If not passed, the project ID from credentials
+        will be used.
+    :type project_id: str
     """
-    def __init__(self, base_log_folder, gcs_log_folder, filename_template):
+    def __init__(
+        self,
+        *,
+        base_log_folder: str,
+        gcs_log_folder: str,
+        filename_template: str,
+        gcp_key_path: Optional[str] = None,
+        gcp_keyfile_dict: Optional[dict] = None,
+        # See: https://github.com/PyCQA/pylint/issues/2377
+        gcp_scopes: Optional[Collection[str]] = _DEFAULT_SCOPESS,  # pylint: disable=unsubscriptable-object
+        project_id: Optional[str] = None,
+    ):
         super().__init__(base_log_folder, filename_template)
         self.remote_base = gcs_log_folder
         self.log_relative_path = ''
         self._hook = None
         self.closed = False
         self.upload_on_close = True
+        self.gcp_key_path = gcp_key_path
+        self.gcp_keyfile_dict = gcp_keyfile_dict
+        self.scopes = gcp_scopes
+        self.project_id = project_id
 
     @cached_property
-    def hook(self):
-        """
-        Returns GCS hook.
-        """
-        remote_conn_id = conf.get('logging', 'REMOTE_LOG_CONN_ID')
-        try:
-            from airflow.providers.google.cloud.hooks.gcs import GCSHook
-            return GCSHook(
-                google_cloud_storage_conn_id=remote_conn_id
-            )
-        except Exception as e:  # pylint: disable=broad-except
-            self.log.error(
-                'Could not create a GoogleCloudStorageHook with connection id '
-                '"%s". %s\n\nPlease make sure that airflow[gcp] is installed '
-                'and the GCS connection exists.', remote_conn_id, str(e)
-            )
+    def client(self) -> storage.Client:
+        """Returns GCS Client."""
+        credentials, project_id = get_credentials_and_project_id(
+            key_path=self.gcp_key_path,
+            keyfile_dict=self.gcp_keyfile_dict,
+            scopes=self.scopes,
+            disable_logging=True
+        )
+        return storage.Client(
+            credentials=credentials,
+            client_info=ClientInfo(client_library_version='airflow_v' + version.version),
+            project=self.project_id if self.project_id else project_id
+        )
 
     def set_context(self, ti):
         super().set_context(ti)
@@ -111,7 +149,8 @@ def _read(self, ti, try_number, metadata=None):
         remote_loc = os.path.join(self.remote_base, log_relative_path)
 
         try:
-            remote_log = self.gcs_read(remote_loc)
+            blob = storage.Blob.from_string(remote_loc, self.client)
+            remote_log = blob.download_as_string()
             log = '*** Reading remote log from {}.\n{}\n'.format(
                 remote_loc, remote_log)
             return log, {'end_of_log': True}
@@ -123,19 +162,9 @@ def _read(self, ti, try_number, metadata=None):
             log += local_log
             return log, metadata
 
-    def gcs_read(self, remote_log_location):
-        """
-        Returns the log found at the remote_log_location.
-
-        :param remote_log_location: the log's location in remote storage
-        :type remote_log_location: str (path)
-        """
-        bkt, blob = self.parse_gcs_url(remote_log_location)
-        return self.hook.download(bkt, blob).decode('utf-8')
-
     def gcs_write(self, log, remote_log_location):
         """
-        Writes the log to the remote_log_location. Fails silently if no hook
+        Writes the log to the remote_log_location. Fails silently if no log
         was created.
 
         :param log: the log to write to the remote_log_location
@@ -144,28 +173,16 @@ def gcs_write(self, log, remote_log_location):
         :type remote_log_location: str (path)
         """
         try:
-            old_log = self.gcs_read(remote_log_location)
+            blob = storage.Blob.from_string(remote_log_location, self.client)
+            old_log = blob.download_as_string()
             log = '\n'.join([old_log, log]) if old_log else log
         except Exception as e:  # pylint: disable=broad-except
             if not hasattr(e, 'resp') or e.resp.get('status') != '404':  # pylint: disable=no-member
                 log = '*** Previous log discarded: {}\n\n'.format(str(e)) + log
+                self.log.info("Previous log discarded: %s", e)
 
         try:
-            bkt, blob = self.parse_gcs_url(remote_log_location)
-            self.hook.upload(bkt, blob, data=log)
+            blob = storage.Blob.from_string(remote_log_location, self.client)
+            blob.upload_from_string(log, content_type="text/plain")
         except Exception as e:  # pylint: disable=broad-except
             self.log.error('Could not write logs to %s: %s', remote_log_location, e)
-
-    @staticmethod
-    def parse_gcs_url(gsurl):
-        """
-        Given a Google Cloud Storage URL (gs://<bucket>/<blob>), returns a
-        tuple containing the corresponding bucket and blob.
-        """
-        parsed_url = urlparse(gsurl)
-        if not parsed_url.netloc:
-            raise AirflowException('Please provide a bucket name')
-        else:
-            bucket = parsed_url.netloc
-            blob = parsed_url.path.strip('/')
-            return bucket, blob

docs/howto/write-logs.rst

@@ -194,10 +194,11 @@ example:
     # configuration requirements.
     remote_logging = True
     remote_base_log_folder = gs://my-bucket/path/to/logs
-    remote_log_conn_id = MyGCSConn
 
-#. Install the ``google`` package first, like so: ``pip install 'apache-airflow[google]'``.
-#. Make sure a Google Cloud Platform connection hook has been defined in Airflow. The hook should have read and write access to the Google Cloud Storage bucket defined above in ``remote_base_log_folder``.
+#. 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 a Google Cloud Platform account have read and write access to the Google Cloud Storage bucket defined above in ``remote_base_log_folder``.
+#. 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've defined.
 #. Verify that the Google Cloud Storage viewer is working in the UI. Pull up a newly executed task, and verify that you see something like:
@@ -311,7 +312,7 @@ 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``.
 
-You can set ``stackdriver_key_path`` option in the ``[logging]`` section to specify the path to `the service
+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>`__.
 If omitted, authorization based on `the Application Default Credentials
 <https://cloud.google.com/docs/authentication/production#finding_credentials_automatically>`__ will