Python Logging and Traces Configuration Example

pip install python-logstash-async

# config.py
 
HOST = "@logstash.host"
PORT = "@logstash.sslPort:strip_quotes"
SSL_ENABLE = True

import logging
from logstash_async.handler import AsynchronousLogstashHandler
from logstash_async.handler import LogstashFormatter
 
# Importing configuration variables from config.py
from config import HOST, PORT, SSL_ENABLE
 
def main():
  try:
    # Create the logger and set it's logging level
    logger = logging.getLogger("logstash")
    logger.setLevel(logging.DEBUG)   
    
    handler = AsynchronousLogstashHandler(
      host=HOST, 
      port=PORT, 
      ssl_enable=SSL_ENABLE, 
      ssl_verify=False,
      transport='logstash_async.transport.BeatsTransport',
      database_path='')
    
    # Here you can specify additional formatting on your log record/message
    formatter = LogstashFormatter()
    handler.setFormatter(formatter)
    
    # Assign handler to the logger
    logger.addHandler(handler)
 
    print("Sending logs to Logit.io.")
    
    # Send log records to Logstash 
    logger.error('python-logstash-async: test error message.')
    logger.info('python-logstash-async: test info message.')
    logger.warning('python-logstash-async: test warning message.')
    logger.debug('python-logstash-async: test debug message.')
    
    # Flush and close the handler to ensure all logs are sent
    # If the application crashes before buffered logs are flushed, 
    # some logs could be lost unless explicitly handled through mechanisms 
    # like flush intervals or proper shutdown processes
    handler.flush()  # Flush the log queue
    handler.close()  # Wait for background thread to finish
 
  except Exception as e:
    print(f"Failed to send log to Logit.io: {e}")
  
if __name__ == '__main__':
    main()

python app.py

pip install flask opentelemetry-api opentelemetry-sdk opentelemetry-instrumentation-flask opentelemetry-exporter-otlp

# config.py
 
SERVICE_NAME = "LogitPythonTestApp"
OTLP_ENDPOINT = "https://@opentelemetry.endpointAddress"
PORT = "@opentelemetry.httpsPort"
USERNAME = "@opentelemetry.username"
PASSWORD = "@opentelemetry.password"

# app.py
 
import base64
from datetime import datetime
from opentelemetry import trace
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from flask import Flask
 
app = Flask(__name__)
 
# Importing configuration variables from config.py
from config import SERVICE_NAME, OTLP_ENDPOINT, PORT, USERNAME, PASSWORD
 
def get_basic_auth_header(username, password):
    """Encodes the username and password to base64 for Basic Auth."""
    credentials = f"{username}:{password}"
    auth_bytes = base64.b64encode(credentials.encode('utf-8'))
    auth_header = auth_bytes.decode('utf-8')
    return f"Basic {auth_header}"
 
def init_tracer():
    # Create a Resource to associate with all traces
    resource = Resource.create({"service.name": SERVICE_NAME})
 
    # Set up the TracerProvider and add the OTLP HTTP exporter with Basic Auth
    tracer_provider = TracerProvider(resource=resource)
    trace.set_tracer_provider(tracer_provider)
 
    # Generate the Basic Auth header
    auth_header = get_basic_auth_header(USERNAME, PASSWORD)
  
    #Generate the full endpoint
    endpoint = f"{OTLP_ENDPOINT}:{PORT}/v1/traces"
 
    # Create the OTLP HTTP exporter with the Basic Auth header
    otlp_exporter = OTLPSpanExporter(
        endpoint=endpoint,
        headers={"Authorization": auth_header}
    )
 
    # Add the exporter to the tracer provider
    span_processor = BatchSpanProcessor(otlp_exporter)
    tracer_provider.add_span_processor(span_processor)
 
@app.route('/')
def hello_world():
    # Initialize the tracer
    init_tracer()
    tracer = trace.get_tracer(__name__)
 
    with tracer.start_as_current_span("parent-span") as parent_span:
        now = datetime.now()
        parent_span.set_attribute("FirstMessage", "This is a test!")
        parent_span.set_attribute("SecondMessage", "Message sent at " + now.strftime("%H:%M:%S"))
 
        # Start the second span within the first span
        with tracer.start_as_current_span("child-span") as child_span:
            child_span.set_attribute("FirstMessage", "This is another test!")
    
    with tracer.start_as_current_span("hello-span"):
        return "Hello, World!"
      
 
if __name__ == '__main__':
    app.run()

python app.py