Skip to content

Instantly share code, notes, and snippets.

@snickell
Last active March 26, 2023 13:21
Show Gist options
  • Save snickell/3b664c668b8a94d1ca6e31b0498a8697 to your computer and use it in GitHub Desktop.
Save snickell/3b664c668b8a94d1ca6e31b0498a8697 to your computer and use it in GitHub Desktop.
# 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)
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'
@snickell
Copy link
Author

Running mylilapp results in a nice structured log inside GCP stackdriver:
Screen Shot 2020-05-21 at 5 31 04 AM

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment