Database observability: biểu đồ phải trả lời câu hỏi nào?

Câu hỏi: query chậm do đang làm việc, chờ khóa hay đọc replica chưa bắt kịp?

Cần biết trước: EXPLAIN, khóa và replication. Lab native đã chạy PostgreSQL 18.6, Python 3.14.4, postgres_exporter 0.20.1, Prometheus 3.15.0 và Grafana OSS 13.2.2 trên macOS arm64. Không dùng DB đang phục vụ ứng dụng, không cần Docker hoặc VM. Chưa kiểm Linux hoặc MySQL trong bài này.

Định nghĩa tín hiệu trước dashboard

Ba gauge đầu do lab tự định nghĩa, không phải tên collector mặc định hoặc latency SLO. Lab dùng custom query của exporter 0.20.1; extend.query-path đã deprecated. Với hệ thống dài hạn, ưu tiên collector có sẵn hoặc SQL exporter phù hợp sau khi kiểm schema metric. Không thay exporter mà giữ nguyên dashboard một cách mù quáng. Exporter configuration.

Chuẩn bị fixture và binary

Lưu repl.sh từ bài replication vào thư mục trống. Chưa gọi repl_up: script dưới tự khởi động rồi cleanup hai cluster socket riêng. PostgreSQL 18.6 tools phải có trong PATH; Python phải là 3.14.4. Bộ binary observability được tải vào OBS_TOOLS (mặc định tools/ trong thư mục lab), không cài toàn máy. Cache có thể tái dùng; archive luôn được kiểm SHA-256 trước khi giải nén.

Nguồn binary/checksum: Prometheus 3.15.0, exporter 0.20.1, Grafana 13.2.2. Script này cố ý chỉ nhận Darwin arm64, đúng nền tảng đã chạy.

set -euo pipefail
[ "$(uname -s)/$(uname -m)" = Darwin/arm64 ]
OBS_TOOLS=${OBS_TOOLS:-"$PWD/tools"}
mkdir -p "$OBS_TOOLS"
OBS_TOOLS=$(cd "$OBS_TOOLS" && pwd)
fetch() (
  name=$1
  url=$2
  digest=$3
  cd "$OBS_TOOLS"
  if [ ! -f "$name" ]; then
    trap 'rm -f "$name.part"' EXIT
    curl --proto '=https' --tlsv1.2 -fsSL --max-time 240 "$url" -o "$name.part"
    mv "$name.part" "$name"
  fi
  printf '%s  %s\n' "$digest" "$name" | shasum -a 256 -c -
  tar -xzf "$name"
)
fetch prometheus-3.15.0.darwin-arm64.tar.gz \
  https://github.com/prometheus/prometheus/releases/download/v3.15.0/prometheus-3.15.0.darwin-arm64.tar.gz \
  920df4d17e78b3b0175af144eb318b0c74d1cf7b1d1251b326966f0e81977260
fetch postgres_exporter-0.20.1.darwin-arm64.tar.gz \
  https://github.com/prometheus-community/postgres_exporter/releases/download/v0.20.1/postgres_exporter-0.20.1.darwin-arm64.tar.gz \
  4dd2b9e7ac658556dd2cb143ac472a87ff540ffc2b482d5311ff0ef47b11413c
fetch grafana_13.2.2_34846740809_darwin_arm64.tar.gz \
  https://dl.grafana.com/grafana/release/13.2.2/grafana_13.2.2_34846740809_darwin_arm64.tar.gz \
  9ea91bf7cc92aa34dd59321c44e7a15a22ee6f3dbb491849a6357451d35d5932
printf 'export OBS_TOOLS=%q\n' "$OBS_TOOLS" > tools.env

wiki_observe:
  query: >
    SELECT
      COALESCE((SELECT max(EXTRACT(EPOCH FROM clock_timestamp()-query_start))
        FROM pg_stat_activity WHERE application_name='wiki_observe_slow' AND state='active'),0) AS slow_seconds,
      (SELECT count(*) FROM pg_stat_activity
        WHERE application_name='wiki_observe_waiter' AND wait_event_type='Lock') AS lock_waiters,
      COALESCE((SELECT max(pg_wal_lsn_diff(pg_current_wal_lsn(),replay_lsn))
        FROM pg_stat_replication),0) AS replay_bytes
  metrics:
    - slow_seconds:
        usage: GAUGE
        description: Active fixture query age in seconds, not completed latency
    - lock_waiters:
        usage: GAUGE
        description: Fixture sessions currently waiting for locks
    - replay_bytes:
        usage: GAUGE
        description: Primary WAL bytes beyond replica replay acknowledgement

