06 · 主机监控与多机 Agent：把"机器本身"纳入观测

"""
hosts/models.py — 主机资源指标（由 hosts.collector 后台线程周期写入）
"""
from django.db import models
from django.utils import timezone


class HostMetric(models.Model):
    """一次主机采样：CPU / 内存 / 磁盘 / 网络 / 进程与连接数"""

    hostname = models.CharField('主机名', max_length=128, default='local')
    cpu_percent = models.FloatField('CPU 使用率(%)')
    cpu_cores = models.IntegerField('逻辑核心数', default=0)
    load_avg = models.FloatField('负载(1min)', default=0.0)
    mem_percent = models.FloatField('内存使用率(%)')
    mem_used_mb = models.FloatField('已用内存(MB)', default=0.0)
    mem_total_mb = models.FloatField('总内存(MB)', default=0.0)
    disk_percent = models.FloatField('系统盘使用率(%)', default=0.0)
    disk_used_gb = models.FloatField('系统盘已用(GB)', default=0.0)
    disk_total_gb = models.FloatField('系统盘总量(GB)', default=0.0)
    net_sent_kbps = models.FloatField('上行速率(KB/s)', default=0.0)
    net_recv_kbps = models.FloatField('下行速率(KB/s)', default=0.0)
    proc_count = models.IntegerField('进程数', default=0)
    tcp_conns = models.IntegerField('TCP 连接数', default=0)
    simulated = models.BooleanField('是否模拟数据', default=False)
    created_at = models.DateTimeField('采集时间', default=timezone.now, db_index=True)

    class Meta:
        verbose_name = '主机指标'
        verbose_name_plural = verbose_name
        ordering = ['-created_at']

    def __str__(self):
        return f'{self.hostname} cpu={self.cpu_percent}% @{self.created_at:%H:%M:%S}'

"""
hosts/collector.py — 主机指标采集线程

默认使用 psutil 读取真实系统指标（CPU / 内存 / 磁盘 / 网络速率 / 进程 / 连接）；
psutil 未安装时自动退化为按正弦波生成的模拟指标，保证平台在任何环境可用。

每 OBSERVABILITY.HOST_INTERVAL_SEC 秒采样一次写入 HostMetric，
并顺手清理超过 RETENTION_DAYS 的过期采集数据。
"""
import logging
import os
import random
import threading
import time
from datetime import timedelta

from django.conf import settings
from django.utils import timezone

try:
    import psutil
    if psutil:
        psutil.cpu_percent(interval=None)  # 预热：首次调用恒为 0，丢弃
except ImportError:
    psutil = None

logger = logging.getLogger(__name__)

MON = settings.OBSERVABILITY

# 网络速率需要上一次采样的累计字节数（采集线程与手动采样并发调用，加锁保护）
_last_net = {'sent': None, 'recv': None, 'ts': None}
_last_net_lock = threading.Lock()

# 资产自动登记节流：同一主机 5 分钟内不重复 update last_seen（10 篇接入 ops.Asset）
_asset_last_seen = {}
_asset_lock = threading.Lock()


def _disk_root():
    return os.path.abspath(os.sep)  # Windows -> C:\


def _hostname():
    name = os.environ.get('COMPUTERNAME')
    if not name and hasattr(os, 'uname'):
        name = os.uname().nodename
    return name or 'local'


def sample_once():
    """采样一次并入库，返回 HostMetric 实例"""
    now = timezone.now()
    hostname = _hostname()
    if psutil:
        return _sample_real(now, hostname)
    return _sample_simulated(now, hostname)


def _sample_real(now, hostname):
    cpu = psutil.cpu_percent(interval=None)
    if cpu < 0.5:
        # 进程刚启动时非阻塞读数不可靠（预热后仍接近 0），阻塞 150ms 重测
        cpu = psutil.cpu_percent(interval=0.15)
    mem = psutil.virtual_memory()
    try:
        load1 = psutil.getloadavg()[0]
    except (AttributeError, OSError):
        load1 = cpu / 100.0 * os.cpu_count()

    du = psutil.disk_usage(_disk_root())

    sent_kbps = recv_kbps = 0.0
    try:
        io = psutil.net_io_counters()
        with _last_net_lock:
            if _last_net['sent'] is not None and _last_net['ts'] is not None:
                dt = max(0.001, (now - _last_net['ts']).total_seconds())
                sent_kbps = max(0.0, (io.bytes_sent - _last_net['sent']) / dt / 1024)
                recv_kbps = max(0.0, (io.bytes_recv - _last_net['recv']) / dt / 1024)
            _last_net['sent'], _last_net['recv'], _last_net['ts'] = io.bytes_sent, io.bytes_recv, now
    except Exception:
        pass

    proc_count = len(psutil.pids())
    try:
        tcp = len(psutil.net_connections(kind='inet'))
    except Exception:
        tcp = 0

    return _save(now, hostname, cpu, load1, mem, du, sent_kbps, recv_kbps,
                 proc_count, tcp, simulated=False)


def _sample_simulated(now, hostname):
    """psutil 缺失时的正弦波模拟指标"""
    phase = now.timestamp() / 600
    cpu = 25 + 18 * (1 + __import__('math').sin(phase)) / 2 + random.uniform(0, 10)
    mem_pct = 55 + 10 * random.random()
    total_mb = 8192.0
    used_mb = total_mb * mem_pct / 100
    du_pct = 62.0
    return _save(now, hostname, cpu, cpu / 100,
                 _Mem(used_mb, total_mb, mem_pct),
                 _Disk(du_pct, du_pct * 5.12, 512),
                 random.uniform(10, 200), random.uniform(50, 800),
                 random.randint(120, 260), random.randint(20, 90),
                 simulated=True)


class _Mem:
    def __init__(self, used, total, percent):
        self.used = used
        self.total = total
        self.percent = percent


class _Disk:
    def __init__(self, percent, used_gb, total_gb):
        self.percent = percent
        self.used_gb = used_gb
        self.total_gb = total_gb


def _save(now, hostname, cpu, load1, mem, du, sent_kbps, recv_kbps,
          proc_count, tcp, simulated):
    from .models import HostMetric
    return HostMetric.objects.create(
        hostname=hostname[:128],
        cpu_percent=round(cpu, 1),
        cpu_cores=os.cpu_count() or 0,
        load_avg=round(float(load1), 2),
        mem_percent=round(mem.percent, 1),
        mem_used_mb=round(mem.used / 1024 / 1024, 1),
        mem_total_mb=round(mem.total / 1024 / 1024, 1),
        disk_percent=round(du.percent, 1),
        disk_used_gb=round(du.used / 1024 / 1024 / 1024, 2),
        disk_total_gb=round(du.total / 1024 / 1024 / 1024, 2),
        net_sent_kbps=round(sent_kbps, 1),
        net_recv_kbps=round(recv_kbps, 1),
        proc_count=proc_count,
        tcp_conns=tcp,
        simulated=simulated,
        created_at=now,
    )


def _prune():
    """清理超过保留期的主机采集数据"""
    from .models import HostMetric
    deadline = timezone.now() - timedelta(days=MON['RETENTION_DAYS'])
    HostMetric.objects.filter(created_at__lt=deadline).delete()


_last_prune = [0.0]  # 清理节拍（可变容器，便于测试里重置）


def host_collect_round():
    """一轮本地采集：采样 + 低频清理。异常不在这里吞（LeaseLoop 统一记录并保住循环）"""
    m = sample_once()
    if time.time() - _last_prune[0] > 3600:
        _prune()
        _last_prune[0] = time.time()


def host_collector_loop():
    """后台线程主循环：只有持有租约的进程采集（多副本 worker 不会重复采同一台机器）"""
    from monitor.leadership import LeaseLoop

    LeaseLoop('host-collector', host_collect_round, MON['HOST_INTERVAL_SEC']).run()

"""
monitor/leadership.py — 后台任务租约选主（多副本安全）

用一张租约表（monitor.TaskLease）选主：
- 每个任务一行，进程身份是 "主机名:pid:随机后缀"；
- 抢约/续约都是一条 compare-and-swap 条件 UPDATE，SQLite 的单写者与 Postgres 的
  行锁都保证竞争者中只有一个命中，其余更新 0 行即落选；
- 租约到期自动失效：持有进程被杀/宕机，其它进程在下一次抢约时接管，无需人工介入；
- term（任期）每次易主 +1，便于识别"我是不是已经被取代了"。

边界（诚实版）：这是租约不是围栏（fencing）。旧持有者若在续约间隙里正跑着一轮长任务，
它仍可能把那轮做完——最坏情况是"一个周期内重复执行一次"。四类周期任务都是幂等写
（告警事件按策略去重合并、采集点多一条只是曲线略密），所以可接受。
"""
import atexit
import logging
import os
import socket
import threading
import time
import uuid
from datetime import timedelta

from django.db import IntegrityError, transaction
from django.utils import timezone

logger = logging.getLogger(__name__)

# 租约默认 60 秒：远大于最快任务周期（拨测 5s），又不至于故障后要等太久才接管
DEFAULT_TTL_SEC = max(5, int(os.environ.get('OBS_LEASE_TTL_SEC', '60')))

_holder = None
_holder_lock = threading.Lock()


def holder():
    """本进程身份：主机名:pid:随机后缀（同机多进程、容器重启都能区分开）"""
    global _holder
    if _holder is None:
        with _holder_lock:
            if _holder is None:
                host, pid, tag = socket.gethostname()[:40], os.getpid(), uuid.uuid4().hex[:8]
                _holder = f'{host}:{pid}:{tag}'
    return _holder


def _model():
    from .models import TaskLease
    return TaskLease


def acquire(name, holder_id=None, ttl=None):
    """抢占或续约任务租约，返回 (是否持有, 任期)。

    三步都是单条条件 UPDATE/INSERT：
    1. 建行：首次使用时没有这一行，直接带自己为持有者插入；撞唯一约束就说明
       别的进程抢先建了，本轮落选；
    2. 续约：已经是我的 → 只推到期时间，任期不变；
    3. 抢占：CAS，条件同时写上"持有者与我观察到的一致"和"确实已过期"；
       竞争失败者条件不再成立 → 更新 0 行 → 落选。
    """
    me = holder_id or holder()
    ttl = int(ttl or DEFAULT_TTL_SEC)
    TaskLease = _model()
    now = timezone.now()

    lease = TaskLease.objects.filter(name=name).first()
    if lease is None:
        try:
            with transaction.atomic():
                TaskLease.objects.create(
                    name=name, holder=me, term=1,
                    acquired_at=now, renewed_at=now, expires_at=now + timedelta(seconds=ttl))
            _log_change(name, me, 1, '取得')
            return True, 1
        except IntegrityError:
            return False, 0  # 别的进程同时建了行，本轮落选，下一轮走续约/抢占

    if lease.holder == me:
        updated = TaskLease.objects.filter(name=name, holder=me).update(
            renewed_at=now, expires_at=now + timedelta(seconds=ttl))
        if updated:
            return True, lease.term
        return False, lease.term  # 持有者已被改写（续约间隙里被人抢走）

    if lease.holder and lease.expires_at and lease.expires_at > now:
        return False, lease.term  # 明确还在别人手里，不必尝试 CAS

    # 抢占：CAS 条件必须把"我观察到的持有者 + 已过期"一起写上，
    # 竞争失败者的条件不再成立 → 更新 0 行 → 落选
    qs = TaskLease.objects.filter(name=name, holder=lease.holder)
    if lease.expires_at is None:
        qs = qs.filter(expires_at__isnull=True)
    else:
        qs = qs.filter(expires_at__lte=now)
    taken = qs.update(holder=me, term=lease.term + 1, acquired_at=now,
                      renewed_at=now, expires_at=now + timedelta(seconds=ttl))
    if taken:
        _log_change(name, me, lease.term + 1, '接管')
        return True, lease.term + 1
    return False, lease.term  # CAS 未命中：已有更近的持有者


def release(name, holder_id=None):
    """主动交出租约（正常退出时调用，让对端立刻接管而不是等 TTL 过期）"""
    me = holder_id or holder()
    TaskLease = _model()
    n = TaskLease.objects.filter(name=name, holder=me).update(
        holder='', expires_at=timezone.now(), renewed_at=timezone.now())
    if n:
        logger.info('已交还任务租约 %s（%s）', name, me)
    return bool(n)


def status():
    """全集群租约一览（给 /metrics 与运维页看）"""
    TaskLease = _model()
    now = timezone.now()
    out = []
    for lease in TaskLease.objects.order_by('name'):
        held = bool(lease.holder) and lease.expires_at is not None and lease.expires_at > now
        out.append({
            'name': lease.name,
            'holder': lease.holder or '-',
            'mine': lease.holder == holder(),
            'leading': held,
            'term': lease.term,
            'ttl_sec': round((lease.expires_at - now).total_seconds(), 1) if held else 0,
        })
    return out


_logged = {}


def _log_change(name, me, term, verb):
    """接管/取得只在任期变化时打日志，避免续约刷屏"""
    key = f'{name}:{term}'
    if _logged.get(name) == key:
        return
    _logged[name] = key
    logger.info('任务租约 %s：%s 已%s（term=%d）', name, me, verb, term)


class LeaseLoop:
    """带租约的周期任务骨架：只有持有租约的进程执行 work()，其余热待命。

    续约节拍与任务节拍分开：tick 间隔取 min(interval, ttl/3)，保证远快于租约到期，
    而 work 的实际执行由 next_run 控制——这样 5 秒一次的拨测和 12 小时一次的巡检
    共用同一套骨架。work() 返回数字时作为"下一轮等多久"。
    """

    def __init__(self, name, work, interval, ttl=None, sleep=time.sleep, holder_id=None):
        self.name = name
        self.work = work
        self.interval = float(interval)
        self.ttl = int(ttl or DEFAULT_TTL_SEC)
        self.holder_id = holder_id
        self.sleep = sleep
        self.tick_interval = max(1.0, min(self.interval, self.ttl / 3))
        self.next_run = 0.0
        self.owned = False
        self.term = 0

    def tick(self, now=None):
        """续约一次；到点且持有租约时执行一轮。返回本轮是否真的执行了 work。"""
        now = time.monotonic() if now is None else now
        owned, term = acquire(self.name, holder_id=self.holder_id, ttl=self.ttl)
        self.owned, self.term = owned, term
        if not owned or now < self.next_run:
            return False
        try:
            result = self.work()
        except Exception:  # 任务异常不能带走循环——否则平台自己瞎了
            logger.exception('周期任务 %s 执行异常', self.name)
            result = None
        # work 返回数字即"下轮等待秒数"；bool 是返回值陷阱，不当作间隔
        delay = (float(result) if isinstance(result, (int, float)) and not isinstance(result, bool)
                 else self.interval)
        self.next_run = now + max(0.1, delay)
        return True

    def run(self):
        while True:
            self.tick()
            self.sleep(self.tick_interval)


def register_exit_release(names):
    """进程退出时交还自己持有的租约（对端不必等满 TTL 才能接管）"""
    names = tuple(names)

    def _release_all():
        for name in names:
            try:
                release(name)
            except Exception:
                logger.exception('退出时交还租约 %s 失败', name)

    atexit.register(_release_all)

