Merge pull request #1319 from molecula/pilosa-bench

Makefile

@@ -152,6 +152,9 @@ prerelease-upload:
 install:
 	go install -tags='$(BUILD_TAGS)' -ldflags $(LDFLAGS) $(FLAGS) ./cmd/pilosa
 
+install-bench:
+	go install -tags='$(BUILD_TAGS)' -ldflags $(LDFLAGS) $(FLAGS) ./cmd/pilosa-bench
+
 lattice:
 	git clone git@github.com:molecula/lattice.git
 

cmd/pilosa-bench/main.go

@@ -0,0 +1,313 @@
+// Copyright 2020 Pilosa Corp.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io/ioutil"
+	"log"
+	"math/rand"
+	"net/http"
+	"os"
+	"sort"
+	"strings"
+	"time"
+
+	"github.com/pilosa/pilosa/v2"
+	phttp "github.com/pilosa/pilosa/v2/http"
+	"golang.org/x/sync/errgroup"
+)
+
+func main() {
+	if err := run(context.Background(), os.Args[1:]); err == flag.ErrHelp {
+		os.Exit(1)
+	} else if err != nil {
+		fmt.Fprintln(os.Stderr, err)
+		os.Exit(1)
+	}
+}
+
+func run(ctx context.Context, args []string) (err error) {
+	fs := flag.NewFlagSet("pilosa-bench", flag.ContinueOnError)
+	hostport := fs.String("hostport", "localhost:10101", "")
+	typ := fs.String("type", "row", "query type (row)")
+	n := fs.Int("n", 1000, "number of queries")
+	rate := fs.Int("rate", 1, "number of queries per second")
+	verbose := fs.Bool("v", false, "verbose logging")
+	from := fs.String("from", "", "from time for row-range queries (ISO 8601)")
+	to := fs.String("to", "", "to time for row-range queries (ISO 8601)")
+	if err := fs.Parse(args); err != nil {
+		return err
+	}
+
+	// Parse from/to time.
+	var opt queryOptions
+	if *from != "" {
+		if opt.from, err = time.Parse(time.RFC3339, *from); err != nil {
+			return fmt.Errorf("cannot parse -from time")
+		}
+	}
+	if *to != "" {
+		if opt.to, err = time.Parse(time.RFC3339, *to); err != nil {
+			return fmt.Errorf("cannot parse -to time")
+		}
+	}
+
+	if (*typ == "row-range" || *typ == "topk") && (opt.from.IsZero() || opt.to.IsZero()) {
+		return fmt.Errorf("-from and -to flags must be specified for topk & row-range queries")
+	}
+
+	// Clear time prefix on log.
+	log.SetFlags(0)
+	if !*verbose {
+		log.SetOutput(ioutil.Discard)
+	}
+
+	// Setup PRNG to have consistent values for the same set of data.
+	rand.Seed(0)
+
+	// Setup connection to pilosa.
+	client, err := phttp.NewInternalClient(*hostport, http.DefaultClient)
+	if err != nil {
+		return err
+	}
+
+	// Load all id/keys for each field.
+	log.Printf("loading field identifiers")
+	fieldIDMap, err := loadFields(ctx, client)
+	if err != nil {
+		return fmt.Errorf("cannot load field identifiers: %w", err)
+	} else if len(fieldIDMap) == 0 {
+		return fmt.Errorf("no field identifiers available, please verify data exists")
+	}
+
+	// Generate list of sorted keys.
+	fieldKeys := make([]fieldKey, 0, len(fieldIDMap))
+	for k, f := range fieldIDMap {
+		switch *typ {
+		case "row-bsi":
+			if f.info.Options.Type != "int" {
+				continue
+			}
+		case "row-range", "topk":
+			if f.info.Options.Type != "time" {
+				continue
+			}
+		default:
+			if f.info.Options.Type == "int" || f.info.Options.Type == "time" {
+				continue
+			}
+		}
+
+		fieldKeys = append(fieldKeys, k)
+	}
+	sort.Slice(fieldKeys, func(i, j int) bool {
+		return compareFieldKeys(fieldKeys[i], fieldKeys[j]) == -1
+	})
+
+	// Ensure we have appropriate fields for our query type.
+	if len(fieldKeys) == 0 {
+		return fmt.Errorf("no available fields are appropriate for %q queries", *typ)
+	}
+
+	log.Printf("issuing %d queries at %d query/sec", *n, *rate)
+
+	// Repeatedly issue queries based on available row data.
+	ticker := time.NewTicker(time.Second / time.Duration(*rate))
+	var g errgroup.Group
+	for i := 0; i < *n; i++ {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-ticker.C:
+		}
+
+		key := fieldKeys[rand.Intn(len(fieldKeys))]
+		q, err := generateQuery(*typ, key.index, key.field, fieldIDMap[key].info, fieldIDMap[key].identifiers, opt)
+		if err != nil {
+			return fmt.Errorf("cannot generate query: %w", err)
+		}
+
+		log.Printf("[query] %s", q)
+
+		g.Go(func() error {
+			_, err = client.Query(ctx, key.index, &pilosa.QueryRequest{Index: key.index, Query: q})
+			if err != nil {
+				return err
+			}
+			return nil
+		})
+	}
+
+	return g.Wait()
+}
+
+func generateQuery(typ, index, field string, info *pilosa.FieldInfo, identifiers *pilosa.RowIdentifiers, opt queryOptions) (string, error) {
+	switch typ {
+	case "row":
+		return generateRowQuery(index, field, identifiers), nil
+	case "row-bsi":
+		return generateRowBSIQuery(index, field), nil
+	case "row-range":
+		return generateRowRangeQuery(index, field, identifiers, opt.from, opt.to), nil
+	case "count":
+		return generateCountQuery(index, field, identifiers), nil
+	case "intersect":
+		return generateIntersectQuery(index, field, identifiers), nil
+	case "union":
+		return generateUnionQuery(index, field, identifiers), nil
+	case "difference":
+		return generateDifferenceQuery(index, field, identifiers), nil
+	case "xor":
+		return generateXorQuery(index, field, identifiers), nil
+	case "groupby":
+		return generateGroupByQuery(index, field), nil
+	case "topk":
+		return generateTopKQuery(index, field, opt.from, opt.to), nil
+	default:
+		return "", fmt.Errorf("invalid query type: %q", typ)
+	}
+}
+
+func generateRowQuery(index, field string, identifiers *pilosa.RowIdentifiers) string {
+	if len(identifiers.Rows) > 0 {
+		return fmt.Sprintf("Row(%s=%d)", field, chooseRowID(identifiers))
+	}
+	return fmt.Sprintf("Row(%s=%q)", field, chooseRowKey(identifiers))
+}
+
+func generateRowBSIQuery(index, field string) string {
+	return fmt.Sprintf("Row(%s > 0)", field)
+}
+
+func generateRowRangeQuery(index, field string, identifiers *pilosa.RowIdentifiers, from, to time.Time) string {
+	if len(identifiers.Rows) > 0 {
+		return fmt.Sprintf("Row(%s=%d, from='%s', to='%s')", field, chooseRowID(identifiers), from.Format("2006-01-02T15:04"), to.Format("2006-01-02T15:04"))
+	}
+	return fmt.Sprintf("Row(%s=%q, from='%s', to='%s')", field, chooseRowKey(identifiers), from.Format("2006-01-02T15:04"), to.Format("2006-01-02T15:04"))
+}
+
+func generateRowQueries(index, field string, identifiers *pilosa.RowIdentifiers) string {
+	a := make([]string, rand.Intn(9)+1)
+	for i := range a {
+		a[i] = generateRowQuery(index, field, identifiers)
+	}
+	return strings.Join(a, ", ")
+}
+
+func generateCountQuery(index, field string, identifiers *pilosa.RowIdentifiers) string {
+	return fmt.Sprintf("Count(%s)", generateRowQuery(index, field, identifiers))
+}
+
+func generateIntersectQuery(index, field string, identifiers *pilosa.RowIdentifiers) string {
+	return fmt.Sprintf("Intersect(%s)", generateRowQueries(index, field, identifiers))
+}
+
+func generateUnionQuery(index, field string, identifiers *pilosa.RowIdentifiers) string {
+	return fmt.Sprintf("Union(%s)", generateRowQueries(index, field, identifiers))
+}
+
+func generateDifferenceQuery(index, field string, identifiers *pilosa.RowIdentifiers) string {
+	return fmt.Sprintf("Difference(%s)", generateRowQueries(index, field, identifiers))
+}
+
+func generateXorQuery(index, field string, identifiers *pilosa.RowIdentifiers) string {
+	return fmt.Sprintf("Xor(%s)", generateRowQueries(index, field, identifiers))
+}
+
+func generateGroupByQuery(index, field string) string {
+	return fmt.Sprintf("GroupBy(Rows(%s))", field)
+}
+
+func generateTopKQuery(index, field string, from, to time.Time) string {
+	return fmt.Sprintf("TopK(%s, from='%s', to='%s')", field, from.Format("2006-01-02T15:04"), to.Format("2006-01-02T15:04"))
+}
+
+// loadFields returns a mapping of index/field names to field info & identifiers.
+func loadFields(ctx context.Context, client *phttp.InternalClient) (map[fieldKey]*fieldInfo, error) {
+	indexes, err := client.Schema(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	m := make(map[fieldKey]*fieldInfo)
+	for _, ii := range indexes {
+		for _, f := range ii.Fields {
+			log.Printf("field: index=%s name=%s type=%s", ii.Name, f.Name, f.Options.Type)
+
+			switch f.Options.Type {
+			case "set", "mutex", "time":
+				identifiers, err := fetchFieldIDs(ctx, client, ii.Name, f.Name)
+				if err != nil {
+					return nil, fmt.Errorf("fetch fields: %w", err)
+				} else if len(identifiers.Rows) > 0 || len(identifiers.Keys) > 0 {
+					m[fieldKey{ii.Name, f.Name}] = &fieldInfo{f, identifiers}
+				}
+
+			case "int":
+				m[fieldKey{ii.Name, f.Name}] = &fieldInfo{info: f}
+			}
+		}
+	}
+	return m, nil
+}
+
+// fetchFieldIDs returns a list of field IDs or keys.
+func fetchFieldIDs(ctx context.Context, client *phttp.InternalClient, indexName, fieldName string) (*pilosa.RowIdentifiers, error) {
+	resp, err := client.Query(ctx, indexName, &pilosa.QueryRequest{Index: indexName, Query: `Rows(` + fieldName + `)`})
+	if err != nil {
+		return nil, err
+	}
+
+	switch result := resp.Results[0].(type) {
+	case *pilosa.RowIdentifiers:
+		return result, nil
+	case pilosa.RowIdentifiers:
+		return &result, nil
+	default:
+		return nil, fmt.Errorf("unexpected result type: %T", result)
+	}
+}
+
+func chooseRowID(identifiers *pilosa.RowIdentifiers) uint64 {
+	return identifiers.Rows[rand.Intn(len(identifiers.Rows))]
+}
+
+func chooseRowKey(identifiers *pilosa.RowIdentifiers) string {
+	return identifiers.Keys[rand.Intn(len(identifiers.Keys))]
+}
+
+type fieldKey struct {
+	index string
+	field string
+}
+
+type fieldInfo struct {
+	info        *pilosa.FieldInfo
+	identifiers *pilosa.RowIdentifiers
+}
+
+func compareFieldKeys(x, y fieldKey) int {
+	if cmp := strings.Compare(x.index, y.index); cmp != 0 {
+		return cmp
+	}
+	return strings.Compare(x.field, y.field)
+}
+
+type queryOptions struct {
+	from, to time.Time
+}

scripts/bench_read.sh

@@ -0,0 +1,42 @@
+#!/bin/bash -x
+set -e
+
+# This script runs benchmarks and posts them to Slack's #nightly channel.
+# The caller of the script should `git pull` on the pilosa repo before executing.
+#
+# Environment variables:
+#   - PILOSA_SRC: Path to pilosa src directory.
+#   - SLACK_OAUTH_TOKEN: Token used to post to Slack.
+
+# Require environment variables.
+: "${PILOSA_SRC:?Must set PILOSA_SRC environment variable}"
+: "${SLACK_OAUTH_TOKEN:?Must set SLACK_OAUTH_TOKEN environment variable}"
+
+# Build pilosa into GOBIN.
+make -C $PILOSA_SRC install install-bench
+
+# Retrieve current SHA.
+SHA=$(git -C $PILOSA_SRC rev-parse HEAD)
+
+# Format current date.
+DATE=$(date '+%Y%m%d')
+
+for TYPE in row row-bsi row-range count intersect union difference xor groupby topk 
+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.
+	RBF_PATH=gloat/data/query/${TYPE}/rbf/${DATE}.tar.gz
+	TXSRC=rbf gloat run -v -o "$RBF_PATH" $WORKFLOW_PATH
+
+	ROARING_PATH=gloat/data/query/${TYPE}/roaring/${DATE}.tar.gz
+	TXSRC=roaring gloat run -v -o "$ROARING_PATH" $WORKFLOW_PATH
+
+	# 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
+
+	# Post graph to Slack with SHA.
+	curl -F file=@/tmp/output.png -F channels=C01HBFKRLGH -F "initial_comment=$TITLE" -H "Authorization: Bearer $SLACK_OAUTH_TOKEN" https://slack.com/api/files.upload
+done

scripts/bench_write.sh

@@ -1,4 +1,5 @@
 #!/bin/bash
+set -e
 
 # This script runs benchmarks and posts them to Slack's #nightly channel.
 # The caller of the script should `git pull` on the pilosa repo before executing.
@@ -32,7 +33,7 @@ ROARING_PATH=gloat/data/1m/roaring/${DATE}.tar.gz
 TXSRC=roaring gloat run -v -o $ROARING_PATH $WORKFLOW_PATH
 
 # Generate graph from results.
-gloat graph -layout 5,2 -size 2048,2048 -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
+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
 
 # Post graph to Slack with SHA.
 curl -F file=@/tmp/output.png -F channels=C01HBFKRLGH -F "initial_comment=$TITLE" -H "Authorization: Bearer $SLACK_OAUTH_TOKEN" https://slack.com/api/files.upload

scripts/etc/gloat/query.count.yml

@@ -0,0 +1,9 @@
+name: "Count() Load Testing"
+
+main: "pilosa server --data-dir ~/pilosa.query.${TXSRC} --txsrc ${TXSRC}"
+load: "pilosa-bench -type count -rate 100 -n 3000" 
+
+health_url:    "http://localhost:10101/status"
+health_regexp: "NORMAL"
+
+debug_url: "http://localhost:10101/debug"

scripts/etc/gloat/query.difference.yml

@@ -0,0 +1,9 @@
+name: "Difference() Load Testing"
+
+main: "pilosa server --data-dir ~/pilosa.query.${TXSRC} --txsrc ${TXSRC}"
+load: "pilosa-bench -type difference -rate 10 -n 300" 
+
+health_url:    "http://localhost:10101/status"
+health_regexp: "NORMAL"
+
+debug_url: "http://localhost:10101/debug"

scripts/etc/gloat/query.groupby.yml

@@ -0,0 +1,9 @@
+name: "GroupBy() Load Testing"
+
+main: "pilosa server --data-dir ~/pilosa.query.${TXSRC} --txsrc ${TXSRC}"
+load: "pilosa-bench -type groupby -rate 100 -n 3000" 
+
+health_url:    "http://localhost:10101/status"
+health_regexp: "NORMAL"
+
+debug_url: "http://localhost:10101/debug"

scripts/etc/gloat/query.intersect.yml

@@ -0,0 +1,9 @@
+name: "Intersect() Load Testing"
+
+main: "pilosa server --data-dir ~/pilosa.query.${TXSRC} --txsrc ${TXSRC}"
+load: "pilosa-bench -type intersect -rate 100 -n 3000" 
+
+health_url:    "http://localhost:10101/status"
+health_regexp: "NORMAL"
+
+debug_url: "http://localhost:10101/debug"

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

@@ -0,0 +1,9 @@
+name: "Row(BSI) Load Testing"
+
+main: "pilosa server --data-dir ~/pilosa.query.${TXSRC} --txsrc ${TXSRC}"
+load: "pilosa-bench -type row -rate 100 -n 3000" 
+
+health_url:    "http://localhost:10101/status"
+health_regexp: "NORMAL"
+
+debug_url: "http://localhost:10101/debug"

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

@@ -0,0 +1,9 @@
+name: "Time-based Row() Load Testing"
+
+main: "pilosa server --data-dir ~/pilosa.query.${TXSRC} --txsrc ${TXSRC}"
+load: "pilosa-bench -type row-range -rate 10 -n 300 -from 2020-01-01T00:00:00Z -to 2020-01-31T23:00:00Z" 
+
+health_url:    "http://localhost:10101/status"
+health_regexp: "NORMAL"
+
+debug_url: "http://localhost:10101/debug"

scripts/etc/gloat/query.row.yml

@@ -0,0 +1,9 @@
+name: "Row() Load Testing"
+
+main: "pilosa server --data-dir ~/pilosa.query.${TXSRC} --txsrc ${TXSRC}"
+load: "pilosa-bench -type row -rate 100 -n 3000" 
+
+health_url:    "http://localhost:10101/status"
+health_regexp: "NORMAL"
+
+debug_url: "http://localhost:10101/debug"

scripts/etc/gloat/query.topk.yml

@@ -0,0 +1,9 @@
+name: "Time-based TopK() Load Testing"
+
+main: "pilosa server --data-dir ~/pilosa.query.${TXSRC} --txsrc ${TXSRC}"
+load: "pilosa-bench -type row-range -rate 10 -n 300 -from 2020-01-01T00:00:00Z -to 2020-01-31T23:00:00Z" 
+
+health_url:    "http://localhost:10101/status"
+health_regexp: "NORMAL"
+
+debug_url: "http://localhost:10101/debug"

scripts/etc/gloat/query.union.yml

@@ -0,0 +1,9 @@
+name: "Union() Load Testing"
+
+main: "pilosa server --data-dir ~/pilosa.query.${TXSRC} --txsrc ${TXSRC}"
+load: "pilosa-bench -type union -rate 10 -n 300"
+
+health_url:    "http://localhost:10101/status"
+health_regexp: "NORMAL"
+
+debug_url: "http://localhost:10101/debug"

scripts/etc/gloat/query.xor.yml

@@ -0,0 +1,9 @@
+name: "Xor() Load Testing"
+
+main: "pilosa server --data-dir ~/pilosa.query.${TXSRC} --txsrc ${TXSRC}"
+load: "pilosa-bench -type xor -rate 10 -n 300" 
+
+health_url:    "http://localhost:10101/status"
+health_regexp: "NORMAL"
+
+debug_url: "http://localhost:10101/debug"

scripts/populate_query_db.sh

@@ -0,0 +1,28 @@
+#!/bin/bash
+set -e
+
+# This script generates data query load testing to be run against.
+#
+# Environment variables:
+#   - TXSRC: Transaction store type ("roaring", "rbf")
+#   - CACHEDIR: Path to local GitHub Archive data, if available.
+
+# Require environment variables.
+: "${TXSRC:?Must set TXSRC environment variable}"
+: "${GHCACHEDIR:''}"
+
+echo "Starting pilosa"
+pilosa server --data-dir ~/pilosa.query.${TXSRC} --txsrc ${TXSRC} & pid_pilosa=$!
+sleep 5
+
+echo ""
+echo "Importing GitHub Archive"
+molecula-consumer-github -i events -d id --record-type event --batch-size=100000 \
+	--start-time 2020-01-01T00:00:00Z --end-time 2020-01-31T23:00:00Z \
+	--cache-dir "$GHCACHEDIR"
+
+echo ""
+echo "Import complete, shutting down pilosa"
+
+sleep 5
+kill $pid_pilosa