google_structlog.py

# Use Google Cloud Platform stackdriver with python structlog

from google.cloud.logging import Client
from google.cloud.logging import _helpers
from google.cloud.logging.handlers import CloudLoggingHandler
from google.cloud.logging.handlers.transports.background_thread import _Worker

# pip install python-json-logger
from pythonjsonlogger import jsonlogger

# pip install structlog
import structlog

import datetime
import json
import logging

def monkeypatch_google_enqueue():
  def decode_json_then_enqueue(self, record, message, resource=None, labels=None, trace=None, span_id=None):
    try:
      info = json.loads(message)
    except json.decoder.JSONDecodeError:
      info = { "message": message }
    finally:
      info["python_logger"] = record.name

    queue_entry = {
      "info": info,
      "severity": _helpers._normalize_severity(record.levelno),
      "resource": resource,
      "labels": labels,
      "trace": trace,
      "span_id": span_id,
      "timestamp": datetime.datetime.utcfromtimestamp(record.created),
    }

    self._queue.put_nowait(queue_entry)
  
  _Worker.enqueue = decode_json_then_enqueue

def configure_structlog():
  structlog.configure(
      processors=[
          structlog.stdlib.filter_by_level,
          structlog.stdlib.add_logger_name,
          structlog.stdlib.add_log_level,
          structlog.stdlib.PositionalArgumentsFormatter(),
          structlog.processors.StackInfoRenderer(),
          structlog.processors.format_exc_info,
          structlog.processors.UnicodeDecoder(),
          structlog.stdlib.render_to_log_kwargs,
      ],
      context_class=dict,
      logger_factory=structlog.stdlib.LoggerFactory(),
      wrapper_class=structlog.stdlib.BoundLogger,
      cache_logger_on_first_use=True,
  )

def get_handler(logName):
  handler = CloudLoggingHandler(Client(), logName)
  handler.setFormatter(jsonlogger.JsonFormatter())
  return handler

def setup(log_name=None):
  configure_structlog()
  monkeypatch_google_enqueue()

  if log_name is None:
    try:
      import __main__
      log_name = __main__.__loader__.name.split('.')[0]
    except:
      pass

  handler = get_handler(log_name)
  
  # Add handler to the root logger
  root_logger = logging.getLogger()
  root_logger.addHandler(handler)

mylilapp.py

import google_structlog

google_structlog.setup(log_name="here-is-mylilapp")

# Now you can use structlog to get searchable json details in stackdriver...
import structlog
logger = structlog.get_logger()
logger.error("Uhoh, something bad did", moreinfo="it was bad", years_back_luck=5)

# Of course, you can still use plain ol' logging stdlib to get { "message": ... } objects
import logging
logger = logging.getLogger("yoyo")
logger.error("Regular logging calls will work happily too")

# Now you can search stackdriver with the query:
# logName: 'here-is-mylilapp'

Running mylilapp results in a nice structured log inside GCP stackdriver: