Merge pull request #64 from topfreegames/chore/add-go-stats-reporter

chore: add go stats reporter and pprof server

.gitignore

@@ -38,6 +38,7 @@ vendor/
 *.swo
 .*swp
 
+.vscode
 .idea
 
 .tool-versions

config/default.yaml

@@ -55,6 +55,10 @@ stats:
     host: "localhost:8125"
     prefix: "push"
     buflen: 1
+pprof:
+  enabled: false
+  host: "127.0.0.1"
+  port: 8081
 invalidToken:
   pg:
     table: "test_apns"

config/docker_test.yaml

@@ -54,6 +54,10 @@ stats:
     host: "statsd:8125"
     prefix: "push"
     flushIntervalMs: 5000
+pprof:
+  enabled: false
+  host: "127.0.0.1"
+  port: 8081
 invalidToken:
   handlers:
     - pg

config/test.yaml

@@ -51,7 +51,11 @@ stats:
   statsd:
     host: "localhost:40001"
     prefix: "push"
-    flushIntervalMs: 5000
+    flushIntervalMs: 
+pprof:
+  enabled: false
+  host: "127.0.0.1"
+  port: 8081
 invalidToken:
   handlers:
     - pg

feedback/listener.go

@@ -25,8 +25,11 @@ package feedback
 import (
 	"context"
 	"fmt"
+	"net/http"
+	"net/http/pprof"
 	"os"
 	"os/signal"
+	"runtime"
 	"sync"
 	"syscall"
 	"time"
@@ -75,6 +78,10 @@ func NewListener(
 func (l *Listener) loadConfigurationDefaults() {
 	l.Config.SetDefault("feedbackListeners.gracefulShutdownTimeout", 1)
 	l.Config.SetDefault("stats.flush.s", 5)
+
+	l.Config.SetDefault("pprof.enabled", false)
+	l.Config.SetDefault("pprof.host", "127.0.0.1")
+	l.Config.SetDefault("pprof.port", 8081)
 }
 
 func (l *Listener) configure(statsdClientrOrNil interfaces.StatsDClient) error {
@@ -128,14 +135,20 @@ func (l *Listener) Start() {
 	)
 	log.Info("starting the feedback listener...")
 
+	if l.Config.GetBool("pprof.enabled") {
+		go l.StartPprofServer()
+	}
+
 	go l.Queue.ConsumeLoop(context.Background())
 	l.Broker.Start()
 	l.InvalidTokenHandler.Start()
 
+	go l.reportGoStats()
+
 	statsReporterReportMetricCount(l.StatsReporters,
 		"feedback_listener_restart", 1, "", "")
 
-	sigchan := make(chan os.Signal)
+	sigchan := make(chan os.Signal, 1)
 	signal.Notify(sigchan, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
 
 	flushTicker := time.NewTicker(l.statsFlushTime)
@@ -172,6 +185,23 @@ func (l *Listener) flushStats() {
 		"invalid_token_handler_buffer", bufferSize, "", "")
 }
 
+func (l *Listener) reportGoStats() {
+	for {
+		num := runtime.NumGoroutine()
+		m := &runtime.MemStats{}
+		runtime.ReadMemStats(m)
+		gcTime := m.PauseNs[(m.NumGC+255)%256]
+		for _, statsReporter := range l.StatsReporters {
+			statsReporter.ReportGoStats(
+				num,
+				m.Alloc, m.HeapObjects, m.NextGC,
+				gcTime,
+			)
+		}
+		time.Sleep(30 * time.Second)
+	}
+}
+
 // Cleanup ends the Listener execution
 func (l *Listener) Cleanup() {
 	l.flushStats()
@@ -217,3 +247,21 @@ func WaitTimeout(wg *sync.WaitGroup, timeout time.Duration) bool {
 		return true // timed out
 	}
 }
+
+func (l *Listener) StartPprofServer() error {
+	mux := http.NewServeMux()
+
+	mux.HandleFunc("/debug/pprof/", pprof.Index)
+	mux.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline)
+	mux.HandleFunc("/debug/pprof/profile", pprof.Profile)
+	mux.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
+	mux.HandleFunc("/debug/pprof/trace", pprof.Trace)
+
+	addr := fmt.Sprintf("%s:%d", l.Config.GetString("pprof.host"), l.Config.GetInt("pprof.port"))
+	server := &http.Server{
+		Addr:    addr,
+		Handler: mux,
+	}
+
+	return server.ListenAndServe()
+}

pusher/pusher.go

@@ -118,7 +118,7 @@ func (p *Pusher) Start(ctx context.Context) {
 	go p.Queue.ConsumeLoop(ctx)
 	go p.reportGoStats()
 
-	sigchan := make(chan os.Signal)
+	sigchan := make(chan os.Signal, 1)
 	signal.Notify(sigchan, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
 
 	select {