aiops/backend/controller/snmp_collect.go

package controller

import (
	"context"
	"fmt"
	"log"
	"net/http"
	"strconv"
	"time"

	"ezops/model"
	snmppkg "ezops/pkg/snmp"

	"github.com/gin-gonic/gin"
	"github.com/gosnmp/gosnmp"
)

// collectDevice 采集单台 SNMP 设备并落库 // Logan.Li
func (h *SNMPHandler) collectDevice(device *model.SNMPDevice) error {
	if device.TemplateID == 0 {
		return fmt.Errorf("未绑定 OID 模板")
	}
	var oids []model.SNMPTemplateOID
	h.db.Where("template_id = ?", device.TemplateID).Order("sort_order ASC").Find(&oids)
	if len(oids) == 0 {
		return fmt.Errorf("模板无 OID 定义")
	}

	col, err := h.newCollector(device)
	if err != nil {
		return err
	}

	defs := make([]snmppkg.OIDDefinition, 0, len(oids))
	for _, o := range oids {
		defs = append(defs, snmppkg.OIDDefinition{
			Name: o.Name, OID: o.OID, Unit: o.Unit,
			DataType: o.DataType, IsTable: o.Conversion == "table",
		})
	}

	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(device.Timeout)*time.Second)
	defer cancel()
	result := col.Collect(ctx, defs)

	now := time.Now()
	ts := now.Unix()
	updates := map[string]interface{}{
		"last_collect_at": now,
	}
	if result.Success {
		updates["status"] = model.SNMPDeviceOnline
		updates["last_error"] = ""
		for _, m := range result.Metrics {
			h.db.Create(&model.SNMPMetric{
				DeviceID: device.ID, Name: m.Name, OID: m.OID, Index: m.Index,
				Value: m.Value, RawValue: m.RawValue, Unit: m.Unit,
				DataType: m.DataType, Timestamp: ts,
			})
		}
	} else {
		updates["status"] = model.SNMPDeviceError
		errMsg := result.Error
		if errMsg == "" {
			errMsg = "采集无数据"
		}
		updates["last_error"] = errMsg
	}
	h.db.Model(device).Updates(updates)
	if !result.Success {
		return fmt.Errorf("%s", updates["last_error"])
	}
	return nil
}

func (h *SNMPHandler) newCollector(device *model.SNMPDevice) (*snmppkg.Collector, error) {
	timeout := device.Timeout
	if timeout <= 0 {
		timeout = 10
	}
	retries := device.Retries
	if retries <= 0 {
		retries = 2
	}
	port := uint16(device.Port)
	if port == 0 {
		port = 161
	}
	cfg := &snmppkg.CollectorConfig{
		IP: device.IP, Port: port, Community: device.Community,
		Timeout: time.Duration(timeout) * time.Second, Retries: retries,
		MaxRepetitions: uint8(device.MaxRepetitions),
	}
	switch device.Version {
	case model.SNMPVersionV1:
		cfg.Version = gosnmp.Version1
	case model.SNMPVersionV3:
		cfg.Version = gosnmp.Version3
	default:
		cfg.Version = gosnmp.Version2c
	}
	return snmppkg.NewCollector(cfg)
}

// deviceCollectInterval 设备采集间隔，不低于全局最低值 // Logan.Li
func deviceCollectInterval(device *model.SNMPDevice) int {
	interval := device.CollectInterval
	if interval <= 0 {
		interval = snmppkg.DefaultCollectIntervalSec
	}
	if interval < snmppkg.MinCollectIntervalSec {
		interval = snmppkg.MinCollectIntervalSec
	}
	return interval
}

// shouldCollectDevice 判断是否到达该设备的采集周期 // Logan.Li
func shouldCollectDevice(device *model.SNMPDevice, now time.Time) bool {
	if device.LastCollectAt == nil {
		return true
	}
	interval := time.Duration(deviceCollectInterval(device)) * time.Second
	return now.Sub(*device.LastCollectAt) >= interval
}

// StartCollector 启动 SNMP 后台采集 // Logan.Li
func (h *SNMPHandler) StartCollector(intervalSec int) {
	if intervalSec < snmppkg.MinCollectIntervalSec {
		intervalSec = snmppkg.DefaultCollectIntervalSec
	}
	h.wg.Add(1)
	go func() {
		defer h.wg.Done()
		ticker := time.NewTicker(time.Duration(intervalSec) * time.Second)
		defer ticker.Stop()
		log.Printf("[SNMP] 采集引擎启动，间隔 %ds", intervalSec)
		h.collectAll()
		for {
			select {
			case <-ticker.C:
				h.collectAll()
			case <-h.stopCh:
				log.Printf("[SNMP] 采集引擎停止")
				return
			}
		}
	}()
}

func (h *SNMPHandler) collectAll() {
	var devices []model.SNMPDevice
	h.db.Where("template_id > 0").Find(&devices)
	now := time.Now()
	for i := range devices {
		if !shouldCollectDevice(&devices[i], now) {
			continue
		}
		if err := h.collectDevice(&devices[i]); err != nil {
			log.Printf("[SNMP] 采集失败 %s(%s): %v", devices[i].Name, devices[i].IP, err)
		}
	}
}

// CollectDeviceNow POST /snmp/devices/:id/collect
func (h *SNMPHandler) CollectDeviceNow(c *gin.Context) {
	id, _ := strconv.Atoi(c.Param("id"))
	var device model.SNMPDevice
	if err := h.db.First(&device, id).Error; err != nil {
		c.JSON(http.StatusNotFound, model.Response{Code: 404, Message: "设备不存在"})
		return
	}
	if err := h.collectDevice(&device); err != nil {
		c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: err.Error()})
		return
	}
	h.db.First(&device, id)
	c.JSON(http.StatusOK, model.Response{Code: 200, Data: device, Message: "采集成功"})
}

// ListDeviceMetrics GET /snmp/devices/:id/metrics（仅返回最近一次采集批次） // Logan.Li
func (h *SNMPHandler) ListDeviceMetrics(c *gin.Context) {
	id, _ := strconv.Atoi(c.Param("id"))
	var latestTS int64
	h.db.Model(&model.SNMPMetric{}).Where("device_id = ?", id).
		Select("COALESCE(MAX(timestamp), 0)").Scan(&latestTS)
	var metrics []model.SNMPMetric
	if latestTS > 0 {
		h.db.Where("device_id = ? AND timestamp = ?", id, latestTS).
			Order("name ASC, `index` ASC").Limit(500).Find(&metrics)
	}
	if metrics == nil {
		metrics = []model.SNMPMetric{}
	}
	c.JSON(http.StatusOK, model.Response{Code: 200, Data: gin.H{"list": metrics, "total": len(metrics)}})
}

// StopCollector 停止采集引擎 // Logan.Li
func (h *SNMPHandler) StopCollector() {
	close(h.stopCh)
	h.wg.Wait()
}