Dieser Artikel beschreibt die manuelle Installation von Grafana Loki auf einem Raspberry Pi und zeigt, wie Log-Daten mit Hilfe von Python auf verschiedene Weisen an Loki übertragen werden. Das Ziel ist es, Logs von mehreren Geräten über serielle Schnittstellen zu empfangen, diese an den Loki-Server zu übertragen und mit Grafana zu visualisieren.
Grafana Loki ist eine zentrale Logging-Lösung, die sich hervorragend zum Speichern, Durchsuchen und Visualisieren von Logs eignet. Die Visualisierung der Log-Daten erfolgt in Grafana mithilfe der eigenen Abfragesprache LogQL. Beliebige Log-Daten wie Linux Syslog lassen sich mit dem Zusatzprogramm Promtail überwachen und kontinuierlich über ein HTTP-API an den Loki-Server übertragen. Das HTTP-API zum Übertagen von Log-Daten kann auch von anderen Anwendungen genutzt werden. So existieren für diverse in der Embedded-Entwicklung häufig eingesetzte Werkzeuge wie Python, MicroPython oder Arduino bereits Bibliotheken, die das Loki HTTP-API direkt verwenden.
Grafana und Loki sind Open-Source-Software, die entweder als Service zur Nutzung in der Grafana Cloud oder als Applikation auf einem Computer, einem Raspberry Pi oder einer virtuellen Maschine installiert werden können. Grafana Loki eignet sich sehr gut für grosse Anwendungen und skaliert hervorragend. In diesem Artikel möchte ich hingegen zeigen, wie sich Grafana Loki mit minimalem Aufwand als entwicklungsbegleitendes Werkzeug für Embedded-Projekte einsetzen lässt.
Installation
Grafana und Loki sind zwei Applikationen welche einzeln installiert werden. Loki speichert die Logs und dient als Datenquelle für die Visualisierung mit Grafana.
Beide Programme lassen sich auf unterschiedliche weisen installieren wie Docker, Docker Compose, Helm, Tanka, APT und RPM. Für APT und RPM muss jedoch erst Grafana Labs als Packet Quelle hinzugefügt werden. Dieser Artikel zeigt hingegen, wie die beiden Applikation manuell heruntergeladen werden und die Binaries direkt gestartet werden.
Raspberry Pi vorbereiten
Voraussetzung für Grafana und Loki ist ein 64-Bit Betriebssystem. Dieser Artikel verwendet einen älteren Rasberry Pi 3 Model B+ mit Raspberry Pi OS Lite 64-Bit. Für den remote Zugriff auf den Raspberry Pi wird VSC mit Remote-SSH verwendet.
Nach dem verbinden über SSH, aktualisiere den Raspberry Pi und installiere die PIP Paketverwaltung:
sudo apt update -y && sudo apt upgrade -y sudo apt install python3-pip -y
Loki Installation
Lade hierfür die Version 3.1.0 von Loki herunter und entpacke diese:
wget https://github.com/grafana/loki/releases/download/v3.1.0/loki-linux-arm64.zip sudo unzip loki-linux-arm64.zip -d /opt/loki
Erstelle das YAML-File zur Konfiguration von Loki im gleichen Unterverzeichnis:
sudo nano /opt/loki/loki-config.yaml
File: loki-config.yaml
auth_enabled: false server: http_listen_port: 3100 grpc_listen_port: 9096 common: instance_addr: 127.0.0.1 path_prefix: /tmp/loki storage: filesystem: chunks_directory: /tmp/loki/chunks rules_directory: /tmp/loki/rules replication_factor: 1 ring: kvstore: store: inmemory schema_config: configs: - from: 2024-01-01 store: tsdb object_store: filesystem schema: v13 index: prefix: index_ period: 24h storage_config: filesystem: directory: /tmp/loki/chunks limits_config: retention_period: 365d query_scheduler: max_outstanding_requests_per_tenant: 10000 ruler: alertmanager_url: http://localhost:9093
Erstelle ein Service File um Loki mit systemd zu steuern.
sudo nano /etc/systemd/system/loki.service
File: loki.service
[Unit] Description=Grafana Loki service After=network.target [Service] Type=simple User=root ExecStart=/opt/loki/loki-linux-arm64 -config.file /opt/loki/loki-config.yaml WorkingDirectory=/opt/loki/ Restart=always RestartSec=10 [Install] WantedBy=multi-user.target
Aktualisiere den systemd daemon, aktiviere den Autostart von Loki und starte Loki.
sudo systemctl daemon-reload sudo systemctl enable loki sudo systemctl start loki sudo systemctl status loki
Grafana installieren
Die Installation von Grafana erfolgt gleich wie die Installation von Loki. Lade hierfür die Version 11.1.3 von Grafana herunter und entpacke diese. Ein Konfigurationsfile ist für Grafana nicht erforderlich.
sudo mkdir /opt/grafana wget https://dl.grafana.com/oss/release/grafana-11.1.3.linux-arm64.tar.gz sudo tar -zxvf grafana-11.1.3.linux-arm64.tar.gz -C /opt/grafana --strip-components=1
Erstelle ein Service File um Grafana mit systemd zu steuern.
sudo nano /etc/systemd/system/grafana.service
File: grafana.service
[Unit] Description=Grafana UI After=network.target [Service] Type=simple User=root ExecStart=/opt/grafana/bin/grafana server WorkingDirectory=/opt/grafana/ Restart=always RestartSec=10 [Install] WantedBy=multi-user.target
Aktualisiere den systemd daemon, aktiviere den Autostart von Grafana und starte Grafana.
sudo systemctl daemon-reload sudo systemctl enable grafana sudo systemctl start grafana sudo systemctl status grafana
Mit den folgenden Schritten wird Grafana mit Loki als Datenquelle verbunden:
- Öffne Grafana mit einem Webbrowser: http://<ip address>:3000
- Benutzername: admin
- Passwort: admin
- [Open menu]➔[Add new connection]➔[Loki]➔[Add new data source]
- Connection URL: http://localhost:3100
- ➔[Save & test]
Damit ist die Installation von Grafana und Loki abgeschlossen und wir können beginnen Log-Daten an den Loki-Server zu senden.
Loki-Log mit Python
Mit Python lassen sich Logs auf unterschiedliche Weisen an Loki übertragen. Am Einfachsten geht dies mit dem Python-Packet python-logging-loki welches die Integration in die Python Logging Umgebung bietet. Alternativ können mit Python die API-Calls direkt genutzt werden, wie in dem Beispiel von push-to-loki.py was mehr Flexibilität für spezifische Anwendungen bietet.
Grafana Labs listet in ihrer Dokumentation eine Vielzahl von weiteren Applikationen und Beispiele in verscheidenden Programmiersprachen zum übertragen von Logs an Loki: Send log data to Loki | Grafana Loki documentation
Loki Python-Library
Folgendes Beispiel verwendet das Python Paket python-logging-loki und eignet sich gut zum schnellen Testen von der zuvor installierten Grafana Loki Umgebung.
File: test.py
import logging
import logging_loki # pip install python-logging-loki
from multiprocessing import Queue
handler = logging_loki.LokiQueueHandler(
Queue(),
url="http://localhost:3100/loki/api/v1/push",
tags={"application": "test"},
version="1",
)
log = logging.getLogger("my-logger")
log.addHandler(handler)
log.info("Hello World")
Zum ausführen des Python Skripts erstellen wir eine neue virtuelle Python Umgebung, installieren darin die erforderlichen Pakete und starten das Python Skript wie folgt:
python3 -m venv .venv .venv/bin/pip install python-logging-loki .venv/bin/python test.py
Loki API-Call nativ mit Python
Das Python-Paket python-logging-loki erstellt den Zeitstempel einer Log-Nachricht erst beim übertragen an Loki wodurch Zeitstempel verfälscht werden können. Insbesondere bei grossen ausstehenden Log-Volumen oder wenn ein Loki Server über eine instabile Verbindung wie WLAN oder LTE verwendet wird.
Der Ursprung von diesem Verhalten könnte sein, dass das Python-Paket vermutlich out-of-order verhindern wollte. Loki seit Version 2.4 unterstützt jedoch per default out-of-order: Allow out of order log submission · Issue #1544 · grafana/loki (github.com)
Demzufolge dürfen wir Logs von verschiedenen Applikationen gleichzeitig an Loki übertragen, auch wenn die Logs insgesamt nicht in Zeitlicher Reihenfolge eintreffen, weil diese zum Beispiel gepuffert wurden. Wir könnten das Python-Paket entsprechend Patchen um Zeitstempel beim erstellen der Logs zu erstellen oder wir können die Loki-API in unserer Applikation direkt verwenden.
Mit folgendem Python-Beispiel, basierend auf push-to-loki.py lässt sich die API direkt ansprechen:
import requests
import json
import time
LOKI_URL = 'http://localhost:3100/loki/api/v1/push'
HEADERS = {'Content-type': 'application/json'}
def send_log(message: str, application: str = "test", level: str = "INFO") -> None:
timestamp = str(time.time_ns())
stream = {
'stream': {
'application': application,
'severity': level
},
'values': [[timestamp, message]],
}
payload = {'streams': [stream]}
try:
response = requests.post(LOKI_URL, json=payload, headers=HEADERS)
response.raise_for_status()
except requests.RequestException as e:
print(f"Failed to send log to Loki: {e}")
if __name__ == "__main__":
send_log("Hello World")
Vollständiger USB-Logger in Python
Das folgende Python Skript öffnet und schliesst automatisch alle verfügbaren Seriellen Schnittstellen und überträgt die Log-Daten Zeilenweise an Loki. Die Logs werden in einer Queue zwischengespeichert und übertragen, sobald eine Verbindung zum Loki-Server besteht.
Erstelle eine virtuelle Python Umgebung und installiere die erforderlichen Pakete
sudo python3 -m venv /opt/myLogger/venv sudo /opt/myLogger/venv/bin/pip install requests pySerial
Erstelle im gleichen Verzeichnis die Datei myLogger.py:
sudo nano /opt/myLogger/myLogger.py
File: myLogger.py
import requests # pip install requests
import time
from threading import Thread
from queue import Queue
import serial # pip install pySerial
import selectors
import glob
from os.path import normpath, basename
LOKI_URL = 'http://localhost:3100/loki/api/v1/push'
HEADERS = {'Content-type': 'application/json'}
def send_log(queue, application):
while (log := queue.get()) != 'exit':
time_ns, host, level, msg = log
stream = {
'stream': {
'host': host,
'application': application,
'severity': level
},
'values': [[str(time_ns), msg]],
}
payload = {'streams': [stream]}
try:
response = requests.post(LOKI_URL, json=payload, headers=HEADERS)
response.raise_for_status()
except requests.RequestException:
queue.put((time_ns, host, level, msg))
class Logger:
def __init__(self, application):
self.application = application
self.queue = Queue()
self.thread = Thread(target=send_log, args=(self.queue,self.application))
def __enter__(self):
self.thread.start()
return self
def __exit__(self, type, value, traceback):
self.queue.put('exit')
self.thread.join()
def debug(self, host, msg):
self.queue.put((time.time_ns(), host, 'debug', msg))
def info(self, host, msg):
self.queue.put((time.time_ns(), host, 'info', msg))
def warning(self, host, msg):
self.queue.put((time.time_ns(), host, 'warning', msg))
def error(self, host, msg):
self.queue.put((time.time_ns(), host, 'error', msg))
if __name__ == "__main__":
with Logger('usb_logger') as log:
try:
sel = selectors.DefaultSelector()
ser_ports = {}
while True:
for key, _ in sel.select(timeout=1):
try:
ser = key.fileobj
host = basename(normpath(ser.name))
while ser.in_waiting:
val = ser.readline()
if msg := val.decode("utf-8", "ignore").strip():
log.debug(host, msg)
except serial.SerialException:
pass
except OSError:
pass
available_ports = set(glob.glob("/dev/serial/by-id/*"))
open_ports = set([ ports.name for ports in ser_ports ])
new_ports = available_ports - open_ports
closed_ports = open_ports - available_ports
for port in new_ports:
port_name = basename(normpath(port))
msg = f'*** open {port_name} ***'
print(msg)
log.info(port_name, msg)
new_ser = serial.Serial(port, baudrate=115200, timeout=0.1)
ser_ports[new_ser] = selectors.EVENT_READ
sel.register(new_ser, selectors.EVENT_READ)
for port in closed_ports:
for ser in ser_ports:
if ser.name == port:
port_name = basename(normpath(port))
msg = f'*** close {port_name} ***'
print(msg)
log.info(port_name, msg)
sel.unregister(ser)
ser.close()
del ser_ports[ser]
break
time.sleep(0.1) # secs
except KeyboardInterrupt:
pass
finally:
[ ser.close() for ser in ser_ports.keys() ]
sel.close()
Erstelle ein Service File um myLogger mit systemd zu steuern.
sudo nano /etc/systemd/system/myLogger.service
File: myLogger.service
[Unit] Description=myLogger Service After=multi-user.target [Service] Type=idle ExecStart=/opt/myLogger/venv/bin/python /opt/myLogger/myLogger.py Environment=PYTHONUNBUFFERED=1 [Install] WantedBy=multi-user.target
Aktualisiere den systemd daemon, aktiviere den Autostart von myLogger und starte myLogger.
sudo systemctl daemon-reload sudo systemctl enable myLogger sudo systemctl start myLogger sudo systemctl status myLogger
Grafana Loki Testaufbau
Für den Test verwenden wir zwei Raspberry Pi Pico als Log-Quelle und ein Pi 3 Model B+ als Log-Server. Auf den zwei Pico ist MicroPython installiert und ein Python Skript gibt fortlaufend Sinuswerte als Text aus. Auf dem Raspberry laufen Grafana, Loki und das myLogger.py Python Skript. Das myLogger.py Skript erkennt serielle Schnittstellen automatisch und sendet den empfangenen Log als Text zusammen mit der ID der seriellen Schnittstelle an Loki weiter. Dank der automatischen Detektion lassen sich USB-Log-Quellen im Betrieb flexibel ein- und ausstecken. Selbst bei vertauschten USB-Anschlüssen bleibt dank der ID der gleiche Name für ihre Erkennung bestehen.
Der Aufbau lässt sich flexibel erweitern mit USB-Hubs oder mit Raspberry Pis welche nur das Python-Skript ausführen und den gleichen gemeinsamen Loki Server verwenden. Damit eignet sich dieser Aufbau hervorragend zum überwachen von Dauertests mit vielen parallel laufenden Testgeräten.
File: main.py
import time
import math
# Frequenz der Sinus-Schwingung
frequency = 0.01
# Zeit-Intervall für die Ausgabe (in Sekunden)
interval = 1
# Startzeitpunkt
start_time = time.time()
while True:
# Aktuelle Zeit berechnen
current_time = time.time() - start_time
# Sinuswert berechnen
sin_value = math.sin(2 * math.pi * frequency * current_time)
# Sinuswert ausgeben
print(f"Zeit: {current_time:.2f}s, Sinuswert: {sin_value:.4f}")
# Eine Sekunde warten
time.sleep(interval)
Grafana Dashboard erstellen
In folgendem Beispiel konfigurieren wir ein einfaches Grafana-Dashboard mit Log-Ausgabe, Log-Visualisierung, Host-Filter und Suchfilter. Loki verwendet die Abfragespreche LogQL welche viele Möglichkeiten für das Filtern und Verarbeiten der Log-Daten bietet. Das finale Dashboard zur Visualisierung unserer Testdaten sieht wie folgt aus:
Dashboard Host-Auswahl hinzufügen
Erstelle mit folgendem Vorgehen ein Drop-Down-Menü im Grafana-Dashboard, das alle im Log verfügbaren Hosts auflistet und eine Auswahl ermöglicht. Diese Auswahlvariable lässt sich später in den Abfragen als Filter wiederverwenden und damit einzelne Hosts gezielt analysiert werden.
- [Open menu]➔[Dashboards]➔[Create dashboard]
- [Settings]➔[Variables]➔[New Variable]
- Select variable type: Query
- General – Name: host
- Query – Query type: Label values
- Query – Label: host
- Query – Stream selector:
{service_name="usb_logger"} - Query – Sort: Alphabetical (asc)
- Selection options – Check Multi-value
- Selection options – Check Include All option
- ➔[Apply]
Dashboard Suche hinzufügen
Ergänze das Grafana-Dashboard um ein Suchfeld, das als Filterparameter für Log-Abfragen eingesetzt werden kann.
- [Settings]➔[Variables]➔[New Variable]
- Select variable type: Text box
- General – Name: search
- General – Label: search
- ➔[Apply]
Visualisierung der Log Ausgabe
Erstelle im Grafana Dashboard eine neue Visualisierung vom Typ “Logs” und verwende den folgende LogQL-Abfrage um die Host-Auswahl und die Suche anzuwenden:
{host=~"$host"}
|~ `$search`
Visualisierung mit Pattern-Parser
Erstelle im Grafana Dashboard eine neue Visualisierung vom Typ “Time series” und verwende folgende LogQL-Abfrage. Diese filtert von den ausgewählten Hosts alle Log Einträge mit dem begriff “Sinuswert”, parst den Zahlenwert und stellt diesen grafisch dar.
avg_over_time({host=~"$host"}
|= `Sinuswert`
| pattern `<_> Sinuswert: <x>`
| unwrap x
| __error__=`` [$__interval])
Visualisierung mit Regex
Klassisches Regex ermöglicht die gleiche Visualisierung, allerdings ist das Programmieren weniger intuitiv.
avg_over_time({host=~"$host"}
|= `Sinuswert`
| regexp `.*Sinuswert: (?P<x>[-+]?\d*\.\d+|\d+)`
| unwrap x
| __error__=`` [$__interval])
Fazit
Grafana Loki bietet sich hervorragend als Entwicklungswerkzeug oder für Dauertests im Embedded-Bereich an, wenn eine zentralisierte Logging-Lösung gewünscht ist. Die Installation ist unkompliziert und die Integration mit Grafana intuitiv. Ausserdem lassen sich Unregelmässigkeiten in Logs visuell schnell erfassen. Während LogQL anfangs etwas anspruchsvoll sein kann, bietet es dennoch flexible Möglichkeiten, grosse Log-Mengen effizient zu durchsuchen und zu verarbeiten. Bei Noser wurde Grafana Loki bereits erfolgreich in mehreren Embedded-Projekten eingesetzt.
