Enable holmes as pyroscope client and reports pprof event to pyroscope server

example/pyroscope_rideshare/README.md

@@ -0,0 +1,21 @@
+
+Enable holmes as pyroscope client and reports pprof 
+event to pyroscope server.
+
+note: CAN NOT set TextDump while using holmes as pyroscope client, 
+bcs pyroscope need profile in proto format.
+
+Step 1
+
+``docker run -it -p 4040:4040 pyroscope/pyroscope:latest server``
+
+open browser on [pyroscope admin page](http://localhost:4040/)
+
+Step 2
+run the script `start_client.sh` at `rideshare/`
+
+Step 3
+wait 15 seconds, refresh pyroscope admin page, select 
+`holmes-client` on the `Application` box as the following.
+![admin](./admin.png)
+

example/pyroscope_rideshare/bike/bike.go

@@ -0,0 +1,10 @@
+package bike
+
+import "rideshare/utility"
+
+func OrderBike(searchRadius int64) {
+	utility.FindNearestVehicle(searchRadius, "bike")
+	for i := 0; i < 3; i++ {
+		go utility.AllocMem()
+	}
+}

example/pyroscope_rideshare/car/car.go

@@ -0,0 +1,9 @@
+package car
+
+import (
+	"rideshare/utility"
+)
+
+func OrderCar(searchRadius int64) {
+	utility.FindNearestVehicle(searchRadius, "car")
+}

example/pyroscope_rideshare/go.mod

@@ -0,0 +1,7 @@
+module rideshare
+
+go 1.14
+
+require mosn.io/holmes v1.1.0
+
+replace mosn.io/holmes => ../../

example/pyroscope_rideshare/main.go

@@ -0,0 +1,104 @@
+package main
+
+import (
+	"fmt"
+	"net/http"
+	"os"
+	"time"
+
+	"mosn.io/holmes"
+	"mosn.io/holmes/reporters/pyroscope_reporter"
+	"rideshare/bike"
+	"rideshare/car"
+	"rideshare/scooter"
+)
+
+func bikeRoute(w http.ResponseWriter, r *http.Request) {
+	bike.OrderBike(1)
+	w.Write([]byte("<h1>Bike ordered</h1>"))
+}
+
+func scooterRoute(w http.ResponseWriter, r *http.Request) {
+	scooter.OrderScooter(2)
+	w.Write([]byte("<h1>Scooter ordered</h1>"))
+}
+
+func carRoute(w http.ResponseWriter, r *http.Request) {
+	car.OrderCar(3)
+
+	w.Write([]byte("<h1>Car ordered</h1>"))
+}
+
+func index(w http.ResponseWriter, r *http.Request) {
+	result := "<h1>environment vars:</h1>"
+	for _, env := range os.Environ() {
+		result += env + "<br>"
+	}
+	w.Write([]byte(result))
+}
+
+var h *holmes.Holmes
+
+func InitHolmes() {
+	fmt.Println("holmes initialing")
+	h, _ = holmes.New(
+		holmes.WithCollectInterval("1s"),
+		holmes.WithDumpPath("./log/"),
+		// can not set text in pyroscope client
+	)
+	fmt.Println("holmes initial success")
+	h.
+		EnableCPUDump().
+		EnableGoroutineDump().
+		EnableMemDump().
+		Start()
+	time.Sleep(11 * time.Second)
+	fmt.Println("on running")
+}
+
+func main() {
+	InitHolmes()
+	region := os.Getenv("region")
+	port := os.Getenv("port")
+	fmt.Printf("region is %v port is %v \n", region, port)
+	cfg := pyroscope_reporter.RemoteConfig{
+		//AuthToken:              "",
+		//UpstreamThreads:        4,
+		UpstreamAddress:        "http://localhost:4040",
+		UpstreamRequestTimeout: 3 * time.Second,
+	}
+
+	tags := map[string]string{
+		"region": region,
+	}
+
+	pReporter, err := pyroscope_reporter.NewPyroscopeReporter("holmes-client", tags, cfg, holmes.NewStdLogger())
+	if err != nil {
+		fmt.Printf("NewPyroscopeReporter error %v\n", err)
+		return
+	}
+
+	err = h.Set(
+		holmes.WithProfileReporter(pReporter),
+		holmes.WithGoroutineDump(2, 2, 20, 90, 20*time.Second),
+		holmes.WithCPUDump(2, 2, 80, 20*time.Second),
+		holmes.WithMemDump(1, 2, 80, 20*time.Second),
+		holmes.WithCollectInterval("5s"),
+	)
+	if err != nil {
+		fmt.Printf("fail to set opts on running time.\n")
+		return
+	}
+
+	http.HandleFunc("/", index)
+	http.HandleFunc("/bike", bikeRoute)
+	http.HandleFunc("/scooter", scooterRoute)
+	http.HandleFunc("/car", carRoute)
+	err = http.ListenAndServe(":"+port, nil)
+	if err != nil {
+		panic(err)
+	}
+
+	time.Sleep(1 * time.Minute)
+
+}

example/pyroscope_rideshare/requests.py

@@ -0,0 +1,26 @@
+import random
+import requests
+import time
+
+PORTS = [
+    '15011',
+    '15012',
+    '15013',
+]
+
+VEHICLES = [
+    'bike',
+    'scooter',
+    'car',
+]
+
+if __name__ == "__main__":
+    print(f"starting load generator")
+    time.sleep(3)
+    while True:
+        port = PORTS[random.randint(0, len(PORTS) - 1)]
+        vehicle = VEHICLES[random.randint(0, len(VEHICLES) - 1)]
+        print(f"requesting {vehicle} from {port}")
+        resp = requests.get(f'http://localhost:{port}/{vehicle}')
+        print(f"received {resp}")
+        time.sleep(random.uniform(0.2, 0.4))

example/pyroscope_rideshare/scooter/scooter.go

@@ -0,0 +1,7 @@
+package scooter
+
+import "rideshare/utility"
+
+func OrderScooter(searchRadius int64) {
+	utility.FindNearestVehicle(searchRadius, "scooter")
+}

example/pyroscope_rideshare/start_client.sh

@@ -0,0 +1,10 @@
+
+region=us-east;port=15011 go run main.go &
+region=eu-north;port=15012 go run main.go &
+region=ap-south;port=15013 go run main.go &
+
+echo "wait holmes client init"
+sleep 15s
+echo "init done, start to send request"
+
+python3 requests.py

example/pyroscope_rideshare/utility/utility.go

@@ -0,0 +1,71 @@
+package utility
+
+import (
+	"os"
+	"time"
+)
+
+const durationConstant = time.Duration(200 * time.Millisecond)
+
+func mutexLock(n int64) {
+	var i int64 = 0
+
+	// start time is number of seconds since epoch
+	startTime := time.Now()
+
+	// This changes the amplitude of cpu bars
+	for time.Since(startTime) < time.Duration(n*30)*durationConstant {
+		i++
+	}
+}
+
+func checkDriverAvailability(n int64) {
+	var i int64 = 0
+
+	// start time is number of seconds since epoch
+	startTime := time.Now()
+
+	for time.Since(startTime) < time.Duration(n)*durationConstant {
+		i++
+	}
+
+	// Every other minute this will artificially create make requests.py in eu-north region slow
+	// this is just for demonstration purposes to show how performance impacts show up in the
+	// flamegraph
+	force_mutex_lock := time.Now().Minute()%2 == 0
+	if os.Getenv("REGION") == "eu-north" && force_mutex_lock {
+		mutexLock(n)
+	}
+
+}
+
+func FindNearestVehicle(searchRadius int64, vehicle string) {
+	//pyroscope.TagWrapper(context.Background(), pyroscope.Labels("vehicle", vehicle), func(ctx context.Context) {
+	var i int64 = 0
+
+	startTime := time.Now()
+	for time.Since(startTime) < time.Duration(searchRadius)*durationConstant {
+		i++
+	}
+
+	if vehicle == "car" {
+		checkDriverAvailability(searchRadius)
+		go func() {
+			go AllocMem()
+		}()
+	}
+	if vehicle == "bike" {
+		for i := 1; i < 10; i++ {
+			go func() {
+				time.Sleep(15 * time.Second)
+			}()
+		}
+	}
+	//})
+}
+
+func AllocMem() {
+	var a = make([]byte, 1073741824)
+	_ = a
+	time.Sleep(10 * time.Second)
+}

holmes.go

@@ -72,7 +72,7 @@ type Holmes struct {
 }
 
 type ProfileReporter interface {
-	Report(pType string, filename string, reason string, eventID string) error
+	Report(pType string, filename string, reason string, eventID string, sampleTime time.Time, pprofBytes []byte) error
 }
 
 // New creates a holmes dumper.
@@ -375,7 +375,8 @@ func (h *Holmes) goroutineProfile(gNum int, c grOptions) bool {
 	var buf bytes.Buffer
 	_ = pprof.Lookup("goroutine").WriteTo(&buf, int(h.opts.DumpProfileType)) // nolint: errcheck
 
-	h.ReportProfile(type2name[goroutine], h.writeProfileDataToFile(buf, goroutine, ""), reason, "")
+	h.ReportProfile(type2name[goroutine], h.writeProfileDataToFile(buf, goroutine, ""),
+		reason, "", time.Now(), buf.Bytes())
 	return true
 }
 
@@ -416,7 +417,7 @@ func (h *Holmes) memProfile(rss int, c typeOption) bool {
 	var buf bytes.Buffer
 	_ = pprof.Lookup("heap").WriteTo(&buf, int(h.opts.DumpProfileType)) // nolint: errcheck
 
-	h.ReportProfile(type2name[mem], h.writeProfileDataToFile(buf, mem, ""), reason, "")
+	h.ReportProfile(type2name[mem], h.writeProfileDataToFile(buf, mem, ""), reason, "", time.Now(), buf.Bytes())
 	return true
 }
 
@@ -536,12 +537,14 @@ func (h *Holmes) threadProfile(curThreadNum int, c typeOption) bool {
 
 	_ = pprof.Lookup("threadcreate").WriteTo(&buf, int(h.opts.DumpProfileType)) // nolint: errcheck
 
-	h.ReportProfile(type2name[thread], h.writeProfileDataToFile(buf, thread, eventID), reason, eventID)
+	h.ReportProfile(type2name[thread], h.writeProfileDataToFile(buf, thread, eventID),
+		reason, eventID, time.Now(), buf.Bytes())
 
 	buf.Reset()
 	_ = pprof.Lookup("goroutine").WriteTo(&buf, int(h.opts.DumpProfileType)) // nolint: errcheck
 
-	h.ReportProfile(type2name[goroutine], h.writeProfileDataToFile(buf, goroutine, eventID), reason, eventID)
+	h.ReportProfile(type2name[goroutine], h.writeProfileDataToFile(buf, goroutine, eventID),
+		reason, eventID, time.Now(), buf.Bytes())
 
 	return true
 }
@@ -597,17 +600,22 @@ func (h *Holmes) cpuProfile(curCPUUsage int, c typeOption) bool {
 	time.Sleep(defaultCPUSamplingTime)
 	pprof.StopCPUProfile()
 
-	if h.opts.DumpToLogger {
-		bfCpy, err := ioutil.ReadFile(binFileName)
+	rptOpts, bfCpy := h.opts.GetReporterOpts(), []byte{}
+	if h.opts.DumpToLogger || rptOpts.active == 1 {
+		bfCpy, err = ioutil.ReadFile(binFileName)
 		if err != nil {
 			h.Errorf("encounter error when dumping profile to logger, failed to read cpu profile file: %v", err)
 			return true
 		}
+	}
+
+	if h.opts.DumpToLogger {
 		h.Infof("[Holmes] CPU profile name : " + "::" + binFileName + " \n" + string(bfCpy))
 	}
 
-	if opts := h.opts.GetReporterOpts(); opts.active == 1 {
-		h.ReportProfile(type2name[cpu], binFileName, reason, "")
+	if rptOpts.active == 1 {
+		h.ReportProfile(type2name[cpu], binFileName,
+			reason, "", time.Now(), bfCpy)
 	}
 
 	return true
@@ -723,7 +731,8 @@ func (h *Holmes) gcHeapProfile(gc int, force bool, c typeOption) bool {
 	var buf bytes.Buffer
 	_ = pprof.Lookup("heap").WriteTo(&buf, int(h.opts.DumpProfileType)) // nolint: errcheck
 
-	h.ReportProfile(type2name[gcHeap], h.writeProfileDataToFile(buf, gcHeap, eventID), reason, eventID)
+	h.ReportProfile(type2name[gcHeap], h.writeProfileDataToFile(buf, gcHeap, eventID),
+		reason, eventID, time.Now(), buf.Bytes())
 	return true
 }
 
@@ -784,7 +793,7 @@ func (h *Holmes) EnableProfileReporter() {
 	atomic.StoreInt32(&h.opts.rptOpts.active, 1)
 }
 
-func (h *Holmes) ReportProfile(pType string, filename string, reason string, eventID string) {
+func (h *Holmes) ReportProfile(pType string, filename string, reason string, eventID string, sampleTime time.Time, pprofBytes []byte) {
 	if filename == "" {
 		h.Errorf("dump name is empty, type:%s, reason:%s, eventID:%s", pType, reason, eventID)
 		return
@@ -806,10 +815,12 @@ func (h *Holmes) ReportProfile(pType string, filename string, reason string, eve
 	}
 
 	msg := rptEvent{
-		PType:    pType,
-		FileName: filename,
-		Reason:   reason,
-		EventID:  eventID,
+		PType:      pType,
+		FileName:   filename,
+		Reason:     reason,
+		EventID:    eventID,
+		SampleTime: sampleTime,
+		PprofBytes: pprofBytes,
 	}
 
 	// read channel should be atomic.
@@ -837,8 +848,9 @@ func (h *Holmes) startReporter(ch chan rptEvent) {
 				// drop the event
 				continue
 			}
+
 			// It's supposed to be sending judgment, isn't it?
-			err := opts.reporter.Report(evt.PType, evt.FileName, evt.Reason, evt.EventID) // nolint: errcheck
+			err := opts.reporter.Report(evt.PType, evt.FileName, evt.Reason, evt.EventID, evt.SampleTime, evt.PprofBytes) // nolint: errcheck
 			if err != nil {
 				h.Infof("reporter err:", err)