opensearch-logger

import logging import os from opensearch_logger import OpenSearchHandler # Configure logging logger = logging.getLogger("my_app") logger.setLevel(logging.INFO) # OpenSearch connection details (use environment variables for production) host = os.environ.get('OPENSEARCH_HOST', 'localhost') port = int(os.environ.get('OPENSEARCH_PORT', 9200)) username = os.environ.get('OPENSEARCH_USERNAME', 'admin') # Default user for OpenSearch/Dashboards password = os.environ.get('OPENSEARCH_PASSWORD', 'admin') # Default pass for OpenSearch/Dashboards # Handler configuration try: handler = OpenSearchHandler( host=[{'host': host, 'port': port}], http_auth=(username, password), # Required for authenticated clusters use_ssl=True, # Recommended for production; set to False for local http verify_certs=True, # Recommended; set to False for self-signed certs in dev # ssl_assert_hostname=False, # Often needed for dev/test with IP/localhost # ssl_show_warn=False, # Suppress SSL warnings if verify_certs=False index_name='python-app-logs', # Custom index name # auto_create_index=True # Default is True, handler creates index if it doesn't exist ) # Add the handler to the logger logger.addHandler(handler) # Log some messages logger.info("This is an informational message from my application.") logger.warning("A warning occurred: something unexpected happened.") try: 1 / 0 except ZeroDivisionError: logger.exception("An error occurred during division.") print(f"Logs sent to OpenSearch at {host}:{port}") print("Check your OpenSearch instance for 'python-app-logs' index.") except Exception as e: print(f"Failed to configure OpenSearch logger: {e}") print("Ensure OpenSearch is running and accessible with correct credentials/settings.")