joybuywork/internal/app/pprof_collector.go

package app

import (
	"bytes"
	"context"
	"joybuywork/internal/consts"
	"encoding/json"
	"fmt"
	"net"
	"runtime"
	"runtime/pprof"
	"time"
)

var Info ServiceInfo

func SetUpCollector() {
	Info = ServiceInfo{
		Name:         consts.ServiceName,
		NodeId:       fmt.Sprint(NodeId),
		GoVersion:    runtime.Version(),
		StartAt:      time.Now().Unix(),
		LastReportAt: time.Now().Unix(),
		Ip:           GetLocalIP(),
	}

}

type PProfCollector struct {
}

// 初始化采集器
func NewPProfCollector() *PProfCollector {
	return &PProfCollector{}
}

// 采集运行时指标
func (c *PProfCollector) collectRuntimeMetrics() map[string]any {
	var memStats runtime.MemStats
	runtime.ReadMemStats(&memStats)

	return map[string]any{
		"timestamp":   time.Now().Unix(),
		"heap_alloc":  memStats.HeapAlloc,     // 当前堆内存分配（字节）
		"heap_sys":    memStats.HeapSys,       // 堆内存总申请量
		"goroutines":  runtime.NumGoroutine(), // Goroutine 数量
		"gc_count":    memStats.NumGC,         // GC 次数
		"gc_pause_ns": memStats.PauseTotalNs,  // GC 总暂停时间（纳秒）
	}
}

// 采集 Heap Profile 并存储到 Redis
func (c *PProfCollector) saveHeapProfile() {
	var buf bytes.Buffer
	if err := pprof.WriteHeapProfile(&buf); err != nil {
		Logger.Error(context.Background(), fmt.Sprintf("Failed to collect heap profile: %v", err))
		return
	}

	// 存储为 Redis String（二进制数据）
	key := fmt.Sprintf("joybuywork:app:heap:%v:%s", NodeId, time.Now().Format("20060102_1504"))
	if err := Redis.Cache.Set(context.Background(), key, buf.Bytes(), 30*time.Minute).Err(); err != nil {
		Logger.Error(context.Background(), fmt.Sprintf("Failed to save heap profile to Redis: %v", err))
	} else {
		Logger.Info(context.Background(), fmt.Sprintf("Heap profile saved: %s", key))
	}
}

// 采集 CPU Profile（需谨慎，仅在必要时启用）
func (c *PProfCollector) saveCPUProfile(duration time.Duration) {

	var buf bytes.Buffer
	if err := pprof.StartCPUProfile(&buf); err != nil {
		Logger.Error(context.Background(), fmt.Sprintf("Failed to start CPU profile: %v", err))
		return
	}
	time.Sleep(duration)
	pprof.StopCPUProfile()

	key := fmt.Sprintf("joybuywork:app:cpu:%v:%s", NodeId, time.Now().Format("20060102_1504"))
	if err := Redis.Cache.Set(context.Background(), key, buf.Bytes(), 30*time.Minute).Err(); err != nil {
		Logger.Error(context.Background(), fmt.Sprintf("Failed to save CPU profile to Redis: %v", err))
	} else {
		Logger.Info(context.Background(), fmt.Sprintf("CPU profile saved: %s", key))
	}
}

// 存储运行时指标到 Redis（JSON 格式）
func (c *PProfCollector) saveRuntimeMetrics() {
	ctx := context.Background()

	// 指标存储
	metrics := c.collectRuntimeMetrics()
	jsonData, err := json.Marshal(metrics)
	if err != nil {
		Logger.Error(ctx, fmt.Sprintf("Failed to marshal metrics: %v", err))
		return
	}

	key := fmt.Sprintf("joybuywork:app:metrics:%v:%s", NodeId, time.Now().Format("20060102"))
	if err := Redis.Cache.HSet(context.Background(), key, time.Now().Format("1504"), jsonData).Err(); err != nil {
		Logger.Error(context.Background(), fmt.Sprintf("Failed to save metrics to Redis: %v", err))
	} else {
		Redis.Cache.Expire(ctx, key, 24*time.Hour)
		Logger.Info(context.Background(), fmt.Sprintf("Metrics saved: %s", key))
	}
}

// 启动定时采集任务
func (c *PProfCollector) Start() {
	c.saveRuntimeMetrics()
	c.saveHeapProfile()
	c.saveCPUProfile(30 * time.Second) // 可选：采集 10 秒 CPU 数据
}

type ServiceInfo struct {
	Name         string `json:"name"`           //服务名称
	NodeId       string `json:"node_id"`        //节点id
	GoVersion    string `json:"go_version"`     //GO版本
	StartAt      int64  `json:"start_at"`       //启动时间戳
	LastReportAt int64  `json:"last_report_at"` //最后上报时间戳
	Ip           string `json:"ip"`
}

func HeartBeat() {
	ctx := context.Background()
	// 服务存储
	Info.LastReportAt = time.Now().Unix()
	jsonInfo, err := json.Marshal(Info)
	if err != nil {
		Logger.Error(ctx, fmt.Sprintf("Failed to collect service info to json: %v\n", err))
		return
	}

	infokey := fmt.Sprintf("service_info:%s", Info.Name)
	err = Redis.Cache.HSet(ctx, infokey, Info.NodeId, string(jsonInfo)).Err()
	if err != nil {
		Logger.Error(ctx, fmt.Sprintf("Failed to report service info to Redis: %v\n", err))
	}
	Redis.Cache.Expire(ctx, infokey, 1*time.Hour)
}

func GetLocalIP() string {
	addrs, err := net.InterfaceAddrs()
	if err != nil {
		return ""
	}

	for _, addr := range addrs {
		ipNet, ok := addr.(*net.IPNet)
		if ok && !ipNet.IP.IsLoopback() {
			if ipNet.IP.To4() != nil {
				return ipNet.IP.String()
			}
		}
	}

	return ""
}