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() }