modularize monitor, reduce init usage (#81)

* kill process tree using syscall on windows & cleanup (#80)

* kill process tree using syscall on windows & cleanup

* use job api

* add error check for cmd.Start

* modularize monitor, reduce init usage

* replace slices with sort

* update gopsutil & other dependencies

pkg/monitor/conn/conn_fallback.go

@@ -0,0 +1,26 @@
+//go:build !linux
+
+package conn
+
+import (
+	"context"
+	"syscall"
+
+	"github.com/shirou/gopsutil/v4/net"
+)
+
+func GetState(_ context.Context) ([]uint64, error) {
+	var tcpConnCount, udpConnCount uint64
+
+	conns, _ := net.Connections("all")
+	for i := 0; i < len(conns); i++ {
+		switch conns[i].Type {
+		case syscall.SOCK_STREAM:
+			tcpConnCount++
+		case syscall.SOCK_DGRAM:
+			udpConnCount++
+		}
+	}
+
+	return []uint64{tcpConnCount, udpConnCount}, nil
+}

pkg/monitor/conn/conn_linux.go

@@ -0,0 +1,50 @@
+//go:build linux
+
+package conn
+
+import (
+	"context"
+	"syscall"
+
+	"github.com/dean2021/goss"
+	"github.com/shirou/gopsutil/v4/net"
+)
+
+func GetState(_ context.Context) ([]uint64, error) {
+	var tcpConnCount, udpConnCount uint64
+
+	tcpStat, err := goss.ConnectionsWithProtocol(goss.AF_INET, syscall.IPPROTO_TCP)
+	if err == nil {
+		tcpConnCount = uint64(len(tcpStat))
+	}
+
+	udpStat, err := goss.ConnectionsWithProtocol(goss.AF_INET, syscall.IPPROTO_UDP)
+	if err == nil {
+		udpConnCount = uint64(len(udpStat))
+	}
+
+	tcpStat6, err := goss.ConnectionsWithProtocol(goss.AF_INET6, syscall.IPPROTO_TCP)
+	if err == nil {
+		tcpConnCount += uint64(len(tcpStat6))
+	}
+
+	udpStat6, err := goss.ConnectionsWithProtocol(goss.AF_INET6, syscall.IPPROTO_UDP)
+	if err == nil {
+		udpConnCount += uint64(len(udpStat6))
+	}
+
+	if tcpConnCount < 1 && udpConnCount < 1 {
+		// fallback to parsing files
+		conns, _ := net.Connections("all")
+		for _, conn := range conns {
+			switch conn.Type {
+			case syscall.SOCK_STREAM:
+				tcpConnCount++
+			case syscall.SOCK_DGRAM:
+				udpConnCount++
+			}
+		}
+	}
+
+	return []uint64{tcpConnCount, udpConnCount}, nil
+}

pkg/monitor/cpu/cpu.go

@@ -0,0 +1,50 @@
+package cpu
+
+import (
+	"context"
+	"fmt"
+
+	psCpu "github.com/shirou/gopsutil/v4/cpu"
+)
+
+type CPUHostType string
+
+const CPUHostKey CPUHostType = "cpu"
+
+func GetHost(ctx context.Context) ([]string, error) {
+	ci, err := psCpu.InfoWithContext(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	cpuModelCount := make(map[string]int)
+	for _, c := range ci {
+		cpuModelCount[c.ModelName]++
+	}
+
+	var cpuType string
+	if t, ok := ctx.Value(CPUHostKey).(string); ok {
+		cpuType = t
+	}
+
+	var ch []string
+	u := len(ci) > 1
+	for model, count := range cpuModelCount {
+		if u {
+			ch = append(ch, fmt.Sprintf("%s %d %s Core", model, count, cpuType))
+		} else {
+			ch = append(ch, fmt.Sprintf("%s %d %s Core", model, ci[0].Cores, cpuType))
+		}
+	}
+
+	return ch, nil
+}
+
+func GetState(ctx context.Context) ([]float64, error) {
+	cp, err := psCpu.PercentWithContext(ctx, 0, false)
+	if err != nil {
+		return nil, err
+	}
+
+	return cp, nil
+}

pkg/monitor/disk/disk.go

@@ -0,0 +1,122 @@
+package disk
+
+import (
+	"context"
+	"os/exec"
+	"runtime"
+	"strconv"
+	"strings"
+
+	psDisk "github.com/shirou/gopsutil/v4/disk"
+
+	"github.com/nezhahq/agent/pkg/util"
+)
+
+type DiskKeyType string
+
+const DiskKey DiskKeyType = "disk"
+
+var expectDiskFsTypes = []string{
+	"apfs", "ext4", "ext3", "ext2", "f2fs", "reiserfs", "jfs", "btrfs",
+	"fuseblk", "zfs", "simfs", "ntfs", "fat32", "exfat", "xfs", "fuse.rclone",
+}
+
+func GetHost(ctx context.Context) (uint64, error) {
+	devices, err := getDevices(ctx)
+	if err != nil {
+		return 0, err
+	}
+
+	var total uint64
+	for _, mountPath := range devices {
+		diskUsageOf, err := psDisk.Usage(mountPath)
+		if err == nil {
+			total += diskUsageOf.Total
+		}
+	}
+
+	// Fallback 到这个方法,仅统计根路径,适用于OpenVZ之类的.
+	if runtime.GOOS == "linux" && total == 0 {
+		cmd := exec.Command("df")
+		out, err := cmd.CombinedOutput()
+		if err == nil {
+			s := strings.Split(string(out), "\n")
+			for _, c := range s {
+				info := strings.Fields(c)
+				if len(info) == 6 {
+					if info[5] == "/" {
+						total, _ = strconv.ParseUint(info[1], 0, 64)
+						// 默认获取的是1K块为单位的.
+						total = total * 1024
+					}
+				}
+			}
+		}
+	}
+
+	return total, nil
+}
+
+func GetState(ctx context.Context) (uint64, error) {
+	devices, err := getDevices(ctx)
+	if err != nil {
+		return 0, err
+	}
+
+	var used uint64
+	for _, mountPath := range devices {
+		diskUsageOf, err := psDisk.Usage(mountPath)
+		if err == nil {
+			used += diskUsageOf.Used
+		}
+	}
+
+	// Fallback 到这个方法,仅统计根路径,适用于OpenVZ之类的.
+	if runtime.GOOS == "linux" && used == 0 {
+		cmd := exec.Command("df")
+		out, err := cmd.CombinedOutput()
+		if err == nil {
+			s := strings.Split(string(out), "\n")
+			for _, c := range s {
+				info := strings.Fields(c)
+				if len(info) == 6 {
+					if info[5] == "/" {
+						used, _ = strconv.ParseUint(info[2], 0, 64)
+						// 默认获取的是1K块为单位的.
+						used = used * 1024
+					}
+				}
+			}
+		}
+	}
+
+	return used, nil
+}
+
+func getDevices(ctx context.Context) (map[string]string, error) {
+	devices := make(map[string]string)
+
+	// 如果配置了白名单，使用白名单的列表
+	if s, ok := ctx.Value(DiskKey).([]string); ok && len(s) > 0 {
+		for i, v := range s {
+			devices[strconv.Itoa(i)] = v
+		}
+		return devices, nil
+	}
+
+	// 否则使用默认过滤规则
+	diskList, err := psDisk.Partitions(false)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, d := range diskList {
+		fsType := strings.ToLower(d.Fstype)
+		// 不统计 K8s 的虚拟挂载点：https://github.com/shirou/gopsutil/issues/1007
+		if devices[d.Device] == "" && util.ContainsStr(expectDiskFsTypes, fsType) && !strings.Contains(d.Mountpoint, "/var/lib/kubelet") {
+			devices[d.Device] = d.Mountpoint
+		}
+	}
+
+	return devices, nil
+}

pkg/monitor/gpu/gpu_darwin.go

@@ -0,0 +1,247 @@
+//go:build darwin
+
+package gpu
+
+import (
+	"context"
+	"fmt"
+	"unsafe"
+
+	"github.com/ebitengine/purego"
+
+	"github.com/nezhahq/agent/pkg/util"
+)
+
+type (
+	CFStringEncoding = uint32
+	CFIndex          = int32
+	CFTypeID         = int32
+	CFNumberType     = CFIndex
+	CFTypeRef        = unsafe.Pointer
+	CFStringRef      = unsafe.Pointer
+	CFDictionaryRef  = unsafe.Pointer
+
+	machPort        = uint32
+	ioIterator      = uint32
+	ioObject        = uint32
+	ioRegistryEntry = uint32
+	ioService       = uint32
+	IOOptionBits    = uint32
+)
+
+type (
+	CFStringCreateWithCStringFunc = func(alloc uintptr, cStr string, encoding CFStringEncoding) CFStringRef
+	CFGetTypeIDFunc               = func(cf uintptr) CFTypeID
+	CFStringGetTypeIDFunc         = func() CFTypeID
+	CFStringGetLengthFunc         = func(theString uintptr) int32
+	CFStringGetCStringFunc        = func(cfStr uintptr, buffer *byte, size CFIndex, encoding CFStringEncoding) bool
+	CFDictionaryGetTypeIDFunc     = func() CFTypeID
+	CFDictionaryGetValueFunc      = func(dict, key uintptr) unsafe.Pointer
+	CFDataGetTypeIDFunc           = func() CFTypeID
+	CFDataGetBytePtrFunc          = func(theData uintptr) unsafe.Pointer
+	CFDataGetLengthFunc           = func(theData uintptr) CFIndex
+	CFNumberGetValueFunc          = func(number uintptr, theType CFNumberType, valuePtr uintptr) bool
+	CFReleaseFunc                 = func(cf uintptr)
+
+	IOServiceGetMatchingServicesFunc    = func(mainPort machPort, matching uintptr, existing *ioIterator) ioService
+	IOIteratorNextFunc                  = func(iterator ioIterator) ioObject
+	IOServiceMatchingFunc               = func(name string) CFDictionaryRef
+	IORegistryEntrySearchCFPropertyFunc = func(entry ioRegistryEntry, plane string, key, allocator uintptr, options IOOptionBits) CFTypeRef
+	IOObjectReleaseFunc                 = func(object ioObject) int
+)
+
+const (
+	KERN_SUCCESS   = 0
+	MACH_PORT_NULL = 0
+	IOSERVICE_GPU  = "IOAccelerator"
+	IOSERVICE_PCI  = "IOPCIDevice"
+
+	kIOServicePlane               = "IOService"
+	kIORegistryIterateRecursively = 1
+	kCFStringEncodingUTF8         = 0x08000100
+	kCFNumberIntType              = 9
+)
+
+var (
+	kCFAllocatorDefault uintptr  = 0
+	kIOMainPortDefault  machPort = 0
+)
+
+var (
+	coreFoundation, _ = purego.Dlopen("/System/Library/Frameworks/CoreFoundation.framework/CoreFoundation", purego.RTLD_LAZY|purego.RTLD_GLOBAL)
+	ioKit, _          = purego.Dlopen("/System/Library/Frameworks/IOKit.framework/IOKit", purego.RTLD_LAZY|purego.RTLD_GLOBAL)
+)
+
+var (
+	CFStringCreateWithCString CFStringCreateWithCStringFunc
+	CFGetTypeID               CFGetTypeIDFunc
+	CFStringGetTypeID         CFStringGetTypeIDFunc
+	CFStringGetLength         CFStringGetLengthFunc
+	CFStringGetCString        CFStringGetCStringFunc
+	CFDictionaryGetTypeID     CFDictionaryGetTypeIDFunc
+	CFDictionaryGetValue      CFDictionaryGetValueFunc
+	CFDataGetTypeID           CFDataGetTypeIDFunc
+	CFDataGetBytePtr          CFDataGetBytePtrFunc
+	CFDataGetLength           CFDataGetLengthFunc
+	CFNumberGetValue          CFNumberGetValueFunc
+	CFRelease                 CFReleaseFunc
+
+	IOServiceGetMatchingServices    IOServiceGetMatchingServicesFunc
+	IOIteratorNext                  IOIteratorNextFunc
+	IOServiceMatching               IOServiceMatchingFunc
+	IORegistryEntrySearchCFProperty IORegistryEntrySearchCFPropertyFunc
+	IOObjectRelease                 IOObjectReleaseFunc
+)
+
+var validVendors = []string{
+	"AMD", "Intel", "NVIDIA", "Apple",
+}
+
+func init() {
+	purego.RegisterLibFunc(&CFStringCreateWithCString, coreFoundation, "CFStringCreateWithCString")
+	purego.RegisterLibFunc(&CFGetTypeID, coreFoundation, "CFGetTypeID")
+	purego.RegisterLibFunc(&CFStringGetTypeID, coreFoundation, "CFStringGetTypeID")
+	purego.RegisterLibFunc(&CFStringGetLength, coreFoundation, "CFStringGetLength")
+	purego.RegisterLibFunc(&CFStringGetCString, coreFoundation, "CFStringGetCString")
+	purego.RegisterLibFunc(&CFDictionaryGetTypeID, coreFoundation, "CFDictionaryGetTypeID")
+	purego.RegisterLibFunc(&CFDictionaryGetValue, coreFoundation, "CFDictionaryGetValue")
+	purego.RegisterLibFunc(&CFDataGetTypeID, coreFoundation, "CFDataGetTypeID")
+	purego.RegisterLibFunc(&CFDataGetBytePtr, coreFoundation, "CFDataGetBytePtr")
+	purego.RegisterLibFunc(&CFDataGetLength, coreFoundation, "CFDataGetLength")
+	purego.RegisterLibFunc(&CFNumberGetValue, coreFoundation, "CFNumberGetValue")
+	purego.RegisterLibFunc(&CFRelease, coreFoundation, "CFRelease")
+
+	purego.RegisterLibFunc(&IOServiceGetMatchingServices, ioKit, "IOServiceGetMatchingServices")
+	purego.RegisterLibFunc(&IOIteratorNext, ioKit, "IOIteratorNext")
+	purego.RegisterLibFunc(&IOServiceMatching, ioKit, "IOServiceMatching")
+	purego.RegisterLibFunc(&IORegistryEntrySearchCFProperty, ioKit, "IORegistryEntrySearchCFProperty")
+	purego.RegisterLibFunc(&IOObjectRelease, ioKit, "IOObjectRelease")
+}
+
+func GetHost(_ context.Context) ([]string, error) {
+	models, err := findDevices("model")
+	if err != nil {
+		return nil, err
+	}
+	return util.RemoveDuplicate(models), nil
+}
+
+func GetState(_ context.Context) ([]float64, error) {
+	usage, err := findUtilization("PerformanceStatistics", "Device Utilization %")
+	return []float64{float64(usage)}, err
+}
+
+func findDevices(key string) ([]string, error) {
+	var iterator ioIterator
+	var results []string
+	done := false
+
+	iv := IOServiceGetMatchingServices(kIOMainPortDefault, uintptr(IOServiceMatching(IOSERVICE_GPU)), &iterator)
+	if iv != KERN_SUCCESS {
+		return nil, fmt.Errorf("error retrieving GPU entry")
+	}
+
+	var service ioObject
+	index := 0
+
+	for {
+		service = IOIteratorNext(iterator)
+		if service == MACH_PORT_NULL {
+			break
+		}
+
+		cfStr := CFStringCreateWithCString(kCFAllocatorDefault, key, kCFStringEncodingUTF8)
+		result, _, _ := findProperties(service, uintptr(cfStr), 0)
+		IOObjectRelease(service)
+
+		if util.ContainsStr(validVendors, result) {
+			results = append(results, result)
+			index++
+		} else if key == "model" && !done {
+			IOObjectRelease(iterator)
+			iv = IOServiceGetMatchingServices(kIOMainPortDefault, uintptr(IOServiceMatching(IOSERVICE_PCI)), &iterator)
+			if iv != KERN_SUCCESS {
+				return nil, fmt.Errorf("error retrieving GPU entry")
+			}
+			done = true
+		}
+	}
+
+	IOObjectRelease(iterator)
+	return results, nil
+}
+
+func findUtilization(key, dictKey string) (int, error) {
+	var iterator ioIterator
+	var err error
+	result := 0
+
+	iv := IOServiceGetMatchingServices(kIOMainPortDefault, uintptr(IOServiceMatching(IOSERVICE_GPU)), &iterator)
+	if iv != KERN_SUCCESS {
+		return 0, fmt.Errorf("error retrieving GPU entry")
+	}
+
+	// Only retrieving the utilization of a single GPU here
+	var service ioObject
+	for {
+		service = IOIteratorNext(iterator)
+		if service == MACH_PORT_NULL {
+			break
+		}
+
+		cfStr := CFStringCreateWithCString(kCFAllocatorDefault, key, CFStringEncoding(kCFStringEncodingUTF8))
+		cfDictStr := CFStringCreateWithCString(kCFAllocatorDefault, dictKey, CFStringEncoding(kCFStringEncodingUTF8))
+
+		_, result, err = findProperties(service, uintptr(cfStr), uintptr(cfDictStr))
+
+		CFRelease(uintptr(cfStr))
+		CFRelease(uintptr(cfDictStr))
+
+		if err != nil {
+			IOObjectRelease(service)
+			continue
+		} else if result != 0 {
+			break
+		}
+	}
+
+	IOObjectRelease(service)
+	IOObjectRelease(iterator)
+
+	return result, err
+}
+
+func findProperties(service ioRegistryEntry, key, dictKey uintptr) (string, int, error) {
+	properties := IORegistryEntrySearchCFProperty(service, kIOServicePlane, key, kCFAllocatorDefault, kIORegistryIterateRecursively)
+	ptrValue := uintptr(properties)
+	if properties != nil {
+		switch CFGetTypeID(ptrValue) {
+		// model
+		case CFStringGetTypeID():
+			length := CFStringGetLength(ptrValue) + 1 // null terminator
+			buf := make([]byte, length-1)
+			CFStringGetCString(ptrValue, &buf[0], length, uint32(kCFStringEncodingUTF8))
+			CFRelease(ptrValue)
+			return string(buf), 0, nil
+		case CFDataGetTypeID():
+			length := CFDataGetLength(ptrValue)
+			bin := unsafe.String((*byte)(CFDataGetBytePtr(ptrValue)), length)
+			CFRelease(ptrValue)
+			return bin, 0, nil
+		// PerformanceStatistics
+		case CFDictionaryGetTypeID():
+			cfValue := CFDictionaryGetValue(ptrValue, dictKey)
+			if cfValue != nil {
+				var value int
+				if CFNumberGetValue(uintptr(cfValue), kCFNumberIntType, uintptr(unsafe.Pointer(&value))) {
+					return "", value, nil
+				} else {
+					return "", 0, fmt.Errorf("failed to exec CFNumberGetValue")
+				}
+			} else {
+				return "", 0, fmt.Errorf("failed to exec CFDictionaryGetValue")
+			}
+		}
+	}
+	return "", 0, fmt.Errorf("failed to exec IORegistryEntrySearchCFProperty")
+}

pkg/monitor/gpu/gpu_fallback.go

@@ -0,0 +1,13 @@
+//go:build !darwin && !linux && !windows
+
+package gpu
+
+import "context"
+
+func GetHost(_ context.Context) ([]string, error) {
+	return nil, nil
+}
+
+func GetState(_ context.Context) ([]float64, error) {
+	return nil, nil
+}

pkg/monitor/gpu/gpu_linux.go

@@ -0,0 +1,126 @@
+//go:build linux
+
+package gpu
+
+import (
+	"context"
+	"errors"
+
+	"github.com/nezhahq/agent/pkg/monitor/gpu/vendor"
+)
+
+const (
+	vendorAMD = iota + 1
+	vendorNVIDIA
+)
+
+var vendorType = getVendor()
+
+func getVendor() uint8 {
+	_, err := getNvidiaStat()
+	if err != nil {
+		return vendorAMD
+	} else {
+		return vendorNVIDIA
+	}
+}
+
+func getNvidiaStat() ([]float64, error) {
+	smi := &vendor.NvidiaSMI{
+		BinPath: "/usr/bin/nvidia-smi",
+	}
+	err1 := smi.Start()
+	if err1 != nil {
+		return nil, err1
+	}
+	data, err2 := smi.GatherUsage()
+	if err2 != nil {
+		return nil, err2
+	}
+	return data, nil
+}
+
+func getAMDStat() ([]float64, error) {
+	rsmi := &vendor.ROCmSMI{
+		BinPath: "/opt/rocm/bin/rocm-smi",
+	}
+	err := rsmi.Start()
+	if err != nil {
+		return nil, err
+	}
+	data, err := rsmi.GatherUsage()
+	if err != nil {
+		return nil, err
+	}
+	return data, nil
+}
+
+func getNvidiaHost() ([]string, error) {
+	smi := &vendor.NvidiaSMI{
+		BinPath: "/usr/bin/nvidia-smi",
+	}
+	err := smi.Start()
+	if err != nil {
+		return nil, err
+	}
+	data, err := smi.GatherModel()
+	if err != nil {
+		return nil, err
+	}
+	return data, nil
+}
+
+func getAMDHost() ([]string, error) {
+	rsmi := &vendor.ROCmSMI{
+		BinPath: "/opt/rocm/bin/rocm-smi",
+	}
+	err := rsmi.Start()
+	if err != nil {
+		return nil, err
+	}
+	data, err := rsmi.GatherModel()
+	if err != nil {
+		return nil, err
+	}
+	return data, nil
+}
+
+func GetHost(_ context.Context) ([]string, error) {
+	var gi []string
+	var err error
+
+	switch vendorType {
+	case vendorAMD:
+		gi, err = getAMDHost()
+	case vendorNVIDIA:
+		gi, err = getNvidiaHost()
+	default:
+		return nil, errors.New("invalid vendor")
+	}
+
+	if err != nil {
+		return nil, err
+	}
+
+	return gi, nil
+}
+
+func GetState(_ context.Context) ([]float64, error) {
+	var gs []float64
+	var err error
+
+	switch vendorType {
+	case vendorAMD:
+		gs, err = getAMDStat()
+	case vendorNVIDIA:
+		gs, err = getNvidiaStat()
+	default:
+		return nil, errors.New("invalid vendor")
+	}
+
+	if err != nil {
+		return nil, err
+	}
+
+	return gs, nil
+}

pkg/monitor/gpu/gpu_windows.go

@@ -0,0 +1,170 @@
+//go:build windows
+
+package gpu
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+	"unsafe"
+
+	"github.com/jaypipes/ghw"
+	"golang.org/x/sys/windows"
+)
+
+const (
+	ERROR_SUCCESS  = 0
+	PDH_FMT_DOUBLE = 0x00000200
+	PDH_MORE_DATA  = 0x800007d2
+	PDH_VAILD_DATA = 0x00000000
+	PDH_NEW_DATA   = 0x00000001
+	PDH_NO_DATA    = 0x800007d5
+)
+
+var (
+	modPdh = windows.NewLazySystemDLL("pdh.dll")
+
+	pdhOpenQuery                 = modPdh.NewProc("PdhOpenQuery")
+	pdhCollectQueryData          = modPdh.NewProc("PdhCollectQueryData")
+	pdhGetFormattedCounterArrayW = modPdh.NewProc("PdhGetFormattedCounterArrayW")
+	pdhAddEnglishCounterW        = modPdh.NewProc("PdhAddEnglishCounterW")
+	pdhCloseQuery                = modPdh.NewProc("PdhCloseQuery")
+)
+
+type PDH_FMT_COUNTERVALUE_DOUBLE struct {
+	CStatus     uint32
+	DoubleValue float64
+}
+
+type PDH_FMT_COUNTERVALUE_ITEM_DOUBLE struct {
+	SzName   *uint16
+	FmtValue PDH_FMT_COUNTERVALUE_DOUBLE
+}
+
+func GetHost(_ context.Context) ([]string, error) {
+	var gpuModel []string
+	gi, err := ghw.GPU(ghw.WithDisableWarnings())
+	if err != nil {
+		return nil, err
+	}
+
+	for _, card := range gi.GraphicsCards {
+		if card.DeviceInfo == nil {
+			return nil, errors.New("Cannot find device info")
+		}
+		gpuModel = append(gpuModel, card.DeviceInfo.Product.Name)
+	}
+
+	return gpuModel, nil
+}
+
+func GetState(_ context.Context) ([]float64, error) {
+	counter, err := newWin32PerformanceCounter("gpu_utilization", "\\GPU Engine(*engtype_3D)\\Utilization Percentage")
+	if err != nil {
+		return nil, err
+	}
+	defer pdhCloseQuery.Call(uintptr(counter.Query))
+
+	values, err := getValue(8192, counter)
+	if err != nil {
+		return nil, err
+	}
+	tot := sumArray(values)
+	if tot > 100 {
+		tot = 100
+	}
+	return []float64{tot}, nil
+}
+
+// https://github.com/influxdata/telegraf/blob/master/plugins/inputs/win_perf_counters/performance_query.go
+func getCounterArrayValue(initialBufSize uint32, counter *win32PerformanceCounter) ([]float64, error) {
+	for buflen := initialBufSize; buflen <= 100*1024*1024; buflen *= 2 {
+		time.Sleep(10 * time.Millisecond) // GPU 查询必须设置间隔，否则数据不准
+		s, _, err := pdhCollectQueryData.Call(uintptr(counter.Query))
+		if s != 0 && err != nil {
+			if s == PDH_NO_DATA {
+				return nil, fmt.Errorf("%w: this counter has not data", err)
+			}
+			return nil, err
+		}
+		buf := make([]byte, buflen)
+		size := buflen
+		var itemCount uint32
+		r, _, _ := pdhGetFormattedCounterArrayW.Call(uintptr(counter.Counter), PDH_FMT_DOUBLE, uintptr(unsafe.Pointer(&size)), uintptr(unsafe.Pointer(&itemCount)), uintptr(unsafe.Pointer(&buf[0])))
+		if r == ERROR_SUCCESS {
+			items := (*[1 << 20]PDH_FMT_COUNTERVALUE_ITEM_DOUBLE)(unsafe.Pointer(&buf[0]))[:itemCount:itemCount]
+			values := make([]float64, 0, itemCount)
+			for _, item := range items {
+				if item.FmtValue.CStatus == PDH_VAILD_DATA || item.FmtValue.CStatus == PDH_NEW_DATA {
+					val := item.FmtValue.DoubleValue
+					values = append(values, val)
+				}
+			}
+			return values, nil
+		}
+		if r != PDH_MORE_DATA {
+			return nil, fmt.Errorf("pdhGetFormattedCounterArrayW failed with status 0x%X", r)
+		}
+	}
+
+	return nil, errors.New("buffer limit reached")
+}
+
+func createQuery() (windows.Handle, error) {
+	var query windows.Handle
+	r, _, err := pdhOpenQuery.Call(0, 0, uintptr(unsafe.Pointer(&query)))
+	if r != ERROR_SUCCESS {
+		return 0, fmt.Errorf("pdhOpenQuery failed with status 0x%X: %v", r, err)
+	}
+	return query, nil
+}
+
+type win32PerformanceCounter struct {
+	PostName    string
+	CounterName string
+	Query       windows.Handle
+	Counter     windows.Handle
+}
+
+func newWin32PerformanceCounter(postName, counterName string) (*win32PerformanceCounter, error) {
+	query, err := createQuery()
+	if err != nil {
+		return nil, err
+	}
+	counter := win32PerformanceCounter{
+		Query:       query,
+		PostName:    postName,
+		CounterName: counterName,
+	}
+	r, _, err := pdhAddEnglishCounterW.Call(
+		uintptr(counter.Query),
+		uintptr(unsafe.Pointer(windows.StringToUTF16Ptr(counter.CounterName))),
+		0,
+		uintptr(unsafe.Pointer(&counter.Counter)),
+	)
+	if r != ERROR_SUCCESS {
+		return nil, fmt.Errorf("pdhAddEnglishCounterW failed with status 0x%X: %v", r, err)
+	}
+	return &counter, nil
+}
+
+func getValue(initialBufSize uint32, counter *win32PerformanceCounter) ([]float64, error) {
+	s, _, err := pdhCollectQueryData.Call(uintptr(counter.Query))
+	if s != 0 && err != nil {
+		if s == PDH_NO_DATA {
+			return nil, fmt.Errorf("%w: this counter has not data", err)
+		}
+		return nil, err
+	}
+
+	return getCounterArrayValue(initialBufSize, counter)
+}
+
+func sumArray(arr []float64) float64 {
+	var sum float64
+	for _, value := range arr {
+		sum += value
+	}
+	return sum
+}

pkg/monitor/gpu/vendor/amd_rocm_smi.go

@@ -0,0 +1,107 @@
+package vendor
+
+// Modified from https://github.com/influxdata/telegraf/blob/master/plugins/inputs/amd_rocm_smi/amd_rocm_smi.go
+// Original License: MIT
+
+import (
+	"errors"
+	"os"
+	"os/exec"
+
+	"github.com/tidwall/gjson"
+)
+
+type ROCmSMI struct {
+	BinPath string
+	data    []byte
+}
+
+func (rsmi *ROCmSMI) GatherModel() ([]string, error) {
+	return rsmi.gatherModel()
+}
+
+func (rsmi *ROCmSMI) GatherUsage() ([]float64, error) {
+	return rsmi.gatherUsage()
+}
+
+func (rsmi *ROCmSMI) Start() error {
+	if _, err := os.Stat(rsmi.BinPath); os.IsNotExist(err) {
+		binPath, err := exec.LookPath("rocm-smi")
+		if err != nil {
+			return errors.New("didn't find the adequate tool to query GPU utilization")
+		}
+		rsmi.BinPath = binPath
+	}
+
+	rsmi.data = rsmi.pollROCmSMI()
+	return nil
+}
+
+func (rsmi *ROCmSMI) pollROCmSMI() []byte {
+	cmd := exec.Command(rsmi.BinPath,
+		"-u",
+		"--showproductname",
+		"--json",
+	)
+	gs, err := cmd.CombinedOutput()
+	if err != nil {
+		return nil
+	}
+	return gs
+}
+
+func (rsmi *ROCmSMI) gatherModel() ([]string, error) {
+	m, err := parseModel(rsmi.data)
+	if err != nil {
+		return nil, err
+	}
+
+	return m, nil
+}
+
+func (rsmi *ROCmSMI) gatherUsage() ([]float64, error) {
+	u, err := parseUsage(rsmi.data)
+	if err != nil {
+		return nil, err
+	}
+
+	return u, nil
+}
+
+func parseModel(jsonObject []byte) ([]string, error) {
+	if jsonObject == nil {
+		return nil, nil
+	}
+
+	result := gjson.ParseBytes(jsonObject)
+	if !result.IsObject() {
+		return nil, errors.New("invalid JSON")
+	}
+
+	ret := make([]string, 0)
+	result.ForEach(func(_, value gjson.Result) bool {
+		ret = append(ret, value.Get("Card series").String())
+		return true
+	})
+
+	return ret, nil
+}
+
+func parseUsage(jsonObject []byte) ([]float64, error) {
+	if jsonObject == nil {
+		return nil, nil
+	}
+
+	result := gjson.ParseBytes(jsonObject)
+	if !result.IsObject() {
+		return nil, errors.New("invalid JSON")
+	}
+
+	ret := make([]float64, 0)
+	result.ForEach(func(_, value gjson.Result) bool {
+		ret = append(ret, value.Get("GPU use (%)").Float())
+		return true
+	})
+
+	return ret, nil
+}

pkg/monitor/gpu/vendor/nvidia_smi.go

@@ -0,0 +1,106 @@
+package vendor
+
+// Modified from https://github.com/influxdata/telegraf/blob/master/plugins/inputs/nvidia_smi/nvidia_smi.go
+// Original License: MIT
+
+import (
+	"encoding/xml"
+	"errors"
+	"os"
+	"os/exec"
+	"strconv"
+	"strings"
+)
+
+type NvidiaSMI struct {
+	BinPath string
+	data    []byte
+}
+
+func (smi *NvidiaSMI) GatherModel() ([]string, error) {
+	return smi.gatherModel()
+}
+
+func (smi *NvidiaSMI) GatherUsage() ([]float64, error) {
+	return smi.gatherUsage()
+}
+
+func (smi *NvidiaSMI) Start() error {
+	if _, err := os.Stat(smi.BinPath); os.IsNotExist(err) {
+		binPath, err := exec.LookPath("nvidia-smi")
+		if err != nil {
+			return errors.New("didn't find the adequate tool to query GPU utilization")
+		}
+		smi.BinPath = binPath
+	}
+	smi.data = smi.pollNvidiaSMI()
+	return nil
+}
+
+func (smi *NvidiaSMI) pollNvidiaSMI() []byte {
+	cmd := exec.Command(smi.BinPath,
+		"-q",
+		"-x",
+	)
+	gs, err := cmd.CombinedOutput()
+	if err != nil {
+		return nil
+	}
+	return gs
+}
+
+func (smi *NvidiaSMI) gatherModel() ([]string, error) {
+	var s smistat
+	var models []string
+
+	err := xml.Unmarshal(smi.data, &s)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, gpu := range s.GPUs {
+		models = append(models, gpu.ProductName)
+	}
+
+	return models, nil
+}
+
+func (smi *NvidiaSMI) gatherUsage() ([]float64, error) {
+	var s smistat
+	var percentage []float64
+
+	err := xml.Unmarshal(smi.data, &s)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, gpu := range s.GPUs {
+		gp, _ := parsePercentage(gpu.Utilization.GpuUtil)
+		percentage = append(percentage, gp)
+	}
+
+	return percentage, nil
+}
+
+func parsePercentage(p string) (float64, error) {
+	per := strings.ReplaceAll(p, " ", "")
+
+	t := strings.TrimSuffix(per, "%")
+
+	value, err := strconv.ParseFloat(t, 64)
+	if err != nil {
+		return 0, err
+	}
+
+	return value, nil
+}
+
+type gpu struct {
+	ProductName string `xml:"product_name"`
+	Utilization struct {
+		GpuUtil string `xml:"gpu_util"`
+	} `xml:"utilization"`
+}
+type smistat struct {
+	GPUs []gpu `xml:"gpu"`
+}

pkg/monitor/load/load.go

@@ -0,0 +1,11 @@
+package load
+
+import (
+	"context"
+
+	psLoad "github.com/shirou/gopsutil/v4/load"
+)
+
+func GetState(ctx context.Context) (*psLoad.AvgStat, error) {
+	return psLoad.AvgWithContext(ctx)
+}

pkg/monitor/monitor.go

@@ -1,44 +1,28 @@
 package monitor
 
 import (
-	"fmt"
-	"os/exec"
+	"context"
 	"runtime"
-	"sort"
-	"strconv"
-	"strings"
 	"sync/atomic"
-	"syscall"
 	"time"
 
-	"github.com/dean2021/goss"
-	"github.com/shirou/gopsutil/v4/cpu"
-	"github.com/shirou/gopsutil/v4/disk"
 	"github.com/shirou/gopsutil/v4/host"
-	"github.com/shirou/gopsutil/v4/load"
 	"github.com/shirou/gopsutil/v4/mem"
-	"github.com/shirou/gopsutil/v4/net"
 	"github.com/shirou/gopsutil/v4/process"
-	"github.com/shirou/gopsutil/v4/sensors"
 
 	"github.com/nezhahq/agent/model"
-	"github.com/nezhahq/agent/pkg/gpu"
+	"github.com/nezhahq/agent/pkg/monitor/conn"
+	"github.com/nezhahq/agent/pkg/monitor/cpu"
+	"github.com/nezhahq/agent/pkg/monitor/disk"
+	"github.com/nezhahq/agent/pkg/monitor/gpu"
+	"github.com/nezhahq/agent/pkg/monitor/load"
+	"github.com/nezhahq/agent/pkg/monitor/nic"
+	"github.com/nezhahq/agent/pkg/monitor/temperature"
 	"github.com/nezhahq/agent/pkg/util"
 )
 
 var (
-	Version           string
-	expectDiskFsTypes = []string{
-		"apfs", "ext4", "ext3", "ext2", "f2fs", "reiserfs", "jfs", "btrfs",
-		"fuseblk", "zfs", "simfs", "ntfs", "fat32", "exfat", "xfs", "fuse.rclone",
-	}
-	excludeNetInterfaces = []string{
-		"lo", "tun", "docker", "veth", "br-", "vmbr", "vnet", "kube",
-	}
-	sensorIgnoreList = []string{
-		"PMU tcal", // the calibration sensor on arm macs, value is fixed
-		"noname",
-	}
+	Version     string
 	agentConfig *model.AgentConfig
 )
 
@@ -51,18 +35,25 @@ var (
 // 获取设备数据的最大尝试次数
 const maxDeviceDataFetchAttempts = 3
 
+const (
+	CPU = iota + 1
+	GPU
+	Load
+	Temperatures
+)
+
 // 获取主机数据的尝试次数，Key 为 Host 的属性名
-var hostDataFetchAttempts = map[string]int{
-	"CPU": 0,
-	"GPU": 0,
+var hostDataFetchAttempts = map[uint8]uint8{
+	CPU: 0,
+	GPU: 0,
 }
 
 // 获取状态数据的尝试次数，Key 为 HostState 的属性名
-var statDataFetchAttempts = map[string]int{
-	"CPU":          0,
-	"Load":         0,
-	"GPU":          0,
-	"Temperatures": 0,
+var statDataFetchAttempts = map[uint8]uint8{
+	CPU:          0,
+	GPU:          0,
+	Load:         0,
+	Temperatures: 0,
 }
 
 var (
@@ -95,40 +86,14 @@ func GetHost() *model.Host {
 		ret.BootTime = hi.BootTime
 	}
 
-	cpuModelCount := make(map[string]int)
-	if hostDataFetchAttempts["CPU"] < maxDeviceDataFetchAttempts {
-		ci, err := cpu.Info()
-		if err != nil {
-			hostDataFetchAttempts["CPU"]++
-			printf("cpu.Info error: %v, attempt: %d", err, hostDataFetchAttempts["CPU"])
-		} else {
-			hostDataFetchAttempts["CPU"] = 0
-			for i := 0; i < len(ci); i++ {
-				cpuModelCount[ci[i].ModelName]++
-			}
-			for model, count := range cpuModelCount {
-				if len(ci) > 1 {
-					ret.CPU = append(ret.CPU, fmt.Sprintf("%s %d %s Core", model, count, cpuType))
-				} else {
-					ret.CPU = append(ret.CPU, fmt.Sprintf("%s %d %s Core", model, ci[0].Cores, cpuType))
-				}
-			}
-		}
-	}
+	ctxCpu := context.WithValue(context.Background(), cpu.CPUHostKey, cpuType)
+	ret.CPU = tryHost(ctxCpu, CPU, cpu.GetHost)
 
 	if agentConfig.GPU {
-		if hostDataFetchAttempts["GPU"] < maxDeviceDataFetchAttempts {
-			ret.GPU, err = gpu.GetGPUModel()
-			if err != nil {
-				hostDataFetchAttempts["GPU"]++
-				printf("gpu.GetGPUModel error: %v, attempt: %d", err, hostDataFetchAttempts["GPU"])
-			} else {
-				hostDataFetchAttempts["GPU"] = 0
-			}
-		}
+		ret.GPU = tryHost(context.Background(), GPU, gpu.GetHost)
 	}
 
-	ret.DiskTotal, _ = getDiskTotalAndUsed()
+	ret.DiskTotal = getDiskTotal()
 
 	mv, err := mem.VirtualMemory()
 	if err != nil {
@@ -160,15 +125,9 @@ func GetHost() *model.Host {
 func GetState(skipConnectionCount bool, skipProcsCount bool) *model.HostState {
 	var ret model.HostState
 
-	if statDataFetchAttempts["CPU"] < maxDeviceDataFetchAttempts {
-		cp, err := cpu.Percent(0, false)
-		if err != nil || len(cp) == 0 {
-			statDataFetchAttempts["CPU"]++
-			printf("cpu.Percent error: %v, attempt: %d", err, statDataFetchAttempts["CPU"])
-		} else {
-			statDataFetchAttempts["CPU"] = 0
-			ret.CPU = cp[0]
-		}
+	cp := tryStat(context.Background(), CPU, cpu.GetState)
+	if len(cp) > 0 {
+		ret.CPU = cp[0]
 	}
 
 	vm, err := mem.VirtualMemory()
@@ -190,20 +149,12 @@ func GetState(skipConnectionCount bool, skipProcsCount bool) *model.HostState {
 		}
 	}
 
-	_, ret.DiskUsed = getDiskTotalAndUsed()
+	ret.DiskUsed = getDiskUsed()
 
-	if statDataFetchAttempts["Load"] < maxDeviceDataFetchAttempts {
-		loadStat, err := load.Avg()
-		if err != nil {
-			statDataFetchAttempts["Load"]++
-			printf("load.Avg error: %v, attempt: %d", err, statDataFetchAttempts["Load"])
-		} else {
-			statDataFetchAttempts["Load"] = 0
-			ret.Load1 = loadStat.Load1
-			ret.Load5 = loadStat.Load5
-			ret.Load15 = loadStat.Load15
-		}
-	}
+	loadStat := tryStat(context.Background(), Load, load.GetState)
+	ret.Load1 = loadStat.Load1
+	ret.Load5 = loadStat.Load5
+	ret.Load15 = loadStat.Load15
 
 	var procs []int32
 	if !skipProcsCount {
@@ -220,12 +171,15 @@ func GetState(skipConnectionCount bool, skipProcsCount bool) *model.HostState {
 		ret.Temperatures = temperatureStat
 	}
 
-	ret.GPU = updateGPUStat()
+	ret.GPU = tryStat(context.Background(), GPU, gpu.GetState)
 
 	ret.NetInTransfer, ret.NetOutTransfer = netInTransfer, netOutTransfer
 	ret.NetInSpeed, ret.NetOutSpeed = netInSpeed, netOutSpeed
 	ret.Uptime = uint64(time.Since(cachedBootTime).Seconds())
-	ret.TcpConnCount, ret.UdpConnCount = getConns(skipConnectionCount)
+
+	if !skipConnectionCount {
+		ret.TcpConnCount, ret.UdpConnCount = getConns()
+	}
 
 	return &ret
 }
@@ -233,136 +187,52 @@ func GetState(skipConnectionCount bool, skipProcsCount bool) *model.HostState {
 // TrackNetworkSpeed NIC监控，统计流量与速度
 func TrackNetworkSpeed() {
 	var innerNetInTransfer, innerNetOutTransfer uint64
-	nc, err := net.IOCounters(true)
-	if err == nil {
-		for _, v := range nc {
-			if len(agentConfig.NICAllowlist) > 0 {
-				if !agentConfig.NICAllowlist[v.Name] {
-					continue
-				}
-			} else {
-				if util.ContainsStr(excludeNetInterfaces, v.Name) {
-					continue
-				}
-			}
-			innerNetInTransfer += v.BytesRecv
-			innerNetOutTransfer += v.BytesSent
-		}
-		now := uint64(time.Now().Unix())
-		diff := now - lastUpdateNetStats
-		if diff > 0 {
-			netInSpeed = (innerNetInTransfer - netInTransfer) / diff
-			netOutSpeed = (innerNetOutTransfer - netOutTransfer) / diff
-		}
-		netInTransfer = innerNetInTransfer
-		netOutTransfer = innerNetOutTransfer
-		lastUpdateNetStats = now
+
+	ctx := context.WithValue(context.Background(), nic.NICKey, agentConfig.NICAllowlist)
+	nc, err := nic.GetState(ctx)
+	if err != nil {
+		return
 	}
+
+	innerNetInTransfer = nc[0]
+	innerNetOutTransfer = nc[1]
+
+	now := uint64(time.Now().Unix())
+	diff := now - lastUpdateNetStats
+	if diff > 0 {
+		netInSpeed = (innerNetInTransfer - netInTransfer) / diff
+		netOutSpeed = (innerNetOutTransfer - netOutTransfer) / diff
+	}
+	netInTransfer = innerNetInTransfer
+	netOutTransfer = innerNetOutTransfer
+	lastUpdateNetStats = now
 }
 
-func getDiskTotalAndUsed() (total uint64, used uint64) {
-	devices := make(map[string]string)
+func getDiskTotal() uint64 {
+	ctx := context.WithValue(context.Background(), disk.DiskKey, agentConfig.HardDrivePartitionAllowlist)
+	total, _ := disk.GetHost(ctx)
 
-	if len(agentConfig.HardDrivePartitionAllowlist) > 0 {
-		// 如果配置了白名单，使用白名单的列表
-		for i, v := range agentConfig.HardDrivePartitionAllowlist {
-			devices[strconv.Itoa(i)] = v
-		}
-	} else {
-		// 否则使用默认过滤规则
-		diskList, _ := disk.Partitions(false)
-		for _, d := range diskList {
-			fsType := strings.ToLower(d.Fstype)
-			// 不统计 K8s 的虚拟挂载点：https://github.com/shirou/gopsutil/issues/1007
-			if devices[d.Device] == "" && util.ContainsStr(expectDiskFsTypes, fsType) && !strings.Contains(d.Mountpoint, "/var/lib/kubelet") {
-				devices[d.Device] = d.Mountpoint
-			}
-		}
-	}
-
-	for _, mountPath := range devices {
-		diskUsageOf, err := disk.Usage(mountPath)
-		if err == nil {
-			total += diskUsageOf.Total
-			used += diskUsageOf.Used
-		}
-	}
-
-	// Fallback 到这个方法,仅统计根路径,适用于OpenVZ之类的.
-	if runtime.GOOS == "linux" && total == 0 && used == 0 {
-		cmd := exec.Command("df")
-		out, err := cmd.CombinedOutput()
-		if err == nil {
-			s := strings.Split(string(out), "\n")
-			for _, c := range s {
-				info := strings.Fields(c)
-				if len(info) == 6 {
-					if info[5] == "/" {
-						total, _ = strconv.ParseUint(info[1], 0, 64)
-						used, _ = strconv.ParseUint(info[2], 0, 64)
-						// 默认获取的是1K块为单位的.
-						total = total * 1024
-						used = used * 1024
-					}
-				}
-			}
-		}
-	}
-
-	return
+	return total
 }
 
-func getConns(skipConnectionCount bool) (tcpConnCount, udpConnCount uint64) {
-	if !skipConnectionCount {
-		ss_err := true
-		if runtime.GOOS == "linux" {
-			tcpStat, err_tcp := goss.ConnectionsWithProtocol(goss.AF_INET, syscall.IPPROTO_TCP)
-			udpStat, err_udp := goss.ConnectionsWithProtocol(goss.AF_INET, syscall.IPPROTO_UDP)
-			if err_tcp == nil && err_udp == nil {
-				ss_err = false
-				tcpConnCount = uint64(len(tcpStat))
-				udpConnCount = uint64(len(udpStat))
-			}
-			if strings.Contains(CachedIP, ":") {
-				tcpStat6, err_tcp := goss.ConnectionsWithProtocol(goss.AF_INET6, syscall.IPPROTO_TCP)
-				udpStat6, err_udp := goss.ConnectionsWithProtocol(goss.AF_INET6, syscall.IPPROTO_UDP)
-				if err_tcp == nil && err_udp == nil {
-					ss_err = false
-					tcpConnCount += uint64(len(tcpStat6))
-					udpConnCount += uint64(len(udpStat6))
-				}
-			}
-		}
-		if ss_err {
-			conns, _ := net.Connections("all")
-			for i := 0; i < len(conns); i++ {
-				switch conns[i].Type {
-				case syscall.SOCK_STREAM:
-					tcpConnCount++
-				case syscall.SOCK_DGRAM:
-					udpConnCount++
-				}
-			}
-		}
-	}
-	return tcpConnCount, udpConnCount
+func getDiskUsed() uint64 {
+	ctx := context.WithValue(context.Background(), disk.DiskKey, agentConfig.HardDrivePartitionAllowlist)
+	used, _ := disk.GetState(ctx)
+
+	return used
 }
 
-func updateGPUStat() []float64 {
-	if agentConfig.GPU {
-		if statDataFetchAttempts["GPU"] < maxDeviceDataFetchAttempts {
-			gs, err := gpu.GetGPUStat()
-			if err != nil {
-				statDataFetchAttempts["GPU"]++
-				printf("gpustat.GetGPUStat error: %v, attempt: %d", err, statDataFetchAttempts["GPU"])
-				return nil
-			} else {
-				statDataFetchAttempts["GPU"] = 0
-				return gs
-			}
-		}
+func getConns() (tcpConnCount, udpConnCount uint64) {
+	connStat, err := conn.GetState(context.Background())
+	if err != nil {
+		return
 	}
-	return nil
+
+	if len(connStat) < 2 {
+		return
+	}
+
+	return connStat[0], connStat[1]
 }
 
 func updateTemperatureStat() {
@@ -371,30 +241,44 @@ func updateTemperatureStat() {
 	}
 	defer updateTempStatus.Store(0)
 
-	if statDataFetchAttempts["Temperatures"] < maxDeviceDataFetchAttempts {
-		temperatures, err := sensors.SensorsTemperatures()
+	stat := tryStat(context.Background(), Temperatures, temperature.GetState)
+	temperatureStat = stat
+}
+
+type hostStateFunc[T any] func(context.Context) (T, error)
+
+func tryHost[T any](ctx context.Context, typ uint8, f hostStateFunc[T]) T {
+	var val T
+
+	if hostDataFetchAttempts[typ] < maxDeviceDataFetchAttempts {
+		v, err := f(ctx)
 		if err != nil {
-			statDataFetchAttempts["Temperatures"]++
-			printf("host.SensorsTemperatures error: %v, attempt: %d", err, statDataFetchAttempts["Temperatures"])
+			hostDataFetchAttempts[typ]++
+			printf("monitor error: %v, attempt: %d", err, hostDataFetchAttempts[typ])
+			return val
 		} else {
-			statDataFetchAttempts["Temperatures"] = 0
-			tempStat := []model.SensorTemperature{}
-			for _, t := range temperatures {
-				if t.Temperature > 0 && !util.ContainsStr(sensorIgnoreList, t.SensorKey) {
-					tempStat = append(tempStat, model.SensorTemperature{
-						Name:        t.SensorKey,
-						Temperature: t.Temperature,
-					})
-				}
-			}
-
-			sort.Slice(tempStat, func(i, j int) bool {
-				return tempStat[i].Name < tempStat[j].Name
-			})
-
-			temperatureStat = tempStat
+			val = v
+			hostDataFetchAttempts[typ] = 0
 		}
 	}
+	return val
+}
+
+func tryStat[T any](ctx context.Context, typ uint8, f hostStateFunc[T]) T {
+	var val T
+
+	if statDataFetchAttempts[typ] < maxDeviceDataFetchAttempts {
+		v, err := f(ctx)
+		if err != nil {
+			statDataFetchAttempts[typ]++
+			printf("monitor error: %v, attempt: %d", err, statDataFetchAttempts[typ])
+			return val
+		} else {
+			val = v
+			statDataFetchAttempts[typ] = 0
+		}
+	}
+	return val
 }
 
 func printf(format string, v ...interface{}) {

pkg/monitor/nic/nic.go

@@ -0,0 +1,45 @@
+package nic
+
+import (
+	"context"
+
+	"github.com/shirou/gopsutil/v4/net"
+)
+
+type NICKeyType string
+
+const NICKey NICKeyType = "nic"
+
+var excludeNetInterfaces = map[string]bool{
+	"lo":     true,
+	"tun":    true,
+	"docker": true,
+	"veth":   true,
+	"br-":    true,
+	"vmbr":   true,
+	"vnet":   true,
+	"kube":   true,
+}
+
+func GetState(ctx context.Context) ([]uint64, error) {
+	var netInTransfer, netOutTransfer uint64
+	nc, err := net.IOCountersWithContext(ctx, true)
+	if err != nil {
+		return nil, err
+	}
+
+	allowList := excludeNetInterfaces
+	if m, ok := ctx.Value(NICKey).(map[string]bool); ok && len(m) > 0 {
+		allowList = m
+	}
+
+	for _, v := range nc {
+		if !allowList[v.Name] {
+			continue
+		}
+		netInTransfer += v.BytesRecv
+		netOutTransfer += v.BytesSent
+	}
+
+	return []uint64{netInTransfer, netOutTransfer}, nil
+}

pkg/monitor/temperature/temperature.go

@@ -0,0 +1,40 @@
+package temperature
+
+import (
+	"context"
+	"fmt"
+	"sort"
+
+	"github.com/shirou/gopsutil/v4/sensors"
+
+	"github.com/nezhahq/agent/model"
+	"github.com/nezhahq/agent/pkg/util"
+)
+
+var sensorIgnoreList = []string{
+	"PMU tcal", // the calibration sensor on arm macs, value is fixed
+	"noname",
+}
+
+func GetState(_ context.Context) ([]model.SensorTemperature, error) {
+	temperatures, err := sensors.SensorsTemperatures()
+	if err != nil {
+		return nil, fmt.Errorf("SensorsTemperatures: %v", err)
+	}
+
+	var tempStat []model.SensorTemperature
+	for _, t := range temperatures {
+		if t.Temperature > 0 && !util.ContainsStr(sensorIgnoreList, t.SensorKey) {
+			tempStat = append(tempStat, model.SensorTemperature{
+				Name:        t.SensorKey,
+				Temperature: t.Temperature,
+			})
+		}
+	}
+
+	sort.Slice(tempStat, func(i, j int) bool {
+		return tempStat[i].Name < tempStat[j].Name
+	})
+
+	return tempStat, nil
+}