Dokku uses sudo under the hood which isnt available in the python build pack, so just use docker to fetch info

This commit is contained in:
Peter Stockings
2025-12-22 12:54:52 +11:00
parent edd6bd2ce9
commit 147f38dfb4
2 changed files with 156 additions and 88 deletions

172
app.py
View File

@@ -6,47 +6,33 @@ from flask import Flask, jsonify, render_template
app = Flask(__name__)
DOKKU = "/usr/bin/dokku"
DOCKER = "/usr/bin/docker"
POLL_SECONDS = int(os.getenv("POLL_SECONDS", "10"))
SHOW_CONTAINERS = os.getenv("SHOW_CONTAINERS", "0") == "1"
APP_DOMAIN = os.getenv("APP_DOMAIN", "peterstockings.com")
DOCKER = os.getenv("DOCKER_BIN", "/usr/bin/docker")
SHOW_INFRA = os.getenv("SHOW_INFRA", "1") == "1"
# Optional JSON map: {"gitea":"https://gitea.peterstockings.com", "bloodpressure":"https://bp.peterstockings.com"}
APP_URL_OVERRIDES = {}
try:
APP_URL_OVERRIDES = json.loads(os.getenv("APP_URL_OVERRIDES", "{}"))
except Exception:
APP_URL_OVERRIDES = {}
def sh(cmd: list[str]) -> str:
return subprocess.check_output(cmd, stderr=subprocess.STDOUT, text=True).strip()
def dokku_apps() -> list[str]:
out = sh([DOKKU, "apps:list"])
lines = [l.strip() for l in out.splitlines()]
# Format:
# =====> My Apps
# app1
# app2
return [l for l in lines if l and not l.startswith("====")]
def dokku_urls(app_name: str) -> list[str]:
try:
out = sh([DOKKU, "urls:report", app_name, "--urls"])
return [u.strip() for u in out.split() if u.strip()]
except Exception:
return []
def dokku_ps_report(app_name: str) -> dict:
try:
out = sh([DOKKU, "ps:report", app_name])
# crude parse: "Key: value"
data = {}
for line in out.splitlines():
if ":" in line:
k, v = line.split(":", 1)
data[k.strip()] = v.strip()
return data
except Exception as e:
return {"error": str(e)}
def docker_ps_all() -> list[dict]:
# Name + Image + Status + Ports
fmt = "{{.Names}}\t{{.Image}}\t{{.Status}}\t{{.Ports}}"
out = sh([DOCKER, "ps", "--format", fmt])
rows = []
for line in out.splitlines():
name, image, status, ports = line.split("\t")
rows.append({"name": name, "image": image, "status": status, "ports": ports})
return rows
def docker_stats() -> dict:
"""
Returns a mapping of container_name -> {cpu, mem_used, mem_limit, mem_pct}
"""
# Name + CPU + MemUsage + MemPerc
fmt = "{{.Name}}\t{{.CPUPerc}}\t{{.MemUsage}}\t{{.MemPerc}}"
out = sh([DOCKER, "stats", "--no-stream", "--format", fmt])
stats = {}
@@ -62,52 +48,94 @@ def docker_stats() -> dict:
}
return stats
def docker_ps() -> list[dict]:
fmt = "{{.Names}}\t{{.Image}}\t{{.Status}}\t{{.Ports}}"
out = sh([DOCKER, "ps", "--format", fmt])
rows = []
for line in out.splitlines():
name, image, status, ports = line.split("\t")
rows.append({"name": name, "image": image, "status": status, "ports": ports})
return rows
def docker_inspect_restart_count(container_name: str) -> int:
# RestartCount is useful when stuff is flapping / OOMing
try:
out = sh([DOCKER, "inspect", "-f", "{{.RestartCount}}", container_name])
return int(out.strip())
except Exception:
return 0
def is_app_web_container(name: str) -> bool:
# Dokku apps typically have containers like "<app>.web.1"
return name.endswith(".web.1") and not name.startswith("dokku.")
def infer_app_name(container_name: str) -> str:
# "<app>.web.1" -> "<app>"
return container_name.rsplit(".web.1", 1)[0]
def infer_url(app_name: str) -> str:
if app_name in APP_URL_OVERRIDES:
return APP_URL_OVERRIDES[app_name]
# default
return f"https://{app_name}.{APP_DOMAIN}"
def classify_infra(container_name: str) -> bool:
return (
container_name.startswith("dokku.postgres.")
or container_name.startswith("dokku.redis.")
or container_name.startswith("dokku.mysql.")
or container_name.startswith("dokku.mongodb.")
or container_name == "dokku.minio.storage"
or container_name == "logspout"
)
def collect():
apps = dokku_apps()
ps_rows = docker_ps_all()
stats = docker_stats()
rows = []
apps = []
infra = []
for r in ps_rows:
name = r["name"]
s = stats.get(name, {})
row = {
"container": name,
"image": r["image"],
"status": r["status"],
"ports": r["ports"],
"cpu": s.get("cpu", ""),
"mem_used": s.get("mem_used", ""),
"mem_limit": s.get("mem_limit", ""),
"mem_pct": s.get("mem_pct", ""),
"restarts": docker_inspect_restart_count(name),
}
if is_app_web_container(name):
app_name = infer_app_name(name)
row["app"] = app_name
row["url"] = infer_url(app_name)
apps.append(row)
elif SHOW_INFRA and classify_infra(name):
infra.append(row)
# Sort stable
apps.sort(key=lambda x: x["app"])
infra.sort(key=lambda x: x["container"])
# Simple top-line summary
warnings = []
for a in apps:
urls = dokku_urls(a)
ps = dokku_ps_report(a)
# Your containers are typically named like "<app>.web.1"
container_name = f"{a}.web.1"
cstats = stats.get(container_name, None)
rows.append({
"app": a,
"urls": urls,
"running": ps.get("Running", ""),
"processes": ps.get("Processes", ""),
"status_web_1": ps.get("Status web 1", ""),
"cpu": cstats["cpu"] if cstats else "",
"mem_used": cstats["mem_used"] if cstats else "",
"mem_limit": cstats["mem_limit"] if cstats else "",
"mem_pct": cstats["mem_pct"] if cstats else "",
})
extra_containers = docker_ps() if SHOW_CONTAINERS else []
# mem_pct is like "15.32%"
try:
pct = float(a["mem_pct"].replace("%", "")) if a["mem_pct"] else 0.0
except Exception:
pct = 0.0
if pct >= 85:
warnings.append(f"{a['app']} RAM high ({a['mem_pct']})")
if a["restarts"] >= 3:
warnings.append(f"{a['app']} restarting (restarts={a['restarts']})")
return {
"generated_at": datetime.utcnow().isoformat() + "Z",
"poll_seconds": POLL_SECONDS,
"apps": sorted(rows, key=lambda r: r["app"]),
"containers": extra_containers,
"domain": APP_DOMAIN,
"apps": apps,
"infra": infra,
"warnings": warnings,
}
@app.get("/api/status")
def api_status():
return jsonify(collect())
@app.get("/")
def index():
return render_template("index.html", poll_seconds=POLL_SECONDS)
@@ -115,4 +143,8 @@ def index():
@app.get("/partial/apps")
def partial_apps():
data = collect()
return render_template("apps_table.html", data=data)
return render_template("apps_table.html", data=data)
@app.get("/api/status")
def api_status():
return jsonify(collect())