Skip to content

Instantly share code, notes, and snippets.

@trenchawr
Last active April 7, 2024 01:53
Show Gist options
  • Save trenchawr/6183e7322126b251b0edb5eeb430f74d to your computer and use it in GitHub Desktop.
Save trenchawr/6183e7322126b251b0edb5eeb430f74d to your computer and use it in GitHub Desktop.
aws lamdba compatible logtail handler in python
import logging
import os
import queue
from logging import Logger
from typing import Optional
from logtail import LogtailHandler, handler, flusher, uploader
_logger: Optional[Logger] = None
class CustomLogtailHandler(LogtailHandler):
"""
logtail handler implementation compatible with AWS Lambda functions,
using queue.Queue instead of multiprocessing.queue since the latter uses multiprocessing.SemLock
which is incompatible with aws lambdas
Note: use logtail-python==0.1.4 only, since constructor is being overriden completely
"""
def __init__(self, source_token, level=logging.NOTSET):
# init grandparent but not parent to avoid instantiating the multiprocessing.queue
super(LogtailHandler, self).__init__(level=level)
# override parent initialization
self.source_token = source_token
self.host = handler.DEFAULT_HOST
self.context = handler.DEFAULT_CONTEXT
self.pipe = queue.Queue(maxsize=handler.DEFAULT_BUFFER_CAPACITY)
self.uploader = uploader.Uploader(self.source_token, self.host)
self.drop_extra_events = handler.DEFAULT_DROP_EXTRA_EVENTS
self.include_extra_attributes = handler.DEFAULT_INCLUDE_EXTRA_ATTRIBUTES
self.buffer_capacity = handler.DEFAULT_BUFFER_CAPACITY
self.flush_interval = handler.DEFAULT_FLUSH_INTERVAL
self.raise_exceptions = handler.DEFAULT_RAISE_EXCEPTIONS
self.dropcount = 0
self.flush_thread = flusher.FlushWorker(
self.uploader,
self.pipe,
self.buffer_capacity,
self.flush_interval
)
if self._is_main_process():
self.flush_thread.start()
def get_logger(name: Optional[str] = None) -> Logger:
"""
Creates a lambda specific logger with a standardized log format
"""
global _logger
if _logger:
return _logger
_logger = logging.getLogger(name)
_logger.setLevel(logging.DEBUG)
# create logging formatter
formatter = logging.Formatter('%(asctime)s :: %(name)s :: %(levelname)s :: %(message)s')
# add console handler
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.DEBUG)
console_handler.setFormatter(formatter)
_logger.addHandler(console_handler)
# add logtail handler
logtail_source_token = "YOUR_KEY"
if logtail_source_token:
_logger.debug('logtail source token found! Adding custom logtail handler')
logtail_handler = CustomLogtailHandler(source_token=logtail_source_token)
logtail_handler.setLevel(logging.DEBUG)
logtail_handler.setFormatter(formatter)
_logger.addHandler(logtail_handler)
_logger.debug('custom logtail handler initiated')
else:
_logger.debug('logtail source token not found in secrets. Only console stream will be added')
return _logger
@Yosefjos
Copy link

Yosefjos commented Apr 7, 2024

Doker

@Yosefjos
Copy link

Yosefjos commented Apr 7, 2024


@Yosefjos
Copy link

Yosefjos commented Apr 7, 2024

Uploading 660f211ecc9e6032c1d50925.png…

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