"""Docker container log and health ingestion."""
import subprocess
import json
import logging
from datetime import datetime, timedelta, timezone
from typing import Optional
from ..models import LogEntry, SourceType, Severity
logger = logging.getLogger(__name__)
def _parse_severity(text: str) -> Severity:
"""Heuristic severity detection from log line."""
text_lower = text.lower()
if any(w in text_lower for w in ["critical", "fatal", "panic", "segfault", "killed"]):
return Severity.CRITICAL
if any(w in text_lower for w in ["error", "exception", "failed", "failure", "traceback", "crash"]):
return Severity.ERROR
if any(w in text_lower for w in ["warn", "timeout", "retry", "deprecated"]):
return Severity.WARNING
return Severity.INFO
def _parse_timestamp(line: str) -> Optional[datetime]:
"""Try to extract timestamp from Docker log line.
Docker log format: 2024-01-01T00:00:00.000000000Z level message
"""
try:
# First token should be ISO timestamp
parts = line.split(maxsplit=1)
if parts:
# Try parsing the timestamp part
ts_str = parts[0].rstrip("Z")
for fmt in ["%Y-%m-%dT%H:%M:%S.%f", "%Y-%m-%dT%H:%M:%S"]:
try:
return datetime.strptime(ts_str, fmt).replace(tzinfo=timezone.utc)
except ValueError:
continue
except Exception:
pass
return None
class DockerIngestor:
"""Collect logs and health from Docker containers."""
def get_containers(self, names: Optional[list[str]] = None) -> list[str]:
"""List running container names."""
try:
args = ["docker", "ps", "--format", "{{.Names}}"]
if names:
# Filter by name
result = subprocess.run(
["docker", "ps", "--format", "{{.Names}}:{{.Image}}"],
capture_output=True, text=True, timeout=10
)
containers = []
for line in result.stdout.strip().split("\n"):
if not line:
continue
name, _ = line.split(":", 1)
if name.strip() in names:
containers.append(name.strip())
return containers
else:
result = subprocess.run(args, capture_output=True, text=True, timeout=10)
if result.returncode != 0:
logger.warning("Failed to list containers: %s", result.stderr)
return []
return [n.strip() for n in result.stdout.strip().split("\n") if n.strip()]
except FileNotFoundError:
logger.warning("Docker CLI not found")
return []
except subprocess.TimeoutExpired:
logger.warning("Docker command timed out")
return []
def get_logs(
self,
container: str,
since_minutes: int = 60,
tail: int = 500
) -> list[LogEntry]:
"""Get recent logs from a container."""
try:
since = f"{since_minutes}m"
result = subprocess.run(
[
"docker", "logs",
"--since", since,
"--tail", str(tail),
"--timestamps",
container
],
capture_output=True, text=True, timeout=30
)
entries = []
raw_output = result.stdout + result.stderr # Docker logs may split
for line in raw_output.strip().split("\n"):
if not line.strip():
continue
entries.append(LogEntry(
timestamp=_parse_timestamp(line) or datetime.now(timezone.utc),
source=SourceType.DOCKER,
source_name=container,
severity=_parse_severity(line),
message=line.split(maxsplit=2)[-1] if len(line.split(maxsplit=2)) > 2 else line,
raw=line
))
return entries
except FileNotFoundError:
logger.warning("Docker CLI not found")
return []
except subprocess.TimeoutExpired:
logger.warning("Docker logs timed out for %s", container)
return []
def get_health(self, container: str) -> dict:
"""Get container health status.
Containers without a healthcheck defined have no .State.Health key,
so we query status and health separately to avoid template parse errors.
"""
try:
# Get container state status (always present)
result = subprocess.run(
[
"docker", "inspect", container,
"--format", "{{.State.Status}}|{{.State.FinishedAt}}"
],
capture_output=True, text=True, timeout=10
)
if result.returncode != 0:
return {"status": "unknown", "health": "none", "finished_at": ""}
parts = result.stdout.strip().split("|")
status = parts[0] if len(parts) > 0 else "unknown"
finished_at = parts[1] if len(parts) > 1 else ""
# Try to get health status — may not exist (no healthcheck defined)
health_result = subprocess.run(
[
"docker", "inspect", container,
"--format", "{{if .State.Health}}{{.State.Health.Status}}{{else}}none{{end}}"
],
capture_output=True, text=True, timeout=10
)
health = health_result.stdout.strip() if health_result.returncode == 0 else "none"
if not health:
health = "none"
return {
"status": status,
"health": health,
"finished_at": finished_at,
}
except (FileNotFoundError, subprocess.TimeoutExpired):
return {"status": "error", "health": "error", "finished_at": ""}
def get_all_logs(
self,
containers: Optional[list[str]] = None,
since_minutes: int = 60,
tail: int = 500
) -> list[LogEntry]:
"""Get logs from all specified containers, or all running if none specified."""
if not containers:
containers = self.get_containers()
all_logs = []
for container in containers:
logs = self.get_logs(container, since_minutes, tail)
all_logs.extend(logs)
# Sort by timestamp
all_logs.sort(key=lambda e: e.timestamp)
return all_logs