Add container_stats tool (Gruppe 2); remove rename_container

container_stats(container_name) calls SYNO.Docker.Container/stats,
locates the entry by name (stripping the DSM-added leading slash),
and reports:
  - CPU %  (standard Docker formula: cpu_delta / system_delta * cpus * 100)
  - Memory used / limit  (human-readable)
  - Network I/O rx / tx  (summed across all interfaces)
  - Block I/O read / write  (from io_service_bytes_recursive)

Gracefully handles first-poll (precpu system_cpu_usage absent → 0%).
7 unit tests covering: found, CPU formula, memory format, slash-strip,
not-found, API error, no-precpu fallback.

rename_container removed: DSM Container Manager offers no rename API.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-04-13 18:29:17 +02:00
parent 5edd051830
commit a8da306ce5
3 changed files with 275 additions and 17 deletions
@@ -7,6 +7,7 @@ from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from mcp.server.fastmcp import FastMCP
from mcp_synology_container.config import AppConfig
from mcp_synology_container.dsm_client import DsmClient
@@ -40,9 +41,9 @@ def register_containers(mcp: FastMCP, config: AppConfig, client: DsmClient) -> N
# Filter by project if specified
if project_name:
containers = [
c for c in containers
if c.get("project_name") == project_name
or _container_in_project(c, project_name)
c
for c in containers
if c.get("project_name") == project_name or _container_in_project(c, project_name)
]
if not containers:
return f"No containers found for project '{project_name}'."
@@ -129,6 +130,92 @@ def register_containers(mcp: FastMCP, config: AppConfig, client: DsmClient) -> N
return header + "\n".join(lines)
@mcp.tool()
async def container_stats(container_name: str) -> str:
"""Get live resource usage statistics for a container.
Reports CPU %, RAM used/limit, Network I/O (rx/tx), and Block I/O
(read/write) for the named container.
Args:
container_name: Name of the container (e.g. "jenkins").
"""
try:
data = await client.request("SYNO.Docker.Container", "stats")
except Exception as e:
return f"Error fetching container stats: {e}"
if not data:
return "No stats data returned."
# Response is a dict keyed by container ID hash; each entry has "name"
# with a leading slash (e.g. "/jenkins").
target: dict[str, Any] | None = None
for entry in data.values():
entry_name = entry.get("name", "").lstrip("/")
if entry_name == container_name.lstrip("/"):
target = entry
break
if target is None:
return (
f"Container '{container_name}' not found in stats. "
f"Available: {', '.join(v.get('name', '?').lstrip('/') for v in data.values())}"
)
# ── CPU % ────────────────────────────────────────────────────────────
cpu_stats = target.get("cpu_stats", {})
precpu_stats = target.get("precpu_stats", {})
cpu_usage = cpu_stats.get("cpu_usage", {})
precpu_usage = precpu_stats.get("cpu_usage", {})
cpu_delta = cpu_usage.get("total_usage", 0) - precpu_usage.get("total_usage", 0)
system_delta = cpu_stats.get("system_cpu_usage", 0) - precpu_stats.get(
"system_cpu_usage", 0
)
online_cpus = cpu_stats.get("online_cpus") or len(cpu_usage.get("percpu_usage") or [1])
if system_delta > 0 and cpu_delta >= 0:
cpu_pct = (cpu_delta / system_delta) * online_cpus * 100.0
else:
cpu_pct = 0.0
# ── Memory ───────────────────────────────────────────────────────────
mem_stats = target.get("memory_stats", {})
mem_usage = mem_stats.get("usage", 0)
mem_limit = mem_stats.get("limit", 0)
# ── Network I/O ──────────────────────────────────────────────────────
net_rx = 0
net_tx = 0
for iface in (target.get("networks") or {}).values():
net_rx += iface.get("rx_bytes", 0)
net_tx += iface.get("tx_bytes", 0)
# ── Block I/O ────────────────────────────────────────────────────────
blk_read = 0
blk_write = 0
for entry_io in target.get("blkio_stats", {}).get("io_service_bytes_recursive") or []:
op = entry_io.get("op", "").lower()
val = entry_io.get("value", 0)
if op == "read":
blk_read += val
elif op == "write":
blk_write += val
# ── Format ───────────────────────────────────────────────────────────
from mcp_synology_container.modules.images import _human_size # reuse helper
mem_limit_str = f" / {_human_size(mem_limit)}" if mem_limit else ""
lines = [
f"Stats for {container_name}:",
f" CPU: {cpu_pct:.2f}% ({online_cpus} CPUs)",
f" Memory: {_human_size(mem_usage)}{mem_limit_str}",
f" Net I/O: rx {_human_size(net_rx)} / tx {_human_size(net_tx)}",
f" Block I/O: read {_human_size(blk_read)} / write {_human_size(blk_write)}",
]
return "\n".join(lines)
@mcp.tool()
async def exec_in_container(
container_name: str,