[AIRFLOW-6372] Align WASB remote logging URI scheme (1.10.x)

airflow/config_templates/airflow_local_settings.py

@@ -190,6 +190,9 @@
 
         DEFAULT_LOGGING_CONFIG['handlers'].update(GCS_REMOTE_HANDLERS)
     elif REMOTE_BASE_LOG_FOLDER.startswith('wasb'):
+
+        # If you use URI scheme, the parameter `wasb_container` is redundancy.
+        # URI Scheme format: wasb://<container_name>@<account_name>.blob.core.windows.net/<path>
         WASB_REMOTE_HANDLERS = {
             'task': {
                 'class': 'airflow.utils.log.wasb_task_handler.WasbTaskHandler',

airflow/config_templates/default_airflow.cfg

@@ -39,7 +39,7 @@ dags_folder = {AIRFLOW_HOME}/dags
 # This path must be absolute
 base_log_folder = {AIRFLOW_HOME}/logs
 
-# Airflow can store logs remotely in AWS S3, Google Cloud Storage or Elastic Search.
+# Airflow can store logs remotely in AWS S3, Azure Blob Storage, Google Cloud Storage or Elasticsearch.
 # Users must supply an Airflow connection id that provides access to the storage
 # location. If remote_logging is set to true, see UPDATING.md for additional
 # configuration requirements.

airflow/utils/log/wasb_task_handler.py

@@ -18,6 +18,7 @@
 # under the License.
 import os
 import shutil
+from urllib.parse import urlparse
 
 from cached_property import cached_property
 
@@ -26,7 +27,6 @@
 from airflow.utils.log.file_task_handler import FileTaskHandler
 from azure.common import AzureHttpError
 
-
 class WasbTaskHandler(FileTaskHandler, LoggingMixin):
     """
     WasbTaskHandler is a python log handler that handles and reads
@@ -129,7 +129,8 @@ def wasb_log_exists(self, remote_log_location):
         :return: True if location exists else False
         """
         try:
-            return self.hook.check_for_blob(self.wasb_container, remote_log_location)
+            container_name, blob = self.parse_wasb_url(remote_log_location)
+            return self.hook.check_for_blob(container_name, blob)
         except Exception:
             pass
         return False
@@ -146,7 +147,8 @@ def wasb_read(self, remote_log_location, return_error=False):
         :type return_error: bool
         """
         try:
-            return self.hook.read_file(self.wasb_container, remote_log_location)
+            container_name, blob = self.parse_wasb_url(remote_log_location)
+            return self.hook.read_file(container_name, blob)
         except AzureHttpError:
             msg = 'Could not read logs from {}'.format(remote_log_location)
             self.log.exception(msg)
@@ -172,11 +174,26 @@ def wasb_write(self, log, remote_log_location, append=True):
             log = '\n'.join([old_log, log]) if old_log else log
 
         try:
+            container_name, blob = self.parse_wasb_url(remote_log_location)
             self.hook.load_string(
                 log,
-                self.wasb_container,
-                remote_log_location,
+                container_name,
+                blob,
             )
         except AzureHttpError:
             self.log.exception('Could not write logs to %s',
                                remote_log_location)
+
+    def parse_wasb_url(self, wasburl):
+        """
+        Given a Azure Blob Storage URL (wasb://<container_name>@<account_name>.blob.core.windows.net/<blob>), returns a
+        tuple containing the corresponding container_name and blob.
+        """
+        parsed_url = urlparse(wasburl)
+
+        blob = parsed_url.path.strip('/')
+        if parsed_url.username:
+            container_name = parsed_url.username
+            return container_name, blob
+        else:
+            return self.wasb_container, blob

docs/howto/write-logs.rst

@@ -72,36 +72,19 @@ In the above example, Airflow will try to use ``S3Hook('MyS3Conn')``.
 Writing Logs to Azure Blob Storage
 ----------------------------------
 
-Airflow can be configured to read and write task logs in Azure Blob Storage.
-
 Follow the steps below to enable Azure Blob Storage logging:
 
-#. Airflow's logging system requires a custom ``.py`` file to be located in the ``PYTHONPATH``, so that it's importable from Airflow. Start by creating a directory to store the config file, ``$AIRFLOW_HOME/config`` is recommended.
-#. Create empty files called ``$AIRFLOW_HOME/config/log_config.py`` and ``$AIRFLOW_HOME/config/__init__.py``.
-#. Copy the contents of ``airflow/config_templates/airflow_local_settings.py`` into the ``log_config.py`` file created in ``Step 2``.
-#. Customize the following portions of the template:
-
-    .. code-block:: ini
-
-        # wasb buckets should start with "wasb" just to help Airflow select correct handler
-        REMOTE_BASE_LOG_FOLDER = 'wasb-<whatever you want here>'
-
-        # Rename DEFAULT_LOGGING_CONFIG to LOGGING CONFIG
-        LOGGING_CONFIG = ...
-
-
-#. Make sure a Azure Blob Storage (Wasb) connection hook has been defined in Airflow. The hook should have read and write access to the Azure Blob Storage bucket defined above in ``REMOTE_BASE_LOG_FOLDER``.
-
-#. Update ``$AIRFLOW_HOME/airflow.cfg`` to contain:
-
-    .. code-block:: bash
+.. code-block:: ini
 
-        remote_logging = True
-        logging_config_class = log_config.LOGGING_CONFIG
-        remote_log_conn_id = <name of the Azure Blob Storage connection>
+    [core]
+    # Airflow can store logs remotely in Azure Blob Storage. Users must supply a remote
+    # location URL (starting with either 'wasb://...') and an Airflow connection
+    # id that provides access to the storage location.
+    remote_logging = True
+    remote_base_log_folder = wasb://container-name@account-name.blob.core.windows.net/path/to/logs
+    remote_log_conn_id = MyWasbConn
 
-#. 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.
+#. Make sure a Azure Blob Storage (Wasb) connection hook has been defined in Airflow. The hook should have read and write access to the Azure Blob Storage container defined above in ``REMOTE_BASE_LOG_FOLDER``.
 
 .. _write-logs-gcp:
 
@@ -115,8 +98,9 @@ example:
 
 .. code-block:: ini
 
+
     [core]
-    # Airflow can store logs remotely in AWS S3, Google Cloud Storage or Elastic Search.
+    # Airflow can store logs remotely in AWS S3, Azure Blob Storage, Google Cloud Storage or Elasticsearch.
     # Users must supply an Airflow connection id that provides access to the storage
     # location. If remote_logging is set to true, see UPDATING.md for additional
     # configuration requirements.
@@ -155,7 +139,7 @@ First, to use the handler, ``airflow.cfg`` must be configured as follows:
 .. code-block:: ini
 
     [core]
-    # Airflow can store logs remotely in AWS S3, Google Cloud Storage or Elastic Search.
+    # Airflow can store logs remotely in AWS S3, Azure Blob Storage, Google Cloud Storage or Elasticsearch.
     # Users must supply an Airflow connection id that provides access to the storage
     # location. If remote_logging is set to true, see UPDATING.md for additional
     # configuration requirements.
@@ -172,7 +156,7 @@ To output task logs to stdout in JSON format, the following config could be used
 .. code-block:: ini
 
     [core]
-    # Airflow can store logs remotely in AWS S3, Google Cloud Storage or Elastic Search.
+    # Airflow can store logs remotely in AWS S3, Azure Blob Storage, Google Cloud Storage or Elasticsearch.
     # Users must supply an Airflow connection id that provides access to the storage
     # location. If remote_logging is set to true, see UPDATING.md for additional
     # configuration requirements.
@@ -190,12 +174,12 @@ To output task logs to stdout in JSON format, the following config could be used
 Writing Logs to Elasticsearch over TLS
 ----------------------------------------
 
-To add custom configurations to ElasticSearch (e.g. turning on ``ssl_verify``, adding a custom self-signed cert, etc.) use the ``elasticsearch_configs`` setting in your ``airfow.cfg``
+To add custom configurations to Elasticsearch (e.g. turning on ``ssl_verify``, adding a custom self-signed cert, etc.) use the ``elasticsearch_configs`` setting in your ``airfow.cfg``
 
 .. code-block:: ini
 
     [core]
-    # Airflow can store logs remotely in AWS S3, Google Cloud Storage or Elastic Search.
+    # Airflow can store logs remotely in AWS S3, Azure Blob Storage, Google Cloud Storage or Elasticsearch.
     # Users must supply an Airflow connection id that provides access to the storage
     # location. If remote_logging is set to true, see UPDATING.md for additional
     # configuration requirements.
@@ -205,3 +189,48 @@ To add custom configurations to ElasticSearch (e.g. turning on ``ssl_verify``, a
     use_ssl=True
     verify_certs=True
     ca_certs=/path/to/CA_certs
+
+.. _customize-write-logs-configuration:
+
+Customize Logging Configuration
+----------------------------------------
+
+Airflow can be configured to override default logging behaviors and add custom logging handlers.
+
+Follow the steps below to customize default logging behaviors:
+
+#. Airflow's logging system requires a custom ``.py`` file to be located in the ``PYTHONPATH``, so that it's importable from Airflow. Start by creating a directory to store the config file, ``$AIRFLOW_HOME/config`` is recommended.
+#. Create empty files called ``$AIRFLOW_HOME/config/log_config.py`` and ``$AIRFLOW_HOME/config/__init__.py``.
+#. Copy the contents of ``airflow/config_templates/airflow_local_settings.py`` into the ``log_config.py`` file created in ``Step 2``.
+#. Customize the following portions of the template:
+
+    .. code-block:: ini
+
+        # Rename DEFAULT_LOGGING_CONFIG to LOGGING CONFIG
+        LOGGING_CONFIG = ...
+
+#. Update ``$AIRFLOW_HOME/airflow.cfg`` to contain:
+
+    .. code-block:: bash
+
+        remote_logging = True
+        logging_config_class = log_config.LOGGING_CONFIG
+        remote_log_conn_id = <name of the Remote Logging System connection>
+
+To add custom logging handler. You must implement a log task handler and add related parameters to above template.
+The configuration may look like:
+
+    .. code-block:: ini
+
+        elif REMOTE_BASE_LOG_FOLDER.startswith('smb://'):
+            SMB_REMOTE_HANDLERS = {
+                'task': {
+                    'class': 'airflow.utils.log.smb_task_handler.SmbTaskHandler',
+                    'formatter': 'airflow',
+                    'base_log_folder': os.path.expanduser(BASE_LOG_FOLDER),
+                    'smb_log_folder': REMOTE_BASE_LOG_FOLDER,
+                    'filename_template': FILENAME_TEMPLATE,
+                },
+            }
+
+        LOGGING_CONFIG['handlers'].update(SMB_REMOTE_HANDLERS)