From c94d242097e34f955eb9eff2608d2a6f2cda0d84 Mon Sep 17 00:00:00 2001 From: Alan Bernstein Date: Wed, 21 Oct 2020 10:43:27 -0500 Subject: [PATCH 01/10] Add basic implementation of query history endpoint --- api.go | 12 ++++++++- http/handler.go | 15 +++++++++++ tracker.go | 68 +++++++++++++++++++++++++++++++++++++++++++------ 3 files changed, 86 insertions(+), 9 deletions(-) diff --git a/api.go b/api.go index 4fc631278..2038679c1 100644 --- a/api.go +++ b/api.go @@ -158,7 +158,7 @@ func (api *API) Query(ctx context.Context, req *QueryRequest) (QueryResponse, er if err != nil { return QueryResponse{}, errors.Wrap(err, "parsing") } - defer api.tracker.Finish(api.tracker.Start(req.Query)) + defer api.tracker.Finish(api.tracker.Start(req.Query, api.server.nodeID), time.Now()) execOpts := &execOptions{ Remote: req.Remote, Profile: req.Profile, @@ -2050,6 +2050,14 @@ func (api *API) ActiveQueries(ctx context.Context) ([]ActiveQueryStatus, error) return api.tracker.ActiveQueries(), nil } +func (api *API) PastQueries(ctx context.Context) ([]PastQueryStatus, error) { + if err := api.validate(apiPastQueries); err != nil { + return nil, errors.Wrap(err, "validating api method") + } + x := api.tracker.PastQueries() + return x, nil +} + // TranslateIndexDB is an internal function to load the index keys database // rd is a boltdb file. func (api *API) TranslateIndexDB(ctx context.Context, indexName string, partitionID int, rd io.Reader) error { @@ -2124,6 +2132,7 @@ const ( apiTransactions apiGetTransaction apiActiveQueries + apiPastQueries ) var methodsCommon = map[apiMethod]struct{}{ @@ -2164,4 +2173,5 @@ var methodsNormal = map[apiMethod]struct{}{ apiTransactions: {}, apiGetTransaction: {}, apiActiveQueries: {}, + apiPastQueries: {}, } diff --git a/http/handler.go b/http/handler.go index f9339a751..69af4d81a 100644 --- a/http/handler.go +++ b/http/handler.go @@ -397,8 +397,10 @@ func newRouter(handler *Handler) http.Handler { router.HandleFunc("/transaction/{id}/finish", handler.handlePostFinishTransaction).Methods("POST").Name("PostFinishTransaction") router.HandleFunc("/transactions", handler.handleGetTransactions).Methods("GET").Name("GetTransactions") router.HandleFunc("/queries", handler.handleGetActiveQueries).Methods("GET").Name("GetActiveQueries") + router.HandleFunc("/query-history", handler.handleGetPastQueries).Methods("GET").Name("GetPastQueries") router.HandleFunc("/version", handler.handleGetVersion).Methods("GET").Name("GetVersion") + // /ui endpoints are for UI use; they may change at any time. router.HandleFunc("/ui/usage", handler.handleGetUsage).Methods("GET").Name("GetUsage") router.HandleFunc("/ui/transaction", handler.handleGetTransactionList).Methods("GET").Name("GetTransactionList") router.HandleFunc("/ui/transaction/", handler.handleGetTransactionList).Methods("GET").Name("GetTransactionList") @@ -1184,6 +1186,19 @@ func (h *Handler) handleGetActiveQueries(w http.ResponseWriter, r *http.Request) } } +func (h *Handler) handleGetPastQueries(w http.ResponseWriter, r *http.Request) { + queries, err := h.api.PastQueries(r.Context()) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(queries); err != nil { + h.logger.Printf("encoding GetActiveQueries response: %s", err) + } + +} + type postIndexAttrDiffRequest struct { Blocks []pilosa.AttrBlock `json:"blocks"` } diff --git a/tracker.go b/tracker.go index af5788137..66bf0e924 100644 --- a/tracker.go +++ b/tracker.go @@ -22,22 +22,40 @@ import ( type ActiveQueryStatus struct { Query string `json:"query"` + Node string `json:"node"` Age time.Duration `json:"age"` } +type PastQueryStatus struct { + Query string `json:"query"` + Node string `json:"node"` + Age time.Duration `json:"age"` + Runtime time.Duration `json:"runtime"` +} + type activeQuery struct { query string + node string started time.Time } +type pastQuery struct { + query string + node string + started time.Time + runtime time.Duration +} + type queryStatusUpdate struct { - q *activeQuery - end bool + q *activeQuery + end bool + endTime time.Time } type queryTracker struct { updates chan<- queryStatusUpdate checks chan<- chan<- []*activeQuery + history map[pastQuery]struct{} // TODO not in memory wg sync.WaitGroup stop chan struct{} } @@ -46,9 +64,11 @@ func newQueryTracker() *queryTracker { done := make(chan struct{}) updates := make(chan queryStatusUpdate, 128) checks := make(chan chan<- []*activeQuery) + history := make(map[pastQuery]struct{}) tracker := &queryTracker{ updates: updates, checks: checks, + history: history, stop: done, } tracker.wg.Add(1) @@ -64,6 +84,8 @@ func newQueryTracker() *queryTracker { delete(activeQueries, update.q) } else { activeQueries[update.q] = struct{}{} + pq := pastQuery{update.q.query, update.q.node, update.q.started, update.endTime.Sub(update.q.started)} // TODO move end time to api.go + tracker.history[pq] = struct{}{} } case check := <-checks: out := make([]*activeQuery, len(activeQueries)) @@ -82,15 +104,15 @@ func newQueryTracker() *queryTracker { return tracker } -func (t *queryTracker) Start(query string) *activeQuery { +func (t *queryTracker) Start(query, nodeID string) *activeQuery { now := time.Now() - q := &activeQuery{query, now} - t.updates <- queryStatusUpdate{q, false} + q := &activeQuery{query, nodeID, now} + t.updates <- queryStatusUpdate{q, false, time.Time{}} return q } -func (t *queryTracker) Finish(q *activeQuery) { - t.updates <- queryStatusUpdate{q, true} +func (t *queryTracker) Finish(q *activeQuery, endTime time.Time) { + t.updates <- queryStatusUpdate{q, true, endTime} } func (t *queryTracker) ActiveQueries() []ActiveQueryStatus { @@ -114,11 +136,41 @@ func (t *queryTracker) ActiveQueries() []ActiveQueryStatus { now := time.Now() out := make([]ActiveQueryStatus, len(queries)) for i, v := range queries { - out[i] = ActiveQueryStatus{v.query, now.Sub(v.started)} + out[i] = ActiveQueryStatus{v.query, v.node, now.Sub(v.started)} } return out } +func (t *queryTracker) PastQueries() []PastQueryStatus { + queries := make([]pastQuery, 0, len(t.history)) + for pq, _ := range t.history { + queries = append(queries, pq) + } + // TODO use sort.Sort + sort.Slice(queries, func(i, j int) bool { + switch { + case queries[i].started.Before(queries[j].started): + return true + case queries[i].started.After(queries[j].started): + return false + case queries[i].query < queries[j].query: + return true + case queries[i].query > queries[j].query: + return false + default: + return false + } + }) + + now := time.Now() + out := make([]PastQueryStatus, len(queries)) + for i, v := range queries { + out[i] = PastQueryStatus{v.query, v.node, now.Sub(v.started), v.runtime} + } + return out + +} + func (t *queryTracker) Stop() { close(t.stop) t.wg.Wait() From 65de5df2a9f2c8397d2ae6a4e92bb54143005689 Mon Sep 17 00:00:00 2001 From: Alan Bernstein Date: Wed, 21 Oct 2020 11:07:33 -0500 Subject: [PATCH 02/10] Update tracker test --- tracker.go | 2 +- tracker_test.go | 9 ++++++--- 2 files changed, 7 insertions(+), 4 deletions(-) diff --git a/tracker.go b/tracker.go index 66bf0e924..d33a8fb79 100644 --- a/tracker.go +++ b/tracker.go @@ -143,7 +143,7 @@ func (t *queryTracker) ActiveQueries() []ActiveQueryStatus { func (t *queryTracker) PastQueries() []PastQueryStatus { queries := make([]pastQuery, 0, len(t.history)) - for pq, _ := range t.history { + for pq := range t.history { queries = append(queries, pq) } // TODO use sort.Sort diff --git a/tracker_test.go b/tracker_test.go index 80f6f84f2..4afa1abce 100644 --- a/tracker_test.go +++ b/tracker_test.go @@ -14,7 +14,10 @@ package pilosa -import "testing" +import ( + "testing" + "time" +) func TestQueryTracker(t *testing.T) { tracker := newQueryTracker() @@ -24,7 +27,7 @@ func TestQueryTracker(t *testing.T) { t.Fatalf("expected no active queries; found %v", queries) } - qs := tracker.Start("test query") + qs := tracker.Start("test query", "node0") var queries []ActiveQueryStatus for len(queries) < 1 { @@ -34,7 +37,7 @@ func TestQueryTracker(t *testing.T) { t.Fatalf("unexpected queries: %v", queries) } - tracker.Finish(qs) + tracker.Finish(qs, time.Now()) for len(queries) > 0 { queries = tracker.ActiveQueries() From bbd147d18fac6f012ba29eeb0d8c9ba8398430e4 Mon Sep 17 00:00:00 2001 From: Alan Bernstein Date: Mon, 26 Oct 2020 13:16:29 -0500 Subject: [PATCH 03/10] Replace query history map with ringBuffer --- tracker.go | 68 +++++++++++++++++++++++++++++++------------------ tracker_test.go | 36 ++++++++++++++++++++++++++ 2 files changed, 79 insertions(+), 25 deletions(-) diff --git a/tracker.go b/tracker.go index d33a8fb79..36625b7b8 100644 --- a/tracker.go +++ b/tracker.go @@ -55,16 +55,53 @@ type queryStatusUpdate struct { type queryTracker struct { updates chan<- queryStatusUpdate checks chan<- chan<- []*activeQuery - history map[pastQuery]struct{} // TODO not in memory - wg sync.WaitGroup - stop chan struct{} + history *ringBuffer + + wg sync.WaitGroup + stop chan struct{} +} + +type ringBuffer struct { + queries []pastQuery + start int + count int + mu sync.Mutex +} + +// newRingBuffer initializes an empty RingBuffer of specified capacity. +func newRingBuffer(n int) *ringBuffer { + return &ringBuffer{ + queries: make([]pastQuery, n), + start: 0, + count: 0, + } +} + +// add adds a new element to the queue, overwriting the oldest if it is already full. +func (b *ringBuffer) add(q pastQuery) { + b.mu.Lock() + defer b.mu.Unlock() + + b.queries[(b.start+b.count)%cap(b.queries)] = q + if b.count == cap(b.queries) { + b.start = (b.start + 1) % cap(b.queries) + } else { + b.count++ + } +} + +// slice returns the contents of the RingBuffer, in insertion order. +func (b *ringBuffer) slice() []pastQuery { + b.mu.Lock() + defer b.mu.Unlock() + return append(b.queries[b.start:b.count], b.queries[0:b.start]...) } func newQueryTracker() *queryTracker { done := make(chan struct{}) updates := make(chan queryStatusUpdate, 128) checks := make(chan chan<- []*activeQuery) - history := make(map[pastQuery]struct{}) + history := newRingBuffer(128) tracker := &queryTracker{ updates: updates, checks: checks, @@ -85,7 +122,7 @@ func newQueryTracker() *queryTracker { } else { activeQueries[update.q] = struct{}{} pq := pastQuery{update.q.query, update.q.node, update.q.started, update.endTime.Sub(update.q.started)} // TODO move end time to api.go - tracker.history[pq] = struct{}{} + tracker.history.add(pq) } case check := <-checks: out := make([]*activeQuery, len(activeQueries)) @@ -142,26 +179,7 @@ func (t *queryTracker) ActiveQueries() []ActiveQueryStatus { } func (t *queryTracker) PastQueries() []PastQueryStatus { - queries := make([]pastQuery, 0, len(t.history)) - for pq := range t.history { - queries = append(queries, pq) - } - // TODO use sort.Sort - sort.Slice(queries, func(i, j int) bool { - switch { - case queries[i].started.Before(queries[j].started): - return true - case queries[i].started.After(queries[j].started): - return false - case queries[i].query < queries[j].query: - return true - case queries[i].query > queries[j].query: - return false - default: - return false - } - }) - + queries := t.history.slice() now := time.Now() out := make([]PastQueryStatus, len(queries)) for i, v := range queries { diff --git a/tracker_test.go b/tracker_test.go index 4afa1abce..fa6767e50 100644 --- a/tracker_test.go +++ b/tracker_test.go @@ -15,10 +15,46 @@ package pilosa import ( + "fmt" "testing" "time" ) +func TestRingBuffer(t *testing.T) { + tests := []struct { + start int + count int + queries []string + }{ + {start: 0, count: 0, queries: []string{}}, + {start: 0, count: 1, queries: []string{"0"}}, + {start: 0, count: 2, queries: []string{"0", "1"}}, + {start: 0, count: 3, queries: []string{"0", "1", "2"}}, + {start: 0, count: 4, queries: []string{"0", "1", "2", "3"}}, + {start: 0, count: 5, queries: []string{"0", "1", "2", "3", "4"}}, + {start: 1, count: 5, queries: []string{"1", "2", "3", "4", "5"}}, + {start: 2, count: 5, queries: []string{"2", "3", "4", "5", "6"}}, + {start: 3, count: 5, queries: []string{"3", "4", "5", "6", "7"}}, + {start: 4, count: 5, queries: []string{"4", "5", "6", "7", "8"}}, + {start: 0, count: 5, queries: []string{"5", "6", "7", "8", "9"}}, + {start: 1, count: 5, queries: []string{"6", "7", "8", "9", "10"}}, + {start: 2, count: 5, queries: []string{"7", "8", "9", "10", "11"}}, + } + buffer := newRingBuffer(5) + for k := 0; k < len(tests); k++ { + if !(buffer.start == tests[k].start && buffer.count == tests[k].count) { + t.Fatalf("expected %d %d, found %d %d", tests[k].start, tests[k].count, buffer.start, buffer.count) + } + + for n, q := range buffer.slice() { + if q.query != tests[k].queries[n] { + t.Fatalf("test[%d], buffer[%d] expected querystring '%s', found '%s'", k, n, tests[k].queries[n], q.query) + } + } + buffer.add(pastQuery{query: fmt.Sprintf("%d", k)}) + } +} + func TestQueryTracker(t *testing.T) { tracker := newQueryTracker() defer tracker.Stop() From 542a6ffdb603d794d73d99241dcfc3fe93090aed Mon Sep 17 00:00:00 2001 From: Alan Bernstein Date: Mon, 26 Oct 2020 14:46:23 -0500 Subject: [PATCH 04/10] Gather query history from remote nodes --- api.go | 26 +++++++++++++++++++++++--- client.go | 5 +++++ http/client.go | 35 +++++++++++++++++++++++++++++++---- http/handler.go | 12 ++++++++++-- 4 files changed, 69 insertions(+), 9 deletions(-) diff --git a/api.go b/api.go index 2038679c1..875006289 100644 --- a/api.go +++ b/api.go @@ -2050,12 +2050,32 @@ func (api *API) ActiveQueries(ctx context.Context) ([]ActiveQueryStatus, error) return api.tracker.ActiveQueries(), nil } -func (api *API) PastQueries(ctx context.Context) ([]PastQueryStatus, error) { +func (api *API) PastQueries(ctx context.Context, remote bool) ([]PastQueryStatus, error) { if err := api.validate(apiPastQueries); err != nil { return nil, errors.Wrap(err, "validating api method") } - x := api.tracker.PastQueries() - return x, nil + + clusterQueries := api.tracker.PastQueries() + + if !remote { + nodes := api.cluster.Nodes() + for _, node := range nodes { + if node.ID == api.server.nodeID { + continue + } + nodeQueries, err := api.server.defaultClient.GetPastQueries(ctx, &node.URI) + if err != nil { + return nil, errors.Wrapf(err, "collecting query history from %s", node.URI) + } + clusterQueries = append(clusterQueries, nodeQueries...) + } + } + + sort.Slice(clusterQueries, func(i, j int) bool { + return clusterQueries[i].Age.Seconds() > clusterQueries[j].Age.Seconds() + }) + + return clusterQueries, nil } // TranslateIndexDB is an internal function to load the index keys database diff --git a/client.go b/client.go index f8e2ff8bb..4cd410345 100644 --- a/client.go +++ b/client.go @@ -83,6 +83,7 @@ type InternalClient interface { GetTransaction(ctx context.Context, id string) (*Transaction, error) GetNodeUsage(ctx context.Context, uri *URI) (map[string]NodeUsage, error) + GetPastQueries(ctx context.Context, uri *URI) ([]PastQueryStatus, error) } //=============== @@ -246,3 +247,7 @@ func (n nopInternalClient) GetTransaction(ctx context.Context, id string) (*Tran func (n nopInternalClient) GetNodeUsage(ctx context.Context, uri *URI) (map[string]NodeUsage, error) { return nil, nil } + +func (n nopInternalClient) GetPastQueries(ctx context.Context, uri *URI) ([]PastQueryStatus, error) { + return nil, nil +} diff --git a/http/client.go b/http/client.go index c8ad52f27..801ac9005 100644 --- a/http/client.go +++ b/http/client.go @@ -1278,6 +1278,37 @@ func (c *InternalClient) GetNodeUsage(ctx context.Context, uri *pilosa.URI) (map return nodeUsages, nil } +// GetPastQueries retrieves the query history log for the specified node. +func (c *InternalClient) GetPastQueries(ctx context.Context, uri *pilosa.URI) ([]pilosa.PastQueryStatus, error) { + u := uri.Path("/query-history?remote=true") + req, err := http.NewRequest("GET", u, nil) + if err != nil { + return nil, errors.Wrap(err, "creating request") + } + + req.Header.Set("Accept", "application/json") + req.Header.Set("User-Agent", "pilosa/"+pilosa.Version) + + // Execute request against the host. + resp, err := c.executeRequest(req.WithContext(ctx)) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + // Read body and unmarshal response. + body, err := ioutil.ReadAll(resp.Body) + if err != nil { + return nil, errors.Wrap(err, "reading") + } + + queries := make([]pilosa.PastQueryStatus, 128) + if err := json.Unmarshal(body, &queries); err != nil { + return nil, fmt.Errorf("unmarshal response: %s", err) + } + return queries, nil +} + func (c *InternalClient) FindIndexKeysNode(ctx context.Context, uri *pilosa.URI, index string, keys ...string) (transMap map[string]uint64, err error) { span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.FindIndexKeysNode") defer span.Finish() @@ -1341,10 +1372,6 @@ func (c *InternalClient) FindFieldKeysNode(ctx context.Context, uri *pilosa.URI, return nil, errors.Wrap(err, "marshalling request") } req, err := http.NewRequest("POST", u.String(), bytes.NewReader(reqData)) - if err != nil { - return nil, errors.Wrap(err, "creating request") - } - // Apply headers. req.Header.Set("Content-Length", strconv.Itoa(len(reqData))) req.Header.Set("Content-Type", "application/json") diff --git a/http/handler.go b/http/handler.go index 69af4d81a..be621177b 100644 --- a/http/handler.go +++ b/http/handler.go @@ -706,7 +706,7 @@ func (h *Handler) handleGetUsage(w http.ResponseWriter, r *http.Request) { } } -// handleGetUsage handles GET /ui/shard-distribution requests. +// handleGetShardDistribution handles GET /ui/shard-distribution requests. func (h *Handler) handleGetShardDistribution(w http.ResponseWriter, r *http.Request) { dist := h.api.ShardDistribution(r.Context()) w.Header().Set("Content-Type", "application/json") @@ -1187,11 +1187,19 @@ func (h *Handler) handleGetActiveQueries(w http.ResponseWriter, r *http.Request) } func (h *Handler) handleGetPastQueries(w http.ResponseWriter, r *http.Request) { - queries, err := h.api.PastQueries(r.Context()) + q := r.URL.Query() + remoteStr := q.Get("remote") + var remote bool + if remoteStr == "true" { + remote = true + } + + queries, err := h.api.PastQueries(r.Context(), remote) if err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) return } + w.Header().Set("Content-Type", "application/json") if err := json.NewEncoder(w).Encode(queries); err != nil { h.logger.Printf("encoding GetActiveQueries response: %s", err) From b172f4d05b8ce9e4765f3ef4b094bbe328e157b9 Mon Sep 17 00:00:00 2001 From: Alan Bernstein Date: Mon, 26 Oct 2020 14:46:42 -0500 Subject: [PATCH 05/10] Include index in query history response --- api.go | 2 +- tracker.go | 18 +++++++++++------- 2 files changed, 12 insertions(+), 8 deletions(-) diff --git a/api.go b/api.go index 875006289..6ead76e0e 100644 --- a/api.go +++ b/api.go @@ -158,7 +158,7 @@ func (api *API) Query(ctx context.Context, req *QueryRequest) (QueryResponse, er if err != nil { return QueryResponse{}, errors.Wrap(err, "parsing") } - defer api.tracker.Finish(api.tracker.Start(req.Query, api.server.nodeID), time.Now()) + defer api.tracker.Finish(api.tracker.Start(req.Query, api.server.nodeID, req.Index), time.Now()) execOpts := &execOptions{ Remote: req.Remote, Profile: req.Profile, diff --git a/tracker.go b/tracker.go index 36625b7b8..7d58b2479 100644 --- a/tracker.go +++ b/tracker.go @@ -23,12 +23,14 @@ import ( type ActiveQueryStatus struct { Query string `json:"query"` Node string `json:"node"` + Index string `json:index` Age time.Duration `json:"age"` } type PastQueryStatus struct { Query string `json:"query"` - Node string `json:"node"` + Node string `json:"nodeID"` + Index string `json:index` Age time.Duration `json:"age"` Runtime time.Duration `json:"runtime"` } @@ -36,12 +38,14 @@ type PastQueryStatus struct { type activeQuery struct { query string node string + index string started time.Time } type pastQuery struct { query string node string + index string started time.Time runtime time.Duration } @@ -118,11 +122,11 @@ func newQueryTracker() *queryTracker { select { case update := <-updates: if update.end { + pq := pastQuery{update.q.query, update.q.node, update.q.index, update.q.started, update.endTime.Sub(update.q.started)} + tracker.history.add(pq) delete(activeQueries, update.q) } else { activeQueries[update.q] = struct{}{} - pq := pastQuery{update.q.query, update.q.node, update.q.started, update.endTime.Sub(update.q.started)} // TODO move end time to api.go - tracker.history.add(pq) } case check := <-checks: out := make([]*activeQuery, len(activeQueries)) @@ -141,9 +145,9 @@ func newQueryTracker() *queryTracker { return tracker } -func (t *queryTracker) Start(query, nodeID string) *activeQuery { +func (t *queryTracker) Start(query, nodeID, index string) *activeQuery { now := time.Now() - q := &activeQuery{query, nodeID, now} + q := &activeQuery{query, nodeID, index, now} t.updates <- queryStatusUpdate{q, false, time.Time{}} return q } @@ -173,7 +177,7 @@ func (t *queryTracker) ActiveQueries() []ActiveQueryStatus { now := time.Now() out := make([]ActiveQueryStatus, len(queries)) for i, v := range queries { - out[i] = ActiveQueryStatus{v.query, v.node, now.Sub(v.started)} + out[i] = ActiveQueryStatus{v.query, v.node, v.index, now.Sub(v.started)} } return out } @@ -183,7 +187,7 @@ func (t *queryTracker) PastQueries() []PastQueryStatus { now := time.Now() out := make([]PastQueryStatus, len(queries)) for i, v := range queries { - out[i] = PastQueryStatus{v.query, v.node, now.Sub(v.started), v.runtime} + out[i] = PastQueryStatus{v.query, v.node, v.index, now.Sub(v.started), v.runtime} } return out From 2bc2a32263402bd893b8561f7774594bad879f60 Mon Sep 17 00:00:00 2001 From: Alan Bernstein Date: Mon, 26 Oct 2020 15:15:15 -0500 Subject: [PATCH 06/10] Make query history length configurable --- api.go | 2 +- ctl/server.go | 1 + http/client.go | 2 +- server.go | 12 ++++++++++++ server/config.go | 7 +++++++ server/server.go | 1 + tracker.go | 4 ++-- 7 files changed, 25 insertions(+), 4 deletions(-) diff --git a/api.go b/api.go index 6ead76e0e..588233603 100644 --- a/api.go +++ b/api.go @@ -100,7 +100,7 @@ func NewAPI(opts ...apiOption) (*API, error) { }() } - api.tracker = newQueryTracker() + api.tracker = newQueryTracker(api.server.queryHistoryLength) return api, nil } diff --git a/ctl/server.go b/ctl/server.go index 1a0754486..c548f93f3 100644 --- a/ctl/server.go +++ b/ctl/server.go @@ -36,6 +36,7 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) { flags.BoolVar(&srv.Config.Verbose, "verbose", srv.Config.Verbose, "Enable verbose logging") flags.Uint64Var(&srv.Config.MaxMapCount, "max-map-count", srv.Config.MaxMapCount, "Limits the maximum number of active mmaps. Pilosa will fall back to reading files once this is exhausted. Set below your system's vm.max_map_count.") flags.Uint64Var(&srv.Config.MaxFileCount, "max-file-count", srv.Config.MaxFileCount, "Soft limit on the maximum number of fragment files Pilosa keeps open simultaneously.") + flags.IntVar(&srv.Config.QueryHistoryLength, "query-history-length", srv.Config.QueryHistoryLength, "Number of queries to remember in history.") // TLS SetTLSConfig(flags, "", &srv.Config.TLS.CertificatePath, &srv.Config.TLS.CertificateKeyPath, &srv.Config.TLS.CACertPath, &srv.Config.TLS.SkipVerify, &srv.Config.TLS.EnableClientVerification) diff --git a/http/client.go b/http/client.go index 801ac9005..81cbe6310 100644 --- a/http/client.go +++ b/http/client.go @@ -1302,7 +1302,7 @@ func (c *InternalClient) GetPastQueries(ctx context.Context, uri *pilosa.URI) ([ return nil, errors.Wrap(err, "reading") } - queries := make([]pilosa.PastQueryStatus, 128) + queries := make([]pilosa.PastQueryStatus, 100) if err := json.Unmarshal(body, &queries); err != nil { return nil, fmt.Errorf("unmarshal response: %s", err) } diff --git a/server.go b/server.go index 81cfeffd7..1b876bce3 100644 --- a/server.go +++ b/server.go @@ -86,6 +86,8 @@ type Server struct { // nolint: maligned defaultClient InternalClient dataDir string + + queryHistoryLength int } // Holder returns the holder for server. @@ -361,6 +363,16 @@ func OptServerRBFConfig(cfg *rbfcfg.Config) ServerOption { } } +// OptServerQueryHistoryLength is a functional option on Server +// used to specify the length of the query history buffer that maintains +// the information returned at /query-history. +func OptServerQueryHistoryLength(length int) ServerOption { + return func(s *Server) error { + s.queryHistoryLength = length + return nil + } +} + // NewServer returns a new instance of Server. func NewServer(opts ...ServerOption) (*Server, error) { cluster := newCluster() diff --git a/server/config.go b/server/config.go index c69a5ea26..838ed235f 100644 --- a/server/config.go +++ b/server/config.go @@ -206,6 +206,11 @@ type Config struct { // RBFConfig defines all externally configurable RBF flags. RBFConfig *rbfcfg.Config + + // QueryHistoryLength sets the maximum number of queries that are maintained + // for the /query-history endpoint. This parameter is per-node, and the + // result combines the history from all nodes. + QueryHistoryLength int } // NewConfig returns an instance of Config with default options. @@ -230,6 +235,8 @@ func NewConfig() *Config { ImportWorkerPoolSize: runtime.NumCPU(), RBFConfig: rbfcfg.NewDefaultConfig(), + + QueryHistoryLength: 100, } // Cluster config. diff --git a/server/server.go b/server/server.go index 46a83c453..b510e0fb3 100644 --- a/server/server.go +++ b/server/server.go @@ -412,6 +412,7 @@ func (m *Command) SetupServer() error { pilosa.OptServerTxsrc(m.Config.Txsrc), pilosa.OptServerRowcacheOff(m.Config.RowcacheOff), pilosa.OptServerRBFConfig(m.Config.RBFConfig), + pilosa.OptServerQueryHistoryLength(m.Config.QueryHistoryLength), coordinatorOpt, } diff --git a/tracker.go b/tracker.go index 7d58b2479..b541375e2 100644 --- a/tracker.go +++ b/tracker.go @@ -101,11 +101,11 @@ func (b *ringBuffer) slice() []pastQuery { return append(b.queries[b.start:b.count], b.queries[0:b.start]...) } -func newQueryTracker() *queryTracker { +func newQueryTracker(historyLength int) *queryTracker { done := make(chan struct{}) updates := make(chan queryStatusUpdate, 128) checks := make(chan chan<- []*activeQuery) - history := newRingBuffer(128) + history := newRingBuffer(historyLength) tracker := &queryTracker{ updates: updates, checks: checks, From c5a7a259aac2f499c03c2df6dadc1b44002c6bc1 Mon Sep 17 00:00:00 2001 From: Alan Bernstein Date: Mon, 26 Oct 2020 16:55:30 -0500 Subject: [PATCH 07/10] Make minor fixes to query tracker --- tracker.go | 11 ++++++----- tracker_test.go | 4 ++-- 2 files changed, 8 insertions(+), 7 deletions(-) diff --git a/tracker.go b/tracker.go index b541375e2..330fbe0a1 100644 --- a/tracker.go +++ b/tracker.go @@ -23,14 +23,14 @@ import ( type ActiveQueryStatus struct { Query string `json:"query"` Node string `json:"node"` - Index string `json:index` + Index string `json:"index"` Age time.Duration `json:"age"` } type PastQueryStatus struct { Query string `json:"query"` Node string `json:"nodeID"` - Index string `json:index` + Index string `json:"index"` Age time.Duration `json:"age"` Runtime time.Duration `json:"runtime"` } @@ -86,9 +86,10 @@ func (b *ringBuffer) add(q pastQuery) { b.mu.Lock() defer b.mu.Unlock() - b.queries[(b.start+b.count)%cap(b.queries)] = q - if b.count == cap(b.queries) { - b.start = (b.start + 1) % cap(b.queries) + // len(b.queries) is used here as the *capacity* of the ringBuffer + b.queries[(b.start+b.count)%len(b.queries)] = q + if b.count == len(b.queries) { + b.start = (b.start + 1) % len(b.queries) } else { b.count++ } diff --git a/tracker_test.go b/tracker_test.go index fa6767e50..d2b10e08a 100644 --- a/tracker_test.go +++ b/tracker_test.go @@ -56,14 +56,14 @@ func TestRingBuffer(t *testing.T) { } func TestQueryTracker(t *testing.T) { - tracker := newQueryTracker() + tracker := newQueryTracker(5) defer tracker.Stop() if queries := tracker.ActiveQueries(); len(queries) > 0 { t.Fatalf("expected no active queries; found %v", queries) } - qs := tracker.Start("test query", "node0") + qs := tracker.Start("test query", "node0", "i") var queries []ActiveQueryStatus for len(queries) < 1 { From a8a7e533821508d4e61a2be573a117475382013a Mon Sep 17 00:00:00 2001 From: Alan Bernstein Date: Mon, 26 Oct 2020 17:58:40 -0500 Subject: [PATCH 08/10] Add query-history test --- server/handler_test.go | 129 +++++++++++++++++++++++++++++++++++++++-- 1 file changed, 125 insertions(+), 4 deletions(-) diff --git a/server/handler_test.go b/server/handler_test.go index c115cdaa8..3fcd78d55 100644 --- a/server/handler_test.go +++ b/server/handler_test.go @@ -26,6 +26,7 @@ import ( gohttp "net/http" "net/http/httptest" "reflect" + "sort" "strings" "sync" "testing" @@ -401,7 +402,7 @@ func TestHandler_Endpoints(t *testing.T) { }) t.Run("UI/shard-distribution", func(t *testing.T) { - // This tests the response structure, not the shard distribution. + // This tests the response structure, not the cluster behavior. w := httptest.NewRecorder() h.ServeHTTP(w, test.MustNewHTTPRequest("GET", "/ui/shard-distribution", nil)) if w.Code != gohttp.StatusOK { @@ -1400,7 +1401,7 @@ func TestCluster_TranslateStore(t *testing.T) { cluster.GetNode(0).Config.Gossip.Port = "0" err := cluster.GetNode(0).Start() if err != nil { - t.Fatalf("starting cluster 0: %v", err) + t.Fatalf("starting node 0: %v", err) } defer cluster.GetNode(0).Close() @@ -1417,7 +1418,7 @@ func TestClusterTranslator(t *testing.T) { cluster.GetNode(0).Config.Gossip.Port = "0" err := cluster.GetNode(0).Start() if err != nil { - t.Fatalf("starting cluster 0: %v", err) + t.Fatalf("starting node 0: %v", err) } defer cluster.GetNode(0).Close() cluster.Nodes[1] = test.NewCommandNode(t, false, @@ -1430,7 +1431,7 @@ func TestClusterTranslator(t *testing.T) { cluster.GetNode(1).Config.Gossip.Seeds = []string{cluster.GetNode(0).GossipAddress()} err = cluster.GetNode(1).Start() if err != nil { - t.Fatalf("starting cluster 1: %v", err) + t.Fatalf("starting node 1: %v", err) } defer cluster.GetNode(1).Close() @@ -1471,6 +1472,126 @@ func TestClusterTranslator(t *testing.T) { } } +func TestQueryHistory(t *testing.T) { + cluster := test.MustNewCluster(t, 2) + cluster.Nodes[0] = test.NewCommandNode(t, true) + cluster.GetNode(0).Config.Gossip.Port = "0" + err := cluster.GetNode(0).Start() + if err != nil { + t.Fatalf("starting node 0: %v", err) + } + defer cluster.GetNode(0).Close() + + cluster.Nodes[1] = test.NewCommandNode(t, false) + cluster.GetNode(1).Config.Gossip.Port = "0" + cluster.GetNode(1).Config.Gossip.Seeds = []string{cluster.GetNode(0).GossipAddress()} + err = cluster.GetNode(1).Start() + if err != nil { + t.Fatalf("starting node 1: %v", err) + } + defer cluster.GetNode(1).Close() + + cmd := cluster.GetNode(0) + h := cmd.Handler.(*http.Handler).Handler + + w := httptest.NewRecorder() + + test.Do(t, "POST", cmd.URL()+"/index/i0", "") + test.Do(t, "POST", cmd.URL()+"/index/i0/field/f0", "") + test.Do(t, "POST", cmd.URL()+"/index/i0/query", "Set(0, f0=0)") + test.Do(t, "POST", cmd.URL()+"/index/i0/query", "Set(3000000, f0=0)") + test.Do(t, "POST", cmd.URL()+"/index/i0/query", "TopN(f0)") + + h.ServeHTTP(w, test.MustNewHTTPRequest("GET", "/query-history", nil)) + if w.Code != gohttp.StatusOK { + t.Fatalf("unexpected status code: %d", w.Code) + } + ret := mustJSONDecodeSlice(t, w.Body) + + for n, r := range ret { + fmt.Printf("%d %+v\n", n, r) + } + + // verify result length + if len(ret) != 7 { + // each set query executes on both nodes once + // topn query gets added to history on node0 once, node1 twice + t.Fatalf("expected list of length 7, got %d", len(ret)) + } + + // verify sort order + if !sort.SliceIsSorted(ret, func(i, j int) bool { + // must match the sort in api.PastQueries + return ret[i].(map[string]interface{})["age"].(float64) > ret[j].(map[string]interface{})["age"].(float64) + }) { + t.Fatalf("response list not sorted correctly") + } + + // verify response structure + queryStatus := ret[4].(map[string]interface{}) + expectedStrings := map[string]string{ + "index": "i0", + "nodeID": cluster.GetNode(0).Server.NodeID(), + "query": "TopN(f0)", + } + for k, exp := range expectedStrings { + got, ok := queryStatus[k] + if !ok { + t.Fatalf("response key '%s' not present", k) + } + if exp != got { + t.Fatalf("response value for key '%s' was '%s', expected '%s'", k, got, exp) + } + } + expectedNumKeys := []string{"age", "runtime"} + for _, k := range expectedNumKeys { + got, ok := queryStatus[k] + if !ok { + t.Fatalf("response key '%s' not present", k) + } + gotint, ok := got.(float64) // ugh + if !ok { + t.Fatalf("response value for key '%s' %T instead of float64", k, got) + } + if gotint <= 0 { + t.Fatalf("negative value for key '%s'", k) + } + } + + // verify additional history entries for the TopN call + queryStatus = ret[5].(map[string]interface{}) + expectedStrings = map[string]string{ + "index": "i0", + "nodeID": cluster.GetNode(1).Server.NodeID(), + "query": "TopN(_field=\"f0\")", + } + for k, exp := range expectedStrings { + got, ok := queryStatus[k] + if !ok { + t.Fatalf("response key '%s' not present", k) + } + if exp != got { + t.Fatalf("response value for key '%s' was '%s', expected '%s'", k, got, exp) + } + } + queryStatus = ret[6].(map[string]interface{}) + expectedStrings = map[string]string{ + "index": "i0", + "nodeID": cluster.GetNode(1).Server.NodeID(), + "query": "TopN(_field=\"f0\", ids=[0])", + } + for k, exp := range expectedStrings { + got, ok := queryStatus[k] + if !ok { + t.Fatalf("response key '%s' not present", k) + } + if exp != got { + t.Fatalf("response value for key '%s' was '%s', expected '%s'", k, got, exp) + } + } + +} + func mustJSONDecode(t *testing.T, r io.Reader) (ret map[string]interface{}) { dec := json.NewDecoder(r) err := dec.Decode(&ret) From 0c8e4afe453dbc337d8f8fc14b36af5bb07c1843 Mon Sep 17 00:00:00 2001 From: Alan Bernstein Date: Tue, 27 Oct 2020 09:55:45 -0500 Subject: [PATCH 09/10] Set test nodeIDs to guarantee iteration order in executor --- server/handler_test.go | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/server/handler_test.go b/server/handler_test.go index 3fcd78d55..2d740d622 100644 --- a/server/handler_test.go +++ b/server/handler_test.go @@ -1474,7 +1474,9 @@ func TestClusterTranslator(t *testing.T) { func TestQueryHistory(t *testing.T) { cluster := test.MustNewCluster(t, 2) - cluster.Nodes[0] = test.NewCommandNode(t, true) + cluster.Nodes[0] = test.NewCommandNode(t, true, server.OptCommandServerOptions( + pilosa.OptServerNodeID("1"), + )) cluster.GetNode(0).Config.Gossip.Port = "0" err := cluster.GetNode(0).Start() if err != nil { @@ -1482,7 +1484,9 @@ func TestQueryHistory(t *testing.T) { } defer cluster.GetNode(0).Close() - cluster.Nodes[1] = test.NewCommandNode(t, false) + cluster.Nodes[1] = test.NewCommandNode(t, false, server.OptCommandServerOptions( + pilosa.OptServerNodeID("0"), + )) cluster.GetNode(1).Config.Gossip.Port = "0" cluster.GetNode(1).Config.Gossip.Seeds = []string{cluster.GetNode(0).GossipAddress()} err = cluster.GetNode(1).Start() From 65877c1e56734c2465e5468b8e60b738353b5e4d Mon Sep 17 00:00:00 2001 From: Alan Bernstein Date: Mon, 23 Nov 2020 21:31:40 -0600 Subject: [PATCH 10/10] Store start instead of age, skip remote, fix timing bug --- api.go | 8 +++- server/handler_test.go | 84 +++++++++--------------------------------- tracker.go | 14 +++---- tracker_test.go | 4 +- 4 files changed, 32 insertions(+), 78 deletions(-) diff --git a/api.go b/api.go index 588233603..9be6427c8 100644 --- a/api.go +++ b/api.go @@ -147,6 +147,7 @@ func (api *API) Txf() *TxFactory { // Query parses a PQL query out of the request and executes it. func (api *API) Query(ctx context.Context, req *QueryRequest) (QueryResponse, error) { + start := time.Now() span, ctx := tracing.StartSpanFromContext(ctx, "API.Query") defer span.Finish() @@ -158,7 +159,10 @@ func (api *API) Query(ctx context.Context, req *QueryRequest) (QueryResponse, er if err != nil { return QueryResponse{}, errors.Wrap(err, "parsing") } - defer api.tracker.Finish(api.tracker.Start(req.Query, api.server.nodeID, req.Index), time.Now()) + + if !req.Remote { + defer api.tracker.Finish(api.tracker.Start(req.Query, api.server.nodeID, req.Index, start)) + } execOpts := &execOptions{ Remote: req.Remote, Profile: req.Profile, @@ -2072,7 +2076,7 @@ func (api *API) PastQueries(ctx context.Context, remote bool) ([]PastQueryStatus } sort.Slice(clusterQueries, func(i, j int) bool { - return clusterQueries[i].Age.Seconds() > clusterQueries[j].Age.Seconds() + return clusterQueries[i].Start.After(clusterQueries[j].Start) }) return clusterQueries, nil diff --git a/server/handler_test.go b/server/handler_test.go index 2d740d622..67b9af8e5 100644 --- a/server/handler_test.go +++ b/server/handler_test.go @@ -1510,90 +1510,42 @@ func TestQueryHistory(t *testing.T) { if w.Code != gohttp.StatusOK { t.Fatalf("unexpected status code: %d", w.Code) } - ret := mustJSONDecodeSlice(t, w.Body) - for n, r := range ret { - fmt.Printf("%d %+v\n", n, r) + ret := make([]pilosa.PastQueryStatus, 3) + b, err := ioutil.ReadAll(w.Body) + if err != nil { + t.Fatalf("reading: %v", err) + } + err = json.Unmarshal(b, &ret) + if err != nil { + t.Fatalf("unmarshalling: %v", err) } // verify result length - if len(ret) != 7 { + if len(ret) != 3 { // each set query executes on both nodes once // topn query gets added to history on node0 once, node1 twice - t.Fatalf("expected list of length 7, got %d", len(ret)) + t.Fatalf("expected list of length 3, got %d", len(ret)) } // verify sort order if !sort.SliceIsSorted(ret, func(i, j int) bool { // must match the sort in api.PastQueries - return ret[i].(map[string]interface{})["age"].(float64) > ret[j].(map[string]interface{})["age"].(float64) + return ret[i].Start.After(ret[j].Start) }) { t.Fatalf("response list not sorted correctly") } - // verify response structure - queryStatus := ret[4].(map[string]interface{}) - expectedStrings := map[string]string{ - "index": "i0", - "nodeID": cluster.GetNode(0).Server.NodeID(), - "query": "TopN(f0)", + // verify some response values + if ret[0].Index != "i0" { + t.Fatalf("response value for 'Index' was '%s', expected 'i0'", ret[0].Index) } - for k, exp := range expectedStrings { - got, ok := queryStatus[k] - if !ok { - t.Fatalf("response key '%s' not present", k) - } - if exp != got { - t.Fatalf("response value for key '%s' was '%s', expected '%s'", k, got, exp) - } + if ret[0].Node != cluster.GetNode(0).Server.NodeID() { + t.Fatalf("response value for 'Node' was '%s', expected '%s'", ret[0].Node, cluster.GetNode(0).Server.NodeID()) } - expectedNumKeys := []string{"age", "runtime"} - for _, k := range expectedNumKeys { - got, ok := queryStatus[k] - if !ok { - t.Fatalf("response key '%s' not present", k) - } - gotint, ok := got.(float64) // ugh - if !ok { - t.Fatalf("response value for key '%s' %T instead of float64", k, got) - } - if gotint <= 0 { - t.Fatalf("negative value for key '%s'", k) - } + if ret[0].Query != "TopN(f0)" { + t.Fatalf("response value for 'Query' was '%s', expected 'TopN(f0)'", ret[0].Query) } - - // verify additional history entries for the TopN call - queryStatus = ret[5].(map[string]interface{}) - expectedStrings = map[string]string{ - "index": "i0", - "nodeID": cluster.GetNode(1).Server.NodeID(), - "query": "TopN(_field=\"f0\")", - } - for k, exp := range expectedStrings { - got, ok := queryStatus[k] - if !ok { - t.Fatalf("response key '%s' not present", k) - } - if exp != got { - t.Fatalf("response value for key '%s' was '%s', expected '%s'", k, got, exp) - } - } - queryStatus = ret[6].(map[string]interface{}) - expectedStrings = map[string]string{ - "index": "i0", - "nodeID": cluster.GetNode(1).Server.NodeID(), - "query": "TopN(_field=\"f0\", ids=[0])", - } - for k, exp := range expectedStrings { - got, ok := queryStatus[k] - if !ok { - t.Fatalf("response key '%s' not present", k) - } - if exp != got { - t.Fatalf("response value for key '%s' was '%s', expected '%s'", k, got, exp) - } - } - } func mustJSONDecode(t *testing.T, r io.Reader) (ret map[string]interface{}) { diff --git a/tracker.go b/tracker.go index 330fbe0a1..359850005 100644 --- a/tracker.go +++ b/tracker.go @@ -31,7 +31,7 @@ type PastQueryStatus struct { Query string `json:"query"` Node string `json:"nodeID"` Index string `json:"index"` - Age time.Duration `json:"age"` + Start time.Time `json:"start"` Runtime time.Duration `json:"runtime"` } @@ -146,15 +146,14 @@ func newQueryTracker(historyLength int) *queryTracker { return tracker } -func (t *queryTracker) Start(query, nodeID, index string) *activeQuery { - now := time.Now() - q := &activeQuery{query, nodeID, index, now} +func (t *queryTracker) Start(query, nodeID, index string, start time.Time) *activeQuery { + q := &activeQuery{query, nodeID, index, start} t.updates <- queryStatusUpdate{q, false, time.Time{}} return q } -func (t *queryTracker) Finish(q *activeQuery, endTime time.Time) { - t.updates <- queryStatusUpdate{q, true, endTime} +func (t *queryTracker) Finish(q *activeQuery) { + t.updates <- queryStatusUpdate{q, true, time.Now()} } func (t *queryTracker) ActiveQueries() []ActiveQueryStatus { @@ -185,10 +184,9 @@ func (t *queryTracker) ActiveQueries() []ActiveQueryStatus { func (t *queryTracker) PastQueries() []PastQueryStatus { queries := t.history.slice() - now := time.Now() out := make([]PastQueryStatus, len(queries)) for i, v := range queries { - out[i] = PastQueryStatus{v.query, v.node, v.index, now.Sub(v.started), v.runtime} + out[i] = PastQueryStatus{v.query, v.node, v.index, v.started, v.runtime} } return out diff --git a/tracker_test.go b/tracker_test.go index d2b10e08a..a5d8df418 100644 --- a/tracker_test.go +++ b/tracker_test.go @@ -63,7 +63,7 @@ func TestQueryTracker(t *testing.T) { t.Fatalf("expected no active queries; found %v", queries) } - qs := tracker.Start("test query", "node0", "i") + qs := tracker.Start("test query", "node0", "i", time.Now()) var queries []ActiveQueryStatus for len(queries) < 1 { @@ -73,7 +73,7 @@ func TestQueryTracker(t *testing.T) { t.Fatalf("unexpected queries: %v", queries) } - tracker.Finish(qs, time.Now()) + tracker.Finish(qs) for len(queries) > 0 { queries = tracker.ActiveQueries()