157 lines
4.3 KiB
Python
157 lines
4.3 KiB
Python
from fastapi import FastAPI, Request
|
|
from sqlmodel import Session
|
|
|
|
from arcology.arroyo import Page, engine
|
|
import arcology.html as html
|
|
|
|
from arcology.parse import parse_sexp
|
|
|
|
app = FastAPI()
|
|
|
|
import uvicorn
|
|
|
|
#@click.command(help="start the DRP status servlet")
|
|
#@click.option("--host", "-h", help="the host IP to listen on, defaults to all IPs/interfaces", default="0.0.0.0")
|
|
#@click.option("--port", "-p", help="port to listen on", default=8000)
|
|
def start(host="0.0.0.0", port=8000):
|
|
uvicorn.run("arcology.server:app", host=host, port=port)
|
|
|
|
from prometheus_fastapi_instrumentator import Instrumentator
|
|
|
|
prometheus_instrumentor = Instrumentator()
|
|
# done after adding custom metrics now
|
|
# prometheus_instrumentor.instrument(app).expose(app)
|
|
|
|
from typing import Callable
|
|
from prometheus_fastapi_instrumentator.metrics import Info
|
|
from prometheus_client import Counter
|
|
from arcology.sites import host_to_site
|
|
from arcology.key import ArcologyKey
|
|
|
|
import logging
|
|
logger = logging.getLogger(__name__)
|
|
logger.setLevel("INFO")
|
|
|
|
def http_request_sites_total() -> Callable[[Info], None]:
|
|
METRIC = Counter(
|
|
"http_request_by_site_total",
|
|
"Number of times a site or page has been requested.",
|
|
labelnames=("site", "key", "method", "status", "ua_type")
|
|
)
|
|
|
|
def instrumentation(info: Info) -> None:
|
|
key = ArcologyKey.from_request(info.request)
|
|
|
|
user_agent = info.request.headers.get("User-Agent")
|
|
agent_type = get_agent_type(user_agent)
|
|
|
|
if info.request.url.path.startswith("/metrics"):
|
|
return
|
|
if info.request.url.path.startswith("/static"):
|
|
return
|
|
if info.request.url.path.startswith("/favicon.ico"):
|
|
return
|
|
|
|
if agent_type == "unknown":
|
|
logger.info("Detected unknown user agent: {agent}", dict(agent=user_agent))
|
|
|
|
METRIC.labels(key.site.key, key.key, info.method, info.modified_status, agent_type).inc()
|
|
|
|
return instrumentation
|
|
|
|
prometheus_instrumentor.add(http_request_sites_total())
|
|
|
|
prometheus_instrumentor.instrument(app).expose(app)
|
|
|
|
def get_agent_type(user_agent: str) -> str:
|
|
|
|
if user_agent == "":
|
|
return "no-ua"
|
|
|
|
if "Synapse" in user_agent:
|
|
return "matrix"
|
|
if "Element" in user_agent:
|
|
return "matrix"
|
|
|
|
if "SubwayTooter" in user_agent:
|
|
return "app"
|
|
if "Dalvik" in user_agent:
|
|
return "app"
|
|
if "Nextcloud-android" in user_agent:
|
|
return "app"
|
|
|
|
if "prometheus" in user_agent:
|
|
return "internal"
|
|
if "feediverse" in user_agent:
|
|
return "internal"
|
|
|
|
if "Pleroma" in user_agent:
|
|
return "fedi"
|
|
if "Mastodon/" in user_agent:
|
|
return "fedi"
|
|
if "Akkoma" in user_agent:
|
|
return "fedi"
|
|
if "Friendica" in user_agent:
|
|
return "fedi"
|
|
if "FoundKey" in user_agent:
|
|
return "fedi"
|
|
if "MissKey" in user_agent:
|
|
return "fedi"
|
|
if "CalcKey" in user_agent:
|
|
return "fedi"
|
|
if "gotosocial" in user_agent:
|
|
return "fedi"
|
|
if "Epicyon" in user_agent:
|
|
return "fedi"
|
|
|
|
if "feedparser" in user_agent:
|
|
return "feed"
|
|
if "granary" in user_agent:
|
|
return "feed"
|
|
if "Tiny Tiny RSS" in user_agent:
|
|
return "feed"
|
|
if "Go-NEB" in user_agent:
|
|
return "feed"
|
|
if "Gwene" in user_agent:
|
|
return "feed"
|
|
if "Feedbin" in user_agent:
|
|
return "feed"
|
|
if "SimplePie" in user_agent:
|
|
return "feed"
|
|
if "Elfeed" in user_agent:
|
|
return "feed"
|
|
if "inoreader" in user_agent:
|
|
return "feed"
|
|
if "Reeder" in user_agent:
|
|
return "feed"
|
|
if "Miniflux" in user_agent:
|
|
return "feed"
|
|
|
|
if "Bot" in user_agent:
|
|
return "bot"
|
|
if "bot" in user_agent:
|
|
return "bot"
|
|
if "Poduptime" in user_agent:
|
|
return "bot"
|
|
|
|
if "Chrome/" in user_agent:
|
|
return "browser"
|
|
if "Firefox/" in user_agent:
|
|
return "browser"
|
|
if "DuckDuckGo/" in user_agent:
|
|
return "browser"
|
|
if "Safari/" in user_agent:
|
|
return "browser"
|
|
|
|
return "unknown"
|
|
|
|
from fastapi.staticfiles import StaticFiles
|
|
import os
|
|
|
|
static_directory = os.environ.get('STATIC_FILE_DIR', "arcology/static")
|
|
|
|
app.mount("/static", StaticFiles(directory=static_directory), name="static")
|
|
|
|
import arcology.routing.domains as domains
|
|
app = domains.decorate_app(app)
|