def start_workers():
    """启动：主机采集 / 告警评估 / 拨测 / 定时巡检 线程（幂等、线程安全）"""
    global _started
    if _started:
        return
    with _start_lock:
        if _started:
            return
        from hosts.collector import host_collector_loop
        # from alerts.engine import alert_engine_loop      # 08 篇启用
        # from ops.workers import probe_loop, inspect_loop # 10 篇启用

        threads = (
            ('obs-host-collector', host_collector_loop, 'host-collector'),
        )
        started = 0
        leases = []
        for name, target, lease_name in threads:
            try:
                threading.Thread(target=target, name=name, daemon=True).start()
                started += 1
                leases.append(lease_name)
            except Exception:
                logger.exception('后台线程 %s 启动失败', name)
        _started = True
        logger.info('后台线程已启动：%d/%d（进程身份 %s）',
                    started, len(threads), leadership.holder())
        # 正常退出时交还租约，让其它副本立即接管而不是等 TTL 过期
        leadership.register_exit_release(leases)


def maybe_start_workers():
    # 请求指标缓冲线程在 Web 进程里也必须起：指标由 Web 进程产生
    from . import buffer
    try:
        buffer.start()
    except Exception:
        logger.exception('请求指标缓冲线程启动失败')
    if _should_start():
        try:
            start_workers()
        except Exception:
            logger.exception('后台线程启动失败')

python manage.py makemigrations hosts
python manage.py migrate
python manage.py runserver 8014

# python manage.py shell
from hosts.models import HostMetric
print(HostMetric.objects.count())          # > 0，每 15 秒 +1
print(HostMetric.objects.order_by('-id').values('hostname','cpu_percent','simulated').first())

# ---------------------------------------------------------------------------
# 主机指标（hosts.HostMetric）
# ---------------------------------------------------------------------------

def _host_series(field, minutes, scale=1.0):
    """主机指标序列：多主机时取各主机分钟峰值的最大值（单主机等价于该机读数）"""
    from django.db.models import Max

    from hosts.models import HostMetric
    pts = _bucket(
        HostMetric.objects.filter(created_at__gte=_since(minutes)),
        {'v': Max(field)},
    )
    return [{'t': _fmt(t), 'v': round(float(r['v'] or 0) * scale, 1)} for t, r in pts]


def host_cpu(minutes):
    return _host_series('cpu_percent', minutes)


def host_mem(minutes):
    return _host_series('mem_percent', minutes)


def host_disk(minutes):
    return _host_series('disk_percent', minutes)


def host_net_recv(minutes):
    return _host_series('net_recv_kbps', minutes)


def host_net_sent(minutes):
    return _host_series('net_sent_kbps', minutes)

    'host.cpu_percent': ('主机 CPU 使用率', '%', host_cpu),
    'host.mem_percent': ('主机内存使用率', '%', host_mem),
    'host.disk_percent': ('主机磁盘使用率', '%', host_disk),
    'host.net_recv_kbps': ('下行流量', 'KB/s', host_net_recv),
    'host.net_sent_kbps': ('上行流量', 'KB/s', host_net_sent),

from monitor.registry import series
print(series('host.cpu_percent', 30))

# 接入令牌可访问的路径：仅"写数据"的上报端点 + Prometheus 抓取。
# 读类数据 API（日志检索/导出、APM 查询等）一律只认登录会话——
# 令牌分发在各 Agent 主机上，泄露半径必须限制为"只能写监控数据"。
TOKEN_PATHS = (
    '/api/ingest/',      # 自定义指标 / 主机 Agent 上报
    '/logs/api/ingest/', # 日志接入
    '/rum/beacon/',      # 前端 RUM 上报
)
TOKEN_EXACT_PATHS = ('/metrics',)


def _token_ok(request):
    """校验请求携带的接入令牌（常量时间比较，防时序侧信道）"""
    expected = settings.OBSERVABILITY.get('INGEST_TOKEN') or ''
    if not expected:
        return False
    auth = request.META.get('HTTP_AUTHORIZATION', '')
    if auth.startswith('Bearer ') and secrets.compare_digest(
            auth[len('Bearer '):].strip(), expected):
        return True
    if secrets.compare_digest(request.headers.get('X-OBS-Token', ''), expected):
        return True
    # 兼容旧文档/旧 Agent 的 ?token= 查询串（会进访问日志，不推荐）
    if secrets.compare_digest(request.GET.get('token', ''), expected):
        return True
    return False


