mirror of
https://git.ianrenton.com/ian/spothole.git
synced 2026-09-20 22:37:44 +00:00
45 lines
1.7 KiB
Python
45 lines
1.7 KiB
Python
from collections import deque
|
|
from datetime import datetime, timedelta
|
|
|
|
import pytz
|
|
|
|
from core.prometheus_metrics_handler import api_requests_counter, page_requests_counter
|
|
|
|
|
|
class WebServerMetrics:
|
|
"""Tracker for web server metrics. Stores the times pages and API endpoints were accessed for an hour, so we
|
|
can display the rate of requests per hour, and also updates the equivalent Prometheus counters."""
|
|
|
|
def __init__(self):
|
|
self.status = "Starting"
|
|
self._page_access_times = deque()
|
|
self._api_access_times = deque()
|
|
|
|
def record(self, path: str, status_code: int):
|
|
"""Records data for a request, depending on whether it's a page, API, or other request, and making sure
|
|
the response code isn't 404. Also sets the status of the web server."""
|
|
|
|
if status_code == 404 or path.startswith(("/static/", "/metrics", "/manifest.webmanifest")):
|
|
return
|
|
|
|
self.status = "OK" if status_code < 500 else "Error"
|
|
if path.startswith("/api/"):
|
|
api_requests_counter.inc()
|
|
self._api_access_times.append(datetime.now(pytz.UTC))
|
|
else:
|
|
page_requests_counter.inc()
|
|
self._page_access_times.append(datetime.now(pytz.UTC))
|
|
|
|
def page_requests_per_hour(self) -> int:
|
|
return self._count_and_prune_last_hour(self._page_access_times)
|
|
|
|
def api_requests_per_hour(self) -> int:
|
|
return self._count_and_prune_last_hour(self._api_access_times)
|
|
|
|
@staticmethod
|
|
def _count_and_prune_last_hour(access_times: deque) -> int:
|
|
cutoff = datetime.now(pytz.UTC) - timedelta(hours=1)
|
|
while access_times and access_times[0] < cutoff:
|
|
access_times.popleft()
|
|
return len(access_times)
|