nandi/atop-analyze · e5a43e4

modified atop_analyze.py +27 -4 Viewed

@@ -22,6 +22,7 @@ from __future__ import annotations
22	22
23	23	import argparse
24	24	import base64
	25	+import hashlib
25	26	import json
26	27	import logging
27	28	import os
@@ -213,6 +214,16 @@ def clickhouse_identifier(value: str, option: str) -> str:
213	214	return value
214	215
215	216
	217	+def ingest_version(row: dict[str, object]) -> int:
	218	+ """Return a stable UInt64 version for one immutable source row.
	219	+
	220	+ The value is derived from canonical JSON rather than wall-clock time, so a
	221	+ retried export of the same atop interval has exactly the same version.
	222	+ """
	223	+ encoded = json.dumps(row, sort_keys=True, separators=(",", ":")).encode()
	224	+ return int.from_bytes(hashlib.blake2b(encoded, digest_size=8).digest(), "big")
	225	+
	226	+
216	227	def clickhouse_request(
217	228	url: str, query: str, body: str, user: str \| None, password: str \| None,
218	229	modal_key: str \| None, modal_secret: str \| None,
@@ -280,14 +291,22 @@ def write_clickhouse(
280	291	ncpu UInt16, user_pct Float32, sys_pct Float32, nice_pct Float32,
281	292	idle_pct Float32, wait_pct Float32, irq_pct Float32, softirq_pct Float32,
282	293	steal_pct Float32, busy_pct Float32, load1 Float32, load5 Float32,
283		- load15 Float32, ctxsw UInt64, intr UInt64
284		- ) ENGINE = ReplacingMergeTree ORDER BY (host, epoch)""", "", user, password,
	294	+ load15 Float32, ctxsw UInt64, intr UInt64, ingest_version UInt64
	295	+ ) ENGINE = ReplacingMergeTree(ingest_version) ORDER BY (host, epoch)""", "", user, password,
285	296	modal_key, modal_secret)
286	297	clickhouse_request(url, f"""CREATE TABLE IF NOT EXISTS {processes_target} (
287	298	host LowCardinality(String), epoch DateTime, pid UInt32,
288		- name LowCardinality(String), cpu_pct Float32, interval_seconds UInt32
289		- ) ENGINE = ReplacingMergeTree ORDER BY (host, epoch, pid, name)""", "", user, password,
	299	+ name LowCardinality(String), cpu_pct Float32, interval_seconds UInt32,
	300	+ ingest_version UInt64
	301	+ ) ENGINE = ReplacingMergeTree(ingest_version) ORDER BY (host, epoch, pid, name)""", "", user, password,
290	302	modal_key, modal_secret)
	303	+ # Existing installations used ReplacingMergeTree without a version
	304	+ # column. Adding it is non-destructive; Grafana's argMax queries below
	305	+ # make those legacy tables deterministic immediately as well.
	306	+ clickhouse_request(url, f"ALTER TABLE {samples_target} ADD COLUMN IF NOT EXISTS ingest_version UInt64 DEFAULT 0",
	307	+ "", user, password, modal_key, modal_secret)
	308	+ clickhouse_request(url, f"ALTER TABLE {processes_target} ADD COLUMN IF NOT EXISTS ingest_version UInt64 DEFAULT 0",
	309	+ "", user, password, modal_key, modal_secret)
291	310
292	311	sample_rows = [
293	312	{
@@ -301,11 +320,15 @@ def write_clickhouse(
301	320	}
302	321	for s in samples
303	322	]
	323	+ for row in sample_rows:
	324	+ row["ingest_version"] = ingest_version(row)
304	325	process_rows = [
305	326	{"host": s.host, "epoch": s.epoch, "pid": pid, "name": name,
306	327	"cpu_pct": pct, "interval_seconds": s.interval}
307	328	for s in samples for name, pid, pct in s.procs
308	329	]
	330	+ for row in process_rows:
	331	+ row["ingest_version"] = ingest_version(row)
309	332	clickhouse_request(url, f"INSERT INTO {samples_target} FORMAT JSONEachRow",
310	333	"\n".join(json.dumps(row) for row in sample_rows), user, password,
311	334	modal_key, modal_secret)

  import argparse
  import base64
++import hashlib
  import json
  import logging
  import os
      return value
++def ingest_version(row: dict[str, object]) -> int:
++    """Return a stable UInt64 version for one immutable source row.
++
++    The value is derived from canonical JSON rather than wall-clock time, so a
++    retried export of the same atop interval has exactly the same version.
++    """
++    encoded = json.dumps(row, sort_keys=True, separators=(",", ":")).encode()
++    return int.from_bytes(hashlib.blake2b(encoded, digest_size=8).digest(), "big")
++
++
  def clickhouse_request(
      url: str, query: str, body: str, user: str | None, password: str | None,
      modal_key: str | None, modal_secret: str | None,
              ncpu UInt16, user_pct Float32, sys_pct Float32, nice_pct Float32,
              idle_pct Float32, wait_pct Float32, irq_pct Float32, softirq_pct Float32,
              steal_pct Float32, busy_pct Float32, load1 Float32, load5 Float32,
--            load15 Float32, ctxsw UInt64, intr UInt64
++            load15 Float32, ctxsw UInt64, intr UInt64, ingest_version UInt64
--        ) ENGINE = ReplacingMergeTree ORDER BY (host, epoch)""", "", user, password,
++        ) ENGINE = ReplacingMergeTree(ingest_version) ORDER BY (host, epoch)""", "", user, password,
                             modal_key, modal_secret)
          clickhouse_request(url, f"""CREATE TABLE IF NOT EXISTS {processes_target} (
              host LowCardinality(String), epoch DateTime, pid UInt32,
--            name LowCardinality(String), cpu_pct Float32, interval_seconds UInt32
++            name LowCardinality(String), cpu_pct Float32, interval_seconds UInt32,
--        ) ENGINE = ReplacingMergeTree ORDER BY (host, epoch, pid, name)""", "", user, password,
++            ingest_version UInt64
++        ) ENGINE = ReplacingMergeTree(ingest_version) ORDER BY (host, epoch, pid, name)""", "", user, password,
                             modal_key, modal_secret)