def _path_is_token_api(path):
    """判断路径是否属于令牌可访问的上报端点（精确白名单，防令牌读数据）"""
    if any(path == p or path.startswith(p + '/')
           for p in TOKEN_EXACT_PATHS):
        return True
    return any(path.startswith(p) for p in TOKEN_PATHS)


def ingest_allowed(request):
    """上报接口的准入判断：完整账号会话 或 令牌，二者其一。"""
    user = getattr(request, 'user', None)
    if user is not None and user.is_authenticated:
        return True
    return _token_ok(request)


def require_ingest(view):
    """上报类接口装饰器：未登录且未携带令牌 -> 401"""
    @wraps(view)
    def wrapped(request, *args, **kwargs):
        if not ingest_allowed(request):
            return JsonResponse(
                {'ok': False,
                 'error': 'unauthorized: 需要登录会话，或携带接入令牌'
                          '（X-OBS-Token 头 / Authorization: Bearer）'},
                status=401,
            )
        return view(request, *args, **kwargs)
    return wrapped


def rate_limit(prefix, rate, per=60):
    """按客户端 IP 的固定窗口限速装饰器（进程内 cache 实现）。

    rate 次 / per 秒，超限返回 429。
    """
    from django.core.cache import cache

    def decorator(view):
        @wraps(view)
        def wrapped(request, *args, **kwargs):
            ip = (request.META.get('REMOTE_ADDR') or 'unknown')[:40]
            key = f'obs-rl:{prefix}:{ip}'
            now = int(time.time())
            window = now // per
            bucket_key = f'{key}:{window}'
            try:
                count = cache.get(bucket_key)
                if count is None:
                    cache.add(bucket_key, 1, per + 1)
                    count = 1
                else:
                    count = cache.incr(bucket_key)
            except Exception:
                count = 0  # cache 不可用时不阻断业务
            if count > rate:
                return JsonResponse(
                    {'ok': False, 'error': f'rate limited: 最多 {rate} 次/{per} 秒'},
                    status=429,
                )
            return view(request, *args, **kwargs)
        return wrapped
    return decorator

        # 携带有效令牌且目标确为上报端点时放行（读类 API 与页面路径不认令牌）
        if _path_is_token_api(path) and _token_ok(request):
            return self.get_response(request)

"""
hosts/views.py — 主机页与 Agent 上报接口
"""
import json

from django.http import JsonResponse
from django.utils import timezone
from django.views.decorators.http import require_POST

from monitor.security import rate_limit, require_ingest

from .models import HostMetric

# 数值字段统一校验：NaN/Infinity/超界会污染聚合甚至 500
_NUM_FIELDS = {
    'cpu_percent': (0, 100), 'load_avg': (0, 10000),
    'mem_percent': (0, 100), 'mem_used_mb': (0, 1e8), 'mem_total_mb': (0, 1e8),
    'disk_percent': (0, 100), 'disk_used_gb': (0, 1e6), 'disk_total_gb': (0, 1e6),
    'net_sent_kbps': (0, 1e9), 'net_recv_kbps': (0, 1e9),
}
_INT_FIELDS = {'cpu_cores': (0, 100000), 'proc_count': (0, 100000), 'tcp_conns': (0, 1e7)}


def _clean_numbers(payload):
    """校验并收敛数值字段；非法返回 None"""
    import math
    out = {}
    for field, (lo, hi) in _NUM_FIELDS.items():
        try:
            v = float(payload.get(field, 0))
        except (TypeError, ValueError):
            v = 0.0
        if math.isnan(v) or math.isinf(v) or v < lo or v > hi:
            return None
        out[field] = round(v, 2)
    for field, (lo, hi) in _INT_FIELDS.items():
        try:
            v = int(float(payload.get(field, 0)))
        except (TypeError, ValueError):
            v = 0
        if v < lo or v > hi:
            return None
        out[field] = v
    return out


