opensandbox-group--opensandbox/server/opensandbox_server/services/docker/docker_diagnostics.py

# Copyright 2026 Alibaba Group Holding Ltd.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""
Docker diagnostics mixin for DevOps API.

Provides get_sandbox_logs, get_sandbox_inspect, and get_sandbox_events
by reading Docker container state. Mixed into DockerSandboxService.
"""

from __future__ import annotations

import re
import time


def _parse_since_to_timestamp(since: str) -> int:
    """Parse a human-readable duration string (e.g. '10m', '1h') into a Unix timestamp.

    Docker interprets the ``since`` parameter as an absolute Unix timestamp,
    so we convert the relative duration to ``now - duration``.
    """
    m = re.fullmatch(r"(\d+)\s*([smhd])", since.strip().lower())
    if not m:
        seconds = 600  # default 10m
    else:
        value, unit = int(m.group(1)), m.group(2)
        multipliers = {"s": 1, "m": 60, "h": 3600, "d": 86400}
        seconds = value * multipliers[unit]
    return int(time.time()) - seconds


class DockerDiagnosticsMixin:
    """Mixin that implements diagnostics methods for the Docker backend."""

    def get_sandbox_logs(
        self,
        sandbox_id: str,
        tail: int = 100,
        since: str | None = None,
        container: str | None = None,
    ) -> str:
        # ``container`` is accepted for API parity with the Kubernetes backend;
        # the Docker runtime models a sandbox as a single container, so the
        # parameter is informational only.
        del container
        docker_container = self._get_container_by_sandbox_id(sandbox_id)
        kwargs: dict = {"tail": tail, "timestamps": True}
        if since:
            kwargs["since"] = _parse_since_to_timestamp(since)
        output = docker_container.logs(**kwargs)
        if isinstance(output, bytes):
            output = output.decode("utf-8", errors="replace")
        return output or "(no logs)"

    def get_sandbox_inspect(self, sandbox_id: str) -> str:
        container = self._get_container_by_sandbox_id(sandbox_id)
        attrs = container.attrs or {}
        state = attrs.get("State", {})
        config_section = attrs.get("Config", {})
        network_settings = attrs.get("NetworkSettings", {}) or {}
        host_config = attrs.get("HostConfig", {}) or {}

        lines: list[str] = []
        lines.append(f"Container ID:   {container.id[:12]}")
        lines.append(f"Name:           {container.name}")
        lines.append(f"Image:          {config_section.get('Image', 'N/A')}")
        lines.append(f"Status:         {state.get('Status', 'unknown')}")
        lines.append(f"Running:        {state.get('Running', False)}")
        lines.append(f"Paused:         {state.get('Paused', False)}")
        lines.append(f"OOMKilled:      {state.get('OOMKilled', False)}")
        lines.append(f"Exit Code:      {state.get('ExitCode', 'N/A')}")
        lines.append(f"Started At:     {state.get('StartedAt', 'N/A')}")
        lines.append(f"Finished At:    {state.get('FinishedAt', 'N/A')}")
        if state.get("Error"):
            lines.append(f"Error:          {state['Error']}")

        # Resource limits
        lines.append("")
        lines.append("Resources:")
        nano_cpus = host_config.get("NanoCpus", 0)
        if nano_cpus:
            lines.append(f"  CPU:          {nano_cpus / 1e9:.2f} cores")
        memory = host_config.get("Memory", 0)
        if memory:
            lines.append(f"  Memory:       {memory // (1024 * 1024)} MiB")
        pids_limit = host_config.get("PidsLimit", 0)
        if pids_limit:
            lines.append(f"  PIDs Limit:   {pids_limit}")

        # Network
        lines.append("")
        lines.append("Network:")
        networks = network_settings.get("Networks", {})
        for net_name, net_info in networks.items():
            ip = (net_info or {}).get("IPAddress", "N/A")
            lines.append(f"  {net_name}: {ip}")

        # Ports
        ports = network_settings.get("Ports", {})
        if ports:
            lines.append("")
            lines.append("Ports:")
            for port_key, bindings in (ports or {}).items():
                if bindings:
                    for b in bindings:
                        lines.append(f"  {port_key} -> {b.get('HostIp', '')}:{b.get('HostPort', '')}")
                else:
                    lines.append(f"  {port_key} (not bound)")

        # Labels
        labels = config_section.get("Labels", {})
        if labels:
            lines.append("")
            lines.append("Labels:")
            for k, v in sorted(labels.items()):
                lines.append(f"  {k}={v}")

        # Environment (filter sensitive)
        env_list = config_section.get("Env", [])
        if env_list:
            lines.append("")
            lines.append("Environment:")
            for env_entry in env_list:
                key = env_entry.split("=", 1)[0] if "=" in env_entry else env_entry
                if any(s in key.upper() for s in ("SECRET", "TOKEN", "PASSWORD", "KEY")):
                    lines.append(f"  {key}=***")
                else:
                    lines.append(f"  {env_entry}")

        return "\n".join(lines)

    def get_sandbox_events(self, sandbox_id: str, limit: int = 50) -> str:
        container = self._get_container_by_sandbox_id(sandbox_id)
        attrs = container.attrs or {}
        state = attrs.get("State", {})

        lines: list[str] = []

        # Docker does not have a per-container event history in the same way
        # K8s does. We reconstruct key state transitions from container attrs.
        lines.append(f"Container:  {container.id[:12]} ({container.name})")
        lines.append(f"Status:     {state.get('Status', 'unknown')}")

        created = attrs.get("Created", "N/A")
        lines.append(f"Created:    {created}")
        started = state.get("StartedAt")
        if started and started != "0001-01-01T00:00:00Z":
            lines.append(f"Started:    {started}")
        finished = state.get("FinishedAt")
        if finished and finished != "0001-01-01T00:00:00Z":
            lines.append(f"Finished:   {finished}")

        if state.get("OOMKilled"):
            lines.append("Event:      OOMKilled - container was killed due to out-of-memory")
        exit_code = state.get("ExitCode")
        if exit_code is not None and exit_code != 0:
            lines.append(f"Event:      Exited with code {exit_code}")
        if state.get("Error"):
            lines.append(f"Event:      Error - {state['Error']}")

        # Health check status if available
        health = state.get("Health", {})
        if health:
            lines.append(f"Health:     {health.get('Status', 'N/A')}")
            for entry in (health.get("Log") or [])[-limit:]:
                ts = entry.get("Start", "")
                exit_code_h = entry.get("ExitCode", "")
                output_h = (entry.get("Output") or "").strip()[:200]
                lines.append(f"  [{ts}] exit={exit_code_h} {output_h}")

        if len(lines) <= 3:
            lines.append("(no notable events)")

        return "\n".join(lines)