feat: view the regular cpu usage by using prometheus

go.mod

@@ -4,7 +4,7 @@ require (
 	github.com/Masterminds/sprig/v3 v3.2.3
 	github.com/PaesslerAG/gval v1.2.2
 	github.com/PaesslerAG/jsonpath v0.1.1
-	github.com/Rookiecom/cpuprofile v1.0.2
+	github.com/Rookiecom/cpuprofile v1.0.4
 	github.com/alicebob/miniredis/v2 v2.30.0
 	github.com/apple/foundationdb/bindings/go v0.0.0-20240315202255-8943393e84fc
 	github.com/benbjohnson/clock v1.3.5
@@ -89,7 +89,7 @@ require (
 	github.com/go-playground/validator/v10 v10.19.0 // indirect
 	github.com/go-redis/redis/v7 v7.4.1 // indirect
 	github.com/go-sourcemap/sourcemap v2.1.4+incompatible // indirect
-	github.com/google/pprof v0.0.0-20240528025155-186aa0362fba // indirect
+	github.com/google/pprof v0.0.0-20240711041743-f6c9dda6c6da // indirect
 	github.com/hashicorp/errwrap v1.1.0 // indirect
 	github.com/hashicorp/go-multierror v1.1.1 // indirect
 	github.com/hashicorp/golang-lru v0.5.4 // indirect

go.sum

@@ -21,8 +21,8 @@ github.com/PaesslerAG/gval v1.2.2/go.mod h1:XRFLwvmkTEdYziLdaCeCa5ImcGVrfQbeNUbV
 github.com/PaesslerAG/jsonpath v0.1.0/go.mod h1:4BzmtoM/PI8fPO4aQGIusjGxGir2BzcV0grWtFzq1Y8=
 github.com/PaesslerAG/jsonpath v0.1.1 h1:c1/AToHQMVsduPAa4Vh6xp2U0evy4t8SWp8imEsylIk=
 github.com/PaesslerAG/jsonpath v0.1.1/go.mod h1:lVboNxFGal/VwW6d9JzIy56bUsYAP6tH/x80vjnCseY=
-github.com/Rookiecom/cpuprofile v1.0.2 h1:wXGOmpSyCfoi7D0OB79srblNSw5Ccn0fTy4c537yWy4=
-github.com/Rookiecom/cpuprofile v1.0.2/go.mod h1:JEBcjNHceN6/g4ppueDEqWlbtbMGMyLy0YlT8QnKFbE=
+github.com/Rookiecom/cpuprofile v1.0.4 h1:F2eQVuVlHE5XnpNbAk9oBD9vdHbBusaAaAsbu23OGco=
+github.com/Rookiecom/cpuprofile v1.0.4/go.mod h1:JEBcjNHceN6/g4ppueDEqWlbtbMGMyLy0YlT8QnKFbE=
 github.com/Shopify/sarama v1.19.0/go.mod h1:FVkBWblsNy7DGZRfXLU0O9RCGt5g3g3yEuWXgklEdEo=
 github.com/Shopify/toxiproxy v2.1.4+incompatible/go.mod h1:OXgGpZ6Cli1/URJOF1DMxUHB2q5Ap20/P/eIdh4G0pI=
 github.com/VividCortex/gohistogram v1.0.0 h1:6+hBz+qvs0JOrrNhhmR7lFxo5sINxBCGXrdtl/UvroE=
@@ -223,8 +223,8 @@ github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
 github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/google/pprof v0.0.0-20230207041349-798e818bf904/go.mod h1:uglQLonpP8qtYCYyzA+8c/9qtqgA3qsXGYqCPKARAFg=
-github.com/google/pprof v0.0.0-20240528025155-186aa0362fba h1:ql1qNgCyOB7iAEk8JTNM+zJrgIbnyCKX/wdlyPufP5g=
-github.com/google/pprof v0.0.0-20240528025155-186aa0362fba/go.mod h1:K1liHPHnj73Fdn/EKuT8nrFqBihUSKXoLYU0BuatOYo=
+github.com/google/pprof v0.0.0-20240711041743-f6c9dda6c6da h1:xRmpO92tb8y+Z85iUOMOicpCfaYcv7o3Cg3wKrIpg8g=
+github.com/google/pprof v0.0.0-20240711041743-f6c9dda6c6da/go.mod h1:K1liHPHnj73Fdn/EKuT8nrFqBihUSKXoLYU0BuatOYo=
 github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
 github.com/google/uuid v1.0.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=

internal/server/promMetrics/prom_metrcis.go

@@ -29,6 +29,7 @@ const (
 var (
 	RuleStatusCountGauge *prometheus.GaugeVec
 	RuleStatusGauge      *prometheus.GaugeVec
+	RuleCPUUsageGauge    *prometheus.GaugeVec
 )
 
 func InitServerMetrics() {
@@ -45,12 +46,20 @@ func InitServerMetrics() {
 		Name:      "status",
 		Help:      "gauge of rule status",
 	}, []string{LblRuleIDType})
+
+	RuleCPUUsageGauge = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Namespace: "kuiper",
+		Subsystem: "rule",
+		Name:      "cpuUsage",
+		Help:      "gauge of rule CPU usage",
+	}, []string{LblRuleIDType})
 }
 
 func RegisterMetrics() {
 	InitServerMetrics()
 	prometheus.MustRegister(RuleStatusCountGauge)
 	prometheus.MustRegister(RuleStatusGauge)
+	prometheus.MustRegister(RuleCPUUsageGauge)
 }
 
 func SetRuleStatusCountGauge(isRunning bool, count int) {
@@ -69,3 +78,7 @@ func SetRuleStatus(ruleID string, value int) {
 func RemoveRuleStatus(ruleID string) {
 	RuleStatusGauge.DeleteLabelValues(ruleID)
 }
+
+func SetRuleCPUUsageGauge(ruleID string, value int) {
+	RuleCPUUsageGauge.WithLabelValues(ruleID).Set(float64(value))
+}

internal/server/server.go

@@ -481,22 +481,33 @@ func handleScheduleRule(now time.Time, r *def.Rule, state string) scheduleRuleAc
 }
 
 func startCPUProfiling(ctx context.Context) error {
-	if err := cpuprofile.StartProfilerAndAggregater(ctx, time.Duration(1000)*time.Millisecond); err != nil {
+	cpuprofile.EnableWindowAggregator(30)
+	if err := cpuprofile.StartCPUProfiler(ctx, time.Duration(1000)*time.Millisecond); err != nil {
 		return err
 	}
-	receiveChan := make(chan *cpuprofile.DataSetAggregate, 1024)
-	cpuprofile.RegisterTag("rule", receiveChan)
 	go func(ctx context.Context) {
+		ticker := time.NewTicker(time.Duration(15) * time.Second)
 		for {
 			select {
 			case <-ctx.Done():
+				ticker.Stop()
 				return
-			case data := <-receiveChan:
-				// TODO: support query in future
-				conf.Log.Debugf("cpu profile data: %v", data)
+			case <-ticker.C:
+				data := cpuprofile.GetWindowData()
+				if data == nil {
+					continue
+				}
+				ruleUsage, ok := data["rule"]
+				if !ok {
+					continue
+				}
+				for labelValue, t := range ruleUsage.Stats {
+					promMetrics.SetRuleCPUUsageGauge(labelValue, t)
+				}
 			}
 		}
 	}(ctx)
+
 	return nil
 }