Hi @mattventura, here is stats for these satellites, https://stats.storkspace.net/, it will update once a day (retry several times a day if somehow fail).
storj_stats.py
#!/usr/bin/env python3
"""Generate data.json, data-public.json and nodes.json (stats.storjshare.io style)
from satellite databases running on CockroachDB.
Every query reads historical data with AS OF SYSTEM TIME '-10m' (configurable
with --as-of) to avoid contention with live writes. The "last 30 days" and
"4 hour online" windows are relative to now(), which differs from the read
time by only ~10 minutes.
Usage:
pip install "psycopg[binary]"
python3 storj_stats.py --config config.toml --out ./out
python3 storj_stats.py --config config.toml --as-of -30s
python3 storj_stats.py --print-sql # show the queries, no DB needed
Requires Python 3.11+ (tomllib).
"""
from __future__ import annotations
import argparse
import json
import os
import re
import sys
import tempfile
import tomllib
from decimal import Decimal
from pathlib import Path
# --------------------------------------------------------------------------
# Queries. {asof} is replaced by the --as-of value (validated, e.g. -10m).
# pb.PieceAction: PUT = 1 (upload), GET = 2 (download)
# placement 0 / NULL = every country = public placement
# --------------------------------------------------------------------------
AOST = "AS OF SYSTEM TIME '{asof}'"
Q_STORAGE = f"""
SELECT
COALESCE(SUM(total_bytes), 0) AS storage_total_bytes,
COALESCE(SUM(object_count), 0) AS storage_total_objects,
COALESCE(SUM(remote_segments_count), 0) AS storage_total_segments
FROM bucket_storage_tallies {AOST}
WHERE interval_start = (SELECT MAX(interval_start) FROM bucket_storage_tallies)
"""
Q_BANDWIDTH = f"""
SELECT
COALESCE(SUM(settled + inline) FILTER (WHERE action = 2), 0) AS bandwidth_bytes_downloaded,
COALESCE(SUM(settled + inline) FILTER (WHERE action = 1), 0) AS bandwidth_bytes_uploaded
FROM bucket_bandwidth_rollups {AOST}
WHERE interval_start > now() - INTERVAL '30 days'
AND action IN (1, 2)
"""
# Run on the METABASE database. redundancy is a little-endian packed int8:
# byte4 = required shares, byte7 = total shares.
Q_EXPANSION = f"""
SELECT
COALESCE(SUM(
encrypted_size::numeric
* ((redundancy >> 56) & 255)
/ NULLIF((redundancy >> 32) & 255, 0)
), 0)::bigint AS storage_total_bytes_after_expansion
FROM segments {AOST}
WHERE remote_alias_pieces IS NOT NULL
AND (expires_at IS NULL OR expires_at > now())
"""
Q_STORAGE_PUBLIC = f"""
SELECT
COALESCE(SUM(t.total_bytes), 0) AS storage_total_bytes,
COALESCE(SUM(t.object_count), 0) AS storage_total_objects,
COALESCE(SUM(t.remote_segments_count), 0) AS storage_total_segments
FROM bucket_storage_tallies t
JOIN bucket_metainfos b
ON b.project_id = t.project_id AND b.name = t.bucket_name
{AOST}
WHERE t.interval_start = (SELECT MAX(interval_start) FROM bucket_storage_tallies)
AND COALESCE(b.placement, 0) = 0
"""
Q_BANDWIDTH_PUBLIC = f"""
SELECT
COALESCE(SUM(r.settled + r.inline) FILTER (WHERE r.action = 2), 0) AS bandwidth_bytes_downloaded,
COALESCE(SUM(r.settled + r.inline) FILTER (WHERE r.action = 1), 0) AS bandwidth_bytes_uploaded
FROM bucket_bandwidth_rollups r
JOIN bucket_metainfos b
ON b.project_id = r.project_id AND b.name = r.bucket_name
{AOST}
WHERE r.interval_start > now() - INTERVAL '30 days'
AND r.action IN (1, 2)
AND COALESCE(b.placement, 0) = 0
"""
Q_NODES = f"""
WITH n AS (
SELECT
*,
(disqualified IS NOT NULL) AS is_dq,
(disqualified IS NULL AND exit_finished_at IS NOT NULL) AS is_exited,
(disqualified IS NULL AND exit_finished_at IS NULL) AS is_live
FROM nodes
)
SELECT
COUNT(*) FILTER (WHERE is_live AND last_contact_success > now() - INTERVAL '4 hours') AS active_nodes,
COUNT(*) FILTER (WHERE is_dq) AS disqualified_nodes,
COUNT(*) FILTER (WHERE is_exited) AS exited_nodes,
COUNT(*) FILTER (WHERE is_live AND last_contact_success <= now() - INTERVAL '4 hours') AS offline_nodes,
COUNT(*) FILTER (WHERE is_live AND (unknown_audit_suspended IS NOT NULL
OR offline_suspended IS NOT NULL)) AS suspended_nodes,
COUNT(*) AS total_nodes,
COUNT(*) FILTER (WHERE is_live AND vetted_at IS NOT NULL) AS vetted_nodes,
COUNT(*) FILTER (WHERE is_live AND free_disk >= 0 AND free_disk < 5000000000) AS full_nodes,
COUNT(DISTINCT wallet) FILTER (WHERE is_live
AND last_contact_success > now() - INTERVAL '4 hours'
AND wallet <> '') AS active_wallets
FROM n {AOST}
"""
def sql(template: str, asof: str) -> str:
return template.format(asof=asof).strip()
def validate_asof(value: str) -> str:
if not re.fullmatch(r"-\d+(ms|s|m|h)", value):
raise argparse.ArgumentTypeError("use a negative interval such as -10m, -30s or -1h")
return value
def to_jsonable(v):
if v is None:
return 0
if isinstance(v, Decimal):
return int(v)
return v
def fetch_row(conn, query: str) -> dict:
with conn.cursor() as cur:
cur.execute(query)
cols = [c.name for c in cur.description]
row = cur.fetchone()
return {c: to_jsonable(v) for c, v in zip(cols, row)}
def collect(sat: dict, asof: str, skip_expansion: bool) -> dict:
import psycopg # imported here so --print-sql works without the driver
data, data_public, nodes = {}, {}, {}
with psycopg.connect(sat["satellitedb"], autocommit=True) as conn:
data.update(fetch_row(conn, sql(Q_STORAGE, asof)))
data.update(fetch_row(conn, sql(Q_BANDWIDTH, asof)))
data_public.update(fetch_row(conn, sql(Q_STORAGE_PUBLIC, asof)))
data_public.update(fetch_row(conn, sql(Q_BANDWIDTH_PUBLIC, asof)))
nodes.update(fetch_row(conn, sql(Q_NODES, asof)))
if sat.get("metabase") and not skip_expansion:
with psycopg.connect(sat["metabase"], autocommit=True) as conn:
data.update(fetch_row(conn, sql(Q_EXPANSION, asof)))
else:
data["storage_total_bytes_after_expansion"] = None
# keep the same field order as the published page
order = [
"bandwidth_bytes_downloaded", "bandwidth_bytes_uploaded",
"storage_total_bytes", "storage_total_bytes_after_expansion",
"storage_total_objects", "storage_total_segments",
]
data = {k: data[k] for k in order}
return {"data": data, "data_public": data_public, "nodes": nodes}
def write_json_atomic(path: Path, obj) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
fd, tmp = tempfile.mkstemp(dir=path.parent, prefix=path.name, suffix=".tmp")
try:
with os.fdopen(fd, "w") as f:
json.dump(obj, f, indent=2)
f.write("\n")
os.replace(tmp, path)
except BaseException:
os.unlink(tmp)
raise
def print_sql(asof: str) -> None:
sections = [
("data.json: storage [satellitedb]", Q_STORAGE),
("data.json: bandwidth [satellitedb]", Q_BANDWIDTH),
("data.json: storage_total_bytes_after_expansion [metabase]", Q_EXPANSION),
("data-public.json: storage [satellitedb]", Q_STORAGE_PUBLIC),
("data-public.json: bandwidth [satellitedb]", Q_BANDWIDTH_PUBLIC),
("nodes.json [satellitedb]", Q_NODES),
]
print(f"-- Every query reads AS OF SYSTEM TIME '{asof}'\n")
for title, q in sections:
print(f"-- {title}\n{sql(q, asof)};\n")
def main() -> int:
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("--config", default="config.toml", help="TOML file listing satellites (default: config.toml)")
ap.add_argument("--out", default="out", help="output directory (default: ./out)")
ap.add_argument("--as-of", dest="asof", type=validate_asof, default="-10m",
help="AS OF SYSTEM TIME offset (default: -10m)")
ap.add_argument("--skip-expansion", action="store_true",
help="skip the slow metabase scan; storage_total_bytes_after_expansion becomes null")
ap.add_argument("--print-sql", action="store_true", help="print the queries and exit")
args = ap.parse_args()
asof = args.asof
if args.print_sql:
print_sql(asof)
return 0
with open(args.config, "rb") as f:
cfg = tomllib.load(f)
sats = cfg.get("satellite", [])
if not sats:
print("error: no [[satellite]] entries in config", file=sys.stderr)
return 2
print(f"reading AS OF SYSTEM TIME '{asof}'", file=sys.stderr)
out = {"data": {}, "data_public": {}, "nodes": {}}
failed = False
for sat in sats:
url = sat["url"]
print(f"[{url}] querying...", file=sys.stderr)
try:
res = collect(sat, asof, args.skip_expansion)
except Exception as e: # report every satellite, then fail as a whole
failed = True
print(f"[{url}] FAILED: {e}", file=sys.stderr)
continue
for k in out:
out[k][url] = res[k]
if failed:
print("not writing any file because at least one satellite failed", file=sys.stderr)
return 1
outdir = Path(args.out)
write_json_atomic(outdir / "data.json", out["data"])
write_json_atomic(outdir / "data-public.json", out["data_public"])
write_json_atomic(outdir / "nodes.json", out["nodes"])
print(f"wrote data.json, data-public.json, nodes.json to {outdir}/", file=sys.stderr)
return 0
if __name__ == "__main__":
sys.exit(main())