@require_POST
@rate_limit('host-ingest', 120)
@require_ingest
def api_ingest_host(request):
    """Agent 主机指标上报端点：令牌保护 + 数值校验 + 演示账号拒绝"""
    try:
        payload = json.loads(request.body.decode('utf-8') or '{}')
        if not isinstance(payload, dict):
            return JsonResponse({'ok': False, 'error': 'payload 必须是 JSON 对象'}, status=400)
    except (json.JSONDecodeError, UnicodeDecodeError):
        return JsonResponse({'ok': False, 'error': 'invalid json'}, status=400)

    nums = _clean_numbers(payload)
    if nums is None:
        return JsonResponse({'ok': False, 'error': '数值字段非法（NaN/超界）'}, status=400)

    HostMetric.objects.create(
        hostname=str(payload.get('hostname') or 'unknown')[:128],
        **nums,
        created_at=timezone.now(),
    )
    return JsonResponse({'ok': True})

from django.urls import path

from . import views

urlpatterns = [
    path('api/ingest/host/', views.api_ingest_host, name='host_ingest'),
]

    # 上报 OpenAPI（远程主机 Agent）
    path('api/ingest/host/', hosts_views.api_ingest_host, name='host_ingest'),

#!/usr/bin/env python3
"""
obs_agent.py — 全栈可观测运维平台 · 主机采集 Agent

放在任意要监控的服务器上运行，每 N 秒采集一次本机 psutil 指标，
推送到平台的 /api/ingest/host/ 端点（令牌保护）。仅依赖 psutil，
HTTP 用标准库 urllib，无需安装任何其他包。

用法：
    pip install psutil
    set OBS_SERVER_URL=http://平台IP:8014        (Windows) / export（Linux/macOS）
    set OBS_INGEST_TOKEN=<接入令牌>
    python obs_agent.py
"""
import json
import os
import socket
import sys
import time
import urllib.error
import urllib.request
from urllib.parse import urlsplit

try:
    import psutil
except ImportError:
    print('[obs-agent] 缺少依赖：pip install psutil')
    sys.exit(1)


def _resolve_server():
    """启动时校验平台地址：仅允许 http/https 且必须带主机名（防误配）"""
    raw = os.environ.get('OBS_SERVER_URL', 'http://127.0.0.1:8014')
    parts = urlsplit(raw)
    if parts.scheme not in ('http', 'https'):
        raise SystemExit(f'[obs-agent] OBS_SERVER_URL 仅允许 http/https，当前: {raw!r}')
    if not parts.hostname:
        raise SystemExit(f'[obs-agent] OBS_SERVER_URL 缺少主机名: {raw!r}')
    if parts.path not in ('', '/'):
        raise SystemExit(f'[obs-agent] OBS_SERVER_URL 不应包含路径: {raw!r}')
    netloc = parts.hostname
    if ':' in netloc and not netloc.startswith('['):
        netloc = f'[{netloc}]'  # IPv6 字面量
    if parts.port:
        netloc = f'{netloc}:{parts.port}'
    if parts.username or parts.password:
        raise SystemExit('[obs-agent] OBS_SERVER_URL 不应包含用户名/密码，请用 OBS_INGEST_TOKEN')
    return f'{parts.scheme}://{netloc}'


SERVER = _resolve_server()
ENDPOINT = SERVER + '/api/ingest/host/'   # 唯一上报地址（固定路径，不含任何外部输入）
TOKEN = os.environ.get('OBS_INGEST_TOKEN', '')
try:
    INTERVAL = int(float(os.environ.get('OBS_AGENT_INTERVAL', '15')))
except (TypeError, ValueError):
    INTERVAL = 15
