Merge pull request #22 from yandex/expvar

publish some metrics to expvar (close #13)

ammo/expvar.go

@@ -0,0 +1,55 @@
+package ammo
+
+import (
+	"expvar"
+	"log"
+	"strconv"
+	"sync/atomic"
+	"time"
+)
+
+type Counter struct {
+	i int64
+}
+
+func (c *Counter) String() string {
+	return strconv.FormatInt(atomic.LoadInt64(&c.i), 10)
+}
+
+func (c *Counter) Add(delta int64) {
+	atomic.AddInt64(&c.i, delta)
+}
+
+func (c *Counter) Set(value int64) {
+	atomic.StoreInt64(&c.i, value)
+}
+
+func (c *Counter) Get() int64 {
+	return atomic.LoadInt64(&c.i)
+}
+
+func NewCounter(name string) *Counter {
+	v := &Counter{}
+	expvar.Publish(name, v)
+	return v
+}
+
+var (
+	evPassesLeft = NewCounter("ammo_PassesLeft")
+)
+
+func init() {
+	go func() {
+		passesLeft := evPassesLeft.Get()
+		for _ = range time.NewTicker(1 * time.Second).C {
+			if passesLeft < 0 {
+				return // infinite number of passes
+			}
+			newPassesLeft := evPassesLeft.Get()
+			if newPassesLeft != passesLeft {
+				log.Printf("[AMMO] passes left: %d", newPassesLeft)
+				passesLeft = newPassesLeft
+			}
+		}
+	}()
+}

ammo/http.go

@@ -71,9 +71,11 @@ loop:
 		}
 		ammoFile.Seek(0, 0)
 		if ap.passes == 0 {
-			log.Printf("Restarted ammo from the beginning. Infinite passes.\n")
+			evPassesLeft.Set(-1)
+			//log.Printf("Restarted ammo from the beginning. Infinite passes.\n")
 		} else {
-			log.Printf("Restarted ammo from the beginning. Passes left: %d\n", ap.passes-passNum)
+			evPassesLeft.Set(int64(ap.passes - passNum))
+			//log.Printf("Restarted ammo from the beginning. Passes left: %d\n", ap.passes-passNum)
 		}
 	}
 	log.Println("Ran out of ammo")

cmd/pandora.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"io/ioutil"
 	"log"
+	"net/http"
 	"os"
 	"runtime/pprof"
 	"time"
@@ -47,13 +48,18 @@ func Run() {
 	example := flag.Bool("example", false, "print example config to STDOUT and exit")
 	cpuprofile := flag.String("cpuprofile", "", "write cpu profile to file")
 	memprofile := flag.String("memprofile", "", "write memory profile to this file")
+	expvarHttp := flag.Bool("expvar", false, "start HTTP server with monitoring variables")
 	flag.Parse()
 
 	if *example {
 		fmt.Printf(exampleConfig)
 		return
 	}
 
+	if *expvarHttp {
+		go http.ListenAndServe(":1234", nil)
+	}
+
 	configFileName := "./load.json"
 	if len(flag.Args()) > 0 {
 		configFileName = flag.Args()[0]

engine/expvar.go

@@ -0,0 +1,73 @@
+package engine
+
+import (
+	"expvar"
+	"log"
+	"strconv"
+	"sync/atomic"
+	"time"
+)
+
+type Counter struct {
+	i int64
+}
+
+func (c *Counter) String() string {
+	return strconv.FormatInt(atomic.LoadInt64(&c.i), 10)
+}
+
+func (c *Counter) Add(delta int64) {
+	atomic.AddInt64(&c.i, delta)
+}
+
+func (c *Counter) Set(value int64) {
+	atomic.StoreInt64(&c.i, value)
+}
+
+func (c *Counter) Get() int64 {
+	return atomic.LoadInt64(&c.i)
+}
+
+func NewCounter(name string) *Counter {
+	v := &Counter{}
+	expvar.Publish(name, v)
+	return v
+}
+
+var (
+	evRequests      = NewCounter("engine_Requests")
+	evResponses     = NewCounter("engine_Responses")
+	evUsersStarted  = NewCounter("engine_UsersStarted")
+	evUsersFinished = NewCounter("engine_UsersFinished")
+)
+
+func init() {
+	evReqPS := NewCounter("engine_ReqPS")
+	evResPS := NewCounter("engine_ResPS")
+	evActiveUsers := NewCounter("engine_ActiveUsers")
+	evActiveRequests := NewCounter("engine_ActiveRequests")
+	requests := evRequests.Get()
+	responses := evResponses.Get()
+	go func() {
+		var requestsNew, responsesNew int64
+		for _ = range time.NewTicker(1 * time.Second).C {
+			requestsNew = evRequests.Get()
+			responsesNew = evResponses.Get()
+			rps := responsesNew - responses
+			reqps := requestsNew - requests
+			activeUsers := evUsersStarted.Get() - evUsersFinished.Get()
+			activeRequests := requestsNew - responsesNew
+			log.Printf(
+				"[ENGINE] %d resp/s; %d req/s; %d users; %d active\n",
+				rps, reqps, activeUsers, activeRequests)
+
+			requests = requestsNew
+			responses = responsesNew
+
+			evActiveUsers.Set(activeUsers)
+			evActiveRequests.Set(activeRequests)
+			evReqPS.Set(reqps)
+			evResPS.Set(rps)
+		}
+	}()
+}

engine/user.go

@@ -23,9 +23,11 @@ type User struct {
 }
 
 func (u *User) Run(ctx context.Context) error {
-	log.Printf("Starting user: %s\n", u.Name)
+	//log.Printf("Starting user: %s\n", u.Name)
+	evUsersStarted.Add(1)
 	defer func() {
-		log.Printf("Exit user: %s\n", u.Name)
+		//log.Printf("Exit user: %s\n", u.Name)
+		evUsersFinished.Add(1)
 	}()
 	control := u.Limiter.Control()
 	source := u.Ammunition.Source()
@@ -40,10 +42,12 @@ loop:
 			}
 			_, more = <-control
 			if more {
+				evRequests.Add(1)
 				u.Gun.Shoot(ctx, ammo)
+				evResponses.Add(1)
 				u.Ammunition.Release(ammo)
 			} else {
-				log.Println("Limiter ended.")
+				//log.Println("Limiter ended.")
 				break loop
 			}
 		case <-ctx.Done():