1.A batching handler for the standard logging module
uptimeeye_logs.py
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}})Note: Django: register the handler in
LOGGING['handlers'] and attach it to the django and your app loggers.2.Containers: JSON to stdout
python-json-logger or structlog give you one JSON object per line; the platform agent (Kubernetes, Docker) parses it.
logging setup
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])3.OpenTelemetry
shell
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.pyFields you get
These show up in the fields panel and can be used in every filter:
service,envlevelloggerexceptionwith the formatted traceback- keys passed via
extra
Tips
- Pass structured data via
extra={"fields": {...}}(handler above) or as keyword arguments in structlog — not as f-string fragments in the message. - Gunicorn/uvicorn access logs: attach the same handler to the
gunicorn.access/uvicorn.accessloggers.
FAQ
- Will a slow network block my request handlers?
- No — the handler only appends to an in-memory buffer; the POST happens on a daemon thread or when the buffer is full (bounded by
batch). - Where do I get the ingest key?
- In the app under API Keys → New API Key → type “Log ingest”. The key starts with
ue_ingest_and is shown once. Management keys (ue_live_) are refused by the ingest endpoint. - How do I check that logs arrive?
- Open Logs in the app, pick the 15m range and search for
service:=<your service>. New lines are searchable within about a second; Live tail shows them with a ~6 s delay.
Related
HTTP API / curlThe UptimeEye Logs ingest API: send JSON lines with curl or your own code, choose stream fields, compress with gzip and read the response codes..OpenTelemetryPoint an OpenTelemetry Collector or any OTel SDK at UptimeEye Logs over OTLP/HTTP.KubernetesCollect stdout/stderr of every pod with a Vector DaemonSet and send it to UptimeEye Logs with namespace, container and node attached.Searching logsNow that lines arrive: filters, fields, time ranges and live tail.