++        # Existing installations used ReplacingMergeTree without a version
++        # column. Adding it is non-destructive; Grafana's argMax queries below
++        # make those legacy tables deterministic immediately as well.
++        clickhouse_request(url, f"ALTER TABLE {samples_target} ADD COLUMN IF NOT EXISTS ingest_version UInt64 DEFAULT 0",
++                           "", user, password, modal_key, modal_secret)
++        clickhouse_request(url, f"ALTER TABLE {processes_target} ADD COLUMN IF NOT EXISTS ingest_version UInt64 DEFAULT 0",
++                           "", user, password, modal_key, modal_secret)
      sample_rows = [
          {
          }
          for s in samples
      ]
++    for row in sample_rows:
++        row["ingest_version"] = ingest_version(row)
      process_rows = [
          {"host": s.host, "epoch": s.epoch, "pid": pid, "name": name,
           "cpu_pct": pct, "interval_seconds": s.interval}
          for s in samples for name, pid, pct in s.procs
      ]
++    for row in process_rows:
++        row["ingest_version"] = ingest_version(row)
      clickhouse_request(url, f"INSERT INTO {samples_target} FORMAT JSONEachRow",
                         "\n".join(json.dumps(row) for row in sample_rows), user, password,
                         modal_key, modal_secret)

modified grafana/Dockerfile +2 -2 Viewed

@@ -4,9 +4,9 @@ FROM grafana/grafana:11.6.0
4	4	# Python and the Grafana plugin can be installed, then run Grafana unprivileged.
5	5	USER root
6	6	RUN if [ -f /etc/alpine-release ]; then \
7		- apk add --no-cache python3; \
	7	+ apk add --no-cache python3 socat; \
