Reference
A Python logging handler for efficient and reliable direct log transmission to OpenSearch.
Attributes
module-attribute
urllib3_logger = getLogger('opensearch')
Classes
Context manager to add fields to log records.
Attributes
instance-attribute
added_fields: list[str] = []
Functions
add_log_fields(**values: Any) -> list[str]
Include fields to all log records.
log_fields(func: Optional[Any] = None, **values: Any) -> Any
Decorate to include fields values to all log records.
remove_log_fields(*fields: str) -> None
Remove logged fields.
Modules
Base logger for structured logging.
Classes
Bases: Handler
Abstract base handler for structured logging.
emit(record: LogRecord) -> None
Emit a logging record in a processed format.
send_message(message: Optional[str], record: LogRecord) -> None
Send the log message. This method should be implemented by subclasses.
Structured logger that sends logs to AWS CloudWatch.
Attributes
module-attribute
BUFFER_SIZE = 100
module-attribute
FLUSH_SECONDS = 3.0
module-attribute
RETRY_NUM = 3
Classes
Bases: BaseHandler
Handler that sends log records to AWS CloudWatch.
instance-attribute
buffer_size = BUFFER_SIZE
instance-attribute
flush_seconds = FLUSH_SECONDS
property
log_client: client
Get the boto3 client for CloudWatch logs.
instance-attribute
log_group = log_group
instance-attribute
log_stream = log_stream
close() -> None
Flush the buffer and release any outstanding resource.
flush() -> None
Flush the buffer to CloudWatch.
Could run from Timer's thread or from the main thread. So we should not use json_formatter's routines that access ThreadLocal variables.
send_message(message: Optional[str], record: LogRecord) -> None
Buffer the log message and flush if necessary.
Functions
get_logger(*args: Any, echo_stdout: bool = False, log_group: str, log_stream: str, log_handler: Optional[BaseHandler] = None, **kwargs: Any) -> Logger
Create a logger that stream logs to CloudWatch.
Modules
JSON logging.
Classes
Context manager to add fields to log records.
instance-attribute
added_fields: list[str] = []
Functions
add_log_fields(**values: Any) -> list[str]
Include fields to all log records.
create_logger(log_handler_instance: BaseHandler, level: int, clear_handlers: bool = False) -> Logger
Create a logger that stream logs in JSON format with additional fields.
get_json_formatter() -> JsonFormatter
Get a JSON formatter.
get_logger(application: Optional[str] = sys.argv[0], *, log_handler: BaseHandler, level: int = logging.INFO, clear_handlers: bool = False, **values: Any) -> Logger
Get a JSON logger.
log_fields(func: Optional[Any] = None, **values: Any) -> Any
Decorate to include fields values to all log records.
remove_handlers(_logger: Logger) -> None
Remove any default handlers.
remove_log_fields(*fields: str) -> None
Remove logged fields.
replace_logger_fields(fields_to_log: dict[str, Any]) -> None
Update logger with new fields to log.
set_record_factory() -> None
Set a log record factory.
Structured logger that sends logs to AWS OpenSearch.
Attributes
module-attribute
BUFFER_SIZE = 200
module-attribute
BULK_SIZE = 15
module-attribute
DEFAULT_INDEX_NAME = 'myindex'
module-attribute
DEFAULT_OPENSEARCH_HOST = 'localhost'
module-attribute
FLUSH_SECONDS = 0.8
module-attribute
IGNORED_LOG_RECORD_FIELDS = ['args', 'levelno', 'pathname', 'relativeCreated', 'msecs', 'exc_text']
module-attribute
INDEX_DATE_FORMAT = '%Y.%m.%d'
module-attribute
RETRY_NUM = 3
Classes
Bases: Enum
Index rotation frequency.
class-attribute
instance-attribute
DAILY = 0
class-attribute
instance-attribute
MONTHLY = 10
Bases: BaseHandler
Handler to send log records to AWS OpenSearch.
class-attribute
instance-attribute
DAILY = DAILY
class-attribute
instance-attribute
MONTHLY = MONTHLY
instance-attribute
buffer_size = buffer_size
instance-attribute
flush_seconds = flush_seconds
instance-attribute
index_name = index_name
instance-attribute
index_rotate = IndexRotation[index_rotate]
instance-attribute
opensearch_host = opensearch_host
close() -> None
Flush the buffer on close.
emit(record: LogRecord) -> None
Emit a logging record in a processed format.
flush() -> None
Flush the buffer.
is_connected() -> bool
Return True if we can ping the OpenSearch.
send_message(message: Optional[str], record: LogRecord) -> None
Send the log message to OpenSearch.
Functions
get_logger(*args: Any, opensearch_host: str = DEFAULT_OPENSEARCH_HOST, index_name: str = DEFAULT_INDEX_NAME, index_rotate: Union[IndexRotation, str] = IndexRotation.DAILY, echo_stdout: bool = False, buffer_size: int = BUFFER_SIZE, flush_seconds: float = FLUSH_SECONDS, log_handler: Optional[BaseHandler] = None, **kwargs: Any) -> Logger
Create a logger that stream logs to OpenSearch.
restore_logger() -> None
Flush and remove all handlers.
Modules
JSON serializer.
Classes
Bases: JSONSerializer
Override OpenSearch JSON serializer.
Ignore serialization errors.
default(data: Any) -> Any
Catch all serialization fails and fall to str.
JSON logging to stdout.
Classes
Bases: BaseHandler
Handler to send json log to stdout.
send_message(message: Optional[str], record: LogRecord) -> None
Send the log message to stdout.
Functions
add_stdout_json_handler(logger: Logger) -> None
Add a stdout handler to the logger.
get_logger(application: Optional[str] = None, branch: Optional[str] = None, *, level: int = logging.INFO, clear_handlers: bool = False, **values: Any) -> Logger
Get a logger to send JSON logs to stdout.