python-logstash-async

import logging import os from logstash_async.handler import AsynchronousLogstashHandler from logstash_async.formatter import LogstashFormatter # Configure Logstash host and port (replace with your Logstash details) LOGSTASH_HOST = os.environ.get('LOGSTASH_HOST', 'localhost') LOGSTASH_PORT = int(os.environ.get('LOGSTASH_PORT', '5959')) # Or 5000 for standard TCP/UDP LOGSTASH_DATABASE_PATH = os.environ.get('LOGSTASH_DB_PATH', 'logstash_events.db') # Get a logger instance logger = logging.getLogger('my_app_logger') logger.setLevel(logging.INFO) # Create a Logstash formatter formatter = LogstashFormatter(message_type='python-logstash', extra_prefix='dev', extra={'application': 'my-python-app'}) # Create an asynchronous Logstash handler # It's recommended to specify a database_path for persistence across restarts handler = AsynchronousLogstashHandler( host=LOGSTASH_HOST, port=LOGSTASH_PORT, database_path=LOGSTASH_DATABASE_PATH, # For TCP/Beats with SSL, set ssl_enable=True and configure certs # ssl_enable=True, # ssl_verify=True, # ca_certs='/path/to/ca.crt' ) handler.setFormatter(formatter) logger.addHandler(handler) # Add a console handler for local debugging console_handler = logging.StreamHandler() console_handler.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')) logger.addHandler(console_handler) # Log some messages logger.info('This is an info message from the quickstart.', extra={'user_id': 123}) logger.warning('A warning occurred with some extra data.', extra={'component': 'auth', 'status': 'failed'}) logger.error('An error message with detailed context.') # Ensure all queued logs are sent before exiting # In a real application, this might be handled by an atexit hook or proper shutdown logic logger.info('Flushing pending log events...') handler.close() # This will attempt to flush remaining events