feat(snmp): 实现网络设备自动轮询定时任务
之前只有 arp/mac/interface_poll_interval 字段但没有实际调度, SNMP 设备添加后永远不会被自动采集。 新增 poll_snmp_devices Celery 定时任务: - 遍历所有活跃且配置了 SNMP 凭据的设备 - 按各设备的 poll_interval 判断是否到期,到期才轮询 - 用 asyncio.run 执行异步的 SNMPService.poll_device - Beat 每 60 秒触发检查,新设备 last_polled_at 为空会立即轮询
This commit is contained in:
@@ -32,6 +32,7 @@ scan_single_ip_task = _tasks['scan_single_ip_task']
|
||||
full_network_scan = _tasks['full_network_scan']
|
||||
update_all_statistics = _tasks['update_all_statistics']
|
||||
quick_status_update = _tasks['quick_status_update']
|
||||
poll_snmp_devices = _tasks['poll_snmp_devices']
|
||||
|
||||
# 定时任务配置
|
||||
celery_app.conf.beat_schedule = {
|
||||
@@ -52,4 +53,9 @@ celery_app.conf.beat_schedule = {
|
||||
'task': 'app.tasks.scan_tasks.update_all_statistics',
|
||||
'schedule': 900.0, # 15分钟
|
||||
},
|
||||
# 每60秒检查一次 SNMP 设备轮询(各设备按自己的 poll_interval 判断是否到期)
|
||||
'poll-snmp-devices-every-60s': {
|
||||
'task': 'app.tasks.scan_tasks.poll_snmp_devices',
|
||||
'schedule': 60.0,
|
||||
},
|
||||
}
|
||||
|
||||
@@ -212,11 +212,75 @@ def register_tasks(celery_app):
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
@celery_app.task(bind=True)
|
||||
def poll_snmp_devices(self):
|
||||
"""
|
||||
定时任务:自动轮询所有活跃且配置了 SNMP 凭据的网络设备。
|
||||
|
||||
每个设备按自己的 arp/mac/interface_poll_interval(秒)判断是否到达
|
||||
轮询间隔,到期才执行 SNMPService.poll_device(采集 ARP 表、MAC 地址
|
||||
表、接口信息)。由 Celery Beat 每 60 秒触发一次检查。
|
||||
"""
|
||||
import asyncio
|
||||
from datetime import datetime
|
||||
|
||||
from app.models.snmp import NetworkDevice
|
||||
from app.services.snmp_service import SNMPService
|
||||
|
||||
db = SessionLocal()
|
||||
try:
|
||||
devices = db.query(NetworkDevice).filter(
|
||||
NetworkDevice.is_active == True,
|
||||
NetworkDevice.snmp_credential_id.isnot(None),
|
||||
).all()
|
||||
|
||||
now = datetime.utcnow()
|
||||
|
||||
def _as_naive(dt):
|
||||
"""把 aware/naive datetime 统一转 naive(本地比较用)"""
|
||||
if dt is None:
|
||||
return None
|
||||
if dt.tzinfo is not None:
|
||||
return dt.replace(tzinfo=None)
|
||||
return dt
|
||||
|
||||
polled, skipped = 0, 0
|
||||
for dev in devices:
|
||||
# 取三种采集间隔的最小值作为该设备的轮询周期
|
||||
intervals = [
|
||||
dev.arp_poll_interval,
|
||||
dev.mac_poll_interval,
|
||||
dev.interface_poll_interval,
|
||||
]
|
||||
interval = min([i for i in intervals if i], default=300)
|
||||
|
||||
last = _as_naive(dev.last_polled_at)
|
||||
if last is not None:
|
||||
elapsed = (now - last).total_seconds()
|
||||
if elapsed < interval:
|
||||
skipped += 1
|
||||
continue
|
||||
|
||||
try:
|
||||
# poll_device 是 async 函数,用 asyncio.run 在同步任务中执行
|
||||
asyncio.run(SNMPService.poll_device(db, dev.id))
|
||||
polled += 1
|
||||
logger.info(f"SNMP 自动轮询完成: {dev.name} ({dev.ip_address})")
|
||||
except Exception as e:
|
||||
logger.error(f"SNMP 自动轮询失败 {dev.name} ({dev.ip_address}): {e}", exc_info=True)
|
||||
|
||||
logger.info(f"SNMP 自动轮询检查完成: 设备总数 {len(devices)}, 本轮轮询 {polled}, 未到期跳过 {skipped}")
|
||||
return {'status': 'completed', 'total': len(devices), 'polled': polled, 'skipped': skipped}
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
# 返回任务函数供外部使用
|
||||
return {
|
||||
'scan_network_task': scan_network_task,
|
||||
'scan_single_ip_task': scan_single_ip_task,
|
||||
'full_network_scan': full_network_scan,
|
||||
'update_all_statistics': update_all_statistics,
|
||||
'quick_status_update': quick_status_update
|
||||
'quick_status_update': quick_status_update,
|
||||
'poll_snmp_devices': poll_snmp_devices,
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user