Merge pull request #881 from cwndrws/recalculate-caches-endpoint

Add Recalculate Caches Endpoint

docs/api-reference.md

@@ -412,3 +412,16 @@ Response:
 {"version":"v0.6.0"}
 ```
 
+### Recalculate Caches
+
+`POST /recalculate-caches`
+
+Recalculates the caches on demand. The cache is recalculated every 10
+seconds by default. This endpoint can be used to recalculate the cache
+before the 10 second interval. This should probably only be used in
+integration tests and not in a typical production workflow. Note that
+in a multi-node cluster, the cache is only recalculated on the node
+that receives the request.
+
+Response: `204 No Content`
+

frame.go

@@ -543,6 +543,13 @@ func (f *Frame) Views() []*View {
 	return other
 }
 
+// RecalculateCaches recalculates caches on every view in the frame.
+func (f *Frame) RecalculateCaches() {
+	for _, view := range f.Views() {
+		view.RecalculateCaches()
+	}
+}
+
 // CreateViewIfNotExists returns the named view, creating it if necessary.
 func (f *Frame) CreateViewIfNotExists(name string) (*View, error) {
 	// Don't create inverse views if they are not enabled.

handler.go

@@ -134,6 +134,7 @@ func NewRouter(handler *Handler) *mux.Router {
 	router.HandleFunc("/slices/max", handler.handleGetSliceMax).Methods("GET")
 	router.HandleFunc("/status", handler.handleGetStatus).Methods("GET")
 	router.HandleFunc("/version", handler.handleGetVersion).Methods("GET")
+	router.HandleFunc("/recalculate-caches", handler.handleRecalculateCaches).Methods("POST")
 
 	// TODO: Apply MethodNotAllowed statuses to all endpoints.
 	// Ideally this would be automatic, as described in this (wontfix) ticket:
@@ -1962,6 +1963,11 @@ func (h *Handler) InputJSONDataParser(req map[string]interface{}, index *Index,
 	return setBits, nil
 }
 
+func (h *Handler) handleRecalculateCaches(w http.ResponseWriter, r *http.Request) {
+	h.Holder.RecalculateCaches()
+	w.WriteHeader(http.StatusNoContent)
+}
+
 // GetTimeStamp retrieves unix timestamp from Input data.
 func GetTimeStamp(data map[string]interface{}, timeField string) (int64, error) {
 	tmstamp, ok := data[timeField]

handler_test.go

@@ -1782,3 +1782,19 @@ func MustReadAll(r io.Reader) []byte {
 	}
 	return buf
 }
+
+func TestHandler_RecalculateCaches(t *testing.T) {
+	hldr := test.MustOpenHolder()
+	defer hldr.Close()
+
+	h := test.NewHandler()
+	h.Holder = hldr.Holder
+	h.Cluster = test.NewCluster(1)
+
+	w := httptest.NewRecorder()
+	h.ServeHTTP(w, test.MustNewHTTPRequest("POST", "/recalculate-caches", nil))
+	if w.Code != http.StatusNoContent {
+		t.Fatalf("unexpected status code: %d", w.Code)
+	}
+
+}

holder.go

@@ -361,6 +361,16 @@ func (h *Holder) flushCaches() {
 	}
 }
 
+// RecalculateCaches recalculates caches on every index in the holder. This is
+// probably not practical to call in real-world workloads, but makes writing
+// integration tests much eaiser, since one doesn't have to wait 10 seconds
+// after setting bits to get expected response.
+func (h *Holder) RecalculateCaches() {
+	for _, index := range h.Indexes() {
+		index.RecalculateCaches()
+	}
+}
+
 // setFileLimit attempts to set the open file limit to the FileLimit constant defined above.
 func (h *Holder) setFileLimit() {
 	oldLimit := &syscall.Rlimit{}

index.go

@@ -392,6 +392,13 @@ func (i *Index) Frames() []*Frame {
 	return a
 }
 
+// RecalculateCaches recalculates caches on every frame in the index.
+func (i *Index) RecalculateCaches() {
+	for _, frame := range i.Frames() {
+		frame.RecalculateCaches()
+	}
+}
+
 // CreateFrame creates a frame.
 func (i *Index) CreateFrame(name string, opt FrameOptions) (*Frame, error) {
 	i.mu.Lock()

view.go

@@ -213,6 +213,13 @@ func (v *View) Fragments() []*Fragment {
 	return other
 }
 
+// RecalculateCaches recalculates the cache on every fragment in the view.
+func (v *View) RecalculateCaches() {
+	for _, fragment := range v.Fragments() {
+		fragment.RecalculateCache()
+	}
+}
+
 // CreateFragmentIfNotExists returns a fragment in the view by slice.
 func (v *View) CreateFragmentIfNotExists(slice uint64) (*Fragment, error) {
 	v.mu.Lock()