https://pastein.ru/t/k5M
Загрузка данных
#!/usr/bin/env python3
"""
opensearch_index_stats.py
Получает список индексов из OpenSearch (через _cat/indices), считает общий
размер хранилища и раскладывает объём данных по дням.
День для каждого индекса определяется так:
1) по дате в имени индекса (напр. logs-2026.07.15, app-2026-07-15, ...);
2) если даты в имени нет — по дате создания индекса (creation.date).
Зависимостей нет — только стандартная библиотека Python 3.
Примеры:
export OS_URL=https://opensearch.example.com:9200
export OS_USER=admin
export OS_PASS=secret
python3 opensearch_index_stats.py # все индексы
python3 opensearch_index_stats.py --pattern 'logs-*'
python3 opensearch_index_stats.py --insecure # не проверять TLS-серт
python3 opensearch_index_stats.py --json # машиночитаемый вывод
"""
import argparse
import base64
import datetime as dt
import json
import os
import re
import ssl
import sys
import urllib.error
import urllib.request
from collections import defaultdict
# --- парсинг даты из имени индекса -------------------------------------------
# ловит YYYY.MM.DD / YYYY-MM-DD / YYYY_MM_DD / YYYYMMDD в конце или в теле имени
DATE_RE = re.compile(r"(\d{4})[.\-_]?(\d{2})[.\-_]?(\d{2})")
def parse_args():
p = argparse.ArgumentParser(description="OpenSearch: размер индексов и объём данных за день")
p.add_argument("--url", default=os.environ.get("OS_URL", "http://localhost:9200"),
help="базовый URL OpenSearch (env OS_URL)")
p.add_argument("--user", default=os.environ.get("OS_USER"),
help="имя пользователя (env OS_USER)")
p.add_argument("--password", default=os.environ.get("OS_PASS"),
help="пароль (env OS_PASS)")
p.add_argument("--pattern", default="*",
help="шаблон индексов, напр. 'logs-*' (по умолчанию все, кроме системных)")
p.add_argument("--include-system", action="store_true",
help="включить системные индексы (начинающиеся с точки)")
p.add_argument("--insecure", action="store_true",
help="не проверять TLS-сертификат")
p.add_argument("--top", type=int, default=15,
help="сколько дней показать в разбивке по дням (0 = все)")
p.add_argument("--json", action="store_true",
help="вывести результат в JSON")
p.add_argument("--timeout", type=float, default=30.0, help="таймаут запроса, сек")
return p.parse_args()
def http_get(url, user, password, insecure, timeout):
req = urllib.request.Request(url)
if user is not None:
token = base64.b64encode(f"{user}:{password or ''}".encode()).decode()
req.add_header("Authorization", f"Basic {token}")
ctx = None
if url.lower().startswith("https"):
ctx = ssl.create_default_context()
if insecure:
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
try:
with urllib.request.urlopen(req, timeout=timeout, context=ctx) as r:
return json.loads(r.read().decode())
except urllib.error.HTTPError as e:
sys.exit(f"HTTP {e.code} при запросе {url}: {e.read().decode(errors='replace')[:300]}")
except urllib.error.URLError as e:
sys.exit(f"Не удалось подключиться к {url}: {e.reason}")
def fetch_indices(args):
# bytes=b -> размеры в байтах; h=... -> нужные колонки
cols = "index,docs.count,store.size,pri.store.size,creation.date"
url = (f"{args.url.rstrip('/')}/_cat/indices/{args.pattern}"
f"?format=json&bytes=b&h={cols}")
return http_get(url, args.user, args.password, args.insecure, args.timeout)
def day_of_index(name, creation_epoch_ms):
"""Определить день индекса: сначала из имени, затем из даты создания."""
m = DATE_RE.search(name)
if m:
y, mo, d = map(int, m.groups())
try:
return dt.date(y, mo, d).isoformat(), "name"
except ValueError:
pass
if creation_epoch_ms:
day = dt.datetime.utcfromtimestamp(int(creation_epoch_ms) / 1000).date()
return day.isoformat(), "created"
return "unknown", "none"
def human(nbytes):
n = float(nbytes)
for unit in ("B", "KiB", "MiB", "GiB", "TiB", "PiB"):
if abs(n) < 1024.0 or unit == "PiB":
return f"{n:,.2f} {unit}".replace(",", " ")
n /= 1024.0
def main():
args = parse_args()
rows = fetch_indices(args)
total_size = 0
total_docs = 0
by_day_size = defaultdict(int)
by_day_docs = defaultdict(int)
by_day_count = defaultdict(int)
src_stats = defaultdict(int) # как определён день: name/created/none
n_indices = 0
for r in rows:
name = r.get("index", "")
if not args.include_system and name.startswith("."):
continue
n_indices += 1
size = int(r.get("store.size") or 0)
docs = int(r.get("docs.count") or 0)
total_size += size
total_docs += docs
day, src = day_of_index(name, r.get("creation.date"))
src_stats[src] += 1
by_day_size[day] += size
by_day_docs[day] += docs
by_day_count[day] += 1
# дни с известной датой (для средних)
dated = {d: s for d, s in by_day_size.items() if d != "unknown"}
n_days = len(dated)
avg_per_day = (sum(dated.values()) / n_days) if n_days else 0
avg_docs_per_day = (sum(by_day_docs[d] for d in dated) / n_days) if n_days else 0
if args.json:
out = {
"indices": n_indices,
"total_size_bytes": total_size,
"total_size_human": human(total_size),
"total_docs": total_docs,
"days_with_date": n_days,
"avg_size_per_day_bytes": round(avg_per_day),
"avg_size_per_day_human": human(avg_per_day),
"avg_docs_per_day": round(avg_docs_per_day),
"by_day": {
d: {
"size_bytes": by_day_size[d],
"size_human": human(by_day_size[d]),
"docs": by_day_docs[d],
"indices": by_day_count[d],
} for d in sorted(by_day_size)
},
}
print(json.dumps(out, ensure_ascii=False, indent=2))
return
# --- человекочитаемый отчёт ---
print(f"OpenSearch: {args.url} шаблон: {args.pattern}")
print("=" * 64)
print(f"Индексов: {n_indices}")
print(f"Общий размер: {human(total_size)} ({total_size:,} B)".replace(",", " "))
print(f"Всего документов: {total_docs:,}".replace(",", " "))
print(f"Дней с известной датой: {n_days}")
print(f"Средний объём в день: {human(avg_per_day)}")
print(f"Средне документов/день: {avg_docs_per_day:,.0f}".replace(",", " "))
print(f"Определение дня: из имени={src_stats['name']}, "
f"по created={src_stats['created']}, не определено={src_stats['none']}")
print("=" * 64)
days_sorted = sorted((d for d in by_day_size), reverse=True)
shown = days_sorted if args.top == 0 else days_sorted[: args.top]
print(f"Разбивка по дням (последние {len(shown)}):")
print(f"{'дата':<12} {'размер':>14} {'документы':>16} {'индексов':>9}")
print("-" * 64)
for d in shown:
print(f"{d:<12} {human(by_day_size[d]):>14} "
f"{by_day_docs[d]:>16,} {by_day_count[d]:>9}".replace(",", " "))
if args.top and len(days_sorted) > args.top:
print(f"... ещё {len(days_sorted) - args.top} дн. (используй --top 0 для полного списка)")
if __name__ == "__main__":
main()
