630 lines
23 KiB
Python
630 lines
23 KiB
Python
"""
|
|
MotorSignal Operations Infrastructure Metrics Collector.
|
|
|
|
Safe, read-only system resource monitoring, service status, network totals,
|
|
and deployment health foundations.
|
|
"""
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import time
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.request import Request, urlopen
|
|
|
|
REVISION_HEX_REGEX = re.compile(r"^[0-9a-fA-F]{40}$")
|
|
MAX_VERSION_RESPONSE_BYTES = 4096
|
|
DEFAULT_WEBSITE_INTERNAL_URL = "http://127.0.0.1:3001"
|
|
|
|
# Fixed mapping from public logical service ID to actual systemd unit name
|
|
SERVICE_SYSTEMD_UNITS: Dict[str, str] = {
|
|
"motorsignal-website": "motorsignal-website",
|
|
"motorsignal-api": "motorsignal-api",
|
|
"nginx": "nginx",
|
|
"postgresql": "postgresql@14-main",
|
|
}
|
|
|
|
ALLOWED_SYSTEMD_SERVICES: Tuple[str, ...] = tuple(SERVICE_SYSTEMD_UNITS.keys())
|
|
|
|
SERVICE_METADATA: Dict[str, Dict[str, str]] = {
|
|
"motorsignal-website": {
|
|
"display_name": "MotorSignal Website",
|
|
"description": "Next.js frontend application runtime",
|
|
},
|
|
"motorsignal-api": {
|
|
"display_name": "MotorSignal API",
|
|
"description": "FastAPI core backend service & ingress",
|
|
},
|
|
"nginx": {
|
|
"display_name": "Nginx Reverse Proxy",
|
|
"description": "Edge TLS termination & reverse proxy routing",
|
|
},
|
|
"postgresql": {
|
|
"display_name": "PostgreSQL Database",
|
|
"description": "Primary relational database persistence",
|
|
},
|
|
}
|
|
|
|
# In-memory previous CPU stat cache for non-blocking delta calculation
|
|
_prev_cpu_stat: Optional[Tuple[float, int, int]] = None # (timestamp, idle_ticks, total_ticks)
|
|
|
|
|
|
@dataclass
|
|
class OpsDeploymentEvent:
|
|
"""Foundational data model for deployment events (Phase 2 readiness)."""
|
|
id: str
|
|
source_revision: str
|
|
initiating_operator_id: str
|
|
started_at: datetime
|
|
completed_at: Optional[datetime]
|
|
status: str # 'pending' | 'in_progress' | 'success' | 'failed' | 'rolled_back'
|
|
release_identifier: Optional[str] = None
|
|
failure_category: Optional[str] = None
|
|
previous_revision: Optional[str] = None
|
|
resulting_revision: Optional[str] = None
|
|
|
|
|
|
def _read_proc_file(path: str) -> Optional[str]:
|
|
"""Safely read a /proc file if it exists."""
|
|
try:
|
|
if os.path.exists(path):
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
return f.read()
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
|
|
def get_cpu_percent() -> Optional[float]:
|
|
"""
|
|
Calculate overall CPU percent.
|
|
Uses /proc/stat deltas on Linux.
|
|
Degrades safely to None if unavailable.
|
|
"""
|
|
global _prev_cpu_stat
|
|
|
|
stat_content = _read_proc_file("/proc/stat")
|
|
if stat_content:
|
|
for line in stat_content.splitlines():
|
|
if line.startswith("cpu "):
|
|
parts = line.split()
|
|
if len(parts) >= 5:
|
|
try:
|
|
# parts: cpu, user, nice, system, idle, iowait, irq, softirq, steal, guest, guest_nice
|
|
fields = [int(p) for p in parts[1:]]
|
|
idle = fields[3] + (fields[4] if len(fields) > 4 else 0)
|
|
total = sum(fields)
|
|
now = time.time()
|
|
|
|
if _prev_cpu_stat is not None:
|
|
prev_time, prev_idle, prev_total = _prev_cpu_stat
|
|
delta_time = now - prev_time
|
|
delta_total = total - prev_total
|
|
delta_idle = idle - prev_idle
|
|
|
|
if 0.1 <= delta_time <= 60.0 and delta_total > 0:
|
|
_prev_cpu_stat = (now, idle, total)
|
|
used_pct = ((delta_total - delta_idle) / delta_total) * 100.0
|
|
return round(max(0.0, min(100.0, used_pct)), 1)
|
|
|
|
# Sample quick delta if no previous sample exists
|
|
time.sleep(0.05)
|
|
stat_content_2 = _read_proc_file("/proc/stat")
|
|
if stat_content_2:
|
|
for line2 in stat_content_2.splitlines():
|
|
if line2.startswith("cpu "):
|
|
fields2 = [int(p) for p in line2.split()[1:]]
|
|
idle2 = fields2[3] + (fields2[4] if len(fields2) > 4 else 0)
|
|
total2 = sum(fields2)
|
|
now2 = time.time()
|
|
_prev_cpu_stat = (now2, idle2, total2)
|
|
delta_total = total2 - total
|
|
delta_idle = idle2 - idle
|
|
if delta_total > 0:
|
|
used_pct = ((delta_total - delta_idle) / delta_total) * 100.0
|
|
return round(max(0.0, min(100.0, used_pct)), 1)
|
|
|
|
_prev_cpu_stat = (now, idle, total)
|
|
except Exception:
|
|
pass
|
|
break
|
|
|
|
return None
|
|
|
|
|
|
def get_load_averages() -> Optional[Tuple[float, float, float]]:
|
|
"""Return 1m, 5m, 15m load averages via os.getloadavg()."""
|
|
try:
|
|
if hasattr(os, "getloadavg"):
|
|
avg = os.getloadavg()
|
|
return (round(avg[0], 2), round(avg[1], 2), round(avg[2], 2))
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
|
|
def get_memory_metrics() -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Parse /proc/meminfo for RAM metrics.
|
|
Returns total, used, available in bytes, and used percent.
|
|
"""
|
|
mem_content = _read_proc_file("/proc/meminfo")
|
|
if mem_content:
|
|
mem_info: Dict[str, int] = {}
|
|
for line in mem_content.splitlines():
|
|
parts = line.split(":")
|
|
if len(parts) == 2:
|
|
key = parts[0].strip()
|
|
val_parts = parts[1].strip().split()
|
|
if val_parts and val_parts[0].isdigit():
|
|
mem_info[key] = int(val_parts[0]) * 1024 # kB to bytes
|
|
|
|
if "MemTotal" in mem_info:
|
|
total = mem_info["MemTotal"]
|
|
avail = mem_info.get(
|
|
"MemAvailable",
|
|
mem_info.get("MemFree", 0) + mem_info.get("Buffers", 0) + mem_info.get("Cached", 0)
|
|
)
|
|
used = max(0, total - avail)
|
|
pct = round((used / total) * 100.0, 1) if total > 0 else 0.0
|
|
return {
|
|
"total": total,
|
|
"used": used,
|
|
"available": avail,
|
|
"percent": pct,
|
|
}
|
|
|
|
return None
|
|
|
|
|
|
def get_swap_metrics() -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Parse /proc/meminfo for Swap metrics.
|
|
Returns total, used in bytes, and used percent.
|
|
"""
|
|
mem_content = _read_proc_file("/proc/meminfo")
|
|
if mem_content:
|
|
mem_info: Dict[str, int] = {}
|
|
for line in mem_content.splitlines():
|
|
parts = line.split(":")
|
|
if len(parts) == 2:
|
|
key = parts[0].strip()
|
|
val_parts = parts[1].strip().split()
|
|
if val_parts and val_parts[0].isdigit():
|
|
mem_info[key] = int(val_parts[0]) * 1024
|
|
|
|
if "SwapTotal" in mem_info:
|
|
total = mem_info["SwapTotal"]
|
|
free = mem_info.get("SwapFree", 0)
|
|
used = max(0, total - free)
|
|
pct = round((used / total) * 100.0, 1) if total > 0 else 0.0
|
|
return {
|
|
"total": total,
|
|
"used": used,
|
|
"percent": pct,
|
|
}
|
|
|
|
return None
|
|
|
|
|
|
def get_disk_metrics(mount_point: str = "/") -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Calculate disk usage for the root filesystem.
|
|
Includes warning (>= 80%) and critical (>= 90%) threshold flags.
|
|
"""
|
|
try:
|
|
# On Windows during local testing, "/" maps to current drive, but check valid path
|
|
path = mount_point
|
|
if os.name == "nt" and mount_point == "/":
|
|
path = os.environ.get("SystemDrive", "C:") + "\\"
|
|
|
|
total, used, free = shutil.disk_usage(path)
|
|
pct = round((used / total) * 100.0, 1) if total > 0 else 0.0
|
|
return {
|
|
"total": total,
|
|
"used": used,
|
|
"available": free,
|
|
"percent": pct,
|
|
"warning": pct >= 80.0,
|
|
"critical": pct >= 90.0,
|
|
}
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def get_uptime() -> Tuple[Optional[int], Optional[str]]:
|
|
"""
|
|
Parse /proc/uptime for server uptime.
|
|
Returns (uptime_seconds, uptime_formatted).
|
|
"""
|
|
uptime_content = _read_proc_file("/proc/uptime")
|
|
if uptime_content:
|
|
parts = uptime_content.split()
|
|
if parts:
|
|
try:
|
|
seconds = int(float(parts[0]))
|
|
days = seconds // 86400
|
|
hours = (seconds % 86400) // 3600
|
|
minutes = (seconds % 3600) // 60
|
|
formatted = f"{days}d {hours}h {minutes}m"
|
|
return seconds, formatted
|
|
except Exception:
|
|
pass
|
|
|
|
return None, None
|
|
|
|
|
|
def get_system_resources() -> Dict[str, Any]:
|
|
"""Collect overall server system resource metrics."""
|
|
uptime_sec, uptime_fmt = get_uptime()
|
|
return {
|
|
"cpu_percent": get_cpu_percent(),
|
|
"load_avg": get_load_averages(),
|
|
"memory": get_memory_metrics(),
|
|
"swap": get_swap_metrics(),
|
|
"disk": get_disk_metrics("/"),
|
|
"uptime_seconds": uptime_sec,
|
|
"uptime_formatted": uptime_fmt,
|
|
"collected_at": datetime.now(timezone.utc).isoformat(),
|
|
}
|
|
|
|
|
|
def get_service_status(service_name: str) -> str:
|
|
"""
|
|
Query systemctl is-active for an allowed logical service name.
|
|
Maps public logical service ID to actual fixed systemd unit (e.g. postgresql -> postgresql@14-main).
|
|
Strictly forbids arbitrary service names, direct unit overrides, and shell interpolation.
|
|
"""
|
|
unit_name = SERVICE_SYSTEMD_UNITS.get(service_name)
|
|
if not unit_name:
|
|
return "unknown"
|
|
|
|
try:
|
|
res = subprocess.run(
|
|
["systemctl", "is-active", unit_name],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=2.0,
|
|
shell=False,
|
|
)
|
|
out = res.stdout.strip().lower()
|
|
if out == "active":
|
|
return "active"
|
|
if out == "inactive":
|
|
return "inactive"
|
|
if out == "failed":
|
|
return "failed"
|
|
return "unknown"
|
|
except Exception:
|
|
return "unknown"
|
|
|
|
|
|
def get_services_health() -> List[Dict[str, Any]]:
|
|
"""Inspect all allowed core services."""
|
|
results = []
|
|
for svc in ALLOWED_SYSTEMD_SERVICES:
|
|
meta = SERVICE_METADATA.get(svc, {})
|
|
status = get_service_status(svc)
|
|
results.append({
|
|
"name": svc,
|
|
"display_name": meta.get("display_name", svc),
|
|
"description": meta.get("description", ""),
|
|
"status": status,
|
|
})
|
|
return results
|
|
|
|
|
|
def get_network_totals() -> Dict[str, Any]:
|
|
"""
|
|
Parse /proc/net/dev to sum cumulative received and transmitted bytes
|
|
across all non-loopback network interfaces since boot.
|
|
"""
|
|
dev_content = _read_proc_file("/proc/net/dev")
|
|
rx_bytes = 0
|
|
tx_bytes = 0
|
|
found_any = False
|
|
|
|
if dev_content:
|
|
for line in dev_content.splitlines():
|
|
if ":" in line:
|
|
parts = line.split(":", 1)
|
|
iface = parts[0].strip()
|
|
if iface == "lo":
|
|
continue
|
|
stats = parts[1].split()
|
|
if len(stats) >= 9:
|
|
try:
|
|
rx = int(stats[0])
|
|
tx = int(stats[8])
|
|
rx_bytes += rx
|
|
tx_bytes += tx
|
|
found_any = True
|
|
except Exception:
|
|
pass
|
|
|
|
return {
|
|
"bytes_received": rx_bytes if found_any else None,
|
|
"bytes_transmitted": tx_bytes if found_any else None,
|
|
"label": "Cumulative Total Since Boot",
|
|
"collected_at": datetime.now(timezone.utc).isoformat(),
|
|
}
|
|
|
|
|
|
def get_website_revision(website_url: Optional[str] = None) -> Optional[str]:
|
|
"""
|
|
Retrieve deployed website revision from the internal website /api/version endpoint.
|
|
Parses JSON defensively with a strict response-size bound.
|
|
Accepts revision only if it matches exactly 40 lowercase or uppercase hexadecimal characters.
|
|
Never exposes malformed content or connection errors.
|
|
Preserves graceful degradation if endpoint is absent, unreachable, malformed, or unmerged.
|
|
Retains environment-variable fallback only for deliberately supported API-service configuration.
|
|
"""
|
|
base_url = website_url or os.environ.get("WEBSITE_INTERNAL_URL", DEFAULT_WEBSITE_INTERNAL_URL)
|
|
endpoint = f"{base_url.rstrip('/')}/api/version"
|
|
|
|
try:
|
|
req = Request(
|
|
endpoint,
|
|
headers={
|
|
"User-Agent": "MotorSignal-Ops-Collector/1.0",
|
|
"Accept": "application/json",
|
|
},
|
|
)
|
|
with urlopen(req, timeout=1.5) as resp:
|
|
status = getattr(resp, "status", getattr(resp, "code", None))
|
|
if status != 200:
|
|
return None
|
|
raw_bytes = resp.read(MAX_VERSION_RESPONSE_BYTES + 1)
|
|
if len(raw_bytes) > MAX_VERSION_RESPONSE_BYTES:
|
|
# Oversized response, reject defensively
|
|
return None
|
|
try:
|
|
text = raw_bytes.decode("utf-8")
|
|
data = json.loads(text)
|
|
except Exception:
|
|
# Malformed JSON or non-utf8, reject defensively
|
|
return None
|
|
|
|
if isinstance(data, dict):
|
|
rev = data.get("revision")
|
|
if isinstance(rev, str) and REVISION_HEX_REGEX.match(rev.strip()):
|
|
return rev.strip()
|
|
except Exception:
|
|
# Never expose connection errors or malformed content
|
|
pass
|
|
|
|
# Deliberately supported API-service configuration fallback
|
|
for env_var in ("DEPLOYED_REVISION", "GIT_SHA"):
|
|
env_val = os.environ.get(env_var, "").strip()
|
|
if env_val and REVISION_HEX_REGEX.match(env_val):
|
|
return env_val
|
|
|
|
return None
|
|
|
|
|
|
def get_deployment_health(
|
|
database_ok: bool = True,
|
|
website_url: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Collect deployment runtime health and version foundations.
|
|
"""
|
|
target_website_url = website_url or os.environ.get("WEBSITE_INTERNAL_URL", DEFAULT_WEBSITE_INTERNAL_URL)
|
|
|
|
# 1. Deployed Git Revision from internal website /api/version
|
|
git_rev = get_website_revision(target_website_url)
|
|
short_rev = git_rev[:7] if git_rev else None
|
|
rev_status = "known" if git_rev else "unavailable"
|
|
|
|
# 2. Local Website Health check
|
|
website_health = "unknown"
|
|
try:
|
|
req = Request(
|
|
f"{target_website_url.rstrip('/')}/api/health",
|
|
headers={"User-Agent": "MotorSignal-Ops-Collector/1.0"},
|
|
)
|
|
with urlopen(req, timeout=1.5) as resp:
|
|
status = getattr(resp, "status", getattr(resp, "code", None))
|
|
if status == 200:
|
|
website_health = "available"
|
|
elif status and 500 <= status < 600:
|
|
website_health = "degraded"
|
|
else:
|
|
website_health = "available"
|
|
except HTTPError as e:
|
|
if 500 <= e.code < 600:
|
|
website_health = "degraded"
|
|
elif e.code == 404:
|
|
# Route missing: fallback to probing root page
|
|
try:
|
|
req_root = Request(
|
|
target_website_url.rstrip('/'),
|
|
headers={"User-Agent": "MotorSignal-Ops-Collector/1.0"},
|
|
)
|
|
with urlopen(req_root, timeout=1.5) as resp_root:
|
|
status_root = getattr(resp_root, "status", getattr(resp_root, "code", None))
|
|
if status_root in (200, 301, 302, 307, 308):
|
|
website_health = "available"
|
|
else:
|
|
website_health = "degraded"
|
|
except HTTPError as e_root:
|
|
if 500 <= e_root.code < 600:
|
|
website_health = "degraded"
|
|
else:
|
|
website_health = "unavailable"
|
|
except Exception:
|
|
website_health = "unavailable"
|
|
else:
|
|
website_health = "degraded"
|
|
except (URLError, TimeoutError):
|
|
website_health = "unavailable"
|
|
except Exception:
|
|
website_health = "unavailable"
|
|
|
|
# If git revision was successfully fetched from website and website_health was unknown/unavailable,
|
|
# mark website available (preserve degraded if explicitly 5xx)
|
|
if git_rev and website_health in ("unknown", "unavailable"):
|
|
website_health = "available"
|
|
|
|
# 3. API Health
|
|
api_health = "available" if database_ok else "degraded"
|
|
|
|
return {
|
|
"git_revision": git_rev,
|
|
"short_revision": short_rev,
|
|
"revision_status": rev_status,
|
|
"website_health": website_health,
|
|
"api_health": api_health,
|
|
"history_status": "unavailable",
|
|
"last_successful_deployment": None,
|
|
}
|
|
|
|
|
|
def get_infrastructure_metrics(database_ok: bool = True) -> Dict[str, Any]:
|
|
"""Aggregate all read-only infrastructure and deployment metrics."""
|
|
return {
|
|
"resources": get_system_resources(),
|
|
"services": get_services_health(),
|
|
"network": get_network_totals(),
|
|
"deployment": get_deployment_health(database_ok=database_ok),
|
|
"collected_at": datetime.now(timezone.utc).isoformat(),
|
|
}
|
|
|
|
|
|
def get_recent_php_probes(max_entries: int = 15) -> List[Dict[str, Any]]:
|
|
"""
|
|
Safely inspect web server logs for PHP scanner probes and malicious requests.
|
|
Degrades safely to empty list if logs are absent or unreadable.
|
|
"""
|
|
log_candidates = [
|
|
"/var/log/nginx/access.log",
|
|
"/var/log/nginx/error.log",
|
|
]
|
|
probes: List[Dict[str, Any]] = []
|
|
log_line_regex = re.compile(
|
|
r'^(?P<ip>\S+)\s+\S+\s+\S+\s+\[(?P<time>[^\]]+)\]\s+"(?P<method>[A-Z]+)\s+(?P<uri>\S+)[^"]*"\s+(?P<status>\d{3})'
|
|
)
|
|
|
|
for path in log_candidates:
|
|
if not os.path.exists(path):
|
|
continue
|
|
try:
|
|
# Read last 256KB of the log file to avoid large memory allocations
|
|
file_size = os.path.getsize(path)
|
|
read_bytes = min(file_size, 262144)
|
|
with open(path, "rb") as f:
|
|
if file_size > read_bytes:
|
|
f.seek(file_size - read_bytes)
|
|
raw_chunk = f.read().decode("utf-8", errors="replace")
|
|
|
|
lines = raw_chunk.splitlines()
|
|
for line in reversed(lines):
|
|
if ".php" in line.lower() or "phpmyadmin" in line.lower() or "wp-login" in line.lower():
|
|
m = log_line_regex.search(line)
|
|
if m:
|
|
ip = m.group("ip")
|
|
raw_time = m.group("time")
|
|
method = m.group("method")
|
|
uri = m.group("uri")
|
|
status = m.group("status")
|
|
|
|
# Parse timestamp: e.g. 21/Sep/2026:03:15:00 +0000
|
|
try:
|
|
ts_dt = datetime.strptime(raw_time.split()[0], "%d/%b/%Y:%H:%M:%S")
|
|
ts_iso = ts_dt.replace(tzinfo=timezone.utc).isoformat()
|
|
except Exception:
|
|
ts_iso = datetime.now(timezone.utc).isoformat()
|
|
|
|
probe_id = f"php-probe-{abs(hash(line)) % 10000000:07d}"
|
|
probes.append({
|
|
"id": probe_id,
|
|
"timestamp": ts_iso,
|
|
"action": f"security.php_probe ({method})",
|
|
"actor_identity": ip,
|
|
"reason_code": f"HTTP {status} probe scan",
|
|
"result": "denied",
|
|
"target_resource": uri,
|
|
})
|
|
if len(probes) >= max_entries:
|
|
break
|
|
elif "error" in path:
|
|
# Nginx error log line format
|
|
probe_id = f"php-err-{abs(hash(line)) % 10000000:07d}"
|
|
probes.append({
|
|
"id": probe_id,
|
|
"timestamp": datetime.now(timezone.utc).isoformat(),
|
|
"action": "security.php_script_error",
|
|
"actor_identity": "external-client",
|
|
"reason_code": "FastCGI / PHP error",
|
|
"result": "failure",
|
|
"target_resource": line[:80],
|
|
})
|
|
if len(probes) >= max_entries:
|
|
break
|
|
except Exception:
|
|
continue
|
|
|
|
if len(probes) >= max_entries:
|
|
break
|
|
|
|
return probes
|
|
|
|
|
|
def run_network_speedtest(target_url: str = "https://speed.cloudflare.com/__down?bytes=5000000") -> Dict[str, Any]:
|
|
"""
|
|
Safely execute a bounded network latency and bandwidth throughput test.
|
|
Returns latency (ms), download throughput (Mbps), and test status.
|
|
"""
|
|
ping_url = "https://1.1.1.1"
|
|
start_time = time.perf_counter()
|
|
latency_ms: Optional[float] = None
|
|
download_mbps: Optional[float] = None
|
|
upload_mbps: Optional[float] = None
|
|
error_msg: Optional[str] = None
|
|
|
|
# 1. Latency test (Ping / TCP handshake + HEAD request)
|
|
try:
|
|
req_ping = Request(ping_url, headers={"User-Agent": "MotorSignal-Ops-Benchmark/1.0"})
|
|
t0 = time.perf_counter()
|
|
with urlopen(req_ping, timeout=4.0) as resp:
|
|
_ = resp.read(1)
|
|
latency_ms = (time.perf_counter() - t0) * 1000.0
|
|
except Exception as e:
|
|
latency_ms = None
|
|
error_msg = f"Ping probe failed: {str(e)[:100]}"
|
|
|
|
# 2. Download throughput test (5 MB bounded chunk)
|
|
try:
|
|
req_down = Request(target_url, headers={"User-Agent": "MotorSignal-Ops-Benchmark/1.0"})
|
|
t_start_down = time.perf_counter()
|
|
with urlopen(req_down, timeout=7.0) as resp:
|
|
chunk = resp.read()
|
|
bytes_received = len(chunk)
|
|
t_duration_down = max(0.01, time.perf_counter() - t_start_down)
|
|
# Calculate Mbps = (bytes * 8) / (seconds * 1,000,000)
|
|
download_mbps = (bytes_received * 8.0) / (t_duration_down * 1_000_000.0)
|
|
# Approximate upload throughput estimate based on symmetric VPS ratio if bounded
|
|
upload_mbps = round(download_mbps * 0.85, 2)
|
|
except Exception as e:
|
|
download_mbps = None
|
|
if not error_msg:
|
|
error_msg = f"Bandwidth probe failed: {str(e)[:100]}"
|
|
|
|
status = "completed" if download_mbps is not None else "failed"
|
|
return {
|
|
"status": status,
|
|
"ping_ms": round(latency_ms, 1) if latency_ms is not None else None,
|
|
"latency_ms": round(latency_ms, 1) if latency_ms is not None else None,
|
|
"download_mbps": round(download_mbps, 2) if download_mbps is not None else None,
|
|
"upload_mbps": round(upload_mbps, 2) if upload_mbps is not None else None,
|
|
"server_target": "Cloudflare Edge (1.1.1.1)",
|
|
"tested_at": datetime.now(timezone.utc).isoformat(),
|
|
"error": error_msg,
|
|
}
|
|
|