8	8	else \
9		- apt-get update && apt-get install --no-install-recommends -y python3 && rm -rf /var/lib/apt/lists/*; \
	9	+ apt-get update && apt-get install --no-install-recommends -y python3 socat && rm -rf /var/lib/apt/lists/*; \
10	10	fi \
11	11	&& ln -sf "$(command -v python3)" /usr/local/bin/python \
12	12	&& grafana cli plugins install grafana-clickhouse-datasource \

  # Python and the Grafana plugin can be installed, then run Grafana unprivileged.
  USER root
  RUN if [ -f /etc/alpine-release ]; then \
--      apk add --no-cache python3; \
++      apk add --no-cache python3 socat; \
      else \
--      apt-get update && apt-get install --no-install-recommends -y python3 && rm -rf /var/lib/apt/lists/*; \
++      apt-get update && apt-get install --no-install-recommends -y python3 socat && rm -rf /var/lib/apt/lists/*; \
      fi \
      && ln -sf "$(command -v python3)" /usr/local/bin/python \
      && grafana cli plugins install grafana-clickhouse-datasource \

modified grafana/provisioning/dashboards/json/atop-overview.json +3 -3 Viewed

@@ -7,7 +7,7 @@
7	7	"fieldConfig": {"defaults": {"unit": "percent"}, "overrides": []},
8	8	"gridPos": {"h": 9, "w": 12, "x": 0, "y": 0},
9	9	"id": 1,
10		- "targets": [{"format": 1, "query": "SELECT epoch AS time, busy_pct AS busy, wait_pct AS iowait FROM atop_samples WHERE $__timeFilter(epoch) ORDER BY epoch", "refId": "A"}],
	10	+ "targets": [{"format": 1, "query": "SELECT epoch AS time, argMax(busy_pct, ingest_version) AS busy, argMax(wait_pct, ingest_version) AS iowait FROM atop_samples WHERE $__timeFilter(epoch) GROUP BY host, epoch ORDER BY epoch", "refId": "A"}],
11	11	"title": "CPU busy and I/O wait",
12	12	"type": "timeseries"
13	13	},
@@ -16,7 +16,7 @@
16	16	"fieldConfig": {"defaults": {"unit": "short"}, "overrides": []},
17	17	"gridPos": {"h": 9, "w": 12, "x": 12, "y": 0},
18	18	"id": 2,
19		- "targets": [{"format": 1, "query": "SELECT epoch AS time, load1 AS load_1m, load5 AS load_5m, load15 AS load_15m FROM atop_samples WHERE $__timeFilter(epoch) ORDER BY epoch", "refId": "A"}],
	19	+ "targets": [{"format": 1, "query": "SELECT epoch AS time, argMax(load1, ingest_version) AS load_1m, argMax(load5, ingest_version) AS load_5m, argMax(load15, ingest_version) AS load_15m FROM atop_samples WHERE $__timeFilter(epoch) GROUP BY host, epoch ORDER BY epoch", "refId": "A"}],
20	20	"title": "Load average",
21	21	"type": "timeseries"
22	22	},
@@ -26,7 +26,7 @@
26	26	"gridPos": {"h": 9, "w": 24, "x": 0, "y": 9},
27	27	"id": 3,
28	28	"options": {"showHeader": true},
29		- "targets": [{"format": 1, "query": "SELECT name AS process, round(avg(cpu_pct), 2) AS avg_cpu_pct, round(max(cpu_pct), 2) AS peak_cpu_pct FROM atop_processes WHERE $__timeFilter(epoch) GROUP BY name ORDER BY avg_cpu_pct DESC LIMIT 15", "refId": "A"}],
	29	+ "targets": [{"format": 1, "query": "SELECT name AS process, round(avg(cpu_pct), 2) AS avg_cpu_pct, round(max(cpu_pct), 2) AS peak_cpu_pct FROM (SELECT host, epoch, pid, name, argMax(cpu_pct, ingest_version) AS cpu_pct FROM atop_processes WHERE $__timeFilter(epoch) GROUP BY host, epoch, pid, name) GROUP BY name ORDER BY avg_cpu_pct DESC LIMIT 15", "refId": "A"}],
30	30	"title": "Top processes",
31	31	"type": "table"
32	32	}

        "fieldConfig": {"defaults": {"unit": "percent"}, "overrides": []},
        "gridPos": {"h": 9, "w": 12, "x": 0, "y": 0},
        "id": 1,
--      "targets": [{"format": 1, "query": "SELECT epoch AS time, busy_pct AS busy, wait_pct AS iowait FROM atop_samples WHERE $__timeFilter(epoch) ORDER BY epoch", "refId": "A"}],
++      "targets": [{"format": 1, "query": "SELECT epoch AS time, argMax(busy_pct, ingest_version) AS busy, argMax(wait_pct, ingest_version) AS iowait FROM atop_samples WHERE $__timeFilter(epoch) GROUP BY host, epoch ORDER BY epoch", "refId": "A"}],
        "title": "CPU busy and I/O wait",
        "type": "timeseries"
      },
        "fieldConfig": {"defaults": {"unit": "short"}, "overrides": []},
        "gridPos": {"h": 9, "w": 12, "x": 12, "y": 0},
        "id": 2,
--      "targets": [{"format": 1, "query": "SELECT epoch AS time, load1 AS load_1m, load5 AS load_5m, load15 AS load_15m FROM atop_samples WHERE $__timeFilter(epoch) ORDER BY epoch", "refId": "A"}],
++      "targets": [{"format": 1, "query": "SELECT epoch AS time, argMax(load1, ingest_version) AS load_1m, argMax(load5, ingest_version) AS load_5m, argMax(load15, ingest_version) AS load_15m FROM atop_samples WHERE $__timeFilter(epoch) GROUP BY host, epoch ORDER BY epoch", "refId": "A"}],
        "title": "Load average",
        "type": "timeseries"
      },
        "gridPos": {"h": 9, "w": 24, "x": 0, "y": 9},
        "id": 3,
        "options": {"showHeader": true},
--      "targets": [{"format": 1, "query": "SELECT name AS process, round(avg(cpu_pct), 2) AS avg_cpu_pct, round(max(cpu_pct), 2) AS peak_cpu_pct FROM atop_processes WHERE $__timeFilter(epoch) GROUP BY name ORDER BY avg_cpu_pct DESC LIMIT 15", "refId": "A"}],
++      "targets": [{"format": 1, "query": "SELECT name AS process, round(avg(cpu_pct), 2) AS avg_cpu_pct, round(max(cpu_pct), 2) AS peak_cpu_pct FROM (SELECT host, epoch, pid, name, argMax(cpu_pct, ingest_version) AS cpu_pct FROM atop_processes WHERE $__timeFilter(epoch) GROUP BY host, epoch, pid, name) GROUP BY name ORDER BY avg_cpu_pct DESC LIMIT 15", "refId": "A"}],
        "title": "Top processes",
        "type": "table"
      }

modified justfile +4 -0 Viewed

@@ -23,6 +23,10 @@ test-today:
23	23	logs:
24	24	journalctl -u atop-modal-export.service --no-pager -n 100
25	25
	26	+# Inspect recent logs emitted by the remote Modal ingestion function.
	27	+logs-modal:
	28	+ modal app logs atop-analyze --since 24h --tail 100 --timestamps
	29	+
26	30	# Show the Modal app's local-entrypoint arguments.
27	31	help:
28	32	modal run modal_atop_analyze.py --help

  logs:
    journalctl -u atop-modal-export.service --no-pager -n 100
++# Inspect recent logs emitted by the remote Modal ingestion function.
++logs-modal:
++  modal app logs atop-analyze --since 24h --tail 100 --timestamps
++
  # Show the Modal app's local-entrypoint arguments.
  help:
    modal run modal_atop_analyze.py --help

modified modal_grafana.py +20 -5 Viewed

@@ -54,7 +54,7 @@ image = (
54	54	},
55	55	)
56	56	@modal.concurrent(max_inputs=100)
57		-@modal.web_server(3000, startup_timeout=120)
	57	+@modal.web_server(8080, startup_timeout=120)
58	58	def grafana() -> None:
59	59	"""Run a public Grafana login page; its data-source credentials stay server-side."""
60	60	process = subprocess.Popen(
@@ -63,15 +63,30 @@ def grafana() -> None:
63	63	# environment, since the web-server proxy reaches the service over IPv4.
64	64	env={**os.environ, "GF_SERVER_HTTP_ADDR": "0.0.0.0"},
65	65	)
66		- # Grafana takes a few seconds to bind. Do not return from the Modal web
67		- # function until its port is accepting connections; otherwise the proxy can
68		- # route the first request before a backend exists and leave it hanging.
	66	+ # Start the external listener only after Grafana accepts connections.
69	67	for _ in range(60):
70	68	if process.poll() is not None:
71	69	raise RuntimeError(f"Grafana exited during startup ({process.returncode})")
72	70	with socket.socket() as probe:
73	71	probe.settimeout(0.25)
74	72	if probe.connect_ex(("127.0.0.1", 3000)) == 0:
	73	+ print("Grafana backend ready on port 3000", flush=True)
	74	+ break
	75	+ time.sleep(0.5)
	76	+ else:
	77	+ raise RuntimeError("Grafana did not listen on port 3000 within 30 seconds")
	78	+
	79	+ proxy = subprocess.Popen([
	80	+ "socat", "TCP-LISTEN:8080,bind=0.0.0.0,reuseaddr,fork",
	81	+ "TCP:127.0.0.1:3000",
	82	+ ])
	83	+ for _ in range(20):
	84	+ if proxy.poll() is not None:
	85	+ raise RuntimeError(f"Grafana proxy exited during startup ({proxy.returncode})")
	86	+ with socket.socket() as probe:
	87	+ probe.settimeout(0.25)
	88	+ if probe.connect_ex(("127.0.0.1", 8080)) == 0:
	89	+ print("Grafana proxy ready on port 8080", flush=True)
75	90	return
76	91	time.sleep(0.5)
77		- raise RuntimeError("Grafana did not listen on port 3000 within 30 seconds")
	92	+ raise RuntimeError("Grafana proxy did not listen on port 8080 within 10 seconds")

      },
  )
  @modal.concurrent(max_inputs=100)
--@modal.web_server(3000, startup_timeout=120)
++@modal.web_server(8080, startup_timeout=120)
  def grafana() -> None:
      """Run a public Grafana login page; its data-source credentials stay server-side."""
      process = subprocess.Popen(
          # environment, since the web-server proxy reaches the service over IPv4.
          env={**os.environ, "GF_SERVER_HTTP_ADDR": "0.0.0.0"},
      )
--    # Grafana takes a few seconds to bind. Do not return from the Modal web
++    # Start the external listener only after Grafana accepts connections.
--    # function until its port is accepting connections; otherwise the proxy can
--    # route the first request before a backend exists and leave it hanging.
      for _ in range(60):
          if process.poll() is not None:
              raise RuntimeError(f"Grafana exited during startup ({process.returncode})")
          with socket.socket() as probe:
              probe.settimeout(0.25)
              if probe.connect_ex(("127.0.0.1", 3000)) == 0:
++                print("Grafana backend ready on port 3000", flush=True)
++                break
++        time.sleep(0.5)
++    else:
++        raise RuntimeError("Grafana did not listen on port 3000 within 30 seconds")
++
++    proxy = subprocess.Popen([
++        "socat", "TCP-LISTEN:8080,bind=0.0.0.0,reuseaddr,fork",
++        "TCP:127.0.0.1:3000",
++    ])
++    for _ in range(20):
++        if proxy.poll() is not None:
++            raise RuntimeError(f"Grafana proxy exited during startup ({proxy.returncode})")
++        with socket.socket() as probe:
++            probe.settimeout(0.25)
++            if probe.connect_ex(("127.0.0.1", 8080)) == 0:
++                print("Grafana proxy ready on port 8080", flush=True)
                  return
          time.sleep(0.5)
--    raise RuntimeError("Grafana did not listen on port 3000 within 30 seconds")
++    raise RuntimeError("Grafana proxy did not listen on port 8080 within 10 seconds")

Make ClickHouse ingestion idempotent and expose Grafana through proxy