Compare commits
1 Commits
dev
..
001cbafe56
| Author | SHA1 | Date | |
|---|---|---|---|
 bvandeusen
|
001cbafe56 |
@@ -14,17 +14,6 @@ on:
|
||||
# pull_request intentionally absent — push on [dev, main] already fires CI for
|
||||
# every dev commit and dev→main PRs. Single-operator repo, no fork PRs.
|
||||
|
||||
# Serialize CI runs per ref so the publish lane never shares the runner's docker
|
||||
# daemon with another run. Two publishes racing on one daemon evict each other's
|
||||
# freshly-built image mid-push, breaking the post-build tag/push (issue #1093).
|
||||
# cancel-in-progress:false is deliberate — rule 46 requires EVERY dev/main push to
|
||||
# publish its own immutable :<sha> image, so a superseded run must still run to
|
||||
# completion (never cancelled) to emit its SHA. (Forgejo honors `concurrency:` —
|
||||
# CI-runner's build workflows use it.)
|
||||
concurrency:
|
||||
group: ci-${{ github.ref }}
|
||||
cancel-in-progress: false
|
||||
|
||||
jobs:
|
||||
# Fast-fail lint lane. ruff is pre-installed in the ci-python image, so this
|
||||
# runs with NO dependency install and surfaces lint bounces in seconds.
|
||||
@@ -126,25 +115,14 @@ jobs:
|
||||
run: |
|
||||
set -euxo pipefail
|
||||
IMAGE=git.fabledsword.com/bvandeusen/steward
|
||||
# The moving tag for this ref: dev→:dev, main→:latest (rule 46). main IS
|
||||
# the production line, so :latest tracks main's tip; there is no :main.
|
||||
MOVING=""
|
||||
docker build -t "$IMAGE:${{ github.sha }}" .
|
||||
docker push "$IMAGE:${{ github.sha }}"
|
||||
if [ "${{ github.ref }}" = "refs/heads/dev" ]; then
|
||||
MOVING="dev"
|
||||
docker tag "$IMAGE:${{ github.sha }}" "$IMAGE:dev"
|
||||
docker push "$IMAGE:dev"
|
||||
elif [ "${{ github.ref }}" = "refs/heads/main" ]; then
|
||||
MOVING="latest"
|
||||
fi
|
||||
# Tag BOTH targets at build time (mirrors CI-runner's build workflows):
|
||||
# one `docker build -t :<sha> -t :<moving>` points both tags at the image
|
||||
# atomically, so we never run a post-push `docker tag` of an image a
|
||||
# concurrent run could have evicted from the shared daemon (issue #1093).
|
||||
if [ -n "$MOVING" ]; then
|
||||
docker build -t "$IMAGE:${{ github.sha }}" -t "$IMAGE:$MOVING" .
|
||||
docker push "$IMAGE:${{ github.sha }}"
|
||||
docker push "$IMAGE:$MOVING"
|
||||
else
|
||||
docker build -t "$IMAGE:${{ github.sha }}" .
|
||||
docker push "$IMAGE:${{ github.sha }}"
|
||||
docker tag "$IMAGE:${{ github.sha }}" "$IMAGE:latest"
|
||||
docker push "$IMAGE:latest"
|
||||
fi
|
||||
- name: Prune dangling layers
|
||||
if: always()
|
||||
|
||||
@@ -74,15 +74,8 @@ token = a1b2c3d4...
|
||||
interval_seconds = 30
|
||||
hostname = myhost # optional; defaults to uname -n
|
||||
mounts = /, /mnt/data # optional; defaults to all real mounts (excluding tmpfs/devtmpfs/etc.)
|
||||
docker_logs_enabled = true # optional; on by default — per-host container-log kill-switch
|
||||
docker_log_exclude = watchtower, noisy-svc # optional; container names whose logs this host skips collecting
|
||||
```
|
||||
|
||||
Container-log collection is on by default (no config needed). The two keys above
|
||||
are the per-host opt-outs; the operator-facing global toggle + exclude live in
|
||||
Settings and are enforced server-side (the push model has no channel to command
|
||||
an agent), so they don't require touching a host's conf.
|
||||
|
||||
### Agent internals (function list, not classes)
|
||||
|
||||
- `read_config(path)` — parses the conf file into a dict.
|
||||
@@ -92,17 +85,10 @@ an agent), so they don't require touching a host's conf.
|
||||
- `collect_load()` — reads `/proc/loadavg`, returns `[1m, 5m, 15m]`.
|
||||
- `collect_uptime()` — reads `/proc/uptime`, returns seconds since boot (int).
|
||||
- `collect_metadata()` — `os.uname()` for kernel + arch, `/etc/os-release` for distro. Called once at startup and cached.
|
||||
- `collect_docker_logs(socket, containers, state, exclude)` — (m79) per running
|
||||
container, fetches new log lines over the Docker socket with an incremental
|
||||
`since` cursor kept per container in `state` (a container's first interval
|
||||
seeds from a short tail). Demuxes the multiplexed stream, parses each line's
|
||||
RFC3339 timestamp, caps the whole batch at a byte limit (a marker line records
|
||||
a truncation; the deferred lines come next interval). Folded into the sample as
|
||||
`docker_logs`; omitted when empty.
|
||||
- `build_payload()` — assembles a snapshot from all collectors into one dict.
|
||||
- `post_payload(url, token, payloads)` — POSTs a list of samples, returns success/failure.
|
||||
- `RingBuffer(maxlen=20)` — tiny FIFO wrapper, drops oldest when full.
|
||||
- `main_loop()` — the 30s loop: collect → try POST → on failure push to buffer + backoff → on success flush buffer. Container logs are stripped from a sample before it's buffered (metrics survive an outage; stale logs are dropped).
|
||||
- `main_loop()` — the 30s loop: collect → try POST → on failure push to buffer + backoff → on success flush buffer.
|
||||
|
||||
**Target: ~300 lines total including docstrings.** More than that is a smell that the agent is over-scoping.
|
||||
|
||||
@@ -269,7 +255,6 @@ Content-Type: application/json
|
||||
- **`metadata` is sent on every POST**, not just on change. Server-side diff detects actual changes and only writes on change. Cost per POST is one dict — negligible. Benefit: server can cleanly detect agent restarts.
|
||||
- **Raw bytes, not percentages, for memory and storage.** Percentages are derived server-side. Changing the "what counts as used" math doesn't require re-releasing the agent.
|
||||
- **CPU is the one exception** — reported as a percentage because it's inherently a derivative (delta over time), not a snapshot. The agent must sample twice to compute it.
|
||||
- **Container logs (m79) ride in the same push** as `docker_logs`: a list of `{container, stream, ts, line}` records — incremental since the previous interval. This pushes the *same direction* as metrics, so batched log history needs no inbound channel (only sub-second live-follow would). The server ingests them into `docker_logs`, enforces the global toggle + exclude list on ingest, and bounds storage with a per-container size+age ring. `docker_logs` is omitted when there's nothing new.
|
||||
|
||||
### Server expansion into `PluginMetric` rows
|
||||
|
||||
|
||||
@@ -92,55 +92,6 @@ def _derive_events(old_state: dict, new_containers: list) -> list:
|
||||
return events
|
||||
|
||||
|
||||
def _log_rows(log_batches, host_id: str):
|
||||
"""Pure: flatten the agent's log batches → docker_logs row kwargs (no DB).
|
||||
|
||||
`log_batches` is a list of (recorded_at, records) where each record is
|
||||
{"container", "stream", "ts", "line"}. A line's own Docker timestamp wins;
|
||||
recorded_at is the fallback when the agent couldn't parse one. Drops the
|
||||
agent's advisory truncation marker (container "_steward" — it signals a
|
||||
deferral, not a real container), malformed records, and lineless records;
|
||||
normalises an unknown stream to stdout. Unit-testable in isolation.
|
||||
"""
|
||||
for recorded_at, records in log_batches:
|
||||
if not isinstance(records, list):
|
||||
continue
|
||||
for rec in records:
|
||||
if not isinstance(rec, dict):
|
||||
continue
|
||||
name = rec.get("container")
|
||||
line = rec.get("line")
|
||||
if not name or name == "_steward" or line is None:
|
||||
continue
|
||||
ts = _parse_started_at(rec.get("ts")) or recorded_at
|
||||
stream = rec.get("stream")
|
||||
if stream not in ("stdout", "stderr"):
|
||||
stream = "stdout"
|
||||
yield {"host_id": host_id, "container_name": str(name)[:255],
|
||||
"ts": ts, "stream": stream, "line": str(line)}
|
||||
|
||||
|
||||
async def _persist_logs(session, host, log_batches) -> None:
|
||||
"""Append pushed container log lines (one row per line) for this host.
|
||||
|
||||
Time-series / append-only — the per-container size+age ring (retention)
|
||||
bounds growth, so a chatty container just keeps a shorter window. The global
|
||||
toggle + exclude list (Settings, no restart) are enforced here: this is the
|
||||
authoritative drop point, since the push model has no channel to tell an
|
||||
agent to stop collecting.
|
||||
"""
|
||||
from steward.core.settings import get_setting
|
||||
from .models import DockerLog
|
||||
|
||||
if not await get_setting(session, "docker.logs.enabled"):
|
||||
return
|
||||
exclude = set(await get_setting(session, "docker.logs.exclude") or ())
|
||||
for row in _log_rows(log_batches, host.id):
|
||||
if row["container_name"] in exclude:
|
||||
continue
|
||||
session.add(DockerLog(**row))
|
||||
|
||||
|
||||
async def _persist_swarm(session, host, swarm: dict) -> None:
|
||||
"""Upsert this manager's swarm topology; drop rows no longer reported.
|
||||
|
||||
@@ -250,8 +201,7 @@ async def _persist_disk(session, host, disk: dict) -> None:
|
||||
await session.execute(stale)
|
||||
|
||||
|
||||
async def persist_host_docker(session, host, snapshots, swarm=None, disk=None,
|
||||
logs=None) -> None:
|
||||
async def persist_host_docker(session, host, snapshots, swarm=None, disk=None) -> None:
|
||||
"""Upsert containers + time-series + lifecycle events + swarm for one host.
|
||||
|
||||
`snapshots` is a list of (recorded_at: datetime, containers: list[dict]) —
|
||||
@@ -261,9 +211,7 @@ async def persist_host_docker(session, host, snapshots, swarm=None, disk=None,
|
||||
alert pipeline, and lifecycle-event derivation. `swarm` is the newest
|
||||
sample's swarm object (or None off managers) — persisted when present.
|
||||
`disk` is the newest sample's /system/df summary (or None on Docker-less
|
||||
hosts) — persisted when present. `logs` is a list of (recorded_at, records)
|
||||
log batches (m79) — appended one row per line when present, independent of
|
||||
whether this sample also carried container metrics.
|
||||
hosts) — persisted when present.
|
||||
"""
|
||||
from steward.core.alerts import record_metric
|
||||
from .models import DockerContainer, DockerEvent, DockerMetric
|
||||
@@ -272,8 +220,6 @@ async def persist_host_docker(session, host, snapshots, swarm=None, disk=None,
|
||||
await _persist_swarm(session, host, swarm)
|
||||
if disk is not None:
|
||||
await _persist_disk(session, host, disk)
|
||||
if logs:
|
||||
await _persist_logs(session, host, logs)
|
||||
|
||||
if not snapshots:
|
||||
return
|
||||
|
||||
@@ -1,48 +0,0 @@
|
||||
"""Docker container logs table
|
||||
|
||||
Adds docker_logs — one row per container log line, pushed by the host agent and
|
||||
folded into its metrics push. Time-series, host-scoped (container names are only
|
||||
unique within a host). Bounded by a per-container size+age ring in retention, so
|
||||
a chatty container keeps a shorter window rather than growing without limit.
|
||||
Additive create_table + twin indexes (viewer lookup by (host, container, ts);
|
||||
age-cutoff prune by ts).
|
||||
|
||||
Revision ID: docker_009_container_logs
|
||||
Revises: docker_008_bigint_mem
|
||||
Create Date: 2026-07-19
|
||||
"""
|
||||
from typing import Sequence, Union
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
|
||||
revision: str = "docker_009_container_logs"
|
||||
down_revision: Union[str, None] = "docker_008_bigint_mem"
|
||||
branch_labels: Union[str, Sequence[str], None] = None
|
||||
depends_on: Union[str, Sequence[str], None] = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.create_table(
|
||||
"docker_logs",
|
||||
sa.Column("id", sa.String(length=36), nullable=False),
|
||||
sa.Column("host_id", sa.String(length=36), nullable=False),
|
||||
sa.Column("container_name", sa.String(length=255), nullable=False),
|
||||
sa.Column("ts", sa.DateTime(timezone=True), nullable=False),
|
||||
sa.Column("stream", sa.String(length=8), nullable=False, server_default="stdout"),
|
||||
sa.Column("line", sa.Text(), nullable=False, server_default=""),
|
||||
sa.ForeignKeyConstraint(["host_id"], ["hosts.id"], ondelete="CASCADE"),
|
||||
sa.PrimaryKeyConstraint("id"),
|
||||
)
|
||||
op.create_index("ix_docker_logs_host_id", "docker_logs", ["host_id"])
|
||||
op.create_index("ix_docker_logs_container_name", "docker_logs", ["container_name"])
|
||||
op.create_index("ix_docker_logs_host_container_time",
|
||||
"docker_logs", ["host_id", "container_name", "ts"])
|
||||
op.create_index("ix_docker_logs_ts", "docker_logs", ["ts"])
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.drop_index("ix_docker_logs_ts", table_name="docker_logs")
|
||||
op.drop_index("ix_docker_logs_host_container_time", table_name="docker_logs")
|
||||
op.drop_index("ix_docker_logs_container_name", table_name="docker_logs")
|
||||
op.drop_index("ix_docker_logs_host_id", table_name="docker_logs")
|
||||
op.drop_table("docker_logs")
|
||||
@@ -163,43 +163,6 @@ class DockerEvent(Base):
|
||||
)
|
||||
|
||||
|
||||
class DockerLog(Base):
|
||||
"""Container log lines pushed by the host agent — one row per line.
|
||||
|
||||
Time-series, scoped to the reporting host (container names are only unique
|
||||
within a host, same identity as docker_metrics). The agent tails each running
|
||||
container incrementally and folds new lines into its metrics push; `ts` is the
|
||||
line's own Docker timestamp. Bounded by a per-container size+age ring
|
||||
(retention), so a chatty container just keeps a shorter window rather than
|
||||
growing without limit.
|
||||
"""
|
||||
__tablename__ = "docker_logs"
|
||||
|
||||
id: Mapped[str] = mapped_column(
|
||||
String(36), primary_key=True, default=lambda: str(uuid.uuid4())
|
||||
)
|
||||
host_id: Mapped[str] = mapped_column(
|
||||
String(36), ForeignKey("hosts.id", ondelete="CASCADE"), nullable=False, index=True
|
||||
)
|
||||
container_name: Mapped[str] = mapped_column(String(255), nullable=False, index=True)
|
||||
ts: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), nullable=False,
|
||||
default=lambda: datetime.now(timezone.utc),
|
||||
)
|
||||
stream: Mapped[str] = mapped_column(String(8), nullable=False, default="stdout")
|
||||
# stdout | stderr
|
||||
line: Mapped[str] = mapped_column(Text, nullable=False, default="")
|
||||
|
||||
# Viewer filters on (host_id, container_name) and sorts by time; the ring
|
||||
# prune walks the same key newest-first. A second index on `ts` alone serves
|
||||
# the age-cutoff delete. Twin-index idiom, mirroring docker_events.
|
||||
__table_args__ = (
|
||||
Index("ix_docker_logs_host_container_time",
|
||||
"host_id", "container_name", "ts"),
|
||||
Index("ix_docker_logs_ts", "ts"),
|
||||
)
|
||||
|
||||
|
||||
class DockerSwarmService(Base):
|
||||
"""A Swarm service as seen by a manager host: desired vs running replicas.
|
||||
|
||||
|
||||
@@ -38,8 +38,6 @@ async def run_docker_retention(
|
||||
events_days: int,
|
||||
metrics_raw_days: int,
|
||||
metrics_rollup_days: int,
|
||||
logs_retention_days: int = 3,
|
||||
logs_max_bytes_per_container: int = 5_000_000,
|
||||
now: datetime | None = None,
|
||||
) -> dict:
|
||||
"""Roll up + prune Docker time-series. Returns a counts dict for logging.
|
||||
@@ -49,22 +47,18 @@ async def run_docker_retention(
|
||||
re-run is idempotent, then delete those raw rows.
|
||||
2. Prune rolled-up rows older than the rollup window.
|
||||
3. Prune docker_events older than the events window.
|
||||
4. Prune docker_logs with a per-container size+age ring (m79): drop lines
|
||||
older than the age window, then keep only the newest ~cap bytes per
|
||||
(host, container).
|
||||
"""
|
||||
from datetime import timezone
|
||||
|
||||
from sqlalchemy import delete, func, select
|
||||
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
||||
|
||||
from .models import DockerEvent, DockerLog, DockerMetric, DockerMetricHourly
|
||||
from .models import DockerEvent, DockerMetric, DockerMetricHourly
|
||||
|
||||
if now is None:
|
||||
now = datetime.now(timezone.utc)
|
||||
|
||||
rolled = rolled_rows = events_pruned = rollup_pruned = 0
|
||||
logs_age_pruned = logs_size_pruned = 0
|
||||
|
||||
# ── 1. Roll up raw metrics older than the raw window into hourly buckets ──
|
||||
raw_cutoff = _rollup_cutoff(now, metrics_raw_days)
|
||||
@@ -126,35 +120,9 @@ async def run_docker_retention(
|
||||
)
|
||||
events_pruned = res.rowcount or 0
|
||||
|
||||
# ── 4. Container-log ring: age cutoff, then per-container byte cap (m79) ──
|
||||
logs_cutoff = now - timedelta(days=logs_retention_days)
|
||||
res = await session.execute(
|
||||
delete(DockerLog).where(DockerLog.ts < logs_cutoff)
|
||||
)
|
||||
logs_age_pruned = res.rowcount or 0
|
||||
|
||||
# Size ring: per (host, container), sum line bytes newest-first; delete a row
|
||||
# once its strictly-newer siblings already fill the cap. Using the EXCLUSIVE
|
||||
# prefix (running total minus this row) means the newest row always survives,
|
||||
# so a single line larger than the cap is never wiped out.
|
||||
running = func.sum(func.length(DockerLog.line)).over(
|
||||
partition_by=[DockerLog.host_id, DockerLog.container_name],
|
||||
order_by=[DockerLog.ts.desc(), DockerLog.id.desc()],
|
||||
)
|
||||
prefix_excl = (running - func.length(DockerLog.line)).label("prefix_excl")
|
||||
ranked = select(DockerLog.id, prefix_excl).subquery()
|
||||
over_cap = select(ranked.c.id).where(
|
||||
ranked.c.prefix_excl >= logs_max_bytes_per_container)
|
||||
res = await session.execute(
|
||||
delete(DockerLog).where(DockerLog.id.in_(over_cap))
|
||||
)
|
||||
logs_size_pruned = res.rowcount or 0
|
||||
|
||||
return {
|
||||
"buckets_rolled": rolled,
|
||||
"raw_rows_rolled": rolled_rows,
|
||||
"rollup_pruned": rollup_pruned,
|
||||
"events_pruned": events_pruned,
|
||||
"logs_age_pruned": logs_age_pruned,
|
||||
"logs_size_pruned": logs_size_pruned,
|
||||
}
|
||||
|
||||
+6
-143
@@ -3,10 +3,7 @@ from __future__ import annotations
|
||||
import json
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from quart import (
|
||||
Blueprint, current_app, jsonify, redirect, render_template, request,
|
||||
session, url_for,
|
||||
)
|
||||
from quart import Blueprint, current_app, render_template, request
|
||||
from sqlalchemy import func, select
|
||||
|
||||
from steward.auth.middleware import require_role
|
||||
@@ -16,31 +13,13 @@ from steward.core.time_range import parse_range, DEFAULT_RANGE
|
||||
from .dedup import dedup_by_container_id
|
||||
from .swarm_view import build_swarm_services
|
||||
from .models import (
|
||||
DockerContainer, DockerDiskUsage, DockerEvent, DockerImage, DockerLog,
|
||||
DockerMetric, DockerSwarmNode, DockerSwarmService,
|
||||
DockerContainer, DockerDiskUsage, DockerEvent, DockerImage, DockerMetric,
|
||||
DockerSwarmNode, DockerSwarmService,
|
||||
)
|
||||
|
||||
docker_bp = Blueprint("docker", __name__, template_folder="templates")
|
||||
|
||||
|
||||
def _error(status: int, code: str, detail: str | None = None):
|
||||
body: dict = {"ok": False, "error": code}
|
||||
if detail:
|
||||
body["detail"] = detail
|
||||
return jsonify(body), status
|
||||
|
||||
|
||||
def _prune_extra_vars(prune_target: str) -> dict:
|
||||
"""Extra-vars for the bundled docker_prune playbook. "Prune unused images"
|
||||
means ALL unused (docker image prune -a), not just dangling; the full system
|
||||
prune stays conservative (-f, no -a) per the operator's choice (m78). Pure
|
||||
helper so the mapping is unit-tested without the request/DB stack."""
|
||||
extra_vars: dict = {"prune_target": prune_target}
|
||||
if prune_target == "images":
|
||||
extra_vars["prune_all_images"] = True
|
||||
return extra_vars
|
||||
|
||||
|
||||
def _human_bytes(n: int | None) -> str:
|
||||
"""Compact binary size string (e.g. '1.4 GiB', '512 MiB', '0 B')."""
|
||||
if n is None:
|
||||
@@ -520,13 +499,9 @@ async def swarm():
|
||||
async def disk():
|
||||
"""Image/disk usage page: reclaimable space, per-image sizes, stopped count.
|
||||
|
||||
Admins can reclaim space per host via the prune buttons, which fire the
|
||||
audited bundled prune playbook through Ansible (m78) — the collection agent
|
||||
itself stays read-only.
|
||||
Read-only — prune actions are deferred to the cleanup-actions milestone, so
|
||||
this surfaces the numbers and notes where reclaim lives.
|
||||
"""
|
||||
from steward.core.capabilities import has_capability
|
||||
from steward.models.ansible_inventory import AnsibleTarget
|
||||
|
||||
async with current_app.db_sessionmaker() as db:
|
||||
summaries = list((await db.execute(select(DockerDiskUsage))).scalars())
|
||||
images = list((await db.execute(
|
||||
@@ -541,16 +516,6 @@ async def disk():
|
||||
for hid in stopped_rows:
|
||||
stopped_by_host[hid] = stopped_by_host.get(hid, 0) + 1
|
||||
hosts = await _host_map(db, {s.host_id for s in summaries})
|
||||
# Which hosts have a linked Ansible target — gates the prune buttons
|
||||
# (a target is required to route the playbook run to that host).
|
||||
host_ids = {s.host_id for s in summaries}
|
||||
target_by_host: dict[str, str] = {}
|
||||
if host_ids:
|
||||
for hid, tid in (await db.execute(
|
||||
select(AnsibleTarget.host_id, AnsibleTarget.id)
|
||||
.where(AnsibleTarget.host_id.in_(host_ids))
|
||||
)).all():
|
||||
target_by_host[hid] = tid
|
||||
|
||||
images_by_host: dict[str, list] = {}
|
||||
for im in images:
|
||||
@@ -577,60 +542,11 @@ async def disk():
|
||||
},
|
||||
"stopped": stopped_by_host.get(s.host_id, 0),
|
||||
"images": images_by_host.get(s.host_id, []),
|
||||
"has_target": s.host_id in target_by_host,
|
||||
}
|
||||
for s in summaries
|
||||
]
|
||||
host_groups.sort(key=lambda g: g["host_name"].lower())
|
||||
return await render_template(
|
||||
"docker/disk.html", host_groups=host_groups,
|
||||
ansible_available=has_capability("ansible.run_playbook"),
|
||||
)
|
||||
|
||||
|
||||
@docker_bp.post("/disk/<host_id>/prune")
|
||||
@require_role(UserRole.admin)
|
||||
async def disk_prune(host_id: str):
|
||||
"""Reclaim Docker disk on a host by firing the bundled prune playbook via
|
||||
Ansible (audited, admin-gated) — the collection agent stays read-only (m78).
|
||||
`target` selects the scope: containers | images | system.
|
||||
"""
|
||||
from steward.core.capabilities import has_capability, invoke_capability
|
||||
from steward.ansible.sources import BUILTIN_SOURCE_NAME
|
||||
from steward.models.ansible_inventory import AnsibleTarget
|
||||
|
||||
if not has_capability("ansible.run_playbook"):
|
||||
return _error(400, "ansible_unavailable", "Ansible is not available")
|
||||
|
||||
form = await request.form
|
||||
prune_target = (form.get("target", "") or "").strip()
|
||||
if prune_target not in ("containers", "images", "system"):
|
||||
return _error(400, "bad_target", "Unknown prune target")
|
||||
|
||||
async with current_app.db_sessionmaker() as db:
|
||||
target = (await db.execute(
|
||||
select(AnsibleTarget).where(AnsibleTarget.host_id == host_id)
|
||||
)).scalar_one_or_none()
|
||||
if target is None:
|
||||
return _error(400, "no_target",
|
||||
"Link an Ansible target to this host before pruning")
|
||||
|
||||
# extra_vars_map outranks the playbook's own `vars:` defaults.
|
||||
extra_vars = _prune_extra_vars(prune_target)
|
||||
|
||||
actor_role = UserRole(session.get("user_role", "viewer"))
|
||||
run, _source, err = await invoke_capability(
|
||||
"ansible.run_playbook", actor_role,
|
||||
current_app._get_current_object(), # type: ignore[attr-defined]
|
||||
source_name=BUILTIN_SOURCE_NAME,
|
||||
playbook_path="maintenance/docker_prune.yml",
|
||||
inventory_scope=f"steward:target:{target.id}",
|
||||
params={"extra_vars_map": extra_vars},
|
||||
triggered_by=session.get("user_id"),
|
||||
)
|
||||
if err:
|
||||
return _error(400, "prune_failed", err)
|
||||
return redirect(url_for("ansible.run_detail", run_id=run.id))
|
||||
return await render_template("docker/disk.html", host_groups=host_groups)
|
||||
|
||||
|
||||
@docker_bp.get("/container/<host_id>/<name>/history")
|
||||
@@ -647,56 +563,3 @@ async def container_history(host_id: str, name: str):
|
||||
have_data=len(cpu_hist) >= 2,
|
||||
range_key=range_key,
|
||||
)
|
||||
|
||||
|
||||
# Newest lines returned per fetch — the retained window is bounded by the ring,
|
||||
# but a container can still hold thousands of lines; cap what one fetch renders.
|
||||
_LOG_TAIL_DEFAULT = 500
|
||||
|
||||
|
||||
async def _query_logs(db, host_id: str, name: str, stream: str, query: str,
|
||||
limit: int) -> list:
|
||||
"""The recent retained lines for one container, newest-first, optionally
|
||||
filtered by stream and a case-insensitive substring.
|
||||
|
||||
Newest-first is deliberate: the viewer replaces the list on each poll, which
|
||||
resets scroll to the top — so the latest lines stay visible without any
|
||||
scroll handling, and older lines are a scroll away.
|
||||
"""
|
||||
stmt = (select(DockerLog.ts, DockerLog.stream, DockerLog.line)
|
||||
.where(DockerLog.host_id == host_id)
|
||||
.where(DockerLog.container_name == name))
|
||||
if stream in ("stdout", "stderr"):
|
||||
stmt = stmt.where(DockerLog.stream == stream)
|
||||
if query:
|
||||
stmt = stmt.where(DockerLog.line.ilike(f"%{query}%"))
|
||||
stmt = stmt.order_by(DockerLog.ts.desc(), DockerLog.id.desc()).limit(limit)
|
||||
return (await db.execute(stmt)).all()
|
||||
|
||||
|
||||
@docker_bp.get("/container/<host_id>/<name>/logs")
|
||||
@require_role(UserRole.viewer)
|
||||
async def container_logs(host_id: str, name: str):
|
||||
"""Full log-viewer page for one container (lines load via an HTMX fragment)."""
|
||||
async with current_app.db_sessionmaker() as db:
|
||||
host = await db.get(Host, host_id)
|
||||
return await render_template(
|
||||
"docker/container_logs.html", host=host, host_id=host_id, name=name,
|
||||
)
|
||||
|
||||
|
||||
@docker_bp.get("/container/<host_id>/<name>/logs/lines")
|
||||
@require_role(UserRole.viewer)
|
||||
async def container_logs_lines(host_id: str, name: str):
|
||||
"""HTMX fragment: the recent retained log lines, stream + text filtered.
|
||||
Polled every few seconds by the viewer for near-live follow."""
|
||||
stream = request.args.get("stream", "all")
|
||||
query = (request.args.get("q") or "").strip()
|
||||
async with current_app.db_sessionmaker() as db:
|
||||
rows = await _query_logs(db, host_id, name, stream, query, _LOG_TAIL_DEFAULT)
|
||||
return await render_template(
|
||||
"docker/_container_logs_lines.html",
|
||||
lines=[{"ts": r.ts, "stream": r.stream, "line": r.line} for r in rows],
|
||||
name=name, tail=_LOG_TAIL_DEFAULT,
|
||||
filtered=bool(query) or stream in ("stdout", "stderr"),
|
||||
)
|
||||
|
||||
@@ -1,22 +0,0 @@
|
||||
{# docker/_container_logs_lines.html — log-line fragment, HTMX-polled (m79).
|
||||
{{ l.line }} is auto-escaped by Jinja, so log content can't inject markup. #}
|
||||
{% if lines %}
|
||||
{% for l in lines %}
|
||||
<div style="display:flex;gap:0.6rem;white-space:pre-wrap;word-break:break-word;">
|
||||
<span style="color:var(--text-dim);flex-shrink:0;" title="{{ l.ts }}">{{ l.ts.strftime("%m-%d %H:%M:%S") }}</span>
|
||||
<span style="flex-shrink:0;width:3.2rem;color:{% if l.stream == 'stderr' %}var(--red){% else %}var(--text-muted){% endif %};">{{ l.stream }}</span>
|
||||
<span style="flex:1;min-width:0;{% if l.stream == 'stderr' %}color:var(--text);{% endif %}">{{ l.line }}</span>
|
||||
</div>
|
||||
{% endfor %}
|
||||
{% else %}
|
||||
<div style="color:var(--text-muted);padding:1rem 0.25rem;">
|
||||
{% if filtered %}
|
||||
No lines match the current filter.
|
||||
{% else %}
|
||||
No logs collected yet for <code>{{ name }}</code>. Lines appear within a few
|
||||
seconds of the container writing to stdout/stderr — unless it's on the log
|
||||
exclude list or log collection is turned off in
|
||||
<a href="/settings/thresholds/">Settings → Thresholds & Retention</a>.
|
||||
{% endif %}
|
||||
</div>
|
||||
{% endif %}
|
||||
@@ -26,7 +26,6 @@
|
||||
<div style="font-size:0.82rem;color:var(--text-muted);margin-bottom:1.5rem;">
|
||||
{{ container.status }}{% if uptime %} · up {{ uptime }}{% endif %}
|
||||
{% if host %} · on <a href="/hosts/{{ host.id }}" style="color:var(--text-muted);">{{ host.name }}</a>{% endif %}
|
||||
· <a href="/plugins/docker/container/{{ host_id }}/{{ name }}/logs">View logs</a>
|
||||
</div>
|
||||
|
||||
{# ── Facts grid ──────────────────────────────────────────────────────────── #}
|
||||
|
||||
@@ -1,51 +0,0 @@
|
||||
{# docker/container_logs.html — per-container log viewer (m79) #}
|
||||
{% extends "base.html" %}
|
||||
{% from "_macros.html" import crumbs %}
|
||||
{% block title %}Logs — {{ name }} — Docker — Steward{% endblock %}
|
||||
{% block breadcrumb %}{{ crumbs([
|
||||
("Docker", "/plugins/docker/"),
|
||||
(name, "/plugins/docker/container/" ~ host_id ~ "/" ~ name),
|
||||
("Logs", "")]) }}{% endblock %}
|
||||
{% block content %}
|
||||
|
||||
<div style="display:flex;align-items:baseline;gap:0.6rem;margin-bottom:0.35rem;flex-wrap:wrap;">
|
||||
<h1 class="page-title" style="margin-bottom:0;">{{ name }}</h1>
|
||||
<span style="font-size:0.9rem;color:var(--text-muted);">logs</span>
|
||||
</div>
|
||||
<div style="font-size:0.82rem;color:var(--text-muted);margin-bottom:1rem;">
|
||||
Recent lines (newest first) collected by the host agent{% if host %} on
|
||||
<a href="/hosts/{{ host.id }}" style="color:var(--text-muted);">{{ host.name }}</a>{% endif %} —
|
||||
updated every few seconds.
|
||||
<a href="/plugins/docker/container/{{ host_id }}/{{ name }}">← back to container</a>
|
||||
</div>
|
||||
|
||||
{# ── Controls: stream filter + text search (drive the fragment via HTMX) ───── #}
|
||||
<form id="log-controls" onsubmit="return false;"
|
||||
style="display:flex;gap:0.6rem;align-items:center;flex-wrap:wrap;margin-bottom:0.6rem;">
|
||||
<label style="font-size:0.8rem;color:var(--text-muted);display:flex;align-items:center;gap:0.35rem;">
|
||||
Stream
|
||||
<select name="stream" style="padding:0.25rem 0.4rem;">
|
||||
<option value="all">all</option>
|
||||
<option value="stdout">stdout</option>
|
||||
<option value="stderr">stderr</option>
|
||||
</select>
|
||||
</label>
|
||||
<input type="search" name="q" placeholder="Filter lines…" autocomplete="off"
|
||||
aria-label="Filter log lines"
|
||||
style="flex:1;min-width:180px;padding:0.3rem 0.5rem;">
|
||||
</form>
|
||||
|
||||
<div class="card-flush">
|
||||
<div id="log-lines"
|
||||
hx-get="/plugins/docker/container/{{ host_id }}/{{ name }}/logs/lines"
|
||||
hx-trigger="load, every 5s, change from:#log-controls, keyup changed delay:400ms from:#log-controls"
|
||||
hx-include="#log-controls"
|
||||
hx-swap="innerHTML"
|
||||
role="log" aria-live="polite" tabindex="0"
|
||||
style="max-height:70vh;overflow:auto;padding:0.5rem 0.75rem;
|
||||
font-family:ui-monospace,monospace;font-size:0.8rem;line-height:1.5;">
|
||||
<div style="color:var(--text-muted);">Loading…</div>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
{% endblock %}
|
||||
@@ -7,8 +7,8 @@
|
||||
|
||||
<h1 class="page-title" style="margin-bottom:0.4rem;">Image & disk usage</h1>
|
||||
<p style="font-size:0.82rem;color:var(--text-muted);margin-bottom:1.5rem;">
|
||||
Reclaimable = space held by images no container references. Admins can prune
|
||||
per host below; each action runs an audited Ansible playbook on that host.
|
||||
Reclaimable = space held by images no container references. Cleanup actions
|
||||
(prune) arrive in a later release — these are read-only figures for now.
|
||||
</p>
|
||||
|
||||
{% if host_groups %}
|
||||
@@ -47,43 +47,6 @@
|
||||
</div>
|
||||
</div>
|
||||
|
||||
{# ── Cleanup actions (admin only; audited Ansible prune run) ───────────── #}
|
||||
{% if session.user_role == 'admin' %}
|
||||
<div style="margin-bottom:1rem;">
|
||||
{% if ansible_available and g.has_target %}
|
||||
<div style="display:flex;flex-wrap:wrap;gap:0.5rem;align-items:center;">
|
||||
<form method="post" action="/plugins/docker/disk/{{ g.host_id }}/prune" style="margin:0;"
|
||||
data-msg="Remove all STOPPED containers on {{ g.host_name|e }}? This cannot be undone."
|
||||
onsubmit="return confirm(this.dataset.msg);">
|
||||
<input type="hidden" name="target" value="containers">
|
||||
<button type="submit" class="btn btn-sm">Prune stopped containers</button>
|
||||
</form>
|
||||
<form method="post" action="/plugins/docker/disk/{{ g.host_id }}/prune" style="margin:0;"
|
||||
data-msg="Remove ALL unused images on {{ g.host_name|e }} (docker image prune -a)? Any image not used by a container is deleted."
|
||||
onsubmit="return confirm(this.dataset.msg);">
|
||||
<input type="hidden" name="target" value="images">
|
||||
<button type="submit" class="btn btn-sm">Prune unused images</button>
|
||||
</form>
|
||||
<form method="post" action="/plugins/docker/disk/{{ g.host_id }}/prune" style="margin:0;"
|
||||
data-msg="Run a full system prune on {{ g.host_name|e }}? Removes stopped containers, unused networks, dangling images and build cache."
|
||||
onsubmit="return confirm(this.dataset.msg);">
|
||||
<input type="hidden" name="target" value="system">
|
||||
<button type="submit" class="btn btn-sm btn-danger">System prune…</button>
|
||||
</form>
|
||||
</div>
|
||||
<div style="font-size:0.72rem;color:var(--text-muted);margin-top:0.4rem;">
|
||||
Reclaimed space appears on the next agent sample.
|
||||
</div>
|
||||
{% elif not ansible_available %}
|
||||
<div style="font-size:0.74rem;color:var(--text-muted);">Cleanup needs the Ansible runner (currently unavailable).</div>
|
||||
{% else %}
|
||||
<div style="font-size:0.74rem;color:var(--text-muted);">
|
||||
Link an <a href="/hosts/{{ g.host_id }}">Ansible target</a> to this host to enable prune actions.
|
||||
</div>
|
||||
{% endif %}
|
||||
</div>
|
||||
{% endif %}
|
||||
|
||||
{# ── Per-image table ──────────────────────────────────────────────────── #}
|
||||
<div class="card-flush">
|
||||
<table class="table">
|
||||
|
||||
+10
-254
@@ -20,7 +20,7 @@ from collections import deque
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from datetime import datetime, timezone
|
||||
|
||||
AGENT_VERSION = "1.7.0"
|
||||
AGENT_VERSION = "1.6.0"
|
||||
|
||||
# Default path to the local Docker Engine socket. Overridable via the
|
||||
# `docker_socket` config key; collection is silently skipped if it's absent or
|
||||
@@ -34,10 +34,7 @@ class ConfigError(Exception):
|
||||
|
||||
REQUIRED_KEYS = ("url", "token")
|
||||
INT_KEYS = ("interval_seconds",)
|
||||
LIST_KEYS = ("mounts", "docker_log_exclude")
|
||||
# Truthy strings for bool keys; anything else (incl. empty) is False.
|
||||
BOOL_KEYS = ("docker_logs_enabled",)
|
||||
_BOOL_TRUE = ("1", "true", "yes", "on")
|
||||
LIST_KEYS = ("mounts",)
|
||||
|
||||
|
||||
def read_config(path: str) -> dict:
|
||||
@@ -61,8 +58,6 @@ def read_config(path: str) -> dict:
|
||||
raise ConfigError(f"{path}:{lineno}: {key} must be int")
|
||||
elif key in LIST_KEYS:
|
||||
cfg[key] = [v.strip() for v in value.split(",") if v.strip()]
|
||||
elif key in BOOL_KEYS:
|
||||
cfg[key] = value.lower() in _BOOL_TRUE
|
||||
else:
|
||||
cfg[key] = value
|
||||
except FileNotFoundError:
|
||||
@@ -74,10 +69,6 @@ def read_config(path: str) -> dict:
|
||||
|
||||
cfg.setdefault("interval_seconds", 30)
|
||||
cfg.setdefault("docker_socket", DEFAULT_DOCKER_SOCKET)
|
||||
# Container-log collection is on by default (operator preference); a host can
|
||||
# opt out with `docker_logs_enabled = false` or thin it with docker_log_exclude.
|
||||
cfg.setdefault("docker_logs_enabled", True)
|
||||
cfg.setdefault("docker_log_exclude", [])
|
||||
return cfg
|
||||
|
||||
|
||||
@@ -412,15 +403,6 @@ def _rates(cur: dict, prev: dict, dt: float) -> dict:
|
||||
|
||||
DOCKER_API_TIMEOUT = 5.0
|
||||
|
||||
# Container-log collection (m79). Logs ride in the same push as metrics, so one
|
||||
# interval's batch is capped to keep a chatty container from bloating a POST (and
|
||||
# the backoff ring): once the cap is hit a marker line is emitted and collection
|
||||
# stops — the deferred lines come on the next interval. On a container's first
|
||||
# interval we seed from a short tail, then switch to an incremental since-cursor
|
||||
# kept per container in the agent's rate-state.
|
||||
DOCKER_LOG_BATCH_MAX_BYTES = 262144 # 256 KiB of log text per push, all containers
|
||||
DOCKER_LOG_FIRST_TAIL = 50 # lines seeded on a container's first interval
|
||||
|
||||
|
||||
def _dechunk(body: bytes) -> bytes:
|
||||
"""Decode an HTTP/1.1 chunked-transfer body into the raw payload."""
|
||||
@@ -668,198 +650,6 @@ def collect_docker(socket_path: str) -> list:
|
||||
return list(ex.map(lambda c: _collect_one_container(socket_path, c), containers))
|
||||
|
||||
|
||||
# ─── container logs (m79) ─────────────────────────────────────────────────────
|
||||
#
|
||||
# The logs endpoint returns a raw multiplexed byte stream, not JSON, so it needs
|
||||
# a sibling of _docker_request that hands back bytes. Non-TTY containers frame
|
||||
# stdout/stderr with an 8-byte header (stream byte + 4-byte big-endian length);
|
||||
# TTY containers emit unframed bytes. We request timestamps=1, so every line is
|
||||
# prefixed with an RFC3339Nano timestamp we parse for the since-cursor.
|
||||
|
||||
_LOG_STREAM_NAMES = {0: "stdout", 1: "stdout", 2: "stderr"}
|
||||
|
||||
|
||||
def _docker_request_raw(socket_path: str, path: str,
|
||||
timeout: float = DOCKER_API_TIMEOUT) -> bytes:
|
||||
"""GET `path` from the Docker API over the Unix socket; return the raw body.
|
||||
|
||||
Sibling of `_docker_request` for non-JSON endpoints (container logs). Reuses
|
||||
the connect / send / de-chunk scaffolding; raises OSError on any transport
|
||||
problem or non-2xx status so callers can silent-skip. A 200 with an empty
|
||||
body (no new lines since the cursor) returns b"".
|
||||
"""
|
||||
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
||||
sock.settimeout(timeout)
|
||||
try:
|
||||
sock.connect(socket_path)
|
||||
req = (
|
||||
"GET " + path + " HTTP/1.1\r\n"
|
||||
"Host: docker\r\n"
|
||||
"Accept: application/octet-stream\r\n"
|
||||
"Connection: close\r\n"
|
||||
"\r\n"
|
||||
)
|
||||
sock.sendall(req.encode("ascii"))
|
||||
chunks = []
|
||||
while True:
|
||||
buf = sock.recv(65536)
|
||||
if not buf:
|
||||
break
|
||||
chunks.append(buf)
|
||||
finally:
|
||||
sock.close()
|
||||
|
||||
raw = b"".join(chunks)
|
||||
head, _, body = raw.partition(b"\r\n\r\n")
|
||||
header_text = head.decode("latin-1")
|
||||
status_line = header_text.split("\r\n", 1)[0]
|
||||
parts = status_line.split(None, 2)
|
||||
status = int(parts[1]) if len(parts) >= 2 and parts[1].isdigit() else 0
|
||||
if not (200 <= status < 300):
|
||||
raise OSError(f"docker API {path} returned {status}")
|
||||
if "transfer-encoding: chunked" in header_text.lower():
|
||||
body = _dechunk(body)
|
||||
return body
|
||||
|
||||
|
||||
def _demux_docker_logs(raw: bytes) -> list:
|
||||
"""Split a Docker logs stream into [(stream_name, payload_bytes), …].
|
||||
|
||||
Non-TTY containers multiplex with an 8-byte frame header (stream byte in
|
||||
{0,1,2} + 4-byte big-endian length). The stream is treated as framed only if
|
||||
the whole buffer parses as clean back-to-back frames; anything else (a TTY
|
||||
container's raw stream, or a malformed header) falls back to a single stdout
|
||||
blob so no bytes are lost.
|
||||
"""
|
||||
if not raw:
|
||||
return []
|
||||
frames = []
|
||||
i, n = 0, len(raw)
|
||||
while i + 8 <= n:
|
||||
stream = raw[i]
|
||||
if stream > 2 or raw[i + 1] or raw[i + 2] or raw[i + 3]:
|
||||
break # not a valid frame header → not framed
|
||||
size = int.from_bytes(raw[i + 4:i + 8], "big")
|
||||
if i + 8 + size > n:
|
||||
break # frame overruns the buffer → not framed
|
||||
frames.append((_LOG_STREAM_NAMES.get(stream, "stdout"),
|
||||
raw[i + 8:i + 8 + size]))
|
||||
i += 8 + size
|
||||
if frames and i == n:
|
||||
return frames
|
||||
return [("stdout", raw)]
|
||||
|
||||
|
||||
def _parse_log_ts(token: str):
|
||||
"""Parse a Docker RFC3339Nano timestamp token into a datetime, or None.
|
||||
|
||||
Normalises `Z` → `+00:00` and trims Docker's nanosecond precision to the
|
||||
microseconds datetime.fromisoformat accepts (stdlib, Python 3.8+ safe).
|
||||
"""
|
||||
t = token.strip()
|
||||
if not t:
|
||||
return None
|
||||
if t.endswith("Z"):
|
||||
t = t[:-1] + "+00:00"
|
||||
m = re.match(r"^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})(\.\d+)?(.*)$", t)
|
||||
if m:
|
||||
frac = m.group(2) or ""
|
||||
if len(frac) > 7: # "." + up to 6 fractional digits
|
||||
frac = frac[:7]
|
||||
t = m.group(1) + frac + m.group(3)
|
||||
try:
|
||||
return datetime.fromisoformat(t)
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def _parse_container_logs(raw: bytes) -> list:
|
||||
"""Demux + line-split a logs stream → [(dt|None, stream, line), …] in order.
|
||||
|
||||
Each line is timestamp-prefixed (we request timestamps=1); an unparseable
|
||||
prefix yields dt=None with the full raw line kept.
|
||||
"""
|
||||
out = []
|
||||
for stream, payload in _demux_docker_logs(raw):
|
||||
text = payload.decode("utf-8", "replace")
|
||||
for raw_line in text.split("\n"):
|
||||
if not raw_line:
|
||||
continue
|
||||
token, _, msg = raw_line.partition(" ")
|
||||
dt = _parse_log_ts(token)
|
||||
if dt is not None:
|
||||
out.append((dt, stream, msg))
|
||||
else:
|
||||
out.append((None, stream, raw_line))
|
||||
return out
|
||||
|
||||
|
||||
def collect_docker_logs(socket_path: str, containers: list, state: dict,
|
||||
exclude=None,
|
||||
max_bytes: int = DOCKER_LOG_BATCH_MAX_BYTES) -> list:
|
||||
"""New container-log lines since the last interval, for running containers.
|
||||
|
||||
Keeps a per-container since-cursor in `state["docker_log_cursors"]`
|
||||
(name → unix ts) so each interval fetches only what's new; a container's
|
||||
first interval seeds from a short tail. The whole batch is capped at
|
||||
`max_bytes` of log text — once exceeded a truncation marker is appended and
|
||||
collection stops (logs must never balloon a push). `exclude` container names
|
||||
are skipped entirely (a per-host bandwidth opt-out). Best-effort: a
|
||||
per-container transport error just contributes nothing this interval.
|
||||
Returns [{"container", "stream", "ts", "line"}, …].
|
||||
"""
|
||||
exclude_set = set(exclude or ())
|
||||
cursors = state.setdefault("docker_log_cursors", {})
|
||||
running = {c.get("name") for c in (containers or [])
|
||||
if c.get("name") and c.get("status") == "running"}
|
||||
# Forget cursors for containers no longer running so state can't grow without
|
||||
# bound over the agent's lifetime.
|
||||
for gone in [k for k in cursors if k not in running]:
|
||||
del cursors[gone]
|
||||
|
||||
out: list = []
|
||||
total = 0
|
||||
truncated = False
|
||||
for c in (containers or []):
|
||||
name = c.get("name")
|
||||
if not name or name not in running or name in exclude_set:
|
||||
continue
|
||||
prev = cursors.get(name)
|
||||
base = f"/containers/{name}/logs?stdout=1&stderr=1×tamps=1"
|
||||
# since accepts a fractional Unix ts; we still dedup the boundary line.
|
||||
path = (f"{base}&tail={DOCKER_LOG_FIRST_TAIL}" if prev is None
|
||||
else f"{base}&since={prev:.9f}")
|
||||
try:
|
||||
raw = _docker_request_raw(socket_path, path)
|
||||
except (OSError, ValueError):
|
||||
continue
|
||||
newest = prev
|
||||
for dt, stream, line in _parse_container_logs(raw):
|
||||
ts_unix = dt.timestamp() if dt is not None else None
|
||||
if prev is not None and ts_unix is not None and ts_unix <= prev:
|
||||
continue # boundary line already sent last interval
|
||||
if total >= max_bytes:
|
||||
truncated = True
|
||||
break
|
||||
out.append({"container": name, "stream": stream,
|
||||
"ts": dt.isoformat() if dt is not None else None,
|
||||
"line": line})
|
||||
total += len(line)
|
||||
if ts_unix is not None and (newest is None or ts_unix > newest):
|
||||
newest = ts_unix
|
||||
if newest is not None:
|
||||
cursors[name] = newest
|
||||
if truncated:
|
||||
break
|
||||
|
||||
if truncated:
|
||||
out.append({"container": "_steward", "stream": "stderr",
|
||||
"ts": datetime.now(timezone.utc).isoformat(),
|
||||
"line": (f"[steward] log batch truncated at {max_bytes} bytes; "
|
||||
"remaining lines deferred to the next interval")})
|
||||
return out
|
||||
|
||||
|
||||
# ─── swarm (manager-only) ────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@@ -1062,17 +852,13 @@ class RingBuffer:
|
||||
|
||||
|
||||
def build_sample(mounts: list[str], state: dict,
|
||||
docker_socket: str = DEFAULT_DOCKER_SOCKET, *,
|
||||
docker_logs_enabled: bool = True,
|
||||
docker_log_exclude=None) -> dict:
|
||||
docker_socket: str = DEFAULT_DOCKER_SOCKET) -> dict:
|
||||
"""Collect one full sample. Partial samples allowed if a collector fails.
|
||||
|
||||
`state` carries the previous network/disk counters + monotonic timestamp so
|
||||
throughput rates can be derived from deltas (and the per-container log
|
||||
since-cursors); it is mutated in place. `docker_socket` is probed
|
||||
best-effort — the `docker` key is omitted entirely when no containers are
|
||||
found, so non-Docker hosts add nothing to the payload. `docker_logs_enabled`
|
||||
/ `docker_log_exclude` gate incremental container-log collection.
|
||||
throughput rates can be derived from deltas; it is mutated in place.
|
||||
`docker_socket` is probed best-effort — the `docker` key is omitted entirely
|
||||
when no containers are found, so non-Docker hosts add nothing to the payload.
|
||||
"""
|
||||
sample: dict = {"ts": datetime.now(timezone.utc).isoformat()}
|
||||
try:
|
||||
@@ -1154,18 +940,6 @@ def build_sample(mounts: list[str], state: dict,
|
||||
if disk is not None:
|
||||
sample["docker_disk"] = disk
|
||||
|
||||
# Container logs (m79): incremental per-container tail folded into the same
|
||||
# push as metrics. Skipped when there are no containers or logging is off;
|
||||
# the key is omitted when nothing new arrived this interval.
|
||||
if docker and docker_logs_enabled:
|
||||
try:
|
||||
logs = collect_docker_logs(docker_socket, docker, state,
|
||||
exclude=docker_log_exclude)
|
||||
except Exception:
|
||||
logs = []
|
||||
if logs:
|
||||
sample["docker_logs"] = logs
|
||||
|
||||
return sample
|
||||
|
||||
|
||||
@@ -1184,18 +958,6 @@ def build_payload(samples: list[dict], hostname: str, metadata: dict) -> dict:
|
||||
BACKOFF_CAP = 300
|
||||
|
||||
|
||||
def _drop_logs(sample: dict) -> dict:
|
||||
"""Strip container logs from a sample before it's buffered for retry.
|
||||
|
||||
Logs are the one payload we never carry across a backoff — they'd bloat the
|
||||
ring buffer and go stale — while metrics are kept so an outage doesn't lose
|
||||
them. The since-cursor already advanced when the logs were collected, so the
|
||||
dropped lines are simply not re-sent (accepted loss during an outage).
|
||||
"""
|
||||
sample.pop("docker_logs", None)
|
||||
return sample
|
||||
|
||||
|
||||
def next_backoff(current: int) -> int:
|
||||
if current <= 0:
|
||||
return 30
|
||||
@@ -1260,8 +1022,6 @@ def main_loop(conf_path: str) -> int:
|
||||
hostname = cfg.get("hostname") or socket.gethostname()
|
||||
mounts = cfg.get("mounts") or default_mounts()
|
||||
docker_socket = cfg.get("docker_socket") or DEFAULT_DOCKER_SOCKET
|
||||
docker_logs_enabled = cfg.get("docker_logs_enabled", True)
|
||||
docker_log_exclude = cfg.get("docker_log_exclude") or []
|
||||
buffer = RingBuffer(maxlen=20)
|
||||
backoff = 0
|
||||
# Carries previous net/disk counters + monotonic ts for rate computation.
|
||||
@@ -1276,17 +1036,13 @@ def main_loop(conf_path: str) -> int:
|
||||
cfg = read_config(conf_path)
|
||||
mounts = cfg.get("mounts") or default_mounts()
|
||||
docker_socket = cfg.get("docker_socket") or DEFAULT_DOCKER_SOCKET
|
||||
docker_logs_enabled = cfg.get("docker_logs_enabled", True)
|
||||
docker_log_exclude = cfg.get("docker_log_exclude") or []
|
||||
metadata = collect_metadata() # refresh host_ip/distro on reload
|
||||
_log("INFO", "config reloaded")
|
||||
except ConfigError as e:
|
||||
_log("ERROR", f"reload failed: {e}")
|
||||
_reload_requested = False
|
||||
|
||||
sample = build_sample(mounts, rate_state, docker_socket,
|
||||
docker_logs_enabled=docker_logs_enabled,
|
||||
docker_log_exclude=docker_log_exclude)
|
||||
sample = build_sample(mounts, rate_state, docker_socket)
|
||||
buffered = buffer.drain()
|
||||
payload = build_payload(
|
||||
samples=buffered + [sample],
|
||||
@@ -1305,12 +1061,12 @@ def main_loop(conf_path: str) -> int:
|
||||
_log("ERROR", "server rejected payload (400) — dropping sample")
|
||||
elif status == 401:
|
||||
_log("ERROR", "token rejected (401) — check config + UI")
|
||||
buffer.push(_drop_logs(sample))
|
||||
buffer.push(sample)
|
||||
else:
|
||||
_log("WARN", f"POST failed (status={status}); buffering")
|
||||
for s in buffered:
|
||||
buffer.push(_drop_logs(s))
|
||||
buffer.push(_drop_logs(sample))
|
||||
buffer.push(s)
|
||||
buffer.push(sample)
|
||||
backoff = next_backoff(backoff)
|
||||
sleep_for = backoff
|
||||
|
||||
|
||||
@@ -202,7 +202,6 @@ async def ingest():
|
||||
accepted = 0
|
||||
latest_ts: datetime | None = None
|
||||
docker_snapshots: list[tuple[datetime, list]] = []
|
||||
docker_log_batches: list[tuple[datetime, list]] = []
|
||||
latest_swarm: dict | None = None
|
||||
latest_swarm_ts: datetime | None = None
|
||||
latest_disk: dict | None = None
|
||||
@@ -218,11 +217,6 @@ async def ingest():
|
||||
docker = sample.get("docker")
|
||||
if isinstance(docker, list) and docker:
|
||||
docker_snapshots.append((recorded_at, docker))
|
||||
# Container logs are time-series (append every line, not newest-only);
|
||||
# each record carries its own Docker ts, recorded_at is the fallback.
|
||||
docker_logs = sample.get("docker_logs")
|
||||
if isinstance(docker_logs, list) and docker_logs:
|
||||
docker_log_batches.append((recorded_at, docker_logs))
|
||||
# Swarm is current-state, not time-series — keep only the newest
|
||||
# sample's topology (a manager re-reports it every interval).
|
||||
swarm = sample.get("swarm")
|
||||
@@ -246,8 +240,7 @@ async def ingest():
|
||||
# (opportunistic synergy via the capability registry — no hard import,
|
||||
# no-op when docker is disabled). A failure here must never sink the
|
||||
# whole ingest, so the metrics above still land.
|
||||
if (docker_snapshots or latest_swarm is not None
|
||||
or latest_disk is not None or docker_log_batches):
|
||||
if docker_snapshots or latest_swarm is not None or latest_disk is not None:
|
||||
from steward.core.capabilities import has_capability, invoke_capability
|
||||
if has_capability("docker.persist_host_samples"):
|
||||
try:
|
||||
@@ -257,7 +250,7 @@ async def ingest():
|
||||
await invoke_capability(
|
||||
"docker.persist_host_samples", UserRole.admin,
|
||||
session, host, docker_snapshots, latest_swarm,
|
||||
latest_disk, docker_log_batches,
|
||||
latest_disk,
|
||||
)
|
||||
except Exception:
|
||||
current_app.logger.exception(
|
||||
|
||||
@@ -45,10 +45,6 @@ cat > "$CONF_FILE" <<EOF
|
||||
url = $STEWARD_URL
|
||||
token = $AGENT_TOKEN
|
||||
interval_seconds = 30
|
||||
# Container logs are collected by default. To opt this host out entirely:
|
||||
# docker_logs_enabled = false
|
||||
# To skip specific noisy containers (comma-separated names):
|
||||
# docker_log_exclude = watchtower, some-chatty-service
|
||||
EOF
|
||||
chown "root:$AGENT_USER" "$CONF_FILE"
|
||||
chmod 0640 "$CONF_FILE"
|
||||
|
||||
@@ -1,64 +1,29 @@
|
||||
---
|
||||
# description: Reclaim disk on Docker / Swarm nodes — prune stopped containers, unused images, or a full system prune.
|
||||
# description: Reclaim disk on Docker / Swarm nodes by pruning unused images, containers, networks and build cache.
|
||||
# steward:category: maintenance
|
||||
# steward:confirm: true
|
||||
# Reclaim disk on Docker / Docker Swarm nodes by removing unused data.
|
||||
# `prune_target` selects the scope (default `system` preserves the original
|
||||
# behavior for existing manual/scheduled callers that don't set it):
|
||||
# prune_target=containers remove stopped containers only (docker container prune)
|
||||
# prune_target=images remove unused images (docker image prune;
|
||||
# + prune_all_images=true → -a, i.e. ALL unused, not just dangling)
|
||||
# prune_target=system docker system prune (dangling images, stopped
|
||||
# containers, unused networks, build cache). Widen with:
|
||||
# prune_all_images=true also ALL unused images
|
||||
# prune_volumes=true also unused named volumes (data loss risk)
|
||||
- name: Docker prune
|
||||
# Safe by default: prunes dangling images, stopped containers, unused networks
|
||||
# and build cache. Set extra-vars to widen scope:
|
||||
# prune_all_images=true also remove ALL unused images (not just dangling)
|
||||
# prune_volumes=true also remove unused named volumes (data loss risk)
|
||||
- name: Docker system prune
|
||||
hosts: all
|
||||
gather_facts: false
|
||||
become: true
|
||||
vars:
|
||||
prune_target: system
|
||||
prune_all_images: false
|
||||
prune_volumes: false
|
||||
tasks:
|
||||
- name: Validate prune_target
|
||||
ansible.builtin.assert:
|
||||
that: prune_target in ['containers', 'images', 'system']
|
||||
fail_msg: "prune_target must be one of: containers, images, system (got '{{ prune_target }}')"
|
||||
quiet: true
|
||||
|
||||
# ── Stopped containers only ───────────────────────────────────────────────
|
||||
- name: Prune stopped containers
|
||||
ansible.builtin.command:
|
||||
argv: ['docker', 'container', 'prune', '-f']
|
||||
register: container_prune
|
||||
changed_when: "'Total reclaimed space: 0B' not in container_prune.stdout"
|
||||
when: prune_target == 'containers'
|
||||
|
||||
# ── Unused images (dangling, or all unused with -a) ───────────────────────
|
||||
- name: Prune unused images
|
||||
ansible.builtin.command:
|
||||
argv: >-
|
||||
{{ ['docker', 'image', 'prune', '-f']
|
||||
+ (['-a'] if prune_all_images | bool else []) }}
|
||||
register: image_prune
|
||||
changed_when: "'Total reclaimed space: 0B' not in image_prune.stdout"
|
||||
when: prune_target == 'images'
|
||||
|
||||
# ── Full system prune (default) ───────────────────────────────────────────
|
||||
- name: Run docker system prune
|
||||
ansible.builtin.command:
|
||||
argv: >-
|
||||
{{ ['docker', 'system', 'prune', '-f']
|
||||
+ (['-a'] if prune_all_images | bool else [])
|
||||
+ (['--volumes'] if prune_volumes | bool else []) }}
|
||||
register: system_prune
|
||||
changed_when: "'Total reclaimed space: 0B' not in system_prune.stdout"
|
||||
when: prune_target == 'system'
|
||||
register: prune_result
|
||||
changed_when: "'Total reclaimed space: 0B' not in prune_result.stdout"
|
||||
|
||||
- name: Report reclaimed space
|
||||
ansible.builtin.debug:
|
||||
msg: >-
|
||||
{{ ((container_prune.stdout_lines | default([]))
|
||||
+ (image_prune.stdout_lines | default([]))
|
||||
+ (system_prune.stdout_lines | default([]))) | select | list }}
|
||||
msg: "{{ prune_result.stdout_lines | select | list }}"
|
||||
|
||||
@@ -76,14 +76,10 @@ async def _run_docker_retention(session, now: datetime) -> None:
|
||||
raw_days = int(await get_setting(session, "docker.retention.metrics_raw_days") or 7)
|
||||
rollup_days = int(await get_setting(session, "docker.retention.metrics_rollup_days") or 90)
|
||||
events_days = int(await get_setting(session, "docker.retention.events_days") or 30)
|
||||
logs_days = int(await get_setting(session, "docker.logs.retention_days") or 3)
|
||||
logs_cap = int(
|
||||
await get_setting(session, "docker.logs.max_bytes_per_container") or 5_000_000)
|
||||
counts = await invoke_capability(
|
||||
"docker.run_retention", UserRole.viewer, session,
|
||||
events_days=events_days, metrics_raw_days=raw_days,
|
||||
metrics_rollup_days=rollup_days, logs_retention_days=logs_days,
|
||||
logs_max_bytes_per_container=logs_cap, now=now,
|
||||
metrics_rollup_days=rollup_days, now=now,
|
||||
)
|
||||
if counts and any(counts.values()):
|
||||
logger.info("Docker retention: %s", counts)
|
||||
|
||||
@@ -84,14 +84,6 @@ DEFAULTS: dict[str, Any] = {
|
||||
"docker.retention.metrics_raw_days": 7,
|
||||
"docker.retention.metrics_rollup_days": 90,
|
||||
"docker.retention.events_days": 30,
|
||||
# Container logs (m79): on by default for every container (operator
|
||||
# preference). `exclude` names containers the server drops on ingest; the
|
||||
# per-container ring bounds storage (rotate oldest past whichever of ~age or
|
||||
# ~bytes hits first — a chatty container just keeps a shorter window).
|
||||
"docker.logs.enabled": True,
|
||||
"docker.logs.exclude": [],
|
||||
"docker.logs.retention_days": 3,
|
||||
"docker.logs.max_bytes_per_container": 5_000_000,
|
||||
# Host/plugin metrics retention (plugin_metrics): keep a short raw window at
|
||||
# the agent's ~30s cadence, then roll up to hourly averages kept much longer.
|
||||
"metrics.retention.raw_days": 7,
|
||||
|
||||
@@ -133,7 +133,6 @@ _RETENTION_FIELDS = [
|
||||
("docker_metrics_raw_days", "docker.retention.metrics_raw_days"),
|
||||
("docker_metrics_rollup_days", "docker.retention.metrics_rollup_days"),
|
||||
("docker_events_days", "docker.retention.events_days"),
|
||||
("docker_logs_retention_days", "docker.logs.retention_days"),
|
||||
("metrics_raw_days", "metrics.retention.raw_days"),
|
||||
("metrics_rollup_days", "metrics.retention.rollup_days"),
|
||||
]
|
||||
@@ -171,20 +170,6 @@ async def save_thresholds():
|
||||
except (TypeError, ValueError):
|
||||
continue
|
||||
await set_setting(db, key, val)
|
||||
# Container-log controls (m79). Checkbox: present ⇒ on (this is a
|
||||
# full-page form, so absence is a genuine "off"). Exclude: comma-split
|
||||
# names. Size: entered in MB, stored as bytes.
|
||||
await set_setting(db, "docker.logs.enabled", "docker_logs_enabled" in form)
|
||||
exclude = [n.strip() for n in form.get("docker_logs_exclude", "").split(",")
|
||||
if n.strip()]
|
||||
await set_setting(db, "docker.logs.exclude", exclude)
|
||||
max_mb = form.get("docker_logs_max_mb", "")
|
||||
if max_mb != "":
|
||||
try:
|
||||
await set_setting(db, "docker.logs.max_bytes_per_container",
|
||||
max(1, int(max_mb)) * 1_000_000)
|
||||
except (TypeError, ValueError):
|
||||
pass
|
||||
await _reload_app_config()
|
||||
await log_audit(current_app, session.get("user_id"), session.get("username", ""),
|
||||
"settings.saved", detail={"section": "thresholds"})
|
||||
|
||||
@@ -70,54 +70,6 @@
|
||||
"Keep container start/stop/die/health history this long.") }}
|
||||
</div>
|
||||
|
||||
<div class="card" style="max-width:640px;margin-top:1rem;">
|
||||
<h2 class="section-title" style="margin-bottom:0.5rem;">Container logs</h2>
|
||||
<p style="font-size:0.82rem;color:var(--text-muted);margin-bottom:1.25rem;">
|
||||
The host agent tails each container's logs and pushes them here, viewable per
|
||||
container. On by default for every container; storage is bounded by a
|
||||
per-container ring (oldest lines rotate out once the age or size cap is hit,
|
||||
whichever comes first). Applied by the hourly cleanup and on ingest.
|
||||
</p>
|
||||
|
||||
<div class="form-group" style="margin-bottom:1.1rem;">
|
||||
<label style="display:flex;align-items:center;gap:0.5rem;cursor:pointer;">
|
||||
<input type="checkbox" name="docker_logs_enabled" style="width:auto;"
|
||||
{% if settings["docker.logs.enabled"] %}checked{% endif %}>
|
||||
Collect container logs
|
||||
</label>
|
||||
<div style="font-size:0.78rem;color:var(--text-muted);margin-top:0.3rem;">
|
||||
Global kill-switch. When off, pushed log lines are dropped and no new logs are stored.
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div class="form-group" style="margin-bottom:1.1rem;">
|
||||
<label>Exclude containers</label>
|
||||
<div style="margin-top:0.25rem;">
|
||||
<input type="text" name="docker_logs_exclude"
|
||||
value="{{ settings['docker.logs.exclude'] | join(', ') }}"
|
||||
placeholder="watchtower, some-chatty-service" style="width:100%;max-width:420px;">
|
||||
</div>
|
||||
<div style="font-size:0.78rem;color:var(--text-muted);margin-top:0.3rem;">
|
||||
Comma-separated container names whose logs are dropped on ingest (e.g. known-noisy ones).
|
||||
</div>
|
||||
</div>
|
||||
|
||||
{{ days("Log retention", "docker_logs_retention_days", "docker.logs.retention_days",
|
||||
"Keep each container's log lines at most this long before rotating them out.") }}
|
||||
|
||||
<div class="form-group" style="margin-bottom:0.25rem;">
|
||||
<label>Max size per container <span style="color:var(--text-muted);font-size:0.8rem;">(MB)</span></label>
|
||||
<div style="margin-top:0.25rem;">
|
||||
<input type="number" name="docker_logs_max_mb" min="1" step="1"
|
||||
value="{{ (settings['docker.logs.max_bytes_per_container'] // 1000000) or 1 }}"
|
||||
style="max-width:110px;">
|
||||
</div>
|
||||
<div style="font-size:0.78rem;color:var(--text-muted);margin-top:0.3rem;">
|
||||
Newest lines are kept up to this size per container; older lines rotate out first.
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div class="card" style="max-width:640px;margin-top:1rem;">
|
||||
<h2 class="section-title" style="margin-bottom:0.5rem;">Host metrics retention</h2>
|
||||
<p style="font-size:0.82rem;color:var(--text-muted);margin-bottom:1.25rem;">
|
||||
|
||||
@@ -1,21 +1,11 @@
|
||||
"""The bundled first-party playbook source is always present and discoverable."""
|
||||
from pathlib import Path
|
||||
|
||||
import yaml
|
||||
|
||||
from steward.ansible.sources import (
|
||||
BUILTIN_SOURCE_NAME,
|
||||
discover_playbook_meta,
|
||||
discover_playbooks,
|
||||
get_sources,
|
||||
)
|
||||
|
||||
|
||||
def _bundled_content(rel_path: str) -> str:
|
||||
builtin = get_sources({"sources": []})[0]
|
||||
return (Path(builtin["path"]) / rel_path).read_text()
|
||||
|
||||
|
||||
def test_builtin_source_is_first_and_local():
|
||||
sources = get_sources({"sources": []})
|
||||
assert sources[0]["name"] == BUILTIN_SOURCE_NAME
|
||||
@@ -27,33 +17,3 @@ def test_bundled_playbooks_are_discoverable():
|
||||
playbooks = discover_playbooks(builtin["path"])
|
||||
assert "maintenance/docker_prune.yml" in playbooks
|
||||
assert "host_agent/install.yml" in playbooks
|
||||
|
||||
|
||||
def test_docker_prune_playbook_parses_and_keeps_meta():
|
||||
"""The prune playbook stays valid YAML and keeps its self-describing meta
|
||||
(a destructive maintenance run that must prompt for confirmation)."""
|
||||
content = _bundled_content("maintenance/docker_prune.yml")
|
||||
plays = yaml.safe_load(content)
|
||||
assert isinstance(plays, list) and plays # at least one play
|
||||
|
||||
meta = discover_playbook_meta(content)
|
||||
assert meta["confirm"] is True
|
||||
assert meta["category"] == "maintenance"
|
||||
assert meta["description"]
|
||||
|
||||
|
||||
def test_docker_prune_supports_all_three_targets():
|
||||
"""M78 drives the three prune buttons via a single `prune_target` var;
|
||||
`system` is the default so pre-M78 callers (no var set) are unchanged."""
|
||||
content = _bundled_content("maintenance/docker_prune.yml")
|
||||
play = yaml.safe_load(content)[0]
|
||||
assert play["vars"]["prune_target"] == "system"
|
||||
|
||||
# Every target the routes/UI can send has a matching guarded task.
|
||||
guards = {
|
||||
task.get("when")
|
||||
for task in play["tasks"]
|
||||
if isinstance(task.get("when"), str) and "prune_target" in task["when"]
|
||||
}
|
||||
for target in ("containers", "images", "system"):
|
||||
assert f"prune_target == '{target}'" in guards
|
||||
|
||||
@@ -75,28 +75,6 @@ def test_events_and_swarm_tables_exist(app):
|
||||
asyncio.run(_go())
|
||||
|
||||
|
||||
@_NEEDS_DB
|
||||
def test_docker_logs_table_shape(app):
|
||||
"""docker_009 created docker_logs: host-scoped, with the twin indexes the
|
||||
viewer (host, container, ts) and the age-cutoff prune (ts) rely on."""
|
||||
from sqlalchemy import text
|
||||
|
||||
async def _go():
|
||||
async with app.db_sessionmaker() as s:
|
||||
cols = {r[0] for r in (await s.execute(text(
|
||||
"SELECT column_name FROM information_schema.columns "
|
||||
"WHERE table_name = 'docker_logs'"))).all()}
|
||||
assert cols, "docker_logs missing entirely"
|
||||
assert {"id", "host_id", "container_name", "ts", "stream", "line"} <= cols
|
||||
idx = {r[0] for r in (await s.execute(text(
|
||||
"SELECT indexname FROM pg_indexes "
|
||||
"WHERE tablename = 'docker_logs'"))).all()}
|
||||
assert "ix_docker_logs_host_container_time" in idx
|
||||
assert "ix_docker_logs_ts" in idx
|
||||
|
||||
asyncio.run(_go())
|
||||
|
||||
|
||||
def _persist_fn(app):
|
||||
"""Resolve persist_host_docker via the registered capability if the docker
|
||||
plugin is loaded, else import it directly (the import is safe only when the
|
||||
@@ -172,97 +150,6 @@ def test_persist_scopes_containers_by_host(app):
|
||||
assert enrich == ("healthy", 2, "web", 1000) # enrichment round-trips
|
||||
|
||||
|
||||
@_NEEDS_DB
|
||||
def test_persist_logs_stores_lines(app):
|
||||
"""Pushed container logs land in docker_logs, host-scoped, one row per line,
|
||||
ordered by their own ts; the agent's _steward truncation marker is dropped."""
|
||||
from sqlalchemy import text
|
||||
from steward.models.hosts import Host
|
||||
|
||||
persist = _persist_fn(app)
|
||||
now = datetime.now(timezone.utc)
|
||||
log_batches = [(now, [
|
||||
{"container": "web", "stream": "stdout",
|
||||
"ts": "2026-01-01T00:00:01+00:00", "line": "started"},
|
||||
{"container": "web", "stream": "stderr",
|
||||
"ts": "2026-01-01T00:00:02+00:00", "line": "warn: x"},
|
||||
{"container": "_steward", "stream": "stderr", "ts": None, "line": "truncated"},
|
||||
])]
|
||||
|
||||
async def _go():
|
||||
async with app.db_sessionmaker() as s:
|
||||
async with s.begin():
|
||||
await s.execute(text("DELETE FROM docker_logs"))
|
||||
h = Host(id=str(uuid.uuid4()), name="loghost", address="10.0.0.9")
|
||||
s.add(h)
|
||||
await s.flush()
|
||||
hid = h.id
|
||||
# snapshots empty; logs passed as the 6th positional arg.
|
||||
await persist(s, h, [], None, None, log_batches)
|
||||
rows = (await s.execute(text(
|
||||
"SELECT container_name, stream, line FROM docker_logs "
|
||||
"WHERE host_id = :h ORDER BY ts"), {"h": hid})).all()
|
||||
return [tuple(r) for r in rows]
|
||||
|
||||
assert asyncio.run(_go()) == [
|
||||
("web", "stdout", "started"),
|
||||
("web", "stderr", "warn: x"),
|
||||
]
|
||||
|
||||
|
||||
@_NEEDS_DB
|
||||
def test_persist_logs_respects_toggle_and_exclude(app):
|
||||
"""Server-side controls (Settings): an excluded container's lines are dropped
|
||||
while others persist; the global toggle off stores nothing new."""
|
||||
from types import SimpleNamespace
|
||||
from sqlalchemy import text
|
||||
from steward.models.hosts import Host
|
||||
from steward.core.settings import set_setting
|
||||
|
||||
persist = _persist_fn(app)
|
||||
now = datetime.now(timezone.utc)
|
||||
batch = [(now, [
|
||||
{"container": "web", "stream": "stdout", "ts": None, "line": "keep-me"},
|
||||
{"container": "noisy", "stream": "stdout", "ts": None, "line": "drop-me"},
|
||||
])]
|
||||
|
||||
hid = str(uuid.uuid4())
|
||||
# Stand-in host: the logs-only persist path reads only host.id, so a plain
|
||||
# object sidesteps ORM attribute-expiry across the commit boundaries below.
|
||||
host = SimpleNamespace(id=hid, name="loghost3")
|
||||
|
||||
async def _go():
|
||||
async with app.db_sessionmaker() as s:
|
||||
# Back-to-back begin blocks with NO read in between — a SELECT between
|
||||
# them would autobegin a txn and make the next begin() collide.
|
||||
async with s.begin():
|
||||
await s.execute(text("DELETE FROM docker_logs"))
|
||||
await set_setting(s, "docker.logs.enabled", True)
|
||||
await set_setting(s, "docker.logs.exclude", ["noisy"])
|
||||
s.add(Host(id=hid, name="loghost3", address="10.7.7.10"))
|
||||
await s.flush() # host row before its log rows (FK order)
|
||||
await persist(s, host, [], None, None, batch) # web kept, noisy excluded
|
||||
async with s.begin(): # global kill-switch off → nothing stored
|
||||
await set_setting(s, "docker.logs.enabled", False)
|
||||
await persist(s, host, [], None, None, batch)
|
||||
async with s.begin(): # restore defaults for other tests
|
||||
await set_setting(s, "docker.logs.enabled", True)
|
||||
await set_setting(s, "docker.logs.exclude", [])
|
||||
# Reads only after the last begin block (they autobegin, but nothing
|
||||
# opens a begin() after them). Toggle-off added nothing, so the state
|
||||
# here still reflects the exclude phase.
|
||||
kept = {r[0] for r in (await s.execute(text(
|
||||
"SELECT DISTINCT container_name FROM docker_logs WHERE host_id=:h"),
|
||||
{"h": hid})).all()}
|
||||
after_off = (await s.execute(text(
|
||||
"SELECT COUNT(*) FROM docker_logs WHERE host_id=:h"), {"h": hid})).scalar()
|
||||
return kept, after_off
|
||||
|
||||
kept, after_off = asyncio.run(_go())
|
||||
assert kept == {"web"} # excluded 'noisy' dropped on ingest
|
||||
assert after_off == 1 # toggle off added nothing → still just the one 'web' line
|
||||
|
||||
|
||||
@_NEEDS_DB
|
||||
def test_large_memory_values_persist_as_bigint(app):
|
||||
"""A container using >2^31 bytes of RAM must persist. Regression: mem_usage_bytes
|
||||
@@ -564,67 +451,6 @@ def test_retention_rollup_and_prune(app):
|
||||
assert counts["events_pruned"] == 1 and counts["rollup_pruned"] == 1
|
||||
|
||||
|
||||
@_NEEDS_DB
|
||||
def test_retention_logs_ring(app):
|
||||
"""docker_logs ring (m79): lines past the age window are pruned; within it,
|
||||
each container keeps only the newest ~cap bytes; containers are independent."""
|
||||
from datetime import timedelta
|
||||
from sqlalchemy import text
|
||||
from steward.models.hosts import Host
|
||||
|
||||
run_retention = _retention_fn(app)
|
||||
now = datetime(2026, 6, 19, 12, 0, 0, tzinfo=timezone.utc)
|
||||
line40 = "x" * 40 # 40 chars/line; cap=100 keeps 3 lines (excl-prefix 0/40/80)
|
||||
|
||||
async def _go():
|
||||
async with app.db_sessionmaker() as s:
|
||||
async with s.begin():
|
||||
await s.execute(text("DELETE FROM docker_logs"))
|
||||
h = Host(id=str(uuid.uuid4()), name="loghost2", address="10.7.7.8")
|
||||
s.add(h)
|
||||
await s.flush()
|
||||
hid = h.id
|
||||
|
||||
def ins(cn, ts, line):
|
||||
return s.execute(text(
|
||||
"INSERT INTO docker_logs "
|
||||
"(id, host_id, container_name, ts, stream, line) "
|
||||
"VALUES (:id,:h,:cn,:ts,'stdout',:line)"),
|
||||
{"id": str(uuid.uuid4()), "h": hid, "cn": cn,
|
||||
"ts": ts, "line": line})
|
||||
|
||||
# Past the 3-day age window → age-pruned.
|
||||
await ins("old", now - timedelta(days=10), "z")
|
||||
# 5 recent 'web' lines (40 bytes each = 200 > cap 100) → keep 3.
|
||||
for i in range(1, 6):
|
||||
await ins("web", now - timedelta(minutes=6 - i), line40)
|
||||
# 2 recent 'db' lines (80 bytes ≤ cap) → both survive (isolation).
|
||||
await ins("db", now - timedelta(minutes=2), line40)
|
||||
await ins("db", now - timedelta(minutes=1), line40)
|
||||
|
||||
async with s.begin():
|
||||
counts = await run_retention(
|
||||
s, events_days=30, metrics_raw_days=7, metrics_rollup_days=90,
|
||||
logs_retention_days=3, logs_max_bytes_per_container=100, now=now,
|
||||
)
|
||||
|
||||
def n(cn):
|
||||
return s.execute(text(
|
||||
"SELECT COUNT(*) FROM docker_logs WHERE host_id=:h AND container_name=:cn"),
|
||||
{"h": hid, "cn": cn})
|
||||
web = (await n("web")).scalar()
|
||||
db = (await n("db")).scalar()
|
||||
old = (await n("old")).scalar()
|
||||
return counts, web, db, old
|
||||
|
||||
counts, web, db, old = asyncio.run(_go())
|
||||
assert old == 0 # age window
|
||||
assert web == 3 # newest ~100 bytes kept, oldest 2 dropped
|
||||
assert db == 2 # under cap → untouched (per-container)
|
||||
assert counts["logs_age_pruned"] == 1
|
||||
assert counts["logs_size_pruned"] == 2
|
||||
|
||||
|
||||
def test_widget_dedup_collapses_cross_manager_duplicates():
|
||||
"""The same swarm task is reported by every manager (identical container_id);
|
||||
the dashboard widget must count it once. Older agents send no container_id,
|
||||
|
||||
@@ -65,35 +65,3 @@ def test_no_event_when_unchanged():
|
||||
old = {"web": {"status": "running", "health": "healthy", "oom_killed": False, "exit_code": None}}
|
||||
events = _derive_events(old, [_c("web", "running", health="healthy")])
|
||||
assert events == []
|
||||
|
||||
|
||||
# ── container-log row shaping (m79, pure) ─────────────────────────────────────
|
||||
|
||||
def test_log_rows_shapes_and_filters():
|
||||
from datetime import datetime, timezone
|
||||
from plugins.docker.ingest import _log_rows
|
||||
|
||||
rec_at = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
batches = [(rec_at, [
|
||||
{"container": "web", "stream": "stdout",
|
||||
"ts": "2026-01-01T00:00:01+00:00", "line": "hello"},
|
||||
{"container": "web", "stream": "stderr", "ts": None, "line": "no-ts"},
|
||||
{"container": "_steward", "stream": "stderr",
|
||||
"ts": None, "line": "truncated"}, # advisory marker → dropped
|
||||
{"container": "web", "stream": "weird", "ts": None, "line": "bad-stream"},
|
||||
{"container": "", "line": "x"}, # no name → dropped
|
||||
{"container": "web", "ts": None, "line": None}, # no line → dropped
|
||||
"not-a-dict", # malformed → dropped
|
||||
])]
|
||||
rows = list(_log_rows(batches, "host-1"))
|
||||
assert [r["line"] for r in rows] == ["hello", "no-ts", "bad-stream"]
|
||||
assert rows[0]["host_id"] == "host-1" and rows[0]["container_name"] == "web"
|
||||
assert rows[1]["ts"] == rec_at # None ts → recorded_at fallback
|
||||
assert rows[2]["stream"] == "stdout" # unknown stream normalised
|
||||
|
||||
|
||||
def test_log_rows_skips_non_list_records():
|
||||
from datetime import datetime, timezone
|
||||
from plugins.docker.ingest import _log_rows
|
||||
rec_at = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
assert list(_log_rows([(rec_at, None)], "h")) == []
|
||||
|
||||
@@ -30,28 +30,6 @@ def test_routes_module_exposes_new_views():
|
||||
assert r.docker_bp.name == "docker"
|
||||
|
||||
|
||||
def test_disk_prune_view_defined():
|
||||
# M78 admin-gated prune action.
|
||||
assert callable(r.disk_prune)
|
||||
|
||||
|
||||
def test_container_log_views_defined():
|
||||
# M79 per-container log viewer + its HTMX-polled line fragment.
|
||||
assert callable(r.container_logs)
|
||||
assert callable(r.container_logs_lines)
|
||||
|
||||
|
||||
def test_prune_extra_vars_mapping():
|
||||
"""The prune buttons drive one playbook via prune_target; only 'images'
|
||||
widens to ALL unused images (docker image prune -a), system stays -f."""
|
||||
assert r._prune_extra_vars("containers") == {"prune_target": "containers"}
|
||||
assert r._prune_extra_vars("images") == {
|
||||
"prune_target": "images", "prune_all_images": True,
|
||||
}
|
||||
# System prune stays conservative — no prune_all_images key.
|
||||
assert r._prune_extra_vars("system") == {"prune_target": "system"}
|
||||
|
||||
|
||||
def test_human_bytes_formats_binary_units():
|
||||
assert r._human_bytes(None) == "—"
|
||||
assert r._human_bytes(0) == "0 B"
|
||||
|
||||
@@ -401,190 +401,3 @@ def test_read_config_honours_explicit_docker_socket(tmp_path):
|
||||
"docker_socket = /run/user/1000/docker.sock\n")
|
||||
cfg = a.read_config(str(cfg_file))
|
||||
assert cfg["docker_socket"] == "/run/user/1000/docker.sock"
|
||||
|
||||
|
||||
# ── container logs (m79) ──────────────────────────────────────────────────────
|
||||
|
||||
def _frame(stream: int, payload: bytes) -> bytes:
|
||||
"""Build one Docker multiplexed-log frame (8-byte header + payload)."""
|
||||
return bytes([stream, 0, 0, 0]) + len(payload).to_bytes(4, "big") + payload
|
||||
|
||||
|
||||
def test_demux_docker_logs_framed():
|
||||
raw = _frame(1, b"out line\n") + _frame(2, b"err line\n")
|
||||
assert a._demux_docker_logs(raw) == [
|
||||
("stdout", b"out line\n"), ("stderr", b"err line\n")]
|
||||
|
||||
|
||||
def test_demux_docker_logs_tty_fallback():
|
||||
# A TTY container's stream isn't framed — the leading byte ('2') is not a
|
||||
# valid stream id, so the whole blob is treated as one stdout payload.
|
||||
raw = b"2023-11-14T12:00:00.000000000Z hi\n"
|
||||
assert a._demux_docker_logs(raw) == [("stdout", raw)]
|
||||
|
||||
|
||||
def test_demux_docker_logs_empty():
|
||||
assert a._demux_docker_logs(b"") == []
|
||||
|
||||
|
||||
def test_parse_log_ts_handles_nanoseconds_and_z():
|
||||
dt = a._parse_log_ts("2023-11-14T12:00:00.123456789Z")
|
||||
assert dt is not None
|
||||
assert dt.year == 2023 and dt.microsecond == 123456 # ns trimmed to µs
|
||||
assert dt.utcoffset().total_seconds() == 0
|
||||
assert a._parse_log_ts("not-a-time") is None
|
||||
assert a._parse_log_ts("") is None
|
||||
|
||||
|
||||
def test_parse_container_logs_splits_streams_and_ts():
|
||||
raw = (_frame(1, b"2023-11-14T12:00:00.000000001Z hello world\n")
|
||||
+ _frame(2, b"2023-11-14T12:00:01.000000000Z oops\n"))
|
||||
parsed = a._parse_container_logs(raw)
|
||||
assert len(parsed) == 2
|
||||
dt0, s0, l0 = parsed[0]
|
||||
dt1, s1, l1 = parsed[1]
|
||||
assert (s0, l0) == ("stdout", "hello world")
|
||||
assert (s1, l1) == ("stderr", "oops")
|
||||
assert dt1 > dt0
|
||||
|
||||
|
||||
def test_parse_container_logs_keeps_unparseable_line():
|
||||
# No timestamp prefix (e.g. a partial write) → dt is None, full line kept.
|
||||
parsed = a._parse_container_logs(_frame(1, b"no-timestamp-here\n"))
|
||||
assert parsed == [(None, "stdout", "no-timestamp-here")]
|
||||
|
||||
|
||||
def test_collect_docker_logs_since_cursor_and_dedup(monkeypatch):
|
||||
containers = [{"name": "web", "status": "running"}]
|
||||
calls = []
|
||||
t1 = "2023-11-14T12:00:00.000000000Z"
|
||||
t2 = "2023-11-14T12:00:01.000000000Z"
|
||||
t3 = "2023-11-14T12:00:02.000000000Z"
|
||||
|
||||
def fake_raw(socket_path, path, timeout=a.DOCKER_API_TIMEOUT):
|
||||
calls.append(path)
|
||||
if "tail=" in path: # first interval seeds from a tail
|
||||
return _frame(1, f"{t1} a\n".encode()) + _frame(1, f"{t2} b\n".encode())
|
||||
# second interval: daemon re-returns the boundary line (t2) + a new one
|
||||
return _frame(1, f"{t2} b\n".encode()) + _frame(1, f"{t3} c\n".encode())
|
||||
|
||||
monkeypatch.setattr(a, "_docker_request_raw", fake_raw)
|
||||
state: dict = {}
|
||||
first = a.collect_docker_logs("/sock", containers, state)
|
||||
assert [r["line"] for r in first] == ["a", "b"]
|
||||
assert "tail=" in calls[0]
|
||||
|
||||
second = a.collect_docker_logs("/sock", containers, state)
|
||||
# boundary line b (t2) deduped against the cursor; only c (t3) is new
|
||||
assert [r["line"] for r in second] == ["c"]
|
||||
assert "since=" in calls[1]
|
||||
|
||||
|
||||
def test_collect_docker_logs_excludes_and_skips_non_running(monkeypatch):
|
||||
containers = [
|
||||
{"name": "web", "status": "running"},
|
||||
{"name": "noisy", "status": "running"},
|
||||
{"name": "db", "status": "exited"},
|
||||
]
|
||||
seen = []
|
||||
|
||||
def fake_raw(socket_path, path, timeout=a.DOCKER_API_TIMEOUT):
|
||||
seen.append(path)
|
||||
return _frame(1, b"2023-11-14T12:00:00.000000000Z x\n")
|
||||
|
||||
monkeypatch.setattr(a, "_docker_request_raw", fake_raw)
|
||||
out = a.collect_docker_logs("/sock", containers, {}, exclude=["noisy"])
|
||||
# only web is fetched: noisy is excluded, db isn't running
|
||||
assert seen and all("/containers/web/" in p for p in seen)
|
||||
assert {r["container"] for r in out} == {"web"}
|
||||
|
||||
|
||||
def test_collect_docker_logs_byte_cap_truncates(monkeypatch):
|
||||
containers = [{"name": "web", "status": "running"}]
|
||||
blob = b"".join(
|
||||
_frame(1, f"2023-11-14T12:00:{i:02d}.000000000Z {'x' * 20}\n".encode())
|
||||
for i in range(10))
|
||||
monkeypatch.setattr(a, "_docker_request_raw",
|
||||
lambda s, p, timeout=a.DOCKER_API_TIMEOUT: blob)
|
||||
out = a.collect_docker_logs("/sock", containers, {}, max_bytes=50)
|
||||
# Once the cap is crossed a single marker line is appended and we stop.
|
||||
assert out[-1]["container"] == "_steward"
|
||||
assert "truncated" in out[-1]["line"]
|
||||
real = [r for r in out if r["container"] == "web"]
|
||||
assert 0 < len(real) < 10
|
||||
|
||||
|
||||
def test_collect_docker_logs_forgets_gone_container_cursors(monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
a, "_docker_request_raw",
|
||||
lambda s, p, timeout=a.DOCKER_API_TIMEOUT:
|
||||
_frame(1, b"2023-11-14T12:00:00.000000000Z x\n"))
|
||||
state: dict = {}
|
||||
a.collect_docker_logs("/sock", [{"name": "web", "status": "running"}], state)
|
||||
assert "web" in state["docker_log_cursors"]
|
||||
# web is gone next interval → its cursor is pruned so state can't grow forever
|
||||
a.collect_docker_logs("/sock", [{"name": "db", "status": "running"}], state)
|
||||
assert "web" not in state["docker_log_cursors"]
|
||||
assert "db" in state["docker_log_cursors"]
|
||||
|
||||
|
||||
def test_build_sample_includes_logs_when_present(monkeypatch):
|
||||
monkeypatch.setattr(a, "collect_docker",
|
||||
lambda _s: [{"name": "web", "status": "running"}])
|
||||
monkeypatch.setattr(a, "collect_swarm", lambda _s: None)
|
||||
monkeypatch.setattr(a, "collect_disk_usage", lambda _s: None)
|
||||
monkeypatch.setattr(
|
||||
a, "collect_docker_logs",
|
||||
lambda *args, **kw: [{"container": "web", "stream": "stdout",
|
||||
"ts": "t", "line": "hello"}])
|
||||
sample = a.build_sample(["/"], {}, "/var/run/docker.sock")
|
||||
assert sample["docker_logs"][0]["line"] == "hello"
|
||||
|
||||
|
||||
def test_build_sample_omits_logs_when_disabled(monkeypatch):
|
||||
monkeypatch.setattr(a, "collect_docker",
|
||||
lambda _s: [{"name": "web", "status": "running"}])
|
||||
monkeypatch.setattr(a, "collect_swarm", lambda _s: None)
|
||||
monkeypatch.setattr(a, "collect_disk_usage", lambda _s: None)
|
||||
called = {"logs": False}
|
||||
|
||||
def _logs(*args, **kw):
|
||||
called["logs"] = True
|
||||
return [{"container": "web", "line": "x"}]
|
||||
|
||||
monkeypatch.setattr(a, "collect_docker_logs", _logs)
|
||||
sample = a.build_sample(["/"], {}, "/var/run/docker.sock",
|
||||
docker_logs_enabled=False)
|
||||
assert "docker_logs" not in sample
|
||||
assert called["logs"] is False # collection skipped entirely, not just dropped
|
||||
|
||||
|
||||
def test_drop_logs_strips_only_logs():
|
||||
sample = {"ts": "t", "cpu_pct": 5.0,
|
||||
"docker": [{"name": "web"}], "docker_logs": [{"line": "x"}]}
|
||||
out = a._drop_logs(sample)
|
||||
assert "docker_logs" not in out
|
||||
assert out["docker"] == [{"name": "web"}] and out["cpu_pct"] == 5.0 # metrics kept
|
||||
assert a._drop_logs(out) is out # idempotent
|
||||
|
||||
|
||||
def test_read_config_docker_logs_default_on(tmp_path):
|
||||
p = tmp_path / "agent.conf"
|
||||
p.write_text("url = x\ntoken = y\n")
|
||||
cfg = a.read_config(str(p))
|
||||
assert cfg["docker_logs_enabled"] is True
|
||||
assert cfg["docker_log_exclude"] == []
|
||||
|
||||
|
||||
def test_read_config_disables_docker_logs(tmp_path):
|
||||
p = tmp_path / "agent.conf"
|
||||
p.write_text("url = x\ntoken = y\ndocker_logs_enabled = false\n")
|
||||
cfg = a.read_config(str(p))
|
||||
assert cfg["docker_logs_enabled"] is False
|
||||
|
||||
|
||||
def test_read_config_parses_docker_log_exclude(tmp_path):
|
||||
p = tmp_path / "agent.conf"
|
||||
p.write_text("url = x\ntoken = y\ndocker_log_exclude = watchtower, foo\n")
|
||||
cfg = a.read_config(str(p))
|
||||
assert cfg["docker_log_exclude"] == ["watchtower", "foo"]
|
||||
|
||||
Reference in New Issue
Block a user