Benchmarks for zeroeventhub

benchmark/go.mod

@@ -0,0 +1,14 @@
+module github.com/vippsas/zeroeventhub/benchmark
+
+go 1.18
+
+require github.com/vippsas/zeroeventhub/go v0.0.0-20230222084417-fd5ddb1df01e
+
+require (
+	github.com/gorilla/mux v1.8.0 // indirect
+	github.com/pkg/errors v0.9.1 // indirect
+	github.com/sirupsen/logrus v1.8.1 // indirect
+	github.com/stretchr/testify v1.7.0 // indirect
+	golang.org/x/sys v0.0.0-20220412211240-33da011f77ad // indirect
+	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect
+)

benchmark/go.sum

@@ -0,0 +1,24 @@
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/gorilla/mux v1.8.0 h1:i40aqfkR1h2SlN9hojwV5ZA91wcXFOvkdNIeFDP5koI=
+github.com/gorilla/mux v1.8.0/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB71So=
+github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
+github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/sirupsen/logrus v1.8.1 h1:dJKuHgqk1NNQlqoA6BTlM1Wf9DOH3NBjQyu0h9+AZZE=
+github.com/sirupsen/logrus v1.8.1/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
+github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/vippsas/zeroeventhub/go v0.0.0-20230222084417-fd5ddb1df01e h1:nx1qiJi+AaKu+WLVH/4fnRLsz7RTvr2YiKuNlbj5ZRw=
+github.com/vippsas/zeroeventhub/go v0.0.0-20230222084417-fd5ddb1df01e/go.mod h1:jSNYrs0Z2dp9b19LrNNQ1sHO+mJcSj319UYg+hfZfWg=
+golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20220412211240-33da011f77ad h1:ntjMns5wyP/fN65tdBD4g8J5w8n015+iIIs9rtjXkY0=
+golang.org/x/sys v0.0.0-20220412211240-33da011f77ad/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b h1:h8qDotaEPuJATrMmW04NCwg7v22aHH28wwpauUhK9Oo=
+gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

benchmark/main.go

@@ -0,0 +1,117 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"flag"
+	"fmt"
+	zeroeventhub "github.com/vippsas/zeroeventhub/go"
+	"os"
+	"time"
+)
+
+type EventStatsReceiver struct {
+	EventCount int
+	Cursor     string
+}
+
+func (s *EventStatsReceiver) Event(partitionID int, headers map[string]string, Data json.RawMessage) error {
+	s.EventCount++
+	return nil
+}
+
+func (s *EventStatsReceiver) Checkpoint(partitionID int, cursor string) error {
+	s.Cursor = cursor
+	return nil
+}
+
+func worker(url string, tail bool, statsChan chan int) {
+	c := zeroeventhub.NewClient(url, 1)
+	var cursor string
+	if tail {
+		cursor = "_last"
+	} else {
+		cursor = "_first"
+	}
+	for {
+		page := EventStatsReceiver{}
+		cursors := []zeroeventhub.Cursor{
+			{
+				PartitionID: 0,
+				Cursor:      cursor,
+			},
+		}
+		if err := c.FetchEvents(context.TODO(), cursors, 1000, &page); err != nil {
+			fmt.Fprintln(os.Stderr, "Got error: "+err.Error())
+			continue
+		}
+
+		if tail {
+			statsChan <- 1
+			time.Sleep(50 * time.Millisecond)
+		} else {
+			statsChan <- page.EventCount
+		}
+		cursor = page.Cursor
+		if cursor == "" {
+			return
+		}
+	}
+
+}
+
+func main() {
+	threadcount := flag.Int("n", 1, "Number of threads")
+	url := flag.String("u", "", "ZeroEventHub endpoint to call")
+	tail := flag.Bool("t", false, "Tail-mode -- benchmark polling inserts instead of reconstitution")
+	flag.Parse()
+
+	if *tail {
+		fmt.Println("inserts-mode, stats are polls/seq")
+	} else {
+		fmt.Println("reconstitution mode, stats are events/seq")
+	}
+
+	statsChan := make(chan int)
+	for i := 0; i != *threadcount; i++ {
+		go worker(*url, *tail, statsChan)
+	}
+
+	type row struct {
+		t     time.Time
+		total int
+	}
+
+	total := 0
+	tstart := time.Now()
+	lastPrint := tstart
+	var rows []row
+	for {
+		var s int
+		s = <-statsChan
+		total += s
+
+		now := time.Now()
+
+		rows = append(rows, row{now, total})
+
+		dtPrint := now.Sub(lastPrint)
+		if dtPrint > time.Second {
+			tenAgo := len(rows) - 10
+			if tenAgo < 0 {
+				tenAgo = 0
+			}
+
+			windowEnd := rows[len(rows)-1]
+			windowStart := rows[tenAgo]
+			rate := float64(windowEnd.total-windowStart.total) / (windowEnd.t.Sub(windowStart.t).Seconds())
+
+			fmt.Printf("stats total=%d rate/sec=%.2f\n",
+				total,
+				rate,
+			)
+			lastPrint = now
+		}
+	}
+
+}