duckstat 0.1.0

Python SDK for DuckStat monitoring API

duckstat

Python SDK для DuckStat — мониторинг серверов, сайтов и сервисов.

Установка

pip install duckstat

Или напрямую из репозитория:

pip install git+https://github.com/duckstat/python-sdk.git

Быстрый старт

from duckstat import DuckStat

ds = DuckStat("ds_live_ваш_ключ")

# Проверка подключения
ds.ping()

# Список серверов
for agent in ds.agents():
    print(f"{agent.name}: {'online' if agent.online else 'offline'}")
    if agent.cpu is not None:
        print(f"  CPU: {agent.cpu}%  RAM: {agent.ram_pct}%  Disk: {agent.disk_pct}%")

Серверы (Agents)

# Все серверы
agents = ds.agents()

# Конкретный сервер (с docker, ports, processes)
server = ds.agent("uuid")

# История метрик (последний час)
metrics = ds.agent_metrics("uuid", limit=60)
for m in metrics:
    print(f"{m.timestamp}: CPU {m.cpu_usage}%")

# Удобные шорткаты
offline = ds.offline_agents()
hot = ds.overloaded_agents(cpu_threshold=85)

Мониторы

# Все мониторы
monitors = ds.monitors()
for m in monitors:
    print(f"{m.name} ({m.type}): {'UP' if m.is_up else 'DOWN'}")

# История проверок
checks = ds.monitor_checks("uuid", limit=50)

# Аптайм за 7 дней
uptime = ds.monitor_uptime("uuid", range="7d")
print(f"Uptime: {uptime.uptime_pct}%  Avg response: {uptime.avg_response_ms}ms")

# Кто упал?
down = ds.all_down()

Инциденты

# Открытые инциденты
for inc in ds.incidents(status="OPEN"):
    print(f"{inc.monitor_name}: {inc.cause} (с {inc.started_at})")

# Все инциденты
all_incidents = ds.incidents(limit=100)

Webhooks

Для проверки подписи входящих вебхуков:

from duckstat import verify_webhook

# Flask пример
@app.route("/webhook", methods=["POST"])
def handle_webhook():
    sig = request.headers["X-DuckStat-Signature"]
    if not verify_webhook("whsec_ваш_секрет", request.data, sig):
        abort(401)

    event = request.json
    if event["event"] == "incident.open":
        send_alert(event["data"]["monitor"]["name"])
    return "ok"

Примеры

Скрипт мониторинга в cron

from duckstat import DuckStat
import smtplib

ds = DuckStat("ds_live_...")

# Алерт если что-то упало
down = ds.all_down()
if down:
    names = ", ".join(m.name for m in down)
    print(f"DOWN: {names}")
    # send_email(...)

# Алерт если сервер перегружен
for a in ds.overloaded_agents(cpu_threshold=90):
    print(f"OVERLOAD: {a.name} CPU={a.cpu}%")

Сбор метрик в CSV

import csv
from duckstat import DuckStat

ds = DuckStat("ds_live_...")

with open("metrics.csv", "w", newline="") as f:
    w = csv.writer(f)
    w.writerow(["timestamp", "server", "cpu", "ram_pct", "disk_pct", "load"])
    for agent in ds.agents():
        metrics = ds.agent_metrics(agent.id, limit=1000, since="2025-01-01T00:00:00Z")
        for m in metrics:
            w.writerow([m.timestamp, agent.name, m.cpu_usage, m.ram_pct, m.disk_pct, m.load_avg1])

Grafana / Prometheus bridge

from prometheus_client import Gauge, start_http_server
from duckstat import DuckStat
import time

ds = DuckStat("ds_live_...")
cpu_gauge = Gauge("duckstat_cpu", "CPU usage", ["server"])
ram_gauge = Gauge("duckstat_ram_pct", "RAM %", ["server"])

start_http_server(9100)
while True:
    for a in ds.agents():
        if a.cpu is not None:
            cpu_gauge.labels(server=a.name).set(a.cpu)
        if a.ram_pct is not None:
            ram_gauge.labels(server=a.name).set(a.ram_pct)
    time.sleep(30)

API Reference

Метод	Описание
ds.ping()	Проверка ключа
ds.agents()	Список серверов
ds.agent(id)	Детали сервера
ds.agent_metrics(id, limit, since)	История метрик
ds.monitors()	Список мониторов
ds.monitor(id)	Детали монитора
ds.monitor_checks(id, limit, since)	История проверок
ds.monitor_uptime(id, range)	Аптайм
ds.incidents(status, limit)	Инциденты
ds.incident(id)	Детали инцидента
ds.all_down()	Упавшие мониторы
ds.offline_agents()	Офлайн серверы
ds.overloaded_agents(cpu)	Перегруженные серверы
verify_webhook(secret, payload, sig)	Проверка подписи вебхука