A minimal statuspage to show uptime and latency for HTTP endpoints https://status.demo.j-serv.de/
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 

150 lines
5.1 KiB

import argparse
import asyncio
import datetime
import logging
import random
from pathlib import Path
import aiohttp.web
import httpx
from .web import CONFIG
from .web import get_app
from statuspage import __version__
logger = logging.getLogger(__name__)
DEFAULT_TIMEOUT = 30
DEFAULT_JITTER = 5
HOMEPAGE = "https://codeberg.org/hjacobs/minimal-statuspage"
def comma_separated_values(value):
return list(filter(None, value.split(",")))
def parse_args(argv=None):
parser = argparse.ArgumentParser(description=f"Statuspage v{__version__}")
parser.add_argument(
"--port",
type=int,
default=8080,
help="TCP port to start webserver on (default: 8080)",
)
parser.add_argument(
"--version", action="version", version=f"statuspage {__version__}"
)
parser.add_argument(
"--debug", action="store_true", help="Run in debugging mode (log more)"
)
parser.add_argument(
"--sites",
type=comma_separated_values,
default=[],
help="Sites to monitor (comma separated list of URLs)",
)
parser.add_argument(
"--probe-interval", type=float, default=60, help="Probe interval in seconds"
)
parser.add_argument(
"--data-path",
default=".",
help="Path to directory to store monitoring data",
)
# customization options
parser.add_argument(
"--templates-path", help="Path to directory with custom HTML/Jinja2 templates"
)
parser.add_argument(
"--static-assets-path",
help="Path to custom JS/CSS assets (will be mounted as /assets HTTP path)",
)
args = parser.parse_args(argv)
return args
async def background_process(app):
config = app[CONFIG]
probe_results_path = Path(config.data_path) / "probe-results.tsv"
headers = {"User-Agent": f"minimal-statuspage/{__version__} ({HOMEPAGE})"}
while True:
try:
for site in config.sites:
logger.debug(f"Probing {site} ..")
start = datetime.datetime.utcnow()
try:
async with httpx.AsyncClient() as client:
response = await client.get(
site, timeout=DEFAULT_TIMEOUT, headers=headers
)
now = datetime.datetime.utcnow()
duration = now - start
with probe_results_path.open("a") as fd:
fd.write(
"\t".join(
map(
str,
[
now.isoformat(),
site,
"OK"
if response.status_code == 200
else "ERROR",
response.status_code,
duration.total_seconds(),
"",
],
)
)
+ "\n"
)
except Exception as e:
logger.debug(f"Failed to probe {site}: {e}")
now = datetime.datetime.utcnow()
duration = now - start
with probe_results_path.open("a") as fd:
fd.write(
"\t".join(
map(
str,
[
now.isoformat(),
site,
"ERROR",
999,
duration.total_seconds(),
e,
],
)
)
+ "\n"
)
except Exception as e:
logger.exception(f"Failed to process: {e}")
sleep_time = max(0.1, config.probe_interval) + random.gauss(0, DEFAULT_JITTER)
await asyncio.sleep(sleep_time)
async def start_background_tasks(app):
app["background_task"] = asyncio.create_task(background_process(app))
async def cleanup_background_tasks(app):
app["background_task"].cancel()
await app["background_task"]
def main(argv=None):
args = parse_args(argv)
logging.basicConfig(level=logging.DEBUG if args.debug else logging.INFO)
config_str = ", ".join(f"{k}={v}" for k, v in sorted(vars(args).items()))
logger.info(f"Statuspage v{__version__} started with {config_str}")
app = get_app(args)
app.on_startup.append(start_background_tasks)
app.on_cleanup.append(cleanup_background_tasks)
aiohttp.web.run_app(app, port=args.port, handle_signals=False)