INTERVAL = max(5, min(600, INTERVAL))   # 钳制：0/负数会打爆服务端
try:
    HOSTNAME = os.environ.get('OBS_HOST_NAME') or socket.gethostname() or 'unknown'
except Exception:
    HOSTNAME = 'unknown'


def _log(msg):
    print(f'[obs-agent {time.strftime("%H:%M:%S")}] {msg}', flush=True)


# psutil 的 cpu_percent 首次调用恒为 0，先预热一次
psutil.cpu_percent(interval=None)
_disk_root = os.path.abspath(os.sep)
_last_net = {'sent': None, 'recv': None, 'ts': None}


def sample():
    """采集一次本机指标，返回与平台 HostMetric 字段一致的 dict"""
    # ……与 hosts/collector._sample_real 相同的采集逻辑（cpu/mem/disk/net/proc/tcp），
    # 唯一区别：不落库，返回 dict。完整实现见仓库 agent/obs_agent.py 的 sample()
    ...


def push(payload):
    body = json.dumps(payload).encode('utf-8')
    req = urllib.request.Request(
        ENDPOINT,
        data=body,
        headers={'Content-Type': 'application/json', 'X-OBS-Token': TOKEN},
        method='POST',
    )
    with urllib.request.urlopen(req, timeout=10) as resp:
        data = json.loads(resp.read().decode('utf-8'))
        if not data.get('ok'):
            raise RuntimeError(data)


def main():
    if not TOKEN:
        _log('未设置 OBS_INGEST_TOKEN（平台根目录 ingest_token.txt）')
        sys.exit(1)
    _log(f'启动：host={HOSTNAME} -> {SERVER}，间隔 {INTERVAL}s')
    fail = 0
    while True:
        try:
            push(sample())
            if fail:
                _log('恢复上报')
                fail = 0
        except urllib.error.HTTPError as e:
            fail += 1
            # 4xx 是配置错误（令牌/地址），重试没有意义；5xx/网络错误才值得退避重试
            if 400 <= e.code < 500:
                _log(f'上报被拒绝（HTTP {e.code}）：请检查接入令牌 OBS_INGEST_TOKEN 是否正确')
                sys.exit(1)
            if fail <= 3 or fail % 20 == 0:
                _log(f'上报失败（第 {fail} 次）：HTTP {e.code}')
        except Exception as e:
            fail += 1
            if fail <= 3 or fail % 20 == 0:
                _log(f'上报失败（第 {fail} 次）：{e}')
        # 指数退避：连续失败时 15s -> 30s -> 60s -> ... 上限 5 分钟
        time.sleep(min(INTERVAL * (2 ** min(fail, 5)), 300))


if __name__ == '__main__':
    main()

# Windows：把令牌读进环境变量（token 文件在项目根）
$env:OBS_INGEST_TOKEN = (Get-Content .\ingest_token.txt -Raw).Trim()
$env:OBS_HOST_NAME = "test-agent"
python .\agent\obs_agent.py

# macOS / Linux
export OBS_INGEST_TOKEN=$(cat ingest_token.txt)
export OBS_HOST_NAME="test-agent"
python3 agent/obs_agent.py

from hosts.models import HostMetric
print(HostMetric.objects.values('hostname').distinct())   # 两个主机名：本机 + test-agent

# 匿名推送 → 401
curl.exe -s -X POST http://127.0.0.1:8014/api/ingest/host/ -H "Content-Type: application/json" -d "{}"
# 错误令牌 → 401
curl.exe -s -X POST http://127.0.0.1:8014/api/ingest/host/ -H "X-OBS-Token: wrong" -d "{}"
# 正确令牌 → {"ok": true}
curl.exe -s -X POST http://127.0.0.1:8014/api/ingest/host/ -H "X-OBS-Token: <你的令牌>" -H "Content-Type: application/json" -d "{\"hostname\":\"curl-test\",\"cpu_percent\":12.5}"