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,