机房物联网采集：POE 温湿度传感器 UDP 数据包抓包与丢包排查实战

应用层：采集进程 socket 未读走 / 处理逻辑阻塞 / 缓冲满丢弃
────────────────────────────────────────────────────
传输层：UDP 无重传，socket 接收缓冲区溢出（rmem）、recvmsg 不及时
────────────────────────────────────────────────────
网络层：路由/ACL/多播/源地址校验、conntrack 误丢、RPFilter
────────────────────────────────────────────────────
链路层：PoE 端口丢包、交换机缓冲拥塞、VLAN 错放、镜像口截断
────────────────────────────────────────────────────
物理层：PoE 协商降额、线缆、端口错连、双工不匹配

# 采集端监听 UDP 端口
ss -lunp | grep 9000
# 或
netstat -lunp | grep 9000

# 确认设备配置的上报目的 IP/端口（Web/CLI/配置工具）
# 同时确认源端口随机还是固定，影响抓包过滤

# 主机侧抓，保留完整载荷
sudo tcpdump -i eth0 -nn -vv 'udp port 9000' -s0 -B 4096 -w /tmp/udp_cap.pcap

# 同时抓 ICMP（看分片/不可达）
sudo tcpdump -i eth0 -nn '(udp port 9000) or icmp' -s0 -w /tmp/udp_cap.pcap

# 多接口/VLAN 时指定
sudo tcpdump -i any -nn 'vlan and udp port 9000' -s0 -w /tmp/udp_cap.pcap

udp.dstport == 9000
ip.src == 10.20.1.11
udp.length < 64   # 异常小包

# 接收队列是否有积压
cat /proc/net/udp | awk 'NR>1{print $2,$4,$5}'   # hex lport, rx_queue, drops
# 或更直观
ss -lunpem

# nstat / netstat -su
nstat -az | grep -E 'Udp|RcvbufErrors|InErrors|Drops'
netstat -su | grep -E 'receive errors|RcvbufErrors|dropped'

show interfaces counters errors
show poe power-usage
show mac address-table | include <sensor_mac>
show spanning-tree inconsistentports

"""
udp_collector.py - PoE 温湿度传感器 UDP 上报采集
- SO_RCVBUF 调大、非阻塞事件循环批量 recv
- 解析私有帧（示例：头0xAA 0x55 + dev_id + t + h）
- 统计丢包：按设备上报序号缺口 + 周期超时未到
"""

import asyncio, struct, logging, time
from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS

HOST = "0.0.0.0"
PORT = 9000
INFLUX = {"url":"http://localhost:8086","token":"","org":"dc","bucket":"env"}

logging.basicConfig(level=logging.INFO)

def parse_frame(data: bytes):
    if len(data) < 12 or data[0]!=0xAA or data[1]!=0x55:
        return None
    dev_id, seq = struct.unpack_from("<HH", data, 2)
    t_raw, h_raw = struct.unpack_from("<hh", data, 6)
    t = t_raw/10.0; h = h_raw/10.0
    return {"dev_id":dev_id,"seq":seq,"t":t,"h":h}

class Collector:
    def __init__(self):
        self.transport = None
        self.last_seq = {}      # dev_id -> last seq
        self.last_ts = {}       # dev_id -> last recv mono time
        self.expected_period = 5.0
        self.influx = InfluxDBClient(**INFLUX) if False else \
            InfluxDBClient(url=INFLUX["url"],token=INFLUX["token"],org=INFLUX["org"])
        self.wapi = self.influx.write_api(write_options=SYNCHRONOUS)
        self.rx = 0
        self.app_drop = 0

    def connection_made(self, transport):
        self.transport = transport
        sock = transport.get_extra_info("socket")
        if sock:
            sock.setsockopt(__import__("socket").SOL_SOCKET,
                           __import__("socket").SO_RCVBUF, 4<<20)

    def datagram_received(self, data, addr):
        self.rx += 1
        pkt = parse_frame(data)
        if pkt is None:
            logging.debug("bad frame from %s", addr)
            return
        dev = pkt["dev_id"]
        # 序号连续性检查
        ls = self.last_seq.get(dev)
        if ls is not None and pkt["seq"] != (ls+1) & 0xFFFF:
            gap = (pkt["seq"] - ls) & 0xFFFF
            if gap != 1:
                logging.warning("dev %s seq gap: last=%s got=%s", dev, ls, pkt["seq"])
        self.last_seq[dev] = pkt["seq"]
        self.last_ts[dev] = time.monotonic()
        p = Point("env").tag("dev", str(dev)).field("temperature",pkt["t"]).field("humidity",pkt["h"]).field("quality",1)
        try:
            self.wapi.write(bucket=INFLUX["bucket"],org=INFLUX["org"],record=p)
        except Exception as e:
            self.app_drop += 1
            logging.warning("write fail: %s", e)

    def poll_watchdog(self):
        # 周期外未收到即判丢
        now = time.monotonic()
        for dev, ts in self.last_ts.items():
            if now - ts > self.expected_period*3:
                p = Point("env").tag("dev",str(dev)).field("quality",0).field("stale",1)
                self.wapi.write(bucket=INFLUX["bucket"],org=INFLUX["org"],record=p)
                logging.warning("dev %s stale", dev)

async def main():
    loop = asyncio.get_event_loop()
    collector = Collector()
    transport, proto = await loop.create_datagram_endpoint(
        lambda: collector, local_addr=(HOST,PORT))
    try:
        while True:
            await asyncio.sleep(1)
    finally:
        transport.close()

if __name__ == "__main__":
    asyncio.run(main())

# 接收缓冲上限
sysctl -w net.core.rmem_max=16777216
sysctl -w net.core.rmem_default=4194304
# UDP 特定
sysctl -w net.ipv4.udp_rmem_min=16384
# 中断/RPS
irqbalance off + 手动绑核；多队列网卡 RPS/RFS
# RPFilter
sysctl -w net.ipv4.conf.all.rp_filter=2   # loose，或按环境设0/1
# conntrack 对 UDP 短时流无妨，但别让 NOTRACK 缺失导致误丢

// 概念参考，Python 侧已示
setsockopt(SOL_SOCKET, SO_RCVBUF, 4MB);
// 更大后端缓冲：应用层环形队列 + 批处理

# 用 scapy/nping 模拟设备侧多发
nping --udp -p 9000 --rate 500 --count 10000 10.20.1.250
# 或自研发送端回放 pcap
tcpreplay-edit --intf1=eth0 --portmap=9000:9000 /tmp/udp_cap.pcap

发送端日志/计数器 N_send
交换机端口出方向计数
采集端 tcpdump 计数 (tcpdump -i eth0 -nn 'udp port 9000' | wc -l 不准确，用 capinfos)
应用层 rx 计数
平台侧入库点数

# 内核 drop 点跟踪（bcc/bioprobe）
biosnoop / funccount 'udp_rcv'
# 或简单看 tracepoint
perf trace -e 'net:*' -p $(pidof collector)