apian->xcheck

kl7sn · kl7sn · commit 098d6a2cbd7b · 2023-07-20T11:50:12.000+08:00
Signed-off-by: kl7sn &lt;mex7.0828@gmail.com&gt;
diff --git a/xcheck/README.md b/xcheck/README.md
@@ -0,0 +1,19 @@
+# Brief Introduction
+
+To solve the problem of service abnormal OOM, the OOM generated by sudden memory growth cann't be quickly located.
+The purpose of this service is to help users automatically locate the OOM problem and provide the corresponding OOM analysis report.
+
+# Module Introduction
+- **Monitor Module**: Memory changes are detected every second
+- **Forward Module**: The memory change data is sent to the server
+  - Support Webhook
+
+# TODO
+
+## v0.1.0
+- [x] Support trigger indicator hot loading
+  - use `Apply` to reload the trigger indicator
+- [x] Provide default trigger indicator
+  - if cooling time is 0, use the default cooling time [1m]
+  - if memory absolute value is 0, use the default value [128Mib]
+  - if diff is 0, use the default diff [10%]
diff --git a/xcheck/consts.go b/xcheck/consts.go
@@ -0,0 +1,13 @@
+package xcheck
+
+import (
+	"time"
+)
+
+const (
+	defaultCoolingTime = time.Minute
+
+	defaultMemTriggerValue   = 128 // 128MB
+	defaultMemTriggerPercent = 80  // 80%
+	defaultMemTriggerDiff    = 10  // 10%
+)
diff --git a/xcheck/example/main.go b/xcheck/example/main.go
@@ -0,0 +1,49 @@
+package main
+
+import (
+	"fmt"
+	"runtime"
+	"time"
+
+	"github.com/gotomicro/cetus/xcheck"
+)
+
+func main() {
+	a, _ := xcheck.New(
+		xcheck.WithMemOpts(0, 0, 0, 0),
+	)
+	_ = a.EnableMem().Start()
+	go func() {
+		for i := 0; i < 100; i++ {
+			// memoryLeaking()
+			time.Sleep(time.Second)
+		}
+	}()
+
+	time.Sleep(time.Second * 10)
+	// reload the config
+	_ = a.Apply(xcheck.WithMemOpts(1024, 10, 20, time.Hour))
+
+	time.Sleep(time.Hour)
+}
+
+// memoryLeaking 16MB
+func memoryLeaking() {
+	type T struct {
+		v [1 << 20]int
+		t *T
+	}
+
+	var finalizer = func(t *T) {
+		fmt.Println("finalizer called")
+	}
+
+	var x, y T
+
+	// 此SetFinalizer函数调用将使x逃逸到堆上。
+	runtime.SetFinalizer(&x, finalizer)
+
+	// 下面这行将形成一个包含x和y的循环引用值组。
+	// 这有可能造成x和y不可回收。
+	x.t, y.t = &y, &x // y也逃逸到了堆上。
+}
diff --git a/xcheck/go.mod b/xcheck/go.mod
@@ -0,0 +1,33 @@
+module github.com/gotomicro/cetus/xcheck
+
+go 1.19
+
+require (
+	github.com/go-resty/resty/v2 v2.7.0
+	github.com/gotomicro/ego v1.1.11
+	github.com/shirou/gopsutil/v3 v3.21.6
+)
+
+require (
+	github.com/StackExchange/wmi v0.0.0-20190523213315-cbe66965904d // indirect
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/go-logr/logr v1.2.3 // indirect
+	github.com/go-logr/stdr v1.2.2 // indirect
+	github.com/go-ole/go-ole v1.2.6 // indirect
+	github.com/google/go-cmp v0.5.9 // indirect
+	github.com/gotomicro/logrotate v0.0.0-20211108034117-46d53eedc960 // indirect
+	github.com/mitchellh/mapstructure v1.5.0 // indirect
+	github.com/spf13/cast v1.4.1 // indirect
+	github.com/stretchr/testify v1.8.3 // indirect
+	github.com/tklauser/go-sysconf v0.3.6 // indirect
+	github.com/tklauser/numcpus v0.2.2 // indirect
+	go.opentelemetry.io/otel v1.7.0 // indirect
+	go.opentelemetry.io/otel/trace v1.7.0 // indirect
+	go.uber.org/atomic v1.7.0 // indirect
+	go.uber.org/multierr v1.6.0 // indirect
+	go.uber.org/zap v1.21.0 // indirect
+	golang.org/x/net v0.10.0 // indirect
+	golang.org/x/sys v0.8.0 // indirect
+	google.golang.org/grpc v1.46.0 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
+)
diff --git a/xcheck/go.sum b/xcheck/go.sum
diff --git a/xcheck/model/dto/forwarder.go b/xcheck/model/dto/forwarder.go
@@ -0,0 +1,21 @@
+package dto
+
+type Forwarder struct {
+	WebHook Webhook `json:"webhook"`
+}
+
+type Webhook struct {
+	Url     string                 `json:"url"`
+	Headers map[string]string      `json:"headers"`
+	Body    map[string]interface{} `json:"body"`
+}
+
+type AttachInfo struct {
+	CurrentAbs  uint64 `json:"currentAbs,omitempty"`
+	CurrentDiff int    `json:"currentDiff,omitempty"`
+
+	// Rule
+	OptAbs            uint64 `json:"optAbs,omitempty"`
+	OptDiff           uint64 `json:"optDiff,omitempty"`
+	OptCoolingTimeSec int    `json:"optCoolingTimeSec,omitempty"`
+}
diff --git a/xcheck/model/dto/monitor.go b/xcheck/model/dto/monitor.go
@@ -0,0 +1,3 @@
+package dto
+
+type Monitor struct{}
diff --git a/xcheck/module/forward/webhook/webhook.go b/xcheck/module/forward/webhook/webhook.go
@@ -0,0 +1,33 @@
+package webhook
+
+import (
+	"github.com/go-resty/resty/v2"
+	"github.com/gotomicro/ego/core/elog"
+
+	"github.com/gotomicro/cetus/xcheck/model/utils"
+
+	"github.com/gotomicro/cetus/xcheck/model/dto"
+)
+
+func Webhook(fw dto.Webhook, attach dto.AttachInfo) {
+	client := resty.New()
+	r := client.R().SetHeader("Content-Type", "application/json")
+	for k, v := range fw.Headers {
+		r.SetHeader(k, v)
+	}
+	if _, ok := fw.Body["mode"]; ok {
+		ip, err := utils.GetOutBoundIP()
+		if err != nil {
+			elog.Error("forward", elog.FieldEvent("GetOutBoundIP"), elog.FieldErr(err), elog.String("msg", "get outbound ip error"))
+			return
+		}
+		fw.Body["addr"] = ip + ":9003"
+		elog.Info("xcheck", elog.Any("size", attach.CurrentAbs), elog.Any("attach", attach), elog.String("ip", ip), elog.Any("body", fw.Body))
+	}
+	r.SetBody(fw.Body)
+	resp, err := r.Post(fw.Url)
+	if err != nil {
+		elog.Error("forward", elog.FieldEvent("post"), elog.FieldErr(err), elog.Any("fw", fw), elog.String("resp", resp.String()))
+		return
+	}
+}
diff --git a/xcheck/module/monitor/monitor.go b/xcheck/module/monitor/monitor.go
@@ -0,0 +1,28 @@
+package monitor
+
+import (
+	"os"
+
+	"github.com/shirou/gopsutil/v3/process"
+
+	"github.com/gotomicro/cetus/xcheck/model/dto"
+)
+
+func New(m dto.Monitor) {
+
+}
+
+// ReadMemStats unit is MB
+func ReadMemStats() (uint64, float64) {
+	p, err := process.NewProcess(int32(os.Getpid()))
+	if err != nil {
+		return 0, 0
+	}
+	mem, err := p.MemoryInfo()
+	if err != nil {
+		return 0, 0
+	}
+	// fmt.Printf("TotalAlloc = %v MiB", m.TotalAlloc/1024/1024)
+	// usedPercent, _ := strconv.ParseFloat(fmt.Sprintf("%.2f", float64(mem.RSS)/float64(m.Sys)), 64)
+	return mem.RSS / 1024 / 1024, 0
+}
diff --git a/xcheck/module/monitor/monitor_test.go b/xcheck/module/monitor/monitor_test.go
@@ -0,0 +1,22 @@
+package monitor
+
+import (
+	"testing"
+)
+
+func TestReadMemStats(t *testing.T) {
+	tests := []struct {
+		name string
+		want uint64
+	}{
+		// TODO: Add test cases.
+		{"test1", 0},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got, _ := ReadMemStats(); got != tt.want {
+				t.Errorf("ReadMemStats() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
diff --git a/xcheck/options.go b/xcheck/options.go
@@ -0,0 +1,70 @@
+package xcheck
+
+import (
+	"time"
+)
+
+type options struct {
+	memOpts *cubeOption
+}
+
+type Option interface {
+	apply(*options) error
+}
+
+type optionFunc func(*options) error
+
+func (f optionFunc) apply(opts *options) error {
+	return f(opts)
+}
+
+func newOptions() *options {
+	return &options{
+		memOpts: newMemOptions(),
+	}
+}
+
+func newMemOptions() *cubeOption {
+	return newCubeOpts(defaultMemTriggerValue, defaultMemTriggerPercent, defaultMemTriggerDiff, defaultCoolingTime)
+}
+
+func WithMemOpts(value, percent, diff uint64, coolingTime time.Duration) Option {
+	return optionFunc(func(opts *options) error {
+		opts.memOpts.Set(value, percent, diff, coolingTime)
+		return nil
+	})
+}
+
+type cubeOption struct {
+	Enable         bool
+	TriggerValue   uint64
+	TriggerPercent uint64
+	TriggerDiff    uint64
+	CoolingTime    time.Duration
+}
+
+func newCubeOpts(triggerValue, triggerPercent, triggerDiff uint64, coolingTime time.Duration) *cubeOption {
+	return &cubeOption{
+		Enable:         false,
+		TriggerValue:   triggerValue,
+		TriggerPercent: triggerPercent,
+		TriggerDiff:    triggerDiff,
+		CoolingTime:    coolingTime,
+	}
+}
+
+func (cube *cubeOption) Set(value, percent, diff uint64, coolingTime time.Duration) {
+	if coolingTime == 0 {
+		coolingTime = defaultCoolingTime
+	}
+	if value == 0 {
+		value = defaultMemTriggerValue
+	}
+	if diff == 0 {
+		diff = defaultMemTriggerDiff
+	}
+	cube.TriggerValue = value
+	cube.TriggerPercent = percent
+	cube.TriggerDiff = diff
+	cube.CoolingTime = coolingTime
+}
diff --git a/xcheck/xcheck.go b/xcheck/xcheck.go
@@ -0,0 +1,97 @@
+package xcheck
+
+import (
+	"time"
+
+	"github.com/gotomicro/ego/core/elog"
+
+	"github.com/gotomicro/cetus/xcheck/model/dto"
+	"github.com/gotomicro/cetus/xcheck/module/forward/webhook"
+	"github.com/gotomicro/cetus/xcheck/module/monitor"
+)
+
+type xcheck struct {
+	opts           *options
+	memAvg         uint64
+	memCoolingTime time.Time
+
+	Forwarder dto.Forwarder
+}
+
+// New is the entry of xcheck
+// 1. start monitor
+// 2. storage data
+// 3. calculate data
+// 4. generate pprof files
+// 5. upload to oss
+func New(opts ...Option) (*xcheck, error) {
+	xcheck := &xcheck{
+		opts: newOptions(),
+	}
+	for _, o := range opts {
+		if err := o.apply(xcheck.opts); err != nil {
+			return nil, err
+		}
+	}
+	return xcheck, nil
+}
+
+func (a *xcheck) Apply(opts ...Option) error {
+	for _, o := range opts {
+		if err := o.apply(a.opts); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// EnableMem enables the mem dump.
+func (a *xcheck) EnableMem() *xcheck {
+	a.opts.memOpts.Enable = true
+	return a
+}
+
+func (a *xcheck) Start() error {
+	ticker := time.NewTicker(time.Second)
+	go func() {
+		for {
+			select {
+			case <-ticker.C:
+				a.calculateData()
+			}
+		}
+	}()
+	return nil
+}
+
+func (a *xcheck) calculateData() {
+	current, usedPercent := monitor.ReadMemStats()
+
+	if a.memAvg == 0 {
+		a.memAvg = current
+		return
+	}
+	diff := (float64(current) - float64(a.memAvg)) * 100 / float64(a.memAvg)
+
+	elog.Debug("cal", elog.Int("avg", int(a.memAvg)), elog.Int("size", int(current)), elog.Int("diffPercent", int(diff)), elog.Int("usedPercent", int(usedPercent)), elog.Any("memOpts", a.opts.memOpts))
+
+	if current >= a.opts.memOpts.TriggerValue && uint64(diff) >= a.opts.memOpts.TriggerDiff {
+		a.pprof(dto.AttachInfo{
+			CurrentAbs:        current,
+			CurrentDiff:       int(diff),
+			OptAbs:            a.opts.memOpts.TriggerValue,
+			OptDiff:           a.opts.memOpts.TriggerDiff,
+			OptCoolingTimeSec: int(a.opts.memOpts.CoolingTime.Seconds()),
+		})
+	}
+	a.memAvg = (a.memAvg + current) / 2
+}
+
+func (a *xcheck) pprof(attach dto.AttachInfo) {
+	if a.memCoolingTime.After(time.Now()) {
+		elog.Info("coolingTime")
+		return
+	}
+	a.memCoolingTime = time.Now().Add(a.opts.memOpts.CoolingTime)
+	webhook.Webhook(a.Forwarder.WebHook, attach)
+}
diff --git a/xnet/ip.go b/xnet/ip.go
diff --git a/xnet/ip_test.go b/xnet/ip_test.go

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,3 @@`
	`1`	`+package dto`
	`2`	`+`
	`3`	`+type Monitor struct{}`