marge: 合并代码
This commit is contained in:
TsMask
287
src/modules/monitor/service/monitor.impl.go
Normal file
287
src/modules/monitor/service/monitor.impl.go
Normal file
@@ -0,0 +1,287 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"ems.agt/src/framework/logger"
|
||||
"ems.agt/src/modules/monitor/model"
|
||||
"ems.agt/src/modules/monitor/repository"
|
||||
systemService "ems.agt/src/modules/system/service"
|
||||
"github.com/shirou/gopsutil/cpu"
|
||||
"github.com/shirou/gopsutil/disk"
|
||||
"github.com/shirou/gopsutil/load"
|
||||
"github.com/shirou/gopsutil/net"
|
||||
"github.com/shirou/gopsutil/v3/mem"
|
||||
)
|
||||
|
||||
// 实例化服务层 MonitorImpl 结构体
|
||||
var NewMonitorImpl = &MonitorImpl{
|
||||
sysConfigService: systemService.NewSysConfigImpl,
|
||||
monitorRepository: repository.NewMonitorImpl,
|
||||
diskIO: make(chan []disk.IOCountersStat, 2),
|
||||
netIO: make(chan []net.IOCountersStat, 2),
|
||||
}
|
||||
|
||||
// MonitorImpl 服务器系统相关信息 服务层处理
|
||||
type MonitorImpl struct {
|
||||
// 参数配置服务
|
||||
sysConfigService systemService.ISysConfig
|
||||
// 监控服务资源数据信息
|
||||
monitorRepository repository.IMonitor
|
||||
// 磁盘网络IO 数据通道
|
||||
diskIO chan ([]disk.IOCountersStat)
|
||||
netIO chan ([]net.IOCountersStat)
|
||||
}
|
||||
|
||||
// RunMonitor 执行资源监控
|
||||
func (s *MonitorImpl) RunMonitor() {
|
||||
var itemModel model.MonitorBase
|
||||
itemModel.CreateTime = time.Now().UnixMilli()
|
||||
totalPercent, _ := cpu.Percent(3*time.Second, false)
|
||||
if len(totalPercent) == 1 {
|
||||
itemModel.CPU = totalPercent[0]
|
||||
}
|
||||
cpuCount, _ := cpu.Counts(false)
|
||||
|
||||
loadInfo, _ := load.Avg()
|
||||
itemModel.CPULoad1 = loadInfo.Load1
|
||||
itemModel.CPULoad5 = loadInfo.Load5
|
||||
itemModel.CPULoad15 = loadInfo.Load15
|
||||
itemModel.LoadUsage = loadInfo.Load1 / (float64(cpuCount*2) * 0.75) * 100
|
||||
|
||||
memoryInfo, _ := mem.VirtualMemory()
|
||||
itemModel.Memory = memoryInfo.UsedPercent
|
||||
if err := s.monitorRepository.CreateMonitorBase(itemModel); err != nil {
|
||||
logger.Errorf("CreateMonitorBase err: %v", err)
|
||||
}
|
||||
|
||||
// 将当前资源发送到chan中处理保存
|
||||
s.loadDiskIO()
|
||||
s.loadNetIO()
|
||||
|
||||
// 监控系统资源-保留天数
|
||||
storeDays := s.sysConfigService.SelectConfigValueByKey("monitor.sysResource.storeDays")
|
||||
if storeDays != "" {
|
||||
storeDays, _ := strconv.Atoi(storeDays)
|
||||
ltTime := time.Now().AddDate(0, 0, -storeDays).UnixMilli()
|
||||
_ = s.monitorRepository.DelMonitorBase(ltTime)
|
||||
_ = s.monitorRepository.DelMonitorIO(ltTime)
|
||||
_ = s.monitorRepository.DelMonitorNet(ltTime)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *MonitorImpl) loadDiskIO() {
|
||||
ioStat, _ := disk.IOCounters()
|
||||
var diskIOList []disk.IOCountersStat
|
||||
for _, io := range ioStat {
|
||||
diskIOList = append(diskIOList, io)
|
||||
}
|
||||
s.diskIO <- diskIOList
|
||||
}
|
||||
|
||||
func (s *MonitorImpl) loadNetIO() {
|
||||
netStat, _ := net.IOCounters(true)
|
||||
netStatAll, _ := net.IOCounters(false)
|
||||
var netList []net.IOCountersStat
|
||||
netList = append(netList, netStat...)
|
||||
netList = append(netList, netStatAll...)
|
||||
s.netIO <- netList
|
||||
}
|
||||
|
||||
// monitorCancel 监控搜集IO/Network上下文
|
||||
var monitorCancel context.CancelFunc
|
||||
|
||||
// RunMonitorDataCancel 启动资源监控数据存储io/network通道 移除之前的chan上下文后在设置新的均值
|
||||
// interval 采集的平均值（分钟）
|
||||
func (s *MonitorImpl) RunMonitorDataCancel(removeBefore bool, interval float64) {
|
||||
// 是否取消之前的
|
||||
if removeBefore {
|
||||
monitorCancel()
|
||||
}
|
||||
|
||||
// 上下文控制
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
monitorCancel = cancel
|
||||
|
||||
// chanl 通道进行存储数据
|
||||
go s.saveIODataToDB(ctx, interval)
|
||||
go s.saveNetDataToDB(ctx, interval)
|
||||
}
|
||||
|
||||
func (s *MonitorImpl) saveIODataToDB(ctx context.Context, interval float64) {
|
||||
defer close(s.diskIO)
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case ioStat := <-s.diskIO:
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case ioStat2 := <-s.diskIO:
|
||||
var ioList []model.MonitorIO
|
||||
timeMilli := time.Now().UnixMilli()
|
||||
for _, io2 := range ioStat2 {
|
||||
for _, io1 := range ioStat {
|
||||
if io2.Name == io1.Name {
|
||||
var itemIO model.MonitorIO
|
||||
itemIO.CreateTime = timeMilli
|
||||
itemIO.Name = io1.Name
|
||||
if io2.ReadBytes != 0 && io1.ReadBytes != 0 && io2.ReadBytes > io1.ReadBytes {
|
||||
itemIO.Read = int64(float64(io2.ReadBytes-io1.ReadBytes) / interval / 60)
|
||||
}
|
||||
if io2.WriteBytes != 0 && io1.WriteBytes != 0 && io2.WriteBytes > io1.WriteBytes {
|
||||
itemIO.Write = int64(float64(io2.WriteBytes-io1.WriteBytes) / interval / 60)
|
||||
}
|
||||
|
||||
if io2.ReadCount != 0 && io1.ReadCount != 0 && io2.ReadCount > io1.ReadCount {
|
||||
itemIO.Count = int64(float64(io2.ReadCount-io1.ReadCount) / interval / 60)
|
||||
}
|
||||
writeCount := int64(0)
|
||||
if io2.WriteCount != 0 && io1.WriteCount != 0 && io2.WriteCount > io1.WriteCount {
|
||||
writeCount = int64(float64(io2.WriteCount-io1.WriteCount) / interval * 60)
|
||||
}
|
||||
if writeCount > itemIO.Count {
|
||||
itemIO.Count = writeCount
|
||||
}
|
||||
|
||||
if io2.ReadTime != 0 && io1.ReadTime != 0 && io2.ReadTime > io1.ReadTime {
|
||||
itemIO.Time = int64(float64(io2.ReadTime-io1.ReadTime) / interval / 60)
|
||||
}
|
||||
writeTime := int64(0)
|
||||
if io2.WriteTime != 0 && io1.WriteTime != 0 && io2.WriteTime > io1.WriteTime {
|
||||
writeTime = int64(float64(io2.WriteTime-io1.WriteTime) / interval / 60)
|
||||
}
|
||||
if writeTime > itemIO.Time {
|
||||
itemIO.Time = writeTime
|
||||
}
|
||||
ioList = append(ioList, itemIO)
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
if err := s.monitorRepository.BatchCreateMonitorIO(ioList); err != nil {
|
||||
logger.Errorf("BatchCreateMonitorIO err: %v", err)
|
||||
}
|
||||
s.diskIO <- ioStat2
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (s *MonitorImpl) saveNetDataToDB(ctx context.Context, interval float64) {
|
||||
defer close(s.netIO)
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case netStat := <-s.netIO:
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case netStat2 := <-s.netIO:
|
||||
var netList []model.MonitorNetwork
|
||||
timeMilli := time.Now().UnixMilli()
|
||||
for _, net2 := range netStat2 {
|
||||
for _, net1 := range netStat {
|
||||
if net2.Name == net1.Name {
|
||||
var itemNet model.MonitorNetwork
|
||||
itemNet.CreateTime = timeMilli
|
||||
itemNet.Name = net1.Name
|
||||
|
||||
if net2.BytesSent != 0 && net1.BytesSent != 0 && net2.BytesSent > net1.BytesSent {
|
||||
itemNet.Up = float64(net2.BytesSent-net1.BytesSent) / 1024 / interval / 60
|
||||
}
|
||||
if net2.BytesRecv != 0 && net1.BytesRecv != 0 && net2.BytesRecv > net1.BytesRecv {
|
||||
itemNet.Down = float64(net2.BytesRecv-net1.BytesRecv) / 1024 / interval / 60
|
||||
}
|
||||
netList = append(netList, itemNet)
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if err := s.monitorRepository.BatchCreateMonitorNet(netList); err != nil {
|
||||
logger.Errorf("BatchCreateMonitorNet err: %v", err)
|
||||
}
|
||||
s.netIO <- netStat2
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// MonitorData 监控资源信息
|
||||
type MonitorData struct {
|
||||
Date []int64 `json:"date"`
|
||||
Value []any `json:"value"`
|
||||
}
|
||||
|
||||
// SelectMonitorInfo 查询监控资源信息
|
||||
func (s *MonitorImpl) SelectMonitorInfo(query map[string]any) map[string]MonitorData {
|
||||
infoType := query["type"]
|
||||
startTimeMilli := query["startTime"]
|
||||
endTimeMilli := query["endTime"]
|
||||
neType := query["neType"]
|
||||
neId := query["neId"]
|
||||
name := query["name"]
|
||||
|
||||
// 返回数据
|
||||
backdatas := map[string]MonitorData{}
|
||||
|
||||
// 基本信息
|
||||
if infoType == "all" || infoType == "cpu" || infoType == "memory" {
|
||||
rows := s.monitorRepository.SelectMonitorBase(map[string]any{
|
||||
"startTime": startTimeMilli,
|
||||
"endTime": endTimeMilli,
|
||||
"neType": neType,
|
||||
"neId": neId,
|
||||
})
|
||||
// 组装数据
|
||||
var itemData MonitorData
|
||||
for _, base := range rows {
|
||||
itemData.Date = append(itemData.Date, base.CreateTime)
|
||||
itemData.Value = append(itemData.Value, base)
|
||||
}
|
||||
backdatas["base"] = itemData
|
||||
}
|
||||
|
||||
// 磁盘IO
|
||||
if infoType == "all" || infoType == "io" {
|
||||
rows := s.monitorRepository.SelectMonitorIO(map[string]any{
|
||||
"startTime": startTimeMilli,
|
||||
"endTime": endTimeMilli,
|
||||
"neType": neType,
|
||||
"neId": neId,
|
||||
"name": name,
|
||||
})
|
||||
// 组装数据
|
||||
var itemData MonitorData
|
||||
for _, base := range rows {
|
||||
itemData.Date = append(itemData.Date, base.CreateTime)
|
||||
itemData.Value = append(itemData.Value, base)
|
||||
}
|
||||
backdatas["io"] = itemData
|
||||
}
|
||||
|
||||
// 网络
|
||||
if infoType == "all" || infoType == "network" {
|
||||
rows := s.monitorRepository.SelectMonitorNetwork(map[string]any{
|
||||
"startTime": startTimeMilli,
|
||||
"endTime": endTimeMilli,
|
||||
"neType": neType,
|
||||
"neId": neId,
|
||||
"name": name,
|
||||
})
|
||||
// 组装数据
|
||||
var itemData MonitorData
|
||||
for _, base := range rows {
|
||||
itemData.Date = append(itemData.Date, base.CreateTime)
|
||||
itemData.Value = append(itemData.Value, base)
|
||||
}
|
||||
backdatas["network"] = itemData
|
||||
}
|
||||
|
||||
return backdatas
|
||||
}
|
||||
Reference in New Issue
Block a user