Add ingest time & latency stats to query benchmarks

cmd/pilosa-bench/main.go

@@ -16,12 +16,14 @@ package main
 
 import (
 	"context"
+	"expvar"
 	"flag"
 	"fmt"
 	"io/ioutil"
 	"log"
 	"math/rand"
 	"net/http"
+	_ "net/http/pprof"
 	"os"
 	"sort"
 	"strings"
@@ -32,6 +34,14 @@ import (
 	"golang.org/x/sync/errgroup"
 )
 
+var (
+	requestCountVar          = expvar.NewInt("request_count")
+	requestCurrentLatencyVar = expvar.NewFloat("request_current_latency") // seconds
+	requestAvgLatencyVar     = expvar.NewFloat("request_avg_latency")     // seconds
+	requestTotalLatencyVar   = expvar.NewFloat("request_total_latency")   // seconds
+	requestPerSecVar         = expvar.NewFloat("request_per_sec")
+)
+
 func main() {
 	if err := run(context.Background(), os.Args[1:]); err == flag.ErrHelp {
 		os.Exit(1)
@@ -86,6 +96,13 @@ func run(ctx context.Context, args []string) (err error) {
 		return err
 	}
 
+	// Set up HTTP endpoint to provide /debug endpoints.
+	fmt.Println("Serving debug endpoint at http://localhost:7070/debug")
+	go func() { _ = http.ListenAndServe(":7070", nil) }()
+
+	// Run separate goroutine to calculate the current req/sec & latency.
+	go monitor()
+
 	// Load all id/keys for each field.
 	log.Printf("loading field identifiers")
 	fieldIDMap, err := loadFields(ctx, client)
@@ -145,10 +162,15 @@ func run(ctx context.Context, args []string) (err error) {
 		log.Printf("[query] %s", q)
 
 		g.Go(func() error {
+			t := time.Now()
 			_, err = client.Query(ctx, key.index, &pilosa.QueryRequest{Index: key.index, Query: q})
 			if err != nil {
 				return err
 			}
+			elapsed := time.Since(t).Seconds()
+			requestCountVar.Add(1)
+			requestTotalLatencyVar.Add(elapsed)
+			requestAvgLatencyVar.Set(requestTotalLatencyVar.Value() / float64(requestCountVar.Value()))
 			return nil
 		})
 	}
@@ -156,6 +178,29 @@ func run(ctx context.Context, args []string) (err error) {
 	return g.Wait()
 }
 
+// monitor runs in a separate goroutine and updates metrics.
+func monitor() {
+	ticker := time.NewTicker(1 * time.Second)
+	defer ticker.Stop()
+
+	var lastTime time.Time
+	var lastN int64
+	var lastLatency float64
+	for range ticker.C {
+		now, n := time.Now(), requestCountVar.Value()
+		latency := requestTotalLatencyVar.Value()
+
+		if !lastTime.IsZero() {
+			elapsed := lastTime.Sub(now).Seconds()
+			if n > 0 {
+				requestCurrentLatencyVar.Set((lastLatency - latency) / float64(n))
+			}
+			requestPerSecVar.Set(float64(lastN-n) / elapsed)
+		}
+		lastTime, lastN, lastLatency = now, n, latency
+	}
+}
+
 func generateQuery(typ, index, field string, info *pilosa.FieldInfo, identifiers *pilosa.RowIdentifiers, opt queryOptions) (string, error) {
 	switch typ {
 	case "row":

scripts/bench_read.sh

@@ -25,14 +25,21 @@ for TYPE in row row-bsi row-range count intersect union difference xor groupby t
 do
 	WORKFLOW_PATH="${BASH_SOURCE%/*}/etc/gloat/query.${TYPE}.yml"
 	WORKFLOW_NAME="$(gloat workflow name $WORKFLOW_PATH)"
-	TITLE="$WORKFLOW_NAME, $DATE ($SHA)"
-
+	
 	# Execute RBF/Roaring benchmark.
+	STARTTIME=$(date +%s)
 	RBF_PATH=gloat/data/query/${TYPE}/rbf/${DATE}.tar.gz
 	TXSRC=rbf gloat run -v -o "$RBF_PATH" $WORKFLOW_PATH
+	RBF_ELAPSED=$(($(date +%s) - $STARTTIME))
+	RBF_LATENCY=$(gloat metric -n -name request_avg_latency "$RBF_PATH")
 
+	STARTTIME=$(date +%s)
 	ROARING_PATH=gloat/data/query/${TYPE}/roaring/${DATE}.tar.gz
 	TXSRC=roaring gloat run -v -o "$ROARING_PATH" $WORKFLOW_PATH
+	ROARING_ELAPSED=$(($(date +%s) - $STARTTIME))
+	ROARING_LATENCY=$(gloat metric -n -name request_avg_latency "$ROARING_PATH")
+
+	TITLE="$WORKFLOW_NAME, $DATE ($SHA) elapsed rbf=$RBF_ELAPSEDroaring=$ROARING_ELAPSED> latency rbf=$RBF_LATENCY roaring=$ROARING_LATENCY"
 
 	# Generate graph from results.
 	gloat graph -layout 2,5 -size 5120,820 -title "$TITLE" -name utime,stime,heap_alloc,heap_inuse,heap_objects,num_gc,rchar,wchar,syscr,syscw -series rbf,roaring -o /tmp/output.png $RBF_PATH $ROARING_PATH

scripts/etc/gloat/query.count.yml

@@ -8,3 +8,4 @@ health_regexp: "NORMAL"
 
 vars_urls:
   - http://localhost:10101/debug/vars
+  - http://localhost:7070/debug/vars

scripts/etc/gloat/query.difference.yml

@@ -8,3 +8,4 @@ health_regexp: "NORMAL"
 
 vars_urls:
   - http://localhost:10101/debug/vars
+  - http://localhost:7070/debug/vars

scripts/etc/gloat/query.groupby.yml

@@ -8,3 +8,4 @@ health_regexp: "NORMAL"
 
 vars_urls:
   - http://localhost:10101/debug/vars
+  - http://localhost:7070/debug/vars

scripts/etc/gloat/query.intersect.yml

@@ -8,3 +8,4 @@ health_regexp: "NORMAL"
 
 vars_urls:
   - http://localhost:10101/debug/vars
+  - http://localhost:7070/debug/vars

scripts/etc/gloat/query.row-bsi.yml

@@ -8,3 +8,4 @@ health_regexp: "NORMAL"
 
 vars_urls:
   - http://localhost:10101/debug/vars
+  - http://localhost:7070/debug/vars

scripts/etc/gloat/query.row-range.yml

@@ -8,3 +8,4 @@ health_regexp: "NORMAL"
 
 vars_urls:
   - http://localhost:10101/debug/vars
+  - http://localhost:7070/debug/vars

scripts/etc/gloat/query.row.yml

@@ -8,3 +8,4 @@ health_regexp: "NORMAL"
 
 vars_urls:
   - http://localhost:10101/debug/vars
+  - http://localhost:7070/debug/vars

scripts/etc/gloat/query.topk.yml

@@ -8,3 +8,4 @@ health_regexp: "NORMAL"
 
 vars_urls:
   - http://localhost:10101/debug/vars
+  - http://localhost:7070/debug/vars

scripts/etc/gloat/query.union.yml

@@ -8,3 +8,4 @@ health_regexp: "NORMAL"
 
 vars_urls:
   - http://localhost:10101/debug/vars
+  - http://localhost:7070/debug/vars

scripts/etc/gloat/query.xor.yml

@@ -8,3 +8,4 @@ health_regexp: "NORMAL"
 
 vars_urls:
   - http://localhost:10101/debug/vars
+  - http://localhost:7070/debug/vars