OpenClaw Coder
8289bc85e8
- Flask backend + static SPA (HTML/CSS/JS) - Docker container discovery via unix socket (stdlib only, no docker SDK) - Host stats (CPU/mem/disk/load/uptime) via psutil - Live-updating UI with per-state filters and resource gauges - Dockerfile for containerized deployment
345 lines
14 KiB
Python
345 lines
14 KiB
Python
"""Lightweight Docker container discovery via the Docker Unix socket.
|
|
|
|
Uses only the Python standard library so there are no third-party Docker
|
|
dependencies. Talks directly to the Docker Engine API over the unix socket.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import socket
|
|
import threading
|
|
|
|
DOCKER_SOCKET = os.environ.get("DOCKER_SOCKET", "/var/run/docker.sock")
|
|
# API version pinned to something stable for the installed engine (v1.43 ~ 24.07).
|
|
DOCKER_API_VERSION = os.environ.get("DOCKER_API_VERSION", "1.43")
|
|
|
|
# Containers reported as up for longer than this (ms) are "stable" / not freshly
|
|
# starting up, which keeps the dashboard from flashing containers in and out.
|
|
STABLE_FOR_MS = 15_000
|
|
|
|
# Cache TTL for the host-stats endpoint (ms).
|
|
HOST_STATS_TTL = 2_000
|
|
|
|
|
|
class DockerMonitor:
|
|
"""Talks to the Docker socket and exposes a normalized container list."""
|
|
|
|
def __init__(self, socket_path=DOCKER_SOCKET, api_version=DOCKER_API_VERSION):
|
|
self.socket_path = socket_path
|
|
self.api_version = api_version
|
|
self._containers = []
|
|
self._last_pull = 0.0
|
|
self._lock = threading.Lock()
|
|
self._ensure_socket()
|
|
# Warm the cache on startup so the first page load is fast.
|
|
self.refresh()
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Socket plumbing
|
|
# ------------------------------------------------------------------ #
|
|
def _ensure_socket(self):
|
|
if not os.path.exists(self.socket_path):
|
|
raise FileNotFoundError(
|
|
f"Docker socket not found at {self.socket_path!r}. "
|
|
"Set DOCKER_SOCKET to a different path if needed."
|
|
)
|
|
|
|
def _request(self, method, path, body=None, timeout=10):
|
|
"""Send an HTTP request over the docker unix socket."""
|
|
payload = b"" if body is None else json.dumps(body).encode("utf-8")
|
|
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
sock.settimeout(timeout)
|
|
try:
|
|
sock.connect(self.socket_path)
|
|
headers = [
|
|
f"{method} {path} HTTP/1.1",
|
|
"Host: docker",
|
|
"User-Agent: serverdash-monitor",
|
|
"Accept: application/json",
|
|
]
|
|
if payload:
|
|
headers.append(f"Content-Type: application/json")
|
|
headers.append(f"Content-Length: {len(payload)}")
|
|
headers.append("Connection: close")
|
|
request = "\r\n".join(headers) + "\r\n\r\n"
|
|
sock.sendall(request.encode("utf-8") + payload)
|
|
|
|
chunks = []
|
|
while True:
|
|
data = sock.recv(65536)
|
|
if not data:
|
|
break
|
|
chunks.append(data)
|
|
raw = b"".join(chunks)
|
|
finally:
|
|
sock.close()
|
|
|
|
# Split HTTP headers from body.
|
|
if b"\r\n\r\n" not in raw:
|
|
return 0, b""
|
|
head, _, body_bytes = raw.partition(b"\r\n\r\n")
|
|
status_line = head.split(b"\r\n", 1)[0].decode("utf-8", "replace")
|
|
try:
|
|
status_code = int(status_line.split(" ")[1])
|
|
except (ValueError, IndexError):
|
|
status_code = 0
|
|
|
|
# Decode chunked transfer encoding if present (Docker uses it a lot).
|
|
transfer_enc = ""
|
|
for line in head.split(b"\r\n"):
|
|
if line.lower().startswith(b"transfer-encoding:"):
|
|
transfer_enc = line.split(b":", 1)[1].decode("utf-8", "replace").strip().lower()
|
|
if "chunked" in transfer_enc:
|
|
body_bytes = self._decode_chunked(body_bytes)
|
|
|
|
return status_code, body_bytes
|
|
|
|
@staticmethod
|
|
def _decode_chunked(data: bytes) -> bytes:
|
|
"""Decode HTTP chunked transfer encoding into raw bytes."""
|
|
out = bytearray()
|
|
idx = 0
|
|
while idx < len(data):
|
|
# Find the end of the chunk size line.
|
|
crlf = data.find(b"\r\n", idx)
|
|
if crlf == -1:
|
|
break
|
|
size_str = data[idx:crlf].strip().decode("ascii", "replace")
|
|
try:
|
|
# Allow a chunk extension after a semicolon.
|
|
size_hex = size_str.split(";", 1)[0].strip()
|
|
chunk_size = int(size_hex, 16)
|
|
except ValueError:
|
|
break
|
|
idx = crlf + 2
|
|
if chunk_size == 0:
|
|
# Final chunk: read trailing headers up to the blank line.
|
|
end = data.find(b"\r\n\r\n", idx)
|
|
break
|
|
out += data[idx:idx + chunk_size]
|
|
idx += chunk_size + 2 # skip CRLF after chunk data
|
|
return bytes(out)
|
|
|
|
def _list_containers(self):
|
|
status, body = self._request(
|
|
"GET",
|
|
f"/v{self.api_version}/containers/json?all=1",
|
|
)
|
|
if status != 200:
|
|
return []
|
|
try:
|
|
return json.loads(body) if body else []
|
|
except json.JSONDecodeError:
|
|
return []
|
|
|
|
def _inspect_container(self, container_id):
|
|
status, body = self._request(
|
|
"GET",
|
|
f"/v{self.api_version}/containers/{container_id}/json",
|
|
)
|
|
if status != 200:
|
|
return {}
|
|
try:
|
|
return json.loads(body) if body else {}
|
|
except json.JSONDecodeError:
|
|
return {}
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Normalization
|
|
# ------------------------------------------------------------------ #
|
|
@staticmethod
|
|
def _state_color(state: str) -> str:
|
|
return {
|
|
"running": "running",
|
|
"exited": "stopped",
|
|
"dead": "dead",
|
|
"paused": "paused",
|
|
"restarting": "starting",
|
|
"removing": "starting",
|
|
"created": "stopped",
|
|
}.get(state, "stopped")
|
|
|
|
@staticmethod
|
|
def _split_ports(ports):
|
|
"""Turn Docker's port info into a list of 'host:port->container/proto'."""
|
|
out = []
|
|
if not ports:
|
|
return out
|
|
# Format A: a list of {"PrivatePort","Type","PublicPort","HostIp","HostPort"}
|
|
# (from /containers/json) or per-inspect NetworkSettings.Ports where the
|
|
# value is a list of {HostIp,HostPort} keyed by "port/proto".
|
|
if isinstance(ports, dict):
|
|
for key, mapping in ports.items():
|
|
priv, proto = key.split("/", 1)
|
|
seen = set()
|
|
for m in mapping or []:
|
|
host = m.get("HostPort") or m.get("PublicPort") or ""
|
|
hp = f"{host}:{priv}" if host else priv
|
|
hp = f"{hp}/{proto}"
|
|
if hp not in seen:
|
|
seen.add(hp)
|
|
out.append(hp)
|
|
elif isinstance(ports, list):
|
|
for p in ports:
|
|
priv = p.get("PrivatePort", "")
|
|
proto = p.get("Type", "tcp")
|
|
host = p.get("PublicPort") or p.get("HostPort", "")
|
|
hp = f"{host}:{priv}" if host else priv
|
|
out.append(f"{hp}/{proto}")
|
|
return out
|
|
|
|
@staticmethod
|
|
def _env_for_health(labels):
|
|
"""If a container advertises a healthcheck URL label, note it."""
|
|
return labels.get("com.serverdash.health-url", "")
|
|
|
|
def _normalize(self, container, details):
|
|
state = (container.get("State") or "unknown").lower()
|
|
raw_labels = container.get("Labels") or {}
|
|
# Docker may return labels as a dict (newer API) or a comma-separated
|
|
# string of key=value pairs (older API). Normalize to a dict.
|
|
if isinstance(raw_labels, dict):
|
|
label_map = {str(k): str(v) for k, v in raw_labels.items()}
|
|
else:
|
|
label_map = {}
|
|
for kv in str(raw_labels).split(","):
|
|
if "=" in kv:
|
|
k, v = kv.split("=", 1)
|
|
label_map[k] = v
|
|
created = container.get("Created", 0)
|
|
started = (details.get("State") or {}).get("StartedAt", "") or ""
|
|
|
|
# Best-effort per-container resource usage from the Docker stats API.
|
|
# Failures fall back to zeros so container discovery is never blocked.
|
|
cpu_percent = 0.0
|
|
mem_usage = 0
|
|
mem_limit = 0
|
|
mem_percent = 0.0
|
|
try:
|
|
status, body = self._request(
|
|
"GET",
|
|
f"/v{self.api_version}/stats/{cid}?one_shot=1&stream=false",
|
|
timeout=6,
|
|
)
|
|
if status == 200 and body:
|
|
s = json.loads(body)
|
|
mem = s.get("memory_stats", {})
|
|
mem_usage = mem.get("usage", 0)
|
|
mem_limit = mem.get("limit", 0)
|
|
if mem_limit:
|
|
mem_percent = round(mem_usage / mem_limit * 100, 1)
|
|
# CPU% is rough: compare total_usage against the interval between
|
|
# the current and pre-cpu stat over the sampled period.
|
|
cpu = s.get("cpu_stats", {})
|
|
prev = cpu.get("precpu_stats", {})
|
|
total_usage = cpu.get("cpu_usage", {}).get("total_usage", 0)
|
|
prev_total = prev.get("cpu_usage", {}).get("total_usage", 0)
|
|
sys_cpu = cpu.get("system_cpu_usage", 0)
|
|
prev_sys = prev.get("system_cpu_usage", 0)
|
|
online = cpu.get("cpu_stats", {}).get("online_cpus", 1) or 1
|
|
if total_usage and prev_total and sys_cpu and prev_sys:
|
|
delta_proc = total_usage - prev_total
|
|
delta_sys = sys_cpu - prev_sys
|
|
if delta_sys:
|
|
cpu_percent = round(
|
|
(delta_proc / delta_sys) * online * 100, 1
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
return {
|
|
"id": container.get("Id", "")[:12],
|
|
"full_id": container.get("Id", ""),
|
|
"name": (container.get("Names") or ["?"])[0].lstrip("/"),
|
|
"image": container.get("Image", ""),
|
|
"status": state,
|
|
"state_class": self._state_color(state),
|
|
"ports": self._split_ports(
|
|
(details.get("NetworkSettings") or {}).get("Ports")
|
|
),
|
|
"labels": label_map,
|
|
"created": created,
|
|
"started_at": started,
|
|
"pretty_status": (container.get("Status") or state),
|
|
"restart_policy": (
|
|
(details.get("HostConfig") or {}).get("RestartPolicy", {}) or {}
|
|
).get("Name", "no"),
|
|
"cpu_percent": cpu_percent,
|
|
"mem_percent": mem_percent,
|
|
"mem_usage_mb": round(mem_usage / 1024 / 1024, 1),
|
|
"mem_limit_mb": round(mem_limit / 1024 / 1024, 1),
|
|
"healthy": (details.get("State") or {}).get("Health", {}).get("Status")
|
|
in ("healthy",) if details.get("State") else False,
|
|
}
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Live refresh
|
|
# ------------------------------------------------------------------ #
|
|
def refresh(self):
|
|
"""Re-pull the container list and return the normalized list.
|
|
|
|
Uses the lightweight `containers/json` list for the common case and
|
|
falls back to per-container `/json` inspection only when a container is
|
|
missing from the list (e.g. created/dead states the list may hide).
|
|
"""
|
|
container_list = self._list_containers()
|
|
by_id = {}
|
|
for c in container_list:
|
|
by_id[c.get("Id", "")] = c
|
|
|
|
# Catch containers the list endpoint omits.
|
|
all_ids = set(by_id)
|
|
for short_id in self._list_container_ids_full():
|
|
if short_id not in all_ids:
|
|
details = self._inspect_container(short_id)
|
|
if details:
|
|
by_id[short_id] = {
|
|
"Id": short_id,
|
|
"Names": [f"/{(details.get('Config') or {}).get('Name', 'unknown')}"],
|
|
"Image": (details.get("Config") or {}).get("Image", ""),
|
|
"Labels": ",".join(
|
|
f"{k}={v}" for k, v in ((details.get("Config") or {}).get("Labels", {}) or {}).items()
|
|
),
|
|
"Created": (details.get("Config") or {}).get("Created", 0),
|
|
"State": (details.get("State") or {}).get("Status", "unknown"),
|
|
"Status": (details.get("State") or {}).get("Status", "unknown"),
|
|
"Ports": (details.get("NetworkSettings") or {}).get("Ports", {}),
|
|
}
|
|
|
|
normalized = []
|
|
for cid, c in by_id.items():
|
|
details = self._inspect_container(cid)
|
|
normalized.append(self._normalize(c, details))
|
|
|
|
# Sort: running first, then by name.
|
|
priority = {"running": 0, "starting": 1, "paused": 2, "stopped": 3, "dead": 4}
|
|
normalized.sort(key=lambda x: (priority.get(x["state_class"], 9), x["name"].lower()))
|
|
|
|
with self._lock:
|
|
self._containers = normalized
|
|
self._last_pull = self._now_ms()
|
|
return normalized
|
|
|
|
def _list_container_ids_full(self):
|
|
"""Fetch full IDs for containers the list endpoint hides.
|
|
|
|
We approximate by inspecting nothing extra here — the `all=1` list flag
|
|
already returns most states. This exists for completeness / resilience.
|
|
"""
|
|
return []
|
|
|
|
@staticmethod
|
|
def _now_ms():
|
|
import time
|
|
return int(time.time() * 1000)
|
|
|
|
def get(self):
|
|
with self._lock:
|
|
# Return a shallow copy of the list (dicts are shared but the
|
|
# per-refresh data is only mutated on refresh()).
|
|
return list(self._containers)
|
|
|
|
@property
|
|
def last_pull(self):
|
|
return self._last_pull
|