Python-Logs an UptimeEye senden

import atexit, json, logging, os, threading, time, urllib.request

ENDPOINT = "https://logs.uptimeeye.com/v1/ingest/jsonline?_stream_fields=service,env"
KEY = os.environ["UPTIMEEYE_INGEST_KEY"]

class UptimeEyeHandler(logging.Handler):
    """Buffers records and posts them as JSON lines every 2 s or 200 records."""

    def __init__(self, service, env="prod", batch=200, interval=2.0):
        super().__init__()
        self.base = {"service": service, "env": env}
        self.batch, self.interval = batch, interval
        self.buf, self.lock = [], threading.Lock()
        threading.Thread(target=self._loop, daemon=True).start()
        atexit.register(self.flush)

    def emit(self, record):
        entry = {
            **self.base,
            "_time": time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(record.created)) + f".{int(record.msecs):03d}Z",
            "_msg": record.getMessage(),
            "level": record.levelname.lower(),
            "logger": record.name,
            **getattr(record, "fields", {}),
        }
        if record.exc_info:
            entry["exception"] = self.formatter.formatException(record.exc_info) if self.formatter else logging.Formatter().formatException(record.exc_info)
        with self.lock:
            self.buf.append(json.dumps(entry, default=str))
            if len(self.buf) >= self.batch:
                self._post()

    def _loop(self):
        while True:
            time.sleep(self.interval)
            self.flush()

    def flush(self):
        with self.lock:
            self._post()

    def _post(self):
        if not self.buf:
            return
        body = ("\n".join(self.buf) + "\n").encode()
        self.buf = []
        req = urllib.request.Request(ENDPOINT, data=body, method="POST",
            headers={"Authorization": f"Bearer {KEY}", "Content-Type": "application/x-ndjson"})
        try:
            urllib.request.urlopen(req, timeout=10).read()
        except Exception as err:  # never let logging crash the app
            print("uptimeeye logs:", err)

# usage
log = logging.getLogger("checkout")
log.setLevel(logging.INFO)
log.addHandler(UptimeEyeHandler(service="checkout"))
log.info("payment authorized", extra={"fields": {"order_id": "4711", "duration_ms": 812}})

import logging, sys
from pythonjsonlogger.json import JsonFormatter

handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JsonFormatter(
    "%(asctime)s %(levelname)s %(name)s %(message)s",
    rename_fields={"asctime": "time", "levelname": "level", "name": "logger"},
    static_fields={"service": "checkout", "env": "prod"},
    datefmt="%Y-%m-%dT%H:%M:%S%z",
))
logging.basicConfig(level=logging.INFO, handlers=[handler])

pip install opentelemetry-sdk opentelemetry-exporter-otlp-proto-http

export OTEL_SERVICE_NAME=checkout
export OTEL_RESOURCE_ATTRIBUTES=deployment.environment=prod
export OTEL_PYTHON_LOGGING_AUTO_INSTRUMENTATION_ENABLED=true
export OTEL_LOGS_EXPORTER=otlp
export OTEL_EXPORTER_OTLP_LOGS_PROTOCOL=http/protobuf
export OTEL_EXPORTER_OTLP_LOGS_ENDPOINT=https://logs.uptimeeye.com/v1/ingest/otlp/v1/logs
export OTEL_EXPORTER_OTLP_LOGS_HEADERS="Authorization=Bearer%20ue_ingest_XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"

opentelemetry-instrument python app.py