diff --git a/api.go b/api.go index e544d2d3f..84c2f5146 100644 --- a/api.go +++ b/api.go @@ -484,7 +484,7 @@ func (api *API) PostClusterMessage(ctx context.Context, pb proto.Message) error } func (api *API) LocalID(ctx context.Context) string { - return api.Holder.LocalID + return api.Cluster.Node.ID } func (api *API) Schema(ctx context.Context) []*IndexInfo { diff --git a/handler.go b/handler.go index bd47524ed..f9fc73c52 100644 --- a/handler.go +++ b/handler.go @@ -16,7 +16,6 @@ package pilosa import ( "context" - "encoding/csv" "encoding/json" "errors" "expvar" @@ -44,21 +43,7 @@ import ( // Handler represents an HTTP handler. type Handler struct { - Holder *Holder - Broadcaster Broadcaster - BroadcastHandler BroadcastHandler - StatusHandler StatusHandler - - FileSystem FileSystem - - // Local hostname & cluster configuration. - Node *Node - Cluster *Cluster - RemoteClient *http.Client - - Router *mux.Router - NormalRouter *mux.Router - RestrictedRouter *mux.Router + Router *mux.Router // The execution engine for running queries. Executor interface { @@ -295,8 +280,9 @@ func (h *Handler) handleWebUI(w http.ResponseWriter, r *http.Request) { // handleGetSchema handles GET /schema requests. func (h *Handler) handleGetSchema(w http.ResponseWriter, r *http.Request) { + schema := h.API.Schema(r.Context()) if err := json.NewEncoder(w).Encode(getSchemaResponse{ - Indexes: h.Holder.Schema(), + Indexes: schema, }); err != nil { h.Logger.Printf("write schema response error: %s", err) } @@ -304,7 +290,7 @@ func (h *Handler) handleGetSchema(w http.ResponseWriter, r *http.Request) { // handleGetStatus handles GET /status requests. func (h *Handler) handleGetStatus(w http.ResponseWriter, r *http.Request) { - pb, err := h.StatusHandler.ClusterStatus() + status, err := h.API.Status(r.Context()) if err != nil { h.Logger.Printf("cluster status error: %s", err) return @@ -774,13 +760,6 @@ func (h *Handler) handlePostFrameField(w http.ResponseWriter, r *http.Request) { return } - // Retrieve frame by name. - f := h.Holder.Frame(indexName, frameName) - if f == nil { - http.Error(w, ErrFrameNotFound.Error(), http.StatusNotFound) - return - } - field := &Field{ Name: fieldName, Type: req.Type, @@ -788,23 +767,15 @@ func (h *Handler) handlePostFrameField(w http.ResponseWriter, r *http.Request) { Max: req.Max, } - // Create new field. - if err := f.CreateField(field); err != nil { - http.Error(w, err.Error(), http.StatusInternalServerError) + if err := h.API.CreateFrameField(r.Context(), indexName, frameName, field); err != nil { + if err == ErrFrameNotFound { + http.Error(w, err.Error(), http.StatusNotFound) + } else { + http.Error(w, err.Error(), http.StatusInternalServerError) + } return } - // Send the create field message to all nodes. - err := h.Broadcaster.SendSync( - &internal.CreateFieldMessage{ - Index: indexName, - Frame: frameName, - Field: encodeField(field), - }) - if err != nil { - h.Logger.Printf("problem sending CreateField message: %s", err) - } - // Encode response. if err := json.NewEncoder(w).Encode(postFrameFieldResponse{}); err != nil { h.Logger.Printf("response encoding error: %s", err) @@ -825,30 +796,15 @@ func (h *Handler) handleDeleteFrameField(w http.ResponseWriter, r *http.Request) frameName := mux.Vars(r)["frame"] fieldName := mux.Vars(r)["field"] - // Retrieve frame by name. - f := h.Holder.Frame(indexName, frameName) - if f == nil { - http.Error(w, ErrFrameNotFound.Error(), http.StatusNotFound) + if err := h.API.DeleteFrameField(r.Context(), indexName, frameName, fieldName); err != nil { + if err == ErrFrameNotFound { + http.Error(w, err.Error(), http.StatusNotFound) + } else { + http.Error(w, err.Error(), http.StatusInternalServerError) + } return } - // Delete field. - if err := f.DeleteField(fieldName); err != nil { - http.Error(w, err.Error(), http.StatusInternalServerError) - return - } - - // Send the delete field message to all nodes. - err := h.Broadcaster.SendSync( - &internal.DeleteFieldMessage{ - Index: indexName, - Frame: frameName, - Field: fieldName, - }) - if err != nil { - h.Logger.Printf("problem sending DeleteField message: %s", err) - } - // Encode response. if err := json.NewEncoder(w).Encode(deleteFrameFieldResponse{}); err != nil { h.Logger.Printf("response encoding error: %s", err) @@ -859,24 +815,18 @@ func (h *Handler) handleGetFrameFields(w http.ResponseWriter, r *http.Request) { indexName := mux.Vars(r)["index"] frameName := mux.Vars(r)["frame"] - index := h.Holder.index(indexName) - if index == nil { - http.Error(w, ErrIndexNotFound.Error(), http.StatusNotFound) - return - } - - frame := index.frame(frameName) - if frame == nil { - http.Error(w, ErrFrameNotFound.Error(), http.StatusNotFound) - return - } - - fields, err := frame.GetFields() - if err == ErrFrameFieldsNotAllowed { - http.Error(w, err.Error(), http.StatusBadRequest) - return - } else if err != nil { - http.Error(w, err.Error(), http.StatusInternalServerError) + schema, err := h.API.FrameFields(r.Context(), indexName, frameName) + if err != nil { + switch err { + case ErrIndexNotFound: + fallthrough + case ErrFrameNotFound: + http.Error(w, err.Error(), http.StatusNotFound) + case ErrFrameFieldsNotAllowed: + http.Error(w, err.Error(), http.StatusBadRequest) + default: + http.Error(w, err.Error(), http.StatusInternalServerError) + } return } @@ -899,15 +849,16 @@ func (h *Handler) handleGetFrameViews(w http.ResponseWriter, r *http.Request) { indexName := mux.Vars(r)["index"] frameName := mux.Vars(r)["frame"] - // Retrieve views. - f := h.Holder.Frame(indexName, frameName) - if f == nil { - http.Error(w, ErrFrameNotFound.Error(), http.StatusNotFound) + views, err := h.API.FrameViews(r.Context(), indexName, frameName) + if err != nil { + if err == ErrFrameNotFound { + http.Error(w, err.Error(), http.StatusNotFound) + } else { + http.Error(w, err.Error(), http.StatusInternalServerError) + } return } - // Fetch views. - views := f.Views() names := make([]string, len(views)) for i := range views { names[i] = views[i].Name() @@ -925,31 +876,13 @@ func (h *Handler) handleDeleteView(w http.ResponseWriter, r *http.Request) { frameName := mux.Vars(r)["frame"] viewName := mux.Vars(r)["view"] - // Retrieve frame. - f := h.Holder.Frame(indexName, frameName) - if f == nil { - http.Error(w, ErrFrameNotFound.Error(), http.StatusNotFound) - return - } - - // Delete the view. - if err := f.DeleteView(viewName); err != nil { - // Ingore this error because views do not exist on all nodes due to slice distribution. - if err != ErrInvalidView { + if err := h.API.DeleteView(r.Context(), indexName, frameName, viewName); err != nil { + if err == ErrFrameNotFound { + http.Error(w, err.Error(), http.StatusNotFound) + } else { http.Error(w, err.Error(), http.StatusBadRequest) - return } - } - - // Send the delete view message to all nodes. - err := h.Broadcaster.SendSync( - &internal.DeleteViewMessage{ - Index: indexName, - Frame: frameName, - View: viewName, - }) - if err != nil { - h.Logger.Printf("problem sending DeleteView message: %s", err) + return } // Encode response. @@ -1314,28 +1247,13 @@ func (h *Handler) handleGetExportCSV(w http.ResponseWriter, r *http.Request) { return } - // Find the fragment. - f := h.Holder.Fragment(index, frame, view, slice) - if f == nil { - return + if err = h.API.ExportCSV(r.Context(), index, frame, view, slice, w); err != nil { + if err == ErrFragmentNotFound { + http.Error(w, err.Error(), http.StatusNotFound) + } else { + http.Error(w, err.Error(), http.StatusInternalServerError) + } } - - // Wrap writer with a CSV writer. - cw := csv.NewWriter(w) - - // Iterate over each bit. - if err := f.ForEachBit(func(rowID, columnID uint64) error { - return cw.Write([]string{ - strconv.FormatUint(rowID, 10), - strconv.FormatUint(columnID, 10), - }) - }); err != nil { - http.Error(w, err.Error(), http.StatusInternalServerError) - return - } - - // Ensure data is flushed. - cw.Flush() } // handleGetFragmentNodes handles /fragment/nodes requests. @@ -1351,7 +1269,7 @@ func (h *Handler) handleGetFragmentNodes(w http.ResponseWriter, r *http.Request) } // Retrieve fragment owner nodes. - nodes := h.Cluster.FragmentNodes(index, slice) + nodes := h.API.FragmentNodes(r.Context(), index, slice) // Write to response. if err := json.NewEncoder(w).Encode(nodes); err != nil { @@ -1370,9 +1288,9 @@ func (h *Handler) handleGetFragmentData(w http.ResponseWriter, r *http.Request) } // Retrieve fragment from holder. - f := h.Holder.Fragment(q.Get("index"), q.Get("frame"), q.Get("view"), slice) - if f == nil { - http.Error(w, "fragment not found", http.StatusNotFound) + f, err := h.API.FragmentData(r.Context(), q.Get("index"), q.Get("frame"), q.Get("view"), slice) + if err != nil { + http.Error(w, err.Error(), http.StatusNotFound) return } @@ -1392,31 +1310,12 @@ func (h *Handler) handlePostFragmentData(w http.ResponseWriter, r *http.Request) return } - // Retrieve frame. - f := h.Holder.Frame(q.Get("index"), q.Get("frame")) - if f == nil { - http.Error(w, ErrFrameNotFound.Error(), http.StatusNotFound) - return - } - - // Retrieve view. - view, err := f.CreateViewIfNotExists(q.Get("view")) - if err != nil { - http.Error(w, err.Error(), http.StatusInternalServerError) - return - } - - // Retrieve fragment from frame. - frag, err := view.CreateFragmentIfNotExists(slice) - if err != nil { - http.Error(w, err.Error(), http.StatusInternalServerError) - return - } - - // Read fragment in from request body. - if _, err := frag.ReadFrom(r.Body); err != nil { - http.Error(w, err.Error(), http.StatusInternalServerError) - return + if err = h.API.WriteFragmentData(r.Context(), q.Get("index"), q.Get("frame"), q.Get("view"), slice, r.Body); err != nil { + if err == ErrFrameNotFound { + http.Error(w, ErrFrameNotFound.Error(), http.StatusNotFound) + } else { + http.Error(w, err.Error(), http.StatusInternalServerError) + } } } @@ -1432,19 +1331,16 @@ func (h *Handler) handleGetFragmentBlockData(w http.ResponseWriter, r *http.Requ return } - // Retrieve fragment from holder. - f := h.Holder.Fragment(req.Index, req.Frame, req.View, req.Slice) - if f == nil { - http.Error(w, ErrFragmentNotFound.Error(), http.StatusNotFound) + resp, err := h.API.FragmentBlockData(r.Context(), req) + if err != nil { + if err == ErrFragmentNotFound { + http.Error(w, err.Error(), http.StatusNotFound) + } else { + http.Error(w, err.Error(), http.StatusInternalServerError) + } return } - // Read data - var resp internal.BlockDataResponse - if f != nil { - resp.RowIDs, resp.ColumnIDs = f.BlockData(int(req.Block)) - } - // Encode response. buf, err := proto.Marshal(&resp) if err != nil { @@ -1468,16 +1364,16 @@ func (h *Handler) handleGetFragmentBlocks(w http.ResponseWriter, r *http.Request return } - // Retrieve fragment from holder. - f := h.Holder.Fragment(q.Get("index"), q.Get("frame"), q.Get("view"), slice) - if f == nil { - http.Error(w, "fragment not found", http.StatusNotFound) + blocks, err := h.API.FragmentBlocks(r.Context(), q.Get("index"), q.Get("frame"), q.Get("view"), slice) + if err != nil { + if err == ErrFragmentNotFound { + http.Error(w, err.Error(), http.StatusNotFound) + } else { + http.Error(w, err.Error(), http.StatusInternalServerError) + } return } - // Retrieve blocks. - blocks := f.Blocks() - // Encode response. if err := json.NewEncoder(w).Encode(getFragmentBlocksResponse{ Blocks: blocks, @@ -1509,80 +1405,23 @@ func (h *Handler) handlePostFrameRestore(w http.ResponseWriter, r *http.Request) http.Error(w, err.Error(), http.StatusBadRequest) } - // Create a client for the remote cluster. - client := NewInternalHTTPClientFromURI(host, h.RemoteClient) - - // Determine the maximum number of slices. - maxSlices, err := client.MaxSliceByIndex(r.Context()) - if err != nil { - http.Error(w, "cannot determine remote slice count: "+err.Error(), http.StatusInternalServerError) - return - } - - // Retrieve frame. - f := h.Holder.Frame(indexName, frameName) - if f == nil { - http.Error(w, ErrFrameNotFound.Error(), http.StatusNotFound) - return - } - - // Retrieve list of all views. - views, err := client.FrameViews(r.Context(), indexName, frameName) - if err != nil { - http.Error(w, "cannot retrieve frame views: "+err.Error(), http.StatusInternalServerError) - return - } - - // Loop over each slice and import it if this node owns it. - for slice := uint64(0); slice <= maxSlices[indexName]; slice++ { - // Ignore this slice if we don't own it. - if !h.Cluster.OwnsFragment(h.Node.ID, indexName, slice) { - continue - } - - // Loop over view names. - for _, view := range views { - // Create view. - v, err := f.CreateViewIfNotExists(view) - if err != nil { - http.Error(w, err.Error(), http.StatusInternalServerError) - return - } - - // Otherwise retrieve the local fragment. - frag, err := v.CreateFragmentIfNotExists(slice) - if err != nil { - http.Error(w, err.Error(), http.StatusInternalServerError) - return - } - - // Stream backup from remote node. - rd, err := client.BackupSlice(r.Context(), indexName, frameName, view, slice) - if err != nil { - http.Error(w, err.Error(), http.StatusInternalServerError) - return - } else if rd == nil { - continue // slice doesn't exist - } - - // Restore to local frame and always close reader. - if err := func() error { - defer rd.Close() - if _, err := frag.ReadFrom(rd); err != nil { - return err - } - return nil - }(); err != nil { - http.Error(w, err.Error(), http.StatusInternalServerError) - return - } - } + err = h.API.RestoreFrame(r.Context(), indexName, frameName, host) + switch err { + case nil: + break + case ErrFrameNotFound: + fallthrough + case ErrFragmentNotFound: + http.Error(w, err.Error(), http.StatusNotFound) + default: + http.Error(w, err.Error(), http.StatusInternalServerError) } } // handleGetHosts handles /hosts requests. func (h *Handler) handleGetHosts(w http.ResponseWriter, r *http.Request) { - if err := json.NewEncoder(w).Encode(h.Cluster.Nodes); err != nil { + hosts := h.API.ClusterHosts(r.Context()) + if err := json.NewEncoder(w).Encode(hosts); err != nil { h.Logger.Printf("write version response error: %s", err) } } @@ -1766,13 +1605,6 @@ func (h *Handler) handlePostInputDefinition(w http.ResponseWriter, r *http.Reque indexName := mux.Vars(r)["index"] inputDefName := mux.Vars(r)["input-definition"] - // Find index. - index := h.Holder.Index(indexName) - if index == nil { - http.Error(w, ErrIndexNotFound.Error(), http.StatusNotFound) - return - } - // Decode request. var req InputDefinitionInfo err := json.NewDecoder(r.Body).Decode(&req) @@ -1781,36 +1613,26 @@ func (h *Handler) handlePostInputDefinition(w http.ResponseWriter, r *http.Reque return } - if err := req.Validate(); err != nil { - http.Error(w, err.Error(), http.StatusBadRequest) - return - } - - // Encode InputDefinition to its internal representation. - def := req.Encode() - def.Name = inputDefName - - // Create InputDefinition. - _, err = index.CreateInputDefinition(def) - if err == ErrInputDefinitionExists { + err = h.API.CreateInputDefinition(r.Context(), indexName, inputDefName, req) + switch err { + case nil: + break + case ErrIndexNotFound: + http.Error(w, err.Error(), http.StatusNotFound) + case ErrInputDefinitionExists: http.Error(w, err.Error(), http.StatusConflict) - return - } else if err != nil { + case ErrInputDefinitionAttrsRequired: + fallthrough + case ErrInputDefinitionNameRequired: + fallthrough + case ErrInputDefinitionActionRequired: + fallthrough + case ErrInputDefinitionHasPrimaryKey: + fallthrough + case ErrInputDefinitionDupePrimaryKey: + http.Error(w, err.Error(), http.StatusBadRequest) + default: http.Error(w, err.Error(), http.StatusInternalServerError) - return - } - - err = h.Broadcaster.SendSync( - &internal.CreateInputDefinitionMessage{ - Index: indexName, - Definition: def, - }) - if err != nil { - h.Logger.Printf("problem sending CreateInputDefinition message: %s", err) - } - - if err := json.NewEncoder(w).Encode(defaultInputDefinitionResponse{}); err != nil { - h.Logger.Printf("response encoding error: %s", err) } } @@ -1819,16 +1641,18 @@ func (h *Handler) handleGetInputDefinition(w http.ResponseWriter, r *http.Reques indexName := mux.Vars(r)["index"] inputDefName := mux.Vars(r)["input-definition"] - // Find index. - index := h.Holder.Index(indexName) - if index == nil { - http.Error(w, ErrIndexNotFound.Error(), http.StatusNotFound) - return - } - - inputDef, err := index.InputDefinition(inputDefName) + inputDef, err := h.API.InputDefinition(r.Context(), indexName, inputDefName) if err != nil { - http.Error(w, err.Error(), http.StatusNotFound) + switch err { + case nil: + break + case ErrIndexNotFound: + fallthrough + case ErrInputDefinitionNotFound: + http.Error(w, err.Error(), http.StatusNotFound) + default: + http.Error(w, err.Error(), http.StatusInternalServerError) + } return } @@ -1838,7 +1662,6 @@ func (h *Handler) handleGetInputDefinition(w http.ResponseWriter, r *http.Reques }); err != nil { h.Logger.Printf("write status response error: %s", err) } - } // handleDeleteInputDefinition handles DELETE /input-definition request. @@ -1846,28 +1669,20 @@ func (h *Handler) handleDeleteInputDefinition(w http.ResponseWriter, r *http.Req indexName := mux.Vars(r)["index"] inputDefName := mux.Vars(r)["input-definition"] - // Find index. - index := h.Holder.Index(indexName) - if index == nil { - http.Error(w, ErrIndexNotFound.Error(), http.StatusNotFound) + if err := h.API.DeleteInputDefinition(r.Context(), indexName, inputDefName); err != nil { + switch err { + case nil: + break + case ErrIndexNotFound: + fallthrough + case ErrInputDefinitionNotFound: + http.Error(w, err.Error(), http.StatusNotFound) + default: + http.Error(w, err.Error(), http.StatusNotFound) + } return } - // Delete input definition from the index. - if err := index.DeleteInputDefinition(inputDefName); err != nil { - http.Error(w, err.Error(), http.StatusNotFound) - return - } - - err := h.Broadcaster.SendSync( - &internal.DeleteInputDefinitionMessage{ - Index: indexName, - Name: inputDefName, - }) - if err != nil { - h.Logger.Printf("problem sending DeleteInputDefinition message: %s", err) - } - if err := json.NewEncoder(w).Encode(defaultInputDefinitionResponse{}); err != nil { h.Logger.Printf("response encoding error: %s", err) } @@ -1879,13 +1694,6 @@ func (h *Handler) handlePostInput(w http.ResponseWriter, r *http.Request) { indexName := mux.Vars(r)["index"] inputDefName := mux.Vars(r)["input-definition"] - // Find index. - index := h.Holder.Index(indexName) - if index == nil { - http.Error(w, ErrIndexNotFound.Error(), http.StatusNotFound) - return - } - // Decode request. var reqs []interface{} err := json.NewDecoder(r.Body).Decode(&reqs) @@ -1893,28 +1701,27 @@ func (h *Handler) handlePostInput(w http.ResponseWriter, r *http.Request) { http.Error(w, err.Error(), http.StatusBadRequest) return } - for _, req := range reqs { - bits, err := h.InputJSONDataParser(req.(map[string]interface{}), index, inputDefName) - if err == ErrInputDefinitionNotFound { + + if err = h.API.WriteInput(r.Context(), indexName, inputDefName, reqs); err != nil { + switch err { + case nil: + break + case ErrIndexNotFound: + fallthrough + case ErrInputDefinitionNotFound: http.Error(w, err.Error(), http.StatusNotFound) - return - } else if err != nil { + default: http.Error(w, err.Error(), http.StatusBadRequest) - return - } - for fr, bs := range bits { - err := index.InputBits(fr, bs) - if err != nil { - http.Error(w, err.Error(), http.StatusBadRequest) - return - } } + return } + if err := json.NewEncoder(w).Encode(defaultInputDefinitionResponse{}); err != nil { h.Logger.Printf("response encoding error: %s", err) } } +// <<<<<<< b702f70610962341116d9678ce7c5877b7e171ca // handlePostClusterResizeSetCoordinator handles POST /cluster/resize/set-coordinator request. func (h *Handler) handlePostClusterResizeSetCoordinator(w http.ResponseWriter, r *http.Request) { // Decode request. @@ -2109,14 +1916,11 @@ func (h *Handler) InputJSONDataParser(req map[string]interface{}, index *Index, return setBits, nil } +// ======= +// >>>>>>> Moved more of handler to API + func (h *Handler) handleRecalculateCaches(w http.ResponseWriter, r *http.Request) { - err := h.Broadcaster.SendSync(&internal.RecalculateCaches{}) - if err != nil { - w.WriteHeader(http.StatusInternalServerError) - h.writeQueryResponse(w, r, &QueryResponse{Err: err}) - return - } - h.Holder.RecalculateCaches() + h.API.RecalculateCaches(r.Context()) w.WriteHeader(http.StatusNoContent) } @@ -2161,9 +1965,7 @@ func (h *Handler) handlePostClusterMessage(w http.ResponseWriter, r *http.Reques return } - // Forward the error message. - err = h.BroadcastHandler.ReceiveMessage(pb) - if err != nil { + if err := h.API.PostClusterMessage(r.Context(), pb); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } @@ -2174,7 +1976,7 @@ func (h *Handler) handlePostClusterMessage(w http.ResponseWriter, r *http.Reques } func (h *Handler) handleGetID(w http.ResponseWriter, r *http.Request) { - _, err := w.Write([]byte(h.Cluster.Node.ID)) + _, err := w.Write([]byte(h.API.LocalID(r.Context()))) if err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) }