1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95

"""Modal ingestion app for atop history stored on the local machine.

Run from the machine that has ``/var/log/atop``:

    modal run modal_atop_analyze.py --date 20260921 --initialize

The local entrypoint only reads the binary atop log. Parsing and the ClickHouse
write run in Modal; the ClickHouse endpoint and Proxy Token live in the
``atop-clickhouse-proxy`` Modal Secret, never in local shell environment.
"""

from __future__ import annotations

import os
import logging

import modal

from atop_analyze import AtopError, log_path, parse, run_atop, write_clickhouse

# Modal captures container stderr. Configure it explicitly because the runtime
# does not install an INFO-level handler for application loggers by default.
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s %(message)s",
    force=True,
)
app = modal.App("atop-analyze")
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
# ``modal run`` mounts this entrypoint automatically, but imports are not
# automatically included in the remote container. Bake the shared parser into
# the function image so the remote ingest function can import it.
image = modal.Image.debian_slim(python_version="3.12").add_local_file(
    "atop_analyze.py", "/root/atop_analyze.py", copy=True
)
clickhouse_secret = modal.Secret.from_name(
    "atop-clickhouse-proxy",
    required_keys=["CLICKHOUSE_URL", "MODAL_KEY", "MODAL_SECRET"],
)


@app.function(image=image, secrets=[clickhouse_secret], timeout=15 * 60)
def ingest(raw_atop: str, initialize: bool = False) -> dict[str, int]:
    """Parse an atop export and persist its samples through private ClickHouse."""
    logger.info("ingest_started raw_bytes=%d initialize=%s", len(raw_atop.encode()), initialize)
    samples = parse(raw_atop)
    if not samples:
        raise RuntimeError("no usable atop samples in the supplied log range")
    process_rows = sum(len(sample.procs) for sample in samples)
    logger.info(
        "atop_parsed samples=%d process_rows=%d first_epoch=%d last_epoch=%d",
        len(samples), process_rows, samples[0].epoch, samples[-1].epoch,
    )

    write_clickhouse(
        samples,
        os.environ["CLICKHOUSE_URL"],
        os.environ.get("CLICKHOUSE_DATABASE", "default"),
        os.environ.get("CLICKHOUSE_SAMPLE_TABLE", "atop_samples"),
        os.environ.get("CLICKHOUSE_PROCESS_TABLE", "atop_processes"),
        None,
        None,
        os.environ["MODAL_KEY"],
        os.environ["MODAL_SECRET"],
        initialize,
    )
    logger.info("clickhouse_write_complete samples=%d process_rows=%d", len(samples), process_rows)
    return {
        "samples": len(samples),
        "process_rows": process_rows,
    }


@app.local_entrypoint()
def export(
    date: str | None = None,
    file: str | None = None,
    logdir: str = "/var/log/atop",
    begin: str | None = None,
    end: str | None = None,
    initialize: bool = False,
) -> None:
    """Read the local host's atop log, then delegate ingestion to Modal."""
    try:
        path = file or log_path(date, logdir)
        raw_atop = run_atop(path, begin, end)
    except AtopError as error:
        raise RuntimeError(f"could not read local atop history: {error}") from error

    result = ingest.remote(raw_atop, initialize)
    print(
        f"exported {result['samples']} samples and {result['process_rows']} "
        f"process rows from {path}"
    )