Exporter dùng role wiki_monitor có pg_monitor, không superuser; fixture role lab mới có quyền tạo triệu chứng. Gauge không label bằng SQL text, user hoặc request ID, tránh cardinality và lộ nội dung query. Cấu hình scrape_interval: 1s và scrape_timeout: 800ms chỉ dùng cho lab; production cần ngân sách DB/query riêng. Scrape configuration.

Chạy và đối chiếu cùng cửa sổ

Lưu file Python bên dưới; HTTP binds vào 127.0.0.1 với port trống, không dùng cổng DB TCP. Dashboard provisioned có 4 panel, units seconds/sessions/bytes/short, range 30s, refresh 1s; script truy vấn datasource qua Grafana để kiểm đường đọc thực sự.

from __future__ import annotations

import json
import os
import platform
import signal
import socket
import subprocess
import sys
import time
import urllib.parse
import urllib.request
from pathlib import Path
from typing import Any

assert sys.version_info[:3] == (3, 14, 4)
assert platform.system() == "Darwin" and platform.machine() == "arm64"
tools = Path(os.environ["OBS_TOOLS"]).resolve()
prom_dir = tools / "prometheus-3.15.0.darwin-arm64"
exporter_bin = tools / "postgres_exporter-0.20.1.darwin-arm64/postgres_exporter"
grafana_dirs = list(tools.glob("grafana-*"))
grafana_dir = next(
    p for p in grafana_dirs if p.is_dir() and (p / "bin/grafana").exists()
)
root = Path.cwd()
processes: list[subprocess.Popen[str]] = []
listeners: list[int] = []
logs: list[Any] = []


def shell(code: str) -> str:
    return subprocess.run(
        ["bash", "-euo", "pipefail", "-c", ". ./repl.sh; " + code],
        check=True,
        text=True,
        capture_output=True,
        timeout=60,
    ).stdout.strip()


def sql(query: str, node: str = "primary") -> str:
    import shlex

    return shell("repl_sql " + node + " -At -c " + shlex.quote(query))


def http_json(port: int, path: str) -> Any:
    with urllib.request.urlopen(
        f"http://127.0.0.1:{port}{path}", timeout=3
    ) as response:
        return json.load(response)


def wait(check: Any, seconds: float = 20) -> Any:
    end = time.monotonic() + seconds
    while time.monotonic() < end:
        try:
            result = check()
            if result:
                return result
        except OSError, ValueError:
            pass
        time.sleep(0.1)
    raise TimeoutError("observation barrier")


def port() -> int:
    with socket.socket() as sock:
        sock.bind(("127.0.0.1", 0))
        return int(sock.getsockname()[1])


def start(
    name: str, args: list[str], env: dict[str, str] | None = None
) -> subprocess.Popen[str]:
    output = (root / (name + ".log")).open("w")
    logs.append(output)
    proc = subprocess.Popen(
        args,
        stdout=output,
        stderr=subprocess.STDOUT,
        text=True,
        env=env,
        start_new_session=True,
    )
    processes.append(proc)
    return proc


def stop(proc: subprocess.Popen[str]) -> None:
    if proc.poll() is None:
        os.killpg(proc.pid, signal.SIGTERM)
        try:
            proc.wait(timeout=10)
        except subprocess.TimeoutExpired:
            os.killpg(proc.pid, signal.SIGKILL)
            proc.wait(timeout=5)


def write(path: str, body: str) -> None:
    target = root / path
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_text(body)


try:
    for binary, flag, expected_version in [
        (prom_dir / "prometheus", "--version", "3.15.0"),
        (exporter_bin, "--version", "0.20.1"),
        (grafana_dir / "bin/grafana", "--version", "13.2.2"),
    ]:
        result = subprocess.run(
            [str(binary), flag], capture_output=True, text=True, check=True, timeout=10
        )
        assert expected_version in result.stdout + result.stderr
    shell("repl_up; repl_target")
    sql(
        "CREATE ROLE wiki_monitor LOGIN; GRANT pg_monitor TO wiki_monitor; INSERT INTO wiki_repl.orders VALUES (1,'accepted',clock_timestamp())"
    )
    socket_dir = shell('repl_env; printf %s "$REPL_DIR/primary-socket"')
    export_port, prom_port, grafana_port = port(), port(), port()
    listeners.extend([export_port, prom_port, grafana_port])
    assert len({export_port, prom_port, grafana_port}) == 3
    env = {
        k: v
        for k, v in os.environ.items()
        if not k.startswith("PG") and not k.startswith("DATA_SOURCE")
    }
    env.update(
        DATA_SOURCE_NAME=f"postgresql://wiki_monitor@/postgres?host={socket_dir}&sslmode=disable",
        PG_EXPORTER_COLLECTION_TIMEOUT="500ms",
    )
    exporter_args = [
        str(exporter_bin),
        f"--web.listen-address=127.0.0.1:{export_port}",
        "--extend.query-path=queries.yaml",
    ]
    exporter = start("exporter", exporter_args, env)
    write(
        "prometheus.yaml",
        f'global:\n  scrape_interval: 1s\n  scrape_timeout: 800ms\nscrape_configs:\n  - job_name: wiki_pg\n    static_configs:\n      - targets: ["127.0.0.1:{export_port}"]\n',
    )
    subprocess.run(
        [str(prom_dir / "promtool"), "check", "config", "prometheus.yaml"], check=True
    )
    start(
        "prometheus",
        [
            str(prom_dir / "prometheus"),
            "--config.file=prometheus.yaml",
            "--storage.tsdb.path=prom-data",
            "--storage.tsdb.retention.time=1h",
            f"--web.listen-address=127.0.0.1:{prom_port}",
        ],
    )
    write(
        "provisioning/datasources/lab.yaml",
        f"apiVersion: 1\ndatasources:\n  - name: Lab\n    uid: wiki_pg\n    type: prometheus\n    access: proxy\n    url: http://127.0.0.1:{prom_port}\n    isDefault: true\n    editable: false\n",
    )
    write(
        "provisioning/dashboards/lab.yaml",
        f"apiVersion: 1\nproviders:\n  - name: Lab\n    type: file\n    options:\n      path: {root / 'dashboards'}\n",
    )
    expressions = [
        "wiki_observe_slow_seconds",
        "wiki_observe_lock_waiters",
        "wiki_observe_replay_bytes",
        'up{job="wiki_pg"}',
    ]
    titles = [
        "Active query age",
        "Lock waiters",
        "WAL replay backlog",
        "Scrape available",
    ]
    units = ["s", "short", "bytes", "short"]
    dashboard = {
        "uid": "wiki-observe",
        "title": "Isolated database observations",
        "schemaVersion": 39,
        "version": 1,
        "refresh": "1s",
        "time": {"from": "now-30s", "to": "now"},
        "panels": [
            {
                "id": i + 1,
                "title": titles[i],
                "type": "timeseries",
                "datasource": {"type": "prometheus", "uid": "wiki_pg"},
                "gridPos": {"x": (i % 2) * 12, "y": (i // 2) * 8, "w": 12, "h": 8},
                "fieldConfig": {"defaults": {"unit": units[i]}, "overrides": []},
                "targets": [{"expr": expression, "refId": "A"}],
            }
            for i, expression in enumerate(expressions)
        ],
    }
    write("dashboards/lab.json", json.dumps(dashboard))
    write(
        "grafana.ini",
        f"[paths]\ndata = {root / 'graf-data'}\nlogs = {root / 'graf-logs'}\nplugins = {root / 'graf-plugins'}\nprovisioning = {root / 'provisioning'}\n[server]\nhttp_addr = 127.0.0.1\nhttp_port = {grafana_port}\n[auth.anonymous]\nenabled = true\norg_role = Viewer\n[auth]\ndisable_login_form = true\n[analytics]\nreporting_enabled = false\ncheck_for_updates = false\ncheck_for_plugin_updates = false\n[plugins]\npreinstall_disabled = true\npublic_key_retrieval_disabled = true\n",
    )
    start(
        "grafana",
        [
            str(grafana_dir / "bin/grafana"),
            "server",
            "--homepath",
            str(grafana_dir),
            "--config",
            str(root / "grafana.ini"),
        ],
    )
    wait(lambda: http_json(grafana_port, "/api/health").get("database") == "ok", 60)
    loaded = wait(lambda: http_json(grafana_port, "/api/dashboards/uid/wiki-observe"))
    assert [
        p["targets"][0]["expr"] for p in loaded["dashboard"]["panels"]
    ] == expressions

    def metric(expression: str, via_grafana: bool = False) -> list[Any]:
        query = "/api/v1/query?" + urllib.parse.urlencode({"query": expression})
        data = (
            http_json(grafana_port, "/api/datasources/proxy/uid/wiki_pg" + query)
            if via_grafana
            else http_json(prom_port, query)
        )
        assert data["status"] == "success"
        return list(data["data"]["result"])

    def value(expression: str) -> float:
        rows = metric(expression)
        return float(rows[0]["value"][1]) if rows else -1

    wait(lambda: value('up{job="wiki_pg"}') == 1 and value("pg_up") == 1)
    assert all(metric(expression, True) for expression in expressions)
    snapshots: list[dict[str, Any]] = []

    def capture(case: str, expression: str, query: str, minimum: float) -> None:
        wait(lambda: value(expression) >= minimum)
        rows = metric(expression)
        sampled_at = value("timestamp(" + expression + ")")
        observed_at, db_value = sql(
            "SELECT EXTRACT(EPOCH FROM clock_timestamp()),(" + query + ")"
        ).split("|")
        assert abs(float(observed_at) - sampled_at) <= 3
        assert float(db_value) >= minimum
        if case == "slow":
            assert abs(float(db_value) - float(rows[0]["value"][1])) <= 3
        else:
            assert float(db_value) == float(rows[0]["value"][1])
        record = {
            "case": case,
            "metric": expression,
            "value": float(rows[0]["value"][1]),
            "sampled_at": sampled_at,
            "db_at": float(observed_at),
            "db_value": float(db_value),
        }
        snapshots.append(record)
        print("OBS_CASE " + json.dumps(record))

    def job(name: str, query: str) -> subprocess.Popen[str]:
        import shlex

        return start(
            name,
            [
                "bash",
                "-euo",
                "pipefail",
                "-c",
                ". ./repl.sh; repl_sql primary -c "
                + shlex.quote(
                    "SET application_name="
                    + shlex.quote("wiki_observe_" + name)
                    + "; SET statement_timeout='30s'; "
                    + query
                ),
            ],
        )

    slow = job("slow", "SELECT pg_sleep(20)")
    capture(
        "slow",
        expressions[0],
        "SELECT COALESCE(max(EXTRACT(EPOCH FROM clock_timestamp()-query_start)),0) FROM pg_stat_activity WHERE application_name='wiki_observe_slow' AND state='active'",
        1,
    )
    stop(slow)
    holder = job(
        "holder",
        "BEGIN; UPDATE wiki_repl.orders SET status='held' WHERE id=1; SELECT pg_sleep(20); ROLLBACK",
    )
    wait(
        lambda: (
            sql(
                "SELECT count(*) FROM pg_stat_activity WHERE application_name='wiki_observe_holder' AND wait_event='PgSleep'"
            )
            == "1"
        )
    )
    waiter = job("waiter", "UPDATE wiki_repl.orders SET status='waited' WHERE id=1")
    wait(
        lambda: (
            sql(
                "SELECT count(*) FROM pg_stat_activity WHERE application_name='wiki_observe_waiter' AND cardinality(pg_blocking_pids(pid))>0"
            )
            == "1"
        )
    )
    capture(
        "lock",
        expressions[1],
        "SELECT count(*) FROM pg_stat_activity WHERE application_name='wiki_observe_waiter' AND wait_event_type='Lock'",
        1,
    )
    stop(waiter)
    stop(holder)
    sql("SELECT pg_wal_replay_pause()", "replica")
    wait(
        lambda: sql("SELECT pg_get_wal_replay_pause_state()='paused'", "replica") == "t"
    )
    sql("INSERT INTO wiki_repl.orders VALUES(2,'lagged',clock_timestamp())")
    capture(
        "lag",
        expressions[2],
        "SELECT max(pg_wal_lsn_diff(pg_current_wal_lsn(),replay_lsn)) FROM pg_stat_replication",
        1,
    )
    sql("SELECT pg_wal_replay_resume()", "replica")
    wait(
        lambda: (
            sql("SELECT count(*) FROM wiki_repl.orders WHERE id=2", "replica") == "1"
        )
    )
    stop(exporter)
    wait(lambda: value('up{job="wiki_pg"}') == 0)
    wait(lambda: not metric(expressions[0]))
    assert not metric(expressions[0], True)
    assert sql("SELECT 1") == "1"
    print("exporter_down up=0 custom_series=absent database_select=1")
    exporter = start("exporter-recovery", exporter_args, env)
    wait(lambda: value('up{job="wiki_pg"}') == 1 and value("pg_up") == 1)
    assert all(metric(expression, True) for expression in expressions)
    print("recovered up=1 datasource=queried dashboard_panels=4")
    write(
        "observe-results.json",
        json.dumps(
            {
                "environment": {
                    "postgres": "18.6",
                    "exporter": "0.20.1",
                    "prometheus": "3.15.0",
                    "grafana": "13.2.2",
                    "python": platform.python_version(),
                    "os": platform.platform(),
                },
                "snapshots": snapshots,
            },
            indent=2,
        )
        + "\n",
    )
finally:
    for proc in reversed(processes):
        stop(proc)
    for log in logs:
        log.close()
    shell("repl_clean; repl_clean")
    assert all(proc.poll() is not None for proc in processes)
    for listener in listeners:
        with socket.socket() as probe:
            probe.settimeout(1)
            assert probe.connect_ex(("127.0.0.1", listener)) != 0
    print("cleanup owned_processes=closed clusters=removed")

set -euo pipefail
bash tools.sh
. ./tools.env
python3 observe.py

exporter_down up=0 custom_series=absent database_select=1
recovered up=1 datasource=queried dashboard_panels=4
cleanup owned_processes=closed clusters=removed

set -euo pipefail
. ./repl.sh
repl_clean
test ! -e repl.env

Một lượt đo thật

Đo trên macOS arm64 lúc 08:32:46–49 UTC ngày 2026-10-03. Mỗi dòng là một snapshot Prometheus rồi query DB, không phải thời gian hoàn tất query hay benchmark throughput. Cột cuối là DB timestamp trừ scrape timestamp, đơn vị seconds:

Slow age chênh khoảng 0,1s vì query vẫn chạy giữa hai lần lấy số. Lock cùng một waiter và WAL byte backlog bằng nhau trong cửa sổ này. Script kiểm lệch timestamp không quá 3s, slow age không lệch quá 3s, lock/lag bằng nhau; không bảo đảm các giá trị này xuất hiện giống hệt khi chạy lại. Query DB cũng cần kiểm quyền và ảnh hưởng tải. PostgreSQL statistics.

Khi nào cảnh báo cần hành động?

Dashboard nhìn 30s gần nhất; mỗi scrape 1s là snapshot, không phát hiện mọi query ngắn. Active query age tăng vì pg_sleep là ca kiểm đo, chưa phải nguyên nhân chậm thực tế. Query age giảm về 0 sau khi kết thúc cũng không phải completed latency 0. Đọc wait_event cùng timestamp; query thật cần plan/buffers và histogram ứng dụng.

Trong lab, lock_waiters>0 liên tục 3 scrape là lý do xem blocker; không tự kill session. Replay_bytes>0 liên tục 3 scrape dẫn tới kiểm receive/replay và dung lượng WAL. Đây là ngưỡng minh họa, chưa phải alert production: cần ngân sách đọc stale, baseline và owner có hành động rõ. Gauge bytes không dùng rate như counter đơn điệu.

Khi exporter tắt, up=0 và custom series biến mất sau failed scrape; DB vẫn trả SELECT 1. Một dashboard biến missing thành 0 sẽ cho cảm giác khỏe sai. Với mục tiêu cố định, theo dõi cả up==0 và absent(up{job="wiki_pg"}) nếu target bị gỡ; exporter đáp HTTP 200 nhưng DB hỏng cần pg_up/scrape error, chưa đủ chỉ nhìn HTTP. Không dùng or vector(0) để che missing. Prometheus staleness.

Lab đóng process và cluster do nó tạo, còn binary cache và output để đọc lại. Không cấu hình remote write, credential thật hoặc agent trên máy chủ. Dashboard provisioning không thay cho review metric schema khi nâng version. Grafana provisioning.

Học tiếp: deadlock, replica và đọc-sau-ghi.