From 937a1afade122885cec82bcf367a785dbaf8d4fe Mon Sep 17 00:00:00 2001 From: Jason Aten Date: Wed, 7 Oct 2020 14:58:48 -0500 Subject: [PATCH] pilosa/cmd/random-query: generate random queries from existing schema - fixes molecula/molecula#136 - flag -n 0 does continuous queries until process is killed. --- cmd/random-query/main.go | 377 ++++++++++++++++++++++++++++++++++ cmd/random-query/main_test.go | 165 +++++++++++++++ cmd/random-query/vprint.go | 177 ++++++++++++++++ license.exceptions | 2 +- 4 files changed, 720 insertions(+), 1 deletion(-) create mode 100644 cmd/random-query/main.go create mode 100644 cmd/random-query/main_test.go create mode 100644 cmd/random-query/vprint.go diff --git a/cmd/random-query/main.go b/cmd/random-query/main.go new file mode 100644 index 000000000..10fdfa6d5 --- /dev/null +++ b/cmd/random-query/main.go @@ -0,0 +1,377 @@ +// 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" + "math/rand" + nethttp "net/http" + "os" + "strings" + "time" + + "github.com/pilosa/pilosa/v2" + "github.com/pilosa/pilosa/v2/http" +) + +// RandomQueryConfig +type RandomQueryConfig struct { + + // user facing flags + HostPort string // -hostport + TreeDepth int // -d + QueryCount int // -n + Verbose bool // -v + + IndexMap map[string]*Features + + API *pilosa.API + Info []*pilosa.IndexInfo + + BitmapFunc []string + + Rnd *rand.Rand +} + +type API interface { + + // InternalClient + Schema(ctx context.Context) ([]*pilosa.IndexInfo, error) + Query(ctx context.Context, index string, queryRequest *pilosa.QueryRequest) (*pilosa.QueryResponse, error) + + // API for contrast; just a little different: + //Schema(ctx context.Context) []*IndexInfo + //Query(ctx context.Context, req *pilosa.QueryRequest) (pilosa.QueryResponse, error) +} + +// have to wrap because the ugly little differences between InternalClient and API +type wrapper struct { + api *pilosa.API +} + +func (w *wrapper) Schema(ctx context.Context) ([]*pilosa.IndexInfo, error) { + return w.api.Schema(ctx), nil +} + +func (w *wrapper) Query(ctx context.Context, index string, queryRequest *pilosa.QueryRequest) (*pilosa.QueryResponse, error) { + r, err := w.api.Query(ctx, queryRequest) + return &r, err +} + +func wrapApiToInternalClient(api *pilosa.API) *wrapper { + return &wrapper{api: api} +} + +// call DefineFlags before myflags.Parse() +func (cfg *RandomQueryConfig) DefineFlags(fs *flag.FlagSet) { + fs.StringVar(&cfg.HostPort, "hostport", "localhost:10101", "host:port of pilosa to run random queries on.") + fs.IntVar(&cfg.TreeDepth, "d", 4, "depth of random queries to generate.") + fs.IntVar(&cfg.QueryCount, "n", 100, "number of random queries to generate. Set to 0 for inifinite queries.") + fs.BoolVar(&cfg.Verbose, "v", false, "show queries as they are generated") +} + +// call c.ValidateConfig() after myflags.Parse() +func (c *RandomQueryConfig) ValidateConfig() error { + if c.TreeDepth < 1 { + return fmt.Errorf("-d depth must be 1 or greater; saw %v", c.TreeDepth) + } + if c.QueryCount < 0 { + return fmt.Errorf("-n count must be 0 or greater; saw %v", c.QueryCount) + } + return nil +} + +var ProgramName = "random-query" + +func main() { + + myflags := flag.NewFlagSet(ProgramName, flag.ExitOnError) + cfg := NewRandomQueryConfig() + cfg.DefineFlags(myflags) + + err := myflags.Parse(os.Args[1:]) + if err != nil { + fmt.Fprintf(os.Stderr, "\n%v\n", err.Error()) + os.Exit(1) + } + err = cfg.ValidateConfig() + if err != nil { + fmt.Fprintf(os.Stderr, "%s error: %s\n", ProgramName, err) + os.Exit(1) + } + + err = cfg.Run() + + if err != nil { + fmt.Fprintf(os.Stderr, "error: %v\n", err) + os.Exit(1) + } +} + +func (cfg *RandomQueryConfig) Run() (err error) { + remoteClient := nethttp.DefaultClient + cli, err := http.NewInternalClient(cfg.HostPort, remoteClient) + if err != nil { + return err + } + ctx := context.Background() + totalQ := 0 + loops := 0 + t0 := time.Now() + +NewSetup: + err = cfg.Setup(cli) + if err != nil { + return err + } + + if len(cfg.IndexMap) == 0 { + return fmt.Errorf("no rows to query") + } + + var indexes []string + for index := range cfg.IndexMap { + indexes = append(indexes, index) + } + + for j := 0; ; j++ { + if cfg.QueryCount > 0 { + if j >= cfg.QueryCount { + break + } + } else { + // else keep doing queries forever... + if loops > 0 && loops%500 == 0 { + // ...but account for any new data arrived by getting + // the schema and rows again every so often. + loops++ + goto NewSetup + } + } + if totalQ%100 == 0 { + dur := time.Since(t0) + if dur > 0 { + qps := 1e9 * float64(totalQ) / float64(dur) + AlwaysPrintf("totalQueries run: %v elapsed: %v qps: %0.02f", totalQ, dur, qps) + } + } + + index := indexes[rand.Intn(len(indexes))] + + pql, err := cfg.GenQuery(index) + panicOn(err) + + if cfg.Verbose { + fmt.Printf("pql = '%v'\n", pql) + } + + // Query node0. + res, err := cli.Query(ctx, index, &pilosa.QueryRequest{Index: index, Query: pql}) + if err != nil { + AlwaysPrintf("QUERY FAILED! queries before this=%v; err = '%v', pql='%v'", loops, err, pql) + return err + } + if cfg.Verbose { + fmt.Printf("success on pql = '%v'; res='%v'\n", pql, res.Results[0]) + } + totalQ++ + loops++ + } + + return nil +} + +type Features struct { + Slc []IndexFieldRow +} + +func NewRandomQueryConfig() *RandomQueryConfig { + return &RandomQueryConfig{ + IndexMap: make(map[string]*Features), + } +} + +type IndexFieldRow struct { + Index string + Field string + RowID uint64 + RowKey string + IsRowKey bool +} + +// Run a RandomQuery takes a list of RowIDFeatures and ColumnKeyObjects +// and spits back a PQL query +// +func (cfg *RandomQueryConfig) Setup(api API) (err error) { + + ctx := context.Background() + cfg.Info, err = api.Schema(ctx) + if err != nil { + return err + } + for i, ii := range cfg.Info { + _ = i + for k, fld := range ii.Fields { + _ = k + if fld.Options.Type == "set" { + pql := fmt.Sprintf("Rows(%v)", fld.Name) + + res, err := api.Query(ctx, ii.Name, &pilosa.QueryRequest{Index: ii.Name, Query: pql}) + panicOn(err) + if cfg.Verbose { + fmt.Printf("success on pql = '%v'; res='%v'\n", pql, res.Results[0]) + } + // if the option is set to use RowKeys, then must get the Keys instead of the Rows from the RowIdentifiers. + // e.g. + // success on pql = 'Rows(aba)'; res='&pilosa.RowIdentifiers{Rows:[]uint64(nil), Keys:[]string{"aba1", "aba2"} + // success on pql = 'Rows(f)'; res='pilosa.RowIdentifiers{Rows:[]uint64{0x1}, Keys:[]string(nil), field:"f"}' + + switch x := res.Results[0].(type) { + case *pilosa.RowIdentifiers: + // internalClient gets this + cfg.AddResponse(ii.Name, fld.Name, x) + case pilosa.RowIdentifiers: + // test gets this + cfg.AddResponse(ii.Name, fld.Name, &x) + } + } + } + } + + cfg.BitmapFunc = []string{"Union", "Intersect", "Xor", "Not", "Difference"} + seed := int64(42) + cfg.Rnd = rand.New(rand.NewSource(seed)) + + return nil +} + +func (cfg *RandomQueryConfig) AddResponse(index, field string, x *pilosa.RowIdentifiers) { + for _, rowID := range x.Rows { + cfg.AddFeature(index, field, rowID, "", false) + } + for _, rowKey := range x.Keys { + cfg.AddFeature(index, field, 0, rowKey, true) + } +} + +func (cfg *RandomQueryConfig) GenQuery(index string) (pql string, err error) { + + tree := cfg.GenTree(index, cfg.TreeDepth) + + if cfg.Verbose { + fmt.Printf("%v\n", tree.StringIndent(0)) + } + pql = tree.ToPQL() + + // avoid using too much bandwidth, just count the final bitmap. + pql = fmt.Sprintf("Count(%v)", pql) + return +} + +type Tree struct { + Chd []*Tree + + S string +} + +func (tr *Tree) StringIndent(ind int) (s string) { + spc := strings.Repeat(" ", ind) + spc1 := strings.Repeat(" ", ind+1) + var chds []string + leaf := true + if len(tr.Chd) == 0 { + // leaf + } else { + leaf = false + for _, chd := range tr.Chd { + chds = append(chds, chd.StringIndent(ind+1)) + } + } + if leaf { + s += fmt.Sprintf("%v %v\n", spc1, tr.S) + } else { + for i, c := range chds { + if i == 0 { + s += fmt.Sprintf("%v %v\n%v", spc, tr.S, c) + } else { + s += fmt.Sprintf("%v", c) + } + } + } + return +} + +func (cfg *RandomQueryConfig) GenTree(index string, depth int) (tr *Tree) { + if depth == 0 { + slc := cfg.IndexMap[index].Slc + //vv("depth is 0, slc = '%#v'", slc) + r := cfg.Rnd.Intn(len(slc)) + fea := slc[r] + if fea.IsRowKey { + return &Tree{S: fmt.Sprintf("Row(%v='%v')", fea.Field, fea.RowKey)} + } + return &Tree{S: fmt.Sprintf("Row(%v=%v)", fea.Field, fea.RowID)} + } + + r := cfg.Rnd.Intn(len(cfg.BitmapFunc)) + f := cfg.BitmapFunc[r] + tr = &Tree{S: f} + numChild := 2 + switch f { + case "Union", "Intersect", "Xor": + numChild = cfg.Rnd.Intn(8) + 2 + case "Not": + numChild = 1 + case "Difference": + numChild = 2 + } + for i := 0; i < numChild; i++ { + tr.Chd = append(tr.Chd, cfg.GenTree(index, depth-1)) + } + return +} + +func (tr *Tree) ToPQL() (s string) { + + if len(tr.Chd) == 0 { + // leaf + return tr.S + } + + var chds []string + for _, c := range tr.Chd { + chds = append(chds, c.ToPQL()) + } + all := strings.Join(chds, ", ") + return fmt.Sprintf("%v(%v)", tr.S, all) +} + +func (cfg *RandomQueryConfig) AddFeature(index, field string, rowID uint64, rowKey string, isRowKey bool) { + + f, ok := cfg.IndexMap[index] + if !ok { + f = &Features{} + cfg.IndexMap[index] = f + } + f.Slc = append(f.Slc, IndexFieldRow{ + Index: index, + Field: field, + RowID: rowID, + RowKey: rowKey, + IsRowKey: isRowKey, + }) +} diff --git a/cmd/random-query/main_test.go b/cmd/random-query/main_test.go new file mode 100644 index 000000000..646846899 --- /dev/null +++ b/cmd/random-query/main_test.go @@ -0,0 +1,165 @@ +// 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" + "math/rand" + "strconv" + "testing" + + "github.com/pilosa/pilosa/v2" + "github.com/pilosa/pilosa/v2/boltdb" + "github.com/pilosa/pilosa/v2/http" + "github.com/pilosa/pilosa/v2/server" + "github.com/pilosa/pilosa/v2/test" +) + +func Test_RandomQuery(t *testing.T) { + + cfg := NewRandomQueryConfig() + + nNodes := 1 + nReplicas := 1 + + name := t.Name() + var nodeid []string + for i := 0; i < nNodes; i++ { + // work around a bug in the test.MustRunCluster that corrupts + // the .topology file if we only join name with one "_" underscore. + nodeid = append(nodeid, name+"__"+strconv.Itoa(i)) + } + + c := test.MustRunCluster(t, nNodes, + []server.CommandOption{ + server.OptCommandServerOptions( + pilosa.OptServerNodeID(nodeid[0]), + pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore), + pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)), + pilosa.OptServerReplicaN(nReplicas), + )}, + ) + defer c.Close() + + var nodes []*test.Command + var dirs []string + for i := 0; i < nNodes; i++ { + nd := c.GetNode(i) + nodes = append(nodes, nd) + dirs = append(dirs, nd.Server.Holder().Path()) + } + _ = dirs + + ctx := context.Background() + + indexes := []string{"rick", "morty"} + fieldName := []string{"f", "flying_car"} + idx := make([]*pilosa.Index, len(indexes)) + field := make([]*pilosa.Field, len(indexes)) + + var err error + + for i := range indexes { + + idx[i], err = nodes[0].API.CreateIndex(ctx, indexes[i], pilosa.IndexOptions{Keys: true, TrackExistence: true}) + if err != nil { + t.Fatalf("creating index: %v", err) + } + if idx[i].CreatedAt() == 0 { + t.Fatal("index createdAt is empty") + } + + field[i], err = nodes[0].API.CreateField(ctx, indexes[i], fieldName[i], pilosa.OptFieldTypeSet(pilosa.DefaultCacheType, 100)) + if err != nil { + t.Fatalf("creating field: %v", err) + } + if field[i].CreatedAt() == 0 { + t.Fatal("field createdAt is empty") + } + } + + timestamp := int64(0) + + for i := range indexes { + + // Generate some keyed records. + rowIDs := []uint64{} + timestamps := []int64{} + N := 10 + for j := 1; j <= N; j++ { + rowIDs = append(rowIDs, uint64(j)) + timestamps = append(timestamps, timestamp) + } + + var colKeys []string + switch i { + case 0: + // Keys are sharded so ordering is not guaranteed. + colKeys = []string{"col10", "col8", "col9", "col6", "col7", "col4", "col5", "col2", "col3", "col1"} + colKeys = colKeys[:N] + case 1: + colKeys = []string{"col11", "col12"} + N = len(colKeys) + rowIDs = rowIDs[:N] + timestamps = timestamps[:N] + } + + // Import data with keys to the coordinator (node0) and verify that it gets + // translated and forwarded to the owner of shard 0 (node1; because of offsetModHasher) + req := &pilosa.ImportRequest{ + Index: indexes[i], + IndexCreatedAt: idx[i].CreatedAt(), + Field: fieldName[i], + FieldCreatedAt: field[i].CreatedAt(), + + // even though this says Shard: 0, that won't matter. The column keys + // get hashed and that decides the actual shard. + Shard: 0, + RowIDs: rowIDs, + ColumnKeys: colKeys, + Timestamps: timestamps, + } + //vv("rowIDs = '%#v'", rowIDs) + //vv("colKeys = '%#v'", colKeys) + + qcx := nodes[0].API.Txf().NewQcx() + + if err := nodes[0].API.Import(ctx, qcx, req); err != nil { + t.Fatal(err) + } + panicOn(qcx.Finish()) + //qcx.Reset() + } + // end of setup. + + panicOn(cfg.Setup(wrapApiToInternalClient(nodes[0].API))) + + for j := 0; j < 4; j++ { + index := indexes[rand.Intn(len(indexes))] + + pql, err := cfg.GenQuery(index) + panicOn(err) + + //vv("pql = '%v'", pql) + + // Query node0. + res, err := nodes[0].API.Query(ctx, &pilosa.QueryRequest{Index: index, Query: pql}) + if err != nil { + t.Fatal(err) + } + _ = res + //vv("success on pql = '%v'; res='%v'", pql, res.Results[0]) + } +} diff --git a/cmd/random-query/vprint.go b/cmd/random-query/vprint.go new file mode 100644 index 000000000..83b1681f7 --- /dev/null +++ b/cmd/random-query/vprint.go @@ -0,0 +1,177 @@ +// home: https://github.com/glycerine/vprint +// Copyright 2019 Jason E. Aten, Ph.D. All rights reserved. +// License: MIT +// +// MIT License +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in all +// copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +// SOFTWARE. + +package main + +import ( + "fmt" + "io" + "os" + "path" + "runtime" + "runtime/debug" + "sync" + "time" +) + +const RFC3339MsecTz0 = "2006-01-02T15:04:05.000Z07:00" +const RFC3339UsecTz0 = "2006-01-02T15:04:05.000000Z07:00" + +// for tons of debug output +var VerboseVerbose bool = false + +// convience functions for . import +var pp = PP +var vv = VV + +var panicOn = PanicOn + +func init() { + // keeper linter happy + _ = pp + _ = vv +} + +func PanicOn(err error) { + if err != nil { + panic(err) + } +} + +func PP(format string, a ...interface{}) { + if VerboseVerbose { + TSPrintf(format, a...) + } +} + +func VV(format string, a ...interface{}) { + TSPrintf(format, a...) +} + +func AlwaysPrintf(format string, a ...interface{}) { + TSPrintf(format, a...) +} + +var tsPrintfMut sync.Mutex + +// time-stamped printf +func TSPrintf(format string, a ...interface{}) { + tsPrintfMut.Lock() + Printf("# %s %s ", FileLine(3), ts()) + Printf(format+"\n", a...) + tsPrintfMut.Unlock() +} + +// get timestamp for logging purposes +func ts() string { + return time.Now().Format(RFC3339UsecTz0) +} + +// so we can multi write easily, use our own printf +var OurStdout io.Writer = os.Stdout + +// Printf formats according to a format specifier and writes to standard output. +// It returns the number of bytes written and any write error encountered. +func Printf(format string, a ...interface{}) (n int, err error) { + return fmt.Fprintf(OurStdout, format, a...) +} + +func FileLine(depth int) string { + _, fileName, fileLine, ok := runtime.Caller(depth) + var s string + if ok { + s = fmt.Sprintf("%s:%d", path.Base(fileName), fileLine) + } else { + s = "" + } + return s +} + +func stack() string { + return string(debug.Stack()) +} + +func FileExists(name string) bool { + fi, err := os.Stat(name) + if err != nil { + return false + } + if fi.IsDir() { + return false + } + return true +} + +func DirExists(name string) bool { + fi, err := os.Stat(name) + if err != nil { + return false + } + if fi.IsDir() { + return true + } + return false +} + +func FileSize(name string) int64 { + fi, err := os.Stat(name) + if err != nil { + return 0 + } + return fi.Size() +} + +// Caller returns the name of the calling function. +func Caller(upStack int) string { + // elide ourself and runtime.Callers + target := upStack + 2 + + pc := make([]uintptr, target+2) + n := runtime.Callers(0, pc) + + f := runtime.Frame{Function: "unknown"} + if n > 0 { + frames := runtime.CallersFrames(pc[:n]) + for i := 0; i <= target; i++ { + contender, more := frames.Next() + if i == target { + f = contender + } + if !more { + break + } + } + } + return f.Function +} + +// happy linter: +var _ = DirExists +var _ = FileExists +var _ = Caller +var _ = stack +var _ = RFC3339MsecTz0 +var _ = RFC3339UsecTz0 +var _ = AlwaysPrintf +var _ = FileSize diff --git a/license.exceptions b/license.exceptions index 2c8c4162f..46230d311 100644 --- a/license.exceptions +++ b/license.exceptions @@ -22,4 +22,4 @@ ./proto/vdsm/vdsm.proto ./proto/vdsm/vdsm.pb.go ./cmd/pilosa-fsck/vprint.go - +./cmd/random-query/vprint.go