scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-11 07:11:02 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #1013 from alanbernstein/query-activity-api

Browse source 
Add query history endpoint
This commit is contained in:
Alan Bernstein 2020-11-24 08:46:43 -06:00 • committed by GitHub
commit 9e3e0a6be7
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
11 changed files with 324 additions and 25 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

38
api.go
View file

@ -100,7 +100,7 @@ func NewAPI(opts ...apiOption) (*API, error) {
}()
}
api.tracker = newQueryTracker()
api.tracker = newQueryTracker(api.server.queryHistoryLength)
return api, nil
}
@ -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))
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,
@ -2050,6 +2054,34 @@ func (api *API) ActiveQueries(ctx context.Context) ([]ActiveQueryStatus, error)
return api.tracker.ActiveQueries(), nil
}
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")
}
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].Start.After(clusterQueries[j].Start)
})
return clusterQueries, 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 +2156,7 @@ const (
apiTransactions
apiGetTransaction
apiActiveQueries
apiPastQueries
)
var methodsCommon = map[apiMethod]struct{}{
@ -2164,4 +2197,5 @@ var methodsNormal = map[apiMethod]struct{}{
apiTransactions: {},
apiGetTransaction: {},
apiActiveQueries: {},
apiPastQueries: {},
}

5
client.go
View file

@ -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
}

1
ctl/server.go
View file

@ -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)

35
http/client.go
View file

@ -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, 100)
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")

25
http/handler.go
View file

@ -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")
@ -704,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")
@ -1184,6 +1186,27 @@ func (h *Handler) handleGetActiveQueries(w http.ResponseWriter, r *http.Request)
}
}
func (h *Handler) handleGetPastQueries(w http.ResponseWriter, r *http.Request) {
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)
}
}
type postIndexAttrDiffRequest struct {
Blocks []pilosa.AttrBlock `json:"blocks"`
}

12
server.go
View file

@ -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()

7
server/config.go
View file

@ -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.

85
server/handler_test.go
View file

@ -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,82 @@ func TestClusterTranslator(t *testing.T) {
}
}
func TestQueryHistory(t *testing.T) {
cluster := test.MustNewCluster(t, 2)
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 {
t.Fatalf("starting node 0: %v", err)
}
defer cluster.GetNode(0).Close()
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()
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 := 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) != 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 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].Start.After(ret[j].Start)
}) {
t.Fatalf("response list not sorted correctly")
}
// verify some response values
if ret[0].Index != "i0" {
t.Fatalf("response value for 'Index' was '%s', expected 'i0'", ret[0].Index)
}
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())
}
if ret[0].Query != "TopN(f0)" {
t.Fatalf("response value for 'Query' was '%s', expected 'TopN(f0)'", ret[0].Query)
}
}
func mustJSONDecode(t *testing.T, r io.Reader) (ret map[string]interface{}) {
dec := json.NewDecoder(r)
err := dec.Decode(&ret)

1
server/server.go
View file

@ -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,
}

95
tracker.go
View file

@ -22,33 +22,95 @@ 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:"nodeID"`
Index string `json:"index"`
Start time.Time `json:"start"`
Runtime time.Duration `json:"runtime"`
}
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
}
type queryStatusUpdate struct {
q *activeQuery
end bool
q *activeQuery
end bool
endTime time.Time
}
type queryTracker struct {
updates chan<- queryStatusUpdate
checks chan<- chan<- []*activeQuery
wg sync.WaitGroup
stop chan struct{}
history *ringBuffer
wg sync.WaitGroup
stop chan struct{}
}
func newQueryTracker() *queryTracker {
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()
// 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++
}
}
// 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(historyLength int) *queryTracker {
done := make(chan struct{})
updates := make(chan queryStatusUpdate, 128)
checks := make(chan chan<- []*activeQuery)
history := newRingBuffer(historyLength)
tracker := &queryTracker{
updates: updates,
checks: checks,
history: history,
stop: done,
}
tracker.wg.Add(1)
@ -61,6 +123,8 @@ 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{}{}
@ -82,15 +146,14 @@ func newQueryTracker() *queryTracker {
return tracker
}
func (t *queryTracker) Start(query string) *activeQuery {
now := time.Now()
q := &activeQuery{query, now}
t.updates <- queryStatusUpdate{q, false}
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) {
t.updates <- queryStatusUpdate{q, true}
t.updates <- queryStatusUpdate{q, true, time.Now()}
}
func (t *queryTracker) ActiveQueries() []ActiveQueryStatus {
@ -114,11 +177,21 @@ 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, v.index, now.Sub(v.started)}
}
return out
}
func (t *queryTracker) PastQueries() []PastQueryStatus {
queries := t.history.slice()
out := make([]PastQueryStatus, len(queries))
for i, v := range queries {
out[i] = PastQueryStatus{v.query, v.node, v.index, v.started, v.runtime}
}
return out
}
func (t *queryTracker) Stop() {
close(t.stop)
t.wg.Wait()

45
tracker_test.go
View file

@ -14,17 +14,56 @@
package pilosa
import "testing"
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()
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")
qs := tracker.Start("test query", "node0", "i", time.Now())
var queries []ActiveQueryStatus
for len(queries) < 1 {