python-logging-loki

Python logging handler for Loki.
https://grafana.com/loki

Installation

pip install inuits-python-logging-loki

Usage

import logging
import logging_loki


handler = logging_loki.LokiHandler(
    url="https://my-loki-instance/loki/api/v1/push",
    tags={"application": "my-app"},
    headers={"X-Scope-OrgID": "example-id"},
    auth=("username", "password"),
    props_to_labels = ["foo"]
)

logger = logging.getLogger("my-logger")
logger.addHandler(handler)
logger.error(
    "Something happened",
    extra={"tags": {"service": "my-service"}},
)

Example above will send Something happened message along with these labels:

• Default labels from handler
• Message level as serverity
• Logger's name as logger
• Labels from tags item of extra dict
• Property foo from log record will be sent as loki label

Properties to label

Using a dict instead of a list for props_to_labels will enable renaming labels

handler = logging_loki.LokiHandler(
    url="https://my-loki-instance/loki/api/v1/push",
    tags={"application": "my-app"},
    props_to_labels = {
        "otelTraceID": "trace_id"
        "otelSpanID":  "span_id"
    }
)

In this case, the properties otelTraceID & otelSpanID will be renamed to trace_id & span_id loki labels

Non-blocking mode

The given example is blocking (i.e. each call will wait for the message to be sent).
But you can use the built-in QueueHandler and QueueListener to send messages in a separate thread.

import logging.handlers
import logging_loki
from multiprocessing import Queue


queue = Queue(-1)
handler = logging.handlers.QueueHandler(queue)
handler_loki = logging_loki.LokiHandler(
    url="https://my-loki-instance/loki/api/v1/push",
    tags={"application": "my-app"},
    headers={"X-Scope-OrgID": "example-id"},
    auth=("username", "password"),
    props_to_labels: Optional[list[str]] = ["foo"]
)
logging.handlers.QueueListener(queue, handler_loki)

logger = logging.getLogger("my-logger")
logger.addHandler(handler)
logger.error(...)

Or you can use LokiQueueHandler shortcut, which will automatically create listener and handler.

import logging.handlers
import logging_loki
from multiprocessing import Queue


handler = logging_loki.LokiQueueHandler(
    Queue(-1),
    url="https://my-loki-instance/loki/api/v1/push",
    tags={"application": "my-app"},
    auth=("username", "password"),
)

logger = logging.getLogger("my-logger")
logger.addHandler(handler)
logger.error(...)