python-logging-loki

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

Installation

pip install 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: Optional[list[str]] = ["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

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(...)