title: Airflow Google Cloud Logging
body: <p>For <strong>Apache Airflow</strong> v1.10 running in Python2.7, with `pip install airflow[gcp_api] I am trying to setup logging for the Google Cloud. I have the following log_config py file:</p>
  
  <pre><code>GCS_LOG_FOLDER = 'gs://GCSbucket/'
  LOG_LEVEL = conf.get('core', 'LOGGING_LEVEL').upper()
  
  FAB_LOG_LEVEL = conf.get('core', 'FAB_LOGGING_LEVEL').upper()
  
  LOG_FORMAT = conf.get('core', 'LOG_FORMAT')
  
  BASE_LOG_FOLDER = conf.get('core', 'BASE_LOG_FOLDER')
  
  PROCESSOR_LOG_FOLDER = conf.get('scheduler', 'CHILD_PROCESS_LOG_DIRECTORY')
  
  FILENAME_TEMPLATE = conf.get('core', 'LOG_FILENAME_TEMPLATE')
  PROCESSOR_FILENAME_TEMPLATE = conf.get('core', 'LOG_PROCESSOR_FILENAME_TEMPLATE')
  
  # Storage bucket url for remote logging
  # s3 buckets should start with "s3://"
  # gcs buckets should start with "gs://"
  # wasb buckets should start with "wasb"
  # just to help Airflow select correct handler
   REMOTE_BASE_LOG_FOLDER = conf.get('core', 'REMOTE_BASE_LOG_FOLDER')
  
  LOGGING_CONFIG = {
  'version': 1,
  'disable_existing_loggers': False,
  'formatters': {
      'airflow': {
          'format': LOG_FORMAT,
      },
  },
  'handlers': {
      'console': {
          'class': 'airflow.utils.log.logging_mixin.RedirectStdHandler',
          'formatter': 'airflow',
          'stream': 'sys.stdout'
      },
      'task': {
          'class': 'airflow.utils.log.file_task_handler.FileTaskHandler',
          'formatter': 'airflow',
          'base_log_folder': os.path.expanduser(BASE_LOG_FOLDER),
          'filename_template': FILENAME_TEMPLATE,
      },
      'processor': {
          'class': 'airflow.utils.log.file_processor_handler.FileProcessorHandler',
          'formatter': 'airflow',
          'base_log_folder': os.path.expanduser(PROCESSOR_LOG_FOLDER),
          'filename_template': PROCESSOR_FILENAME_TEMPLATE,
      },
      # Add a GCSTaskHandler to the 'handlers' block of the LOGGING_CONFIG variable
      'gcs.task': {
          'class': 'airflow.utils.log.gcs_task_handler.GCSTaskHandler',
          'formatter': 'airflow',
          'base_log_folder': os.path.expanduser(BASE_LOG_FOLDER),
          'gcs_log_folder': GCS_LOG_FOLDER,
          'filename_template': FILENAME_TEMPLATE,
      },
  },
  'loggers': {
      'airflow.processor': {
          'handlers': ['processor'],
          'level': LOG_LEVEL,
          'propagate': False,
      },
      'airflow.task': {
          'handlers': ['gcs.task'],
          'level': LOG_LEVEL,
          'propagate': False,
      },
      'airflow.task_runner': {
          'handlers': ['gcs.task'],
          'level': LOG_LEVEL,
          'propagate': True,
      },
      'flask_appbuilder': {
          'handler': ['console'],
          'level': FAB_LOG_LEVEL,
          'propagate': True,
      }
  },
  'root': {
      'handlers': ['console'],
      'level': LOG_LEVEL,
  }
  }
  
  REMOTE_HANDLERS = {
  's3': {
      'task': {
          'class': 'airflow.utils.log.s3_task_handler.S3TaskHandler',
          'formatter': 'airflow',
          'base_log_folder': os.path.expanduser(BASE_LOG_FOLDER),
          's3_log_folder': REMOTE_BASE_LOG_FOLDER,
          'filename_template': FILENAME_TEMPLATE,
      },
      'processor': {
          'class': 'airflow.utils.log.s3_task_handler.S3TaskHandler',
          'formatter': 'airflow',
          'base_log_folder': os.path.expanduser(PROCESSOR_LOG_FOLDER),
          's3_log_folder': REMOTE_BASE_LOG_FOLDER,
          'filename_template': PROCESSOR_FILENAME_TEMPLATE,
      },
  },
  'gcs': {
      'task': {
          'class': 'airflow.utils.log.gcs_task_handler.GCSTaskHandler',
          'formatter': 'airflow',
          'base_log_folder': os.path.expanduser(BASE_LOG_FOLDER),
          'gcs_log_folder': REMOTE_BASE_LOG_FOLDER,
          'filename_template': FILENAME_TEMPLATE,
      },
      'processor': {
          'class': 'airflow.utils.log.gcs_task_handler.GCSTaskHandler',
          'formatter': 'airflow',
          'base_log_folder': os.path.expanduser(PROCESSOR_LOG_FOLDER),
          'gcs_log_folder': REMOTE_BASE_LOG_FOLDER,
          'filename_template': PROCESSOR_FILENAME_TEMPLATE,
      },
  },
  'wasb': {
      'task': {
          'class': 'airflow.utils.log.wasb_task_handler.WasbTaskHandler',
          'formatter': 'airflow',
          'base_log_folder': os.path.expanduser(BASE_LOG_FOLDER),
          'wasb_log_folder': REMOTE_BASE_LOG_FOLDER,
          'wasb_container': 'airflow-logs',
          'filename_template': FILENAME_TEMPLATE,
          'delete_local_copy': False,
      },
      'processor': {
          'class': 'airflow.utils.log.wasb_task_handler.WasbTaskHandler',
          'formatter': 'airflow',
          'base_log_folder': os.path.expanduser(PROCESSOR_LOG_FOLDER),
          'wasb_log_folder': REMOTE_BASE_LOG_FOLDER,
          'wasb_container': 'airflow-logs',
          'filename_template': PROCESSOR_FILENAME_TEMPLATE,
          'delete_local_copy': False,
      },
  }
  }
  
  REMOTE_LOGGING = conf.get('core', 'remote_logging')
  
  if REMOTE_LOGGING and REMOTE_BASE_LOG_FOLDER.startswith('s3://'):
   LOGGING_CONFIG['handlers'].update(REMOTE_HANDLERS['s3'])
  elif REMOTE_LOGGING and REMOTE_BASE_LOG_FOLDER.startswith('gs://'):
   LOGGING_CONFIG['handlers'].update(REMOTE_HANDLERS['gcs'])
  elif REMOTE_LOGGING and REMOTE_BASE_LOG_FOLDER.startswith('wasb'):
   LOGGING_CONFIG['handlers'].update(REMOTE_HANDLERS['wasb'])
  </code></pre>
  
  <p>My <code>airflow.cfg</code> settings are:</p>
  
  <pre><code>[core]
  remote_logging = True
  remote_base_log_folder = gs:/GCSbucket/logs 
  remote_log_conn_id = google_cloud_default 
  </code></pre>
  
  <p>The error I get is the following:</p>
  
  <pre><code>Traceback (most recent call last):
    File "/usr/local/lib/python2.7/atexit.py", line 24, in _run_exitfuncs
      func(*targs, **kargs)
    File "/usr/local/lib/python2.7/logging/__init__.py", line 1676, in shutdown
      h.close()
    File "/usr/local/lib/python2.7/site-packages/airflow/utils/log/gcs_task_handler.py", line 73, in close
      if self.closed:
  AttributeError: 'GCSTaskHandler' object has no attribute 'closed'
  </code></pre>
  
  <p>Does anybody know what might have gone wrong?
  The tutorial that is being followed is: <a href="https://airflow.readthedocs.io/en/1.10.0/howto/write-logs.html" rel="nofollow noreferrer">https://airflow.readthedocs.io/en/1.10.0/howto/write-logs.html</a></p>
  
  <p>Update: Did some more research in the source code, here I see that the close statement returns nothing, and this is why my application crashes. </p>
  
  <p><a href="https://github.com/apache/incubator-airflow/blob/v1-10-stable/airflow/utils/log/gcs_task_handler.py" rel="nofollow noreferrer">https://github.com/apache/incubator-airflow/blob/v1-10-stable/airflow/utils/log/gcs_task_handler.py</a></p>
  
  <p>Does somebody know why nothing is returend in </p>
  
  <pre><code>def close(self):
      if self.closed:
           return
  </code></pre>