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)