mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-08-28 02:44:59 +00:00
Merge pull request #1207 from travisturner/security-manager-to-api
Remove SecurityManager. Implement api restrictions in api package.
This commit is contained in:
Travis Turner
GitHub
commit
590b7c30ab
12 changed files with 302 additions and 100 deletions
6
Makefile
6
Makefile
|
|
@ -80,8 +80,12 @@ generate-protoc: require-protoc require-protoc-gen-gofast
|
|||
generate-statik: require-statik
|
||||
go generate github.com/pilosa/pilosa/statik
|
||||
|
||||
# `go generate` stringers
|
||||
generate-stringer:
|
||||
go generate github.com/pilosa/pilosa
|
||||
|
||||
# `go generate` all needed packages
|
||||
generate: generate-protoc generate-statik
|
||||
generate: generate-protoc generate-statik generate-stringer
|
||||
|
||||
# Create Docker image from Dockerfile
|
||||
docker:
|
||||
|
|
|
|||
259
api.go
259
api.go
|
|
@ -12,6 +12,8 @@
|
|||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
//go:generate stringer -type=apiMethod
|
||||
|
||||
package pilosa
|
||||
|
||||
import (
|
||||
|
|
@ -59,8 +61,39 @@ func NewAPI() *API {
|
|||
}
|
||||
}
|
||||
|
||||
// validAPIMethods specifies the api methods that are valid for each
|
||||
// cluster state.
|
||||
var validAPIMethods = map[string]map[apiMethod]struct{}{
|
||||
ClusterStateStarting: methodsCommon,
|
||||
ClusterStateNormal: appendMap(methodsCommon, methodsNormal),
|
||||
ClusterStateResizing: appendMap(methodsCommon, methodsResizing),
|
||||
}
|
||||
|
||||
func appendMap(a, b map[apiMethod]struct{}) map[apiMethod]struct{} {
|
||||
r := make(map[apiMethod]struct{})
|
||||
for k, v := range a {
|
||||
r[k] = v
|
||||
}
|
||||
for k, v := range b {
|
||||
r[k] = v
|
||||
}
|
||||
return r
|
||||
}
|
||||
|
||||
func (api *API) validate(f apiMethod) error {
|
||||
state := api.Cluster.State()
|
||||
if _, ok := validAPIMethods[state][f]; ok {
|
||||
return nil
|
||||
}
|
||||
return ApiMethodNotAllowedError{errors.Errorf("api method %s not allowed in state %s", f, state)}
|
||||
}
|
||||
|
||||
// Query parses a PQL query out of the request and executes it.
|
||||
func (api *API) Query(ctx context.Context, req *QueryRequest) (QueryResponse, error) {
|
||||
if err := api.validate(apiQuery); err != nil {
|
||||
return QueryResponse{}, errors.Wrap(err, "validate api method")
|
||||
}
|
||||
|
||||
resp := QueryResponse{}
|
||||
|
||||
q, err := pql.NewParser(strings.NewReader(req.Query)).Parse()
|
||||
|
|
@ -125,6 +158,10 @@ func (api *API) readColumnAttrSets(index *Index, ids []uint64) ([]*ColumnAttrSet
|
|||
|
||||
// CreateIndex makes a new Pilosa index.
|
||||
func (api *API) CreateIndex(ctx context.Context, indexName string, options IndexOptions) (*Index, error) {
|
||||
if err := api.validate(apiCreateIndex); err != nil {
|
||||
return nil, errors.Wrap(err, "validate api method")
|
||||
}
|
||||
|
||||
// Create index.
|
||||
index, err := api.Holder.CreateIndex(indexName, options)
|
||||
if err != nil {
|
||||
|
|
@ -146,6 +183,10 @@ func (api *API) CreateIndex(ctx context.Context, indexName string, options Index
|
|||
|
||||
// Index retrieves the named index.
|
||||
func (api *API) Index(ctx context.Context, indexName string) (*Index, error) {
|
||||
if err := api.validate(apiIndex); err != nil {
|
||||
return nil, errors.Wrap(err, "validate api method")
|
||||
}
|
||||
|
||||
index := api.Holder.Index(indexName)
|
||||
if index == nil {
|
||||
return nil, ErrIndexNotFound
|
||||
|
|
@ -156,6 +197,10 @@ func (api *API) Index(ctx context.Context, indexName string) (*Index, error) {
|
|||
// DeleteIndex removes the named index. If the index is not found it does
|
||||
// nothing and returns no error.
|
||||
func (api *API) DeleteIndex(ctx context.Context, indexName string) error {
|
||||
if err := api.validate(apiDeleteIndex); err != nil {
|
||||
return errors.Wrap(err, "validate api method")
|
||||
}
|
||||
|
||||
// Delete index from the holder.
|
||||
err := api.Holder.DeleteIndex(indexName)
|
||||
if err != nil {
|
||||
|
|
@ -176,6 +221,10 @@ func (api *API) DeleteIndex(ctx context.Context, indexName string) error {
|
|||
|
||||
// CreateFrame makes the named frame in the named index with the given options.
|
||||
func (api *API) CreateFrame(ctx context.Context, indexName string, frameName string, options FrameOptions) (*Frame, error) {
|
||||
if err := api.validate(apiCreateFrame); err != nil {
|
||||
return nil, errors.Wrap(err, "validate api method")
|
||||
}
|
||||
|
||||
// Find index.
|
||||
index := api.Holder.Index(indexName)
|
||||
if index == nil {
|
||||
|
|
@ -207,6 +256,10 @@ func (api *API) CreateFrame(ctx context.Context, indexName string, frameName str
|
|||
// found, an error is returned. If the frame is not found, it is ignored and no
|
||||
// action is taken.
|
||||
func (api *API) DeleteFrame(ctx context.Context, indexName string, frameName string) error {
|
||||
if err := api.validate(apiDeleteFrame); err != nil {
|
||||
return errors.Wrap(err, "validate api method")
|
||||
}
|
||||
|
||||
// Find index.
|
||||
index := api.Holder.Index(indexName)
|
||||
if index == nil {
|
||||
|
|
@ -235,6 +288,10 @@ func (api *API) DeleteFrame(ctx context.Context, indexName string, frameName str
|
|||
// ExportCSV encodes the fragment designated by the index,frame,view,slice as
|
||||
// CSV of the form <row>,<col>
|
||||
func (api *API) ExportCSV(ctx context.Context, indexName string, frameName string, viewName string, slice uint64, w io.Writer) error {
|
||||
if err := api.validate(apiExportCSV); err != nil {
|
||||
return errors.Wrap(err, "validate api method")
|
||||
}
|
||||
|
||||
// Validate that this handler owns the slice.
|
||||
if !api.Cluster.OwnsSlice(api.LocalID(), indexName, slice) {
|
||||
api.Logger.Printf("host does not own slice %s-%s slice:%d", api.URI, indexName, slice)
|
||||
|
|
@ -267,14 +324,22 @@ func (api *API) ExportCSV(ctx context.Context, indexName string, frameName strin
|
|||
}
|
||||
|
||||
// SliceNodes returns the node and all replicas which should contain a slice's data.
|
||||
func (api *API) SliceNodes(ctx context.Context, indexName string, slice uint64) []*Node {
|
||||
return api.Cluster.SliceNodes(indexName, slice)
|
||||
func (api *API) SliceNodes(ctx context.Context, indexName string, slice uint64) ([]*Node, error) {
|
||||
if err := api.validate(apiSliceNodes); err != nil {
|
||||
return nil, errors.Wrap(err, "validate api method")
|
||||
}
|
||||
|
||||
return api.Cluster.SliceNodes(indexName, slice), nil
|
||||
}
|
||||
|
||||
// MarshalFragment returns an object which can write the specified fragment's data
|
||||
// to an io.Writer. The serialized data can be read back into a fragment with
|
||||
// the UnmarshalFragment API call.
|
||||
func (api *API) MarshalFragment(ctx context.Context, indexName string, frameName string, viewName string, slice uint64) (io.WriterTo, error) {
|
||||
if err := api.validate(apiMarshalFragment); err != nil {
|
||||
return nil, errors.Wrap(err, "validate api method")
|
||||
}
|
||||
|
||||
// Retrieve fragment from holder.
|
||||
f := api.Holder.Fragment(indexName, frameName, viewName, slice)
|
||||
if f == nil {
|
||||
|
|
@ -287,6 +352,10 @@ func (api *API) MarshalFragment(ctx context.Context, indexName string, frameName
|
|||
// Reader which was previously written by MarshalFragment to populate the
|
||||
// fragment's data.
|
||||
func (api *API) UnmarshalFragment(ctx context.Context, indexName string, frameName string, viewName string, slice uint64, reader io.ReadCloser) error {
|
||||
if err := api.validate(apiUnmarshalFragment); err != nil {
|
||||
return errors.Wrap(err, "validate api method")
|
||||
}
|
||||
|
||||
// Retrieve frame.
|
||||
f := api.Holder.Frame(indexName, frameName)
|
||||
if f == nil {
|
||||
|
|
@ -316,6 +385,10 @@ func (api *API) UnmarshalFragment(ctx context.Context, indexName string, frameNa
|
|||
// return anything useful. Currently it returns protobuf encoded row and column
|
||||
// ids from a "block" which is a subdivision of a fragment.
|
||||
func (api *API) FragmentBlockData(ctx context.Context, body io.Reader) ([]byte, error) {
|
||||
if err := api.validate(apiFragmentBlockData); err != nil {
|
||||
return nil, errors.Wrap(err, "validate api method")
|
||||
}
|
||||
|
||||
reqBytes, err := ioutil.ReadAll(body)
|
||||
if err != nil {
|
||||
return nil, BadRequestError{errors.Wrap(err, "read body error")}
|
||||
|
|
@ -337,7 +410,7 @@ func (api *API) FragmentBlockData(ctx context.Context, body io.Reader) ([]byte,
|
|||
// Encode response.
|
||||
buf, err := proto.Marshal(&resp)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "merge block response encoding error: %s")
|
||||
return nil, errors.Wrap(err, "merge block response encoding error")
|
||||
}
|
||||
|
||||
return buf, nil
|
||||
|
|
@ -345,6 +418,10 @@ func (api *API) FragmentBlockData(ctx context.Context, body io.Reader) ([]byte,
|
|||
|
||||
// FragmentBlocks returns the checksums and block ids for all blocks in the specified fragment.
|
||||
func (api *API) FragmentBlocks(ctx context.Context, indexName string, frameName string, viewName string, slice uint64) ([]FragmentBlock, error) {
|
||||
if err := api.validate(apiFragmentBlocks); err != nil {
|
||||
return nil, errors.Wrap(err, "validate api method")
|
||||
}
|
||||
|
||||
// Retrieve fragment from holder.
|
||||
f := api.Holder.Fragment(indexName, frameName, viewName, slice)
|
||||
if f == nil {
|
||||
|
|
@ -359,6 +436,10 @@ func (api *API) FragmentBlocks(ctx context.Context, indexName string, frameName
|
|||
// RestoreFrame reads all the data that this host should have for a given frame
|
||||
// from replicas in the cluster and restores that data to it.
|
||||
func (api *API) RestoreFrame(ctx context.Context, indexName string, frameName string, host *URI) error {
|
||||
if err := api.validate(apiRestoreFrame); err != nil {
|
||||
return errors.Wrap(err, "validate api method")
|
||||
}
|
||||
|
||||
// Create a client for the remote cluster.
|
||||
client := NewInternalHTTPClientFromURI(host, api.RemoteClient)
|
||||
|
||||
|
|
@ -433,6 +514,10 @@ func (api *API) Hosts(ctx context.Context) []*Node {
|
|||
|
||||
// CreateInputDefinition is deprecated and will be removed. Do not use it.
|
||||
func (api *API) CreateInputDefinition(ctx context.Context, indexName string, inputDefName string, inputDef InputDefinitionInfo) error {
|
||||
if err := api.validate(apiCreateInputDefinition); err != nil {
|
||||
return errors.Wrap(err, "validate api method")
|
||||
}
|
||||
|
||||
api.Logger.Printf(`CreateInputDefinition is deprecated and will be removed.
|
||||
Please open an issue if you need to continue using it.`)
|
||||
// Find index.
|
||||
|
|
@ -467,6 +552,10 @@ Please open an issue if you need to continue using it.`)
|
|||
|
||||
// InputDefinition is deprecated and will be removed.
|
||||
func (api *API) InputDefinition(ctx context.Context, indexName string, inputDefName string) (*InputDefinition, error) {
|
||||
if err := api.validate(apiInputDefinition); err != nil {
|
||||
return nil, errors.Wrap(err, "validate api method")
|
||||
}
|
||||
|
||||
api.Logger.Printf(`InputDefinition is deprecated and will be removed.`)
|
||||
// Find index.
|
||||
index := api.Holder.Index(indexName)
|
||||
|
|
@ -483,6 +572,10 @@ func (api *API) InputDefinition(ctx context.Context, indexName string, inputDefN
|
|||
|
||||
// DeleteInputDefinition is deprecated and will be removed.
|
||||
func (api *API) DeleteInputDefinition(ctx context.Context, indexName string, inputDefName string) error {
|
||||
if err := api.validate(apiDeleteInputDefinition); err != nil {
|
||||
return errors.Wrap(err, "validate api method")
|
||||
}
|
||||
|
||||
api.Logger.Printf("DeleteInputDefinition is deprecated and will be removed.")
|
||||
// Find index.
|
||||
index := api.Holder.Index(indexName)
|
||||
|
|
@ -508,6 +601,10 @@ func (api *API) DeleteInputDefinition(ctx context.Context, indexName string, inp
|
|||
|
||||
// WriteInput is deprecated and will be removed.
|
||||
func (api *API) WriteInput(ctx context.Context, indexName string, inputDefName string, reqs []interface{}) error {
|
||||
if err := api.validate(apiWriteInput); err != nil {
|
||||
return errors.Wrap(err, "validate api method")
|
||||
}
|
||||
|
||||
api.Logger.Printf("WriteInput is deprecated and will be removed.")
|
||||
// Find index.
|
||||
index := api.Holder.Index(indexName)
|
||||
|
|
@ -532,6 +629,10 @@ func (api *API) WriteInput(ctx context.Context, indexName string, inputDefName s
|
|||
|
||||
// RecalculateCaches forces all TopN caches to be updated. Used mainly for integration tests.
|
||||
func (api *API) RecalculateCaches(ctx context.Context) error {
|
||||
if err := api.validate(apiRecalculateCaches); err != nil {
|
||||
return errors.Wrap(err, "validate api method")
|
||||
}
|
||||
|
||||
err := api.Broadcaster.SendSync(&internal.RecalculateCaches{})
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "broacasting message")
|
||||
|
|
@ -542,7 +643,11 @@ func (api *API) RecalculateCaches(ctx context.Context) error {
|
|||
|
||||
// PostClusterMessage is for internal use. It decodes a protobuf message out of
|
||||
// the body and forwards it to the BroadcastHandler.
|
||||
func (api *API) PostClusterMessage(ctx context.Context, reqBody io.Reader) error {
|
||||
func (api *API) ClusterMessage(ctx context.Context, reqBody io.Reader) error {
|
||||
if err := api.validate(apiClusterMessage); err != nil {
|
||||
return errors.Wrap(err, "validate api method")
|
||||
}
|
||||
|
||||
// Read entire body.
|
||||
body, err := ioutil.ReadAll(reqBody)
|
||||
if err != nil {
|
||||
|
|
@ -575,6 +680,10 @@ func (api *API) Schema(ctx context.Context) []*IndexInfo {
|
|||
|
||||
// CreateField creates a new BSI field in the given index and frame.
|
||||
func (api *API) CreateField(ctx context.Context, indexName string, frameName string, field *Field) error {
|
||||
if err := api.validate(apiCreateField); err != nil {
|
||||
return errors.Wrap(err, "validate api method")
|
||||
}
|
||||
|
||||
// Retrieve frame by name.
|
||||
f := api.Holder.Frame(indexName, frameName)
|
||||
if f == nil {
|
||||
|
|
@ -601,6 +710,10 @@ func (api *API) CreateField(ctx context.Context, indexName string, frameName str
|
|||
|
||||
// DeleteField deletes the given field.
|
||||
func (api *API) DeleteField(ctx context.Context, indexName string, frameName string, fieldName string) error {
|
||||
if err := api.validate(apiDeleteField); err != nil {
|
||||
return errors.Wrap(err, "validate api method")
|
||||
}
|
||||
|
||||
// Retrieve frame by name.
|
||||
f := api.Holder.Frame(indexName, frameName)
|
||||
if f == nil {
|
||||
|
|
@ -627,6 +740,10 @@ func (api *API) DeleteField(ctx context.Context, indexName string, frameName str
|
|||
|
||||
// Fields returns the fields in the given frame.
|
||||
func (api *API) Fields(ctx context.Context, indexName string, frameName string) ([]*Field, error) {
|
||||
if err := api.validate(apiFields); err != nil {
|
||||
return nil, errors.Wrap(err, "validate api method")
|
||||
}
|
||||
|
||||
index := api.Holder.index(indexName)
|
||||
if index == nil {
|
||||
return nil, ErrIndexNotFound
|
||||
|
|
@ -642,6 +759,10 @@ func (api *API) Fields(ctx context.Context, indexName string, frameName string)
|
|||
|
||||
// Views returns the views in the given frame.
|
||||
func (api *API) Views(ctx context.Context, indexName string, frameName string) ([]*View, error) {
|
||||
if err := api.validate(apiViews); err != nil {
|
||||
return nil, errors.Wrap(err, "validate api method")
|
||||
}
|
||||
|
||||
// Retrieve views.
|
||||
f := api.Holder.Frame(indexName, frameName)
|
||||
if f == nil {
|
||||
|
|
@ -655,6 +776,10 @@ func (api *API) Views(ctx context.Context, indexName string, frameName string) (
|
|||
|
||||
// DeleteView removes the given view.
|
||||
func (api *API) DeleteView(ctx context.Context, indexName string, frameName string, viewName string) error {
|
||||
if err := api.validate(apiDeleteView); err != nil {
|
||||
return errors.Wrap(err, "validate api method")
|
||||
}
|
||||
|
||||
// Retrieve frame.
|
||||
f := api.Holder.Frame(indexName, frameName)
|
||||
if f == nil {
|
||||
|
|
@ -685,6 +810,10 @@ func (api *API) DeleteView(ctx context.Context, indexName string, frameName stri
|
|||
|
||||
// IndexAttrDiff
|
||||
func (api *API) IndexAttrDiff(ctx context.Context, indexName string, blocks []AttrBlock) (map[uint64]map[string]interface{}, error) {
|
||||
if err := api.validate(apiIndexAttrDiff); err != nil {
|
||||
return nil, errors.Wrap(err, "validate api method")
|
||||
}
|
||||
|
||||
// Retrieve index from holder.
|
||||
index := api.Holder.Index(indexName)
|
||||
if index == nil {
|
||||
|
|
@ -715,6 +844,10 @@ func (api *API) IndexAttrDiff(ctx context.Context, indexName string, blocks []At
|
|||
}
|
||||
|
||||
func (api *API) FrameAttrDiff(ctx context.Context, indexName string, frameName string, blocks []AttrBlock) (map[uint64]map[string]interface{}, error) {
|
||||
if err := api.validate(apiFrameAttrDiff); err != nil {
|
||||
return nil, errors.Wrap(err, "validate api method")
|
||||
}
|
||||
|
||||
// Retrieve index from holder.
|
||||
f := api.Holder.Frame(indexName, frameName)
|
||||
if f == nil {
|
||||
|
|
@ -746,6 +879,10 @@ func (api *API) FrameAttrDiff(ctx context.Context, indexName string, frameName s
|
|||
|
||||
// Import bulk imports data into a particular index,frame,slice.
|
||||
func (api *API) Import(ctx context.Context, req internal.ImportRequest) error {
|
||||
if err := api.validate(apiImport); err != nil {
|
||||
return errors.Wrap(err, "validate api method")
|
||||
}
|
||||
|
||||
_, frame, err := api.indexFrame(req.Index, req.Frame, req.Slice)
|
||||
if err != nil {
|
||||
return err
|
||||
|
|
@ -771,6 +908,10 @@ func (api *API) Import(ctx context.Context, req internal.ImportRequest) error {
|
|||
|
||||
// ImportValue bulk imports values into a particular field.
|
||||
func (api *API) ImportValue(ctx context.Context, req internal.ImportValueRequest) error {
|
||||
if err := api.validate(apiImportValue); err != nil {
|
||||
return errors.Wrap(err, "validate api method")
|
||||
}
|
||||
|
||||
_, frame, err := api.indexFrame(req.Index, req.Frame, req.Slice)
|
||||
if err != nil {
|
||||
return err
|
||||
|
|
@ -786,6 +927,10 @@ func (api *API) ImportValue(ctx context.Context, req internal.ImportValueRequest
|
|||
|
||||
// ModifyIndexTimeQuantum changes the default time quantum on the given index.
|
||||
func (api *API) ModifyIndexTimeQuantum(ctx context.Context, indexName string, timeQuantum TimeQuantum) error {
|
||||
if err := api.validate(apiModifyIndexTimeQuantum); err != nil {
|
||||
return errors.Wrap(err, "validate api method")
|
||||
}
|
||||
|
||||
// Retrieve index by name.
|
||||
index := api.Holder.Index(indexName)
|
||||
if index == nil {
|
||||
|
|
@ -799,6 +944,10 @@ func (api *API) ModifyIndexTimeQuantum(ctx context.Context, indexName string, ti
|
|||
// ModifyFrameTimeQuantum changes the time quantum on the given frame. TODO:
|
||||
// what happens if there is already data in the frame?
|
||||
func (api *API) ModifyFrameTimeQuantum(ctx context.Context, indexName string, frameName string, timeQuantum TimeQuantum) error {
|
||||
if err := api.validate(apiModifyFrameTimeQuantum); err != nil {
|
||||
return errors.Wrap(err, "validate api method")
|
||||
}
|
||||
|
||||
// Retrieve index by name.
|
||||
frame := api.Holder.Frame(indexName, frameName)
|
||||
if frame == nil {
|
||||
|
|
@ -933,6 +1082,10 @@ func (api *API) inputJSONDataParser(req map[string]interface{}, index *Index, na
|
|||
|
||||
// SetCoordinator makes a new Node the cluster coordinator.
|
||||
func (api *API) SetCoordinator(ctx context.Context, id string) (oldNode, newNode *Node, err error) {
|
||||
if err := api.validate(apiSetCoordinator); err != nil {
|
||||
return nil, nil, errors.Wrap(err, "validate api method")
|
||||
}
|
||||
|
||||
oldNode = api.Cluster.nodeByID(api.Cluster.Coordinator)
|
||||
newNode = api.Cluster.nodeByID(id)
|
||||
if newNode == nil {
|
||||
|
|
@ -959,6 +1112,10 @@ func (api *API) SetCoordinator(ctx context.Context, id string) (oldNode, newNode
|
|||
// RemoveNode puts the cluster into the "RESIZING" state and begins the job of
|
||||
// removing the given node.
|
||||
func (api *API) RemoveNode(id string) (*Node, error) {
|
||||
if err := api.validate(apiRemoveNode); err != nil {
|
||||
return nil, errors.Wrap(err, "validate api method")
|
||||
}
|
||||
|
||||
removeNode := api.Cluster.nodeByID(id)
|
||||
if removeNode == nil {
|
||||
return nil, errors.Wrap(ErrNodeIDNotExists, "finding node to remove")
|
||||
|
|
@ -974,6 +1131,10 @@ func (api *API) RemoveNode(id string) (*Node, error) {
|
|||
|
||||
// ResizeAbort stops the current resize job.
|
||||
func (api *API) ResizeAbort() error {
|
||||
if err := api.validate(apiResizeAbort); err != nil {
|
||||
return errors.Wrap(err, "validate api method")
|
||||
}
|
||||
|
||||
if !api.Cluster.IsCoordinator() {
|
||||
return ErrNodeNotCoordinator
|
||||
}
|
||||
|
|
@ -992,3 +1153,93 @@ func (api *API) State() string {
|
|||
func (api *API) Version() string {
|
||||
return strings.TrimPrefix(Version, "v")
|
||||
}
|
||||
|
||||
type apiMethod int
|
||||
|
||||
// API validation constants.
|
||||
const (
|
||||
apiClusterMessage apiMethod = iota
|
||||
apiCreateField
|
||||
apiCreateFrame
|
||||
apiCreateIndex
|
||||
apiCreateInputDefinition
|
||||
apiDeleteField
|
||||
apiDeleteFrame
|
||||
apiDeleteIndex
|
||||
apiDeleteInputDefinition
|
||||
apiDeleteView
|
||||
apiExportCSV
|
||||
apiFields
|
||||
apiFragmentBlockData
|
||||
apiFragmentBlocks
|
||||
apiFrameAttrDiff
|
||||
//apiHosts // not implemented
|
||||
apiImport
|
||||
apiImportValue
|
||||
apiIndex
|
||||
apiIndexAttrDiff
|
||||
apiInputDefinition
|
||||
//apiLocalID // not implemented
|
||||
//apiLongQueryTime // not implemented
|
||||
apiMarshalFragment
|
||||
//apiMaxInverseSlices // not implemented
|
||||
//apiMaxSlices // not implemented
|
||||
apiModifyFrameTimeQuantum
|
||||
apiModifyIndexTimeQuantum
|
||||
apiQuery
|
||||
apiRecalculateCaches
|
||||
apiRemoveNode
|
||||
apiResizeAbort
|
||||
apiRestoreFrame
|
||||
//apiSchema // not implemented
|
||||
apiSetCoordinator
|
||||
apiSliceNodes
|
||||
//apiState // not implemented
|
||||
//apiStatsWithTags // not implemented
|
||||
apiUnmarshalFragment
|
||||
//apiVersion // not implemented
|
||||
apiViews
|
||||
apiWriteInput
|
||||
)
|
||||
|
||||
var methodsCommon = map[apiMethod]struct{}{
|
||||
apiClusterMessage: struct{}{},
|
||||
apiMarshalFragment: struct{}{},
|
||||
apiSetCoordinator: struct{}{},
|
||||
}
|
||||
|
||||
var methodsResizing = map[apiMethod]struct{}{
|
||||
apiResizeAbort: struct{}{},
|
||||
}
|
||||
|
||||
var methodsNormal = map[apiMethod]struct{}{
|
||||
apiCreateField: struct{}{},
|
||||
apiCreateFrame: struct{}{},
|
||||
apiCreateIndex: struct{}{},
|
||||
apiCreateInputDefinition: struct{}{},
|
||||
apiDeleteField: struct{}{},
|
||||
apiDeleteFrame: struct{}{},
|
||||
apiDeleteIndex: struct{}{},
|
||||
apiDeleteInputDefinition: struct{}{},
|
||||
apiDeleteView: struct{}{},
|
||||
apiExportCSV: struct{}{},
|
||||
apiFields: struct{}{},
|
||||
apiFragmentBlockData: struct{}{},
|
||||
apiFragmentBlocks: struct{}{},
|
||||
apiFrameAttrDiff: struct{}{},
|
||||
apiImport: struct{}{},
|
||||
apiImportValue: struct{}{},
|
||||
apiIndex: struct{}{},
|
||||
apiIndexAttrDiff: struct{}{},
|
||||
apiInputDefinition: struct{}{},
|
||||
apiModifyFrameTimeQuantum: struct{}{},
|
||||
apiModifyIndexTimeQuantum: struct{}{},
|
||||
apiQuery: struct{}{},
|
||||
apiRecalculateCaches: struct{}{},
|
||||
apiRemoveNode: struct{}{},
|
||||
apiRestoreFrame: struct{}{},
|
||||
apiSliceNodes: struct{}{},
|
||||
apiUnmarshalFragment: struct{}{},
|
||||
apiViews: struct{}{},
|
||||
apiWriteInput: struct{}{},
|
||||
}
|
||||
|
|
|
|||
16
apimethod_string.go
Normal file
16
apimethod_string.go
Normal file
|
|
@ -0,0 +1,16 @@
|
|||
// Code generated by "stringer -type=apiMethod"; DO NOT EDIT.
|
||||
|
||||
package pilosa
|
||||
|
||||
import "fmt"
|
||||
|
||||
const _apiMethod_name = "apiClusterMessageapiCreateFieldapiCreateFrameapiCreateIndexapiCreateInputDefinitionapiDeleteFieldapiDeleteFrameapiDeleteIndexapiDeleteInputDefinitionapiDeleteViewapiExportCSVapiFieldsapiFragmentBlockDataapiFragmentBlocksapiFrameAttrDiffapiImportapiImportValueapiIndexapiIndexAttrDiffapiInputDefinitionapiMarshalFragmentapiModifyFrameTimeQuantumapiModifyIndexTimeQuantumapiQueryapiRecalculateCachesapiRemoveNodeapiResizeAbortapiRestoreFrameapiSetCoordinatorapiSliceNodesapiUnmarshalFragmentapiViewsapiWriteInput"
|
||||
|
||||
var _apiMethod_index = [...]uint16{0, 17, 31, 45, 59, 83, 97, 111, 125, 149, 162, 174, 183, 203, 220, 236, 245, 259, 267, 283, 301, 319, 344, 369, 377, 397, 410, 424, 439, 456, 469, 489, 497, 510}
|
||||
|
||||
func (i apiMethod) String() string {
|
||||
if i < 0 || i >= apiMethod(len(_apiMethod_index)-1) {
|
||||
return fmt.Sprintf("apiMethod(%d)", i)
|
||||
}
|
||||
return _apiMethod_name[_apiMethod_index[i]:_apiMethod_index[i+1]]
|
||||
}
|
||||
12
cluster.go
12
cluster.go
|
|
@ -264,7 +264,6 @@ type Cluster struct {
|
|||
// Close management
|
||||
wg sync.WaitGroup
|
||||
closing chan struct{}
|
||||
prefect SecurityManager
|
||||
|
||||
Logger Logger
|
||||
|
||||
|
|
@ -285,8 +284,7 @@ func NewCluster() *Cluster {
|
|||
closing: make(chan struct{}),
|
||||
joining: make(chan struct{}),
|
||||
|
||||
Logger: NopLogger,
|
||||
prefect: &NopSecurityManager{},
|
||||
Logger: NopLogger,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -433,19 +431,11 @@ func (c *Cluster) setState(state string) {
|
|||
var doCleanup bool
|
||||
|
||||
switch state {
|
||||
case ClusterStateResizing:
|
||||
c.prefect.SetRestricted()
|
||||
case ClusterStateNormal:
|
||||
c.prefect.SetNormal()
|
||||
// Don't change routing for these states:
|
||||
// - ClusterStateStarting
|
||||
|
||||
// If state is RESIZING -> NORMAL then run cleanup.
|
||||
if c.state == ClusterStateResizing {
|
||||
doCleanup = true
|
||||
}
|
||||
default:
|
||||
panic(fmt.Sprintf("invalid cluster state: %s", state))
|
||||
}
|
||||
|
||||
c.state = state
|
||||
|
|
|
|||
59
handler.go
59
handler.go
|
|
@ -43,9 +43,7 @@ import (
|
|||
type Handler struct {
|
||||
Router *mux.Router
|
||||
|
||||
FileSystem FileSystem
|
||||
NormalRouter *mux.Router
|
||||
RestrictedRouter *mux.Router
|
||||
FileSystem FileSystem
|
||||
|
||||
// The execution engine for running queries.
|
||||
Executor interface {
|
||||
|
|
@ -83,29 +81,11 @@ func NewHandler() *Handler {
|
|||
FileSystem: NopFileSystem,
|
||||
Logger: NopLogger,
|
||||
}
|
||||
BuildRouters(handler)
|
||||
handler.Router = NewRouter(handler)
|
||||
handler.populateValidators()
|
||||
return handler
|
||||
}
|
||||
|
||||
// BuildRouters creates Gorilla Mux http routers for both normal and restricted endpoints.
|
||||
func BuildRouters(handler *Handler) {
|
||||
router := mux.NewRouter()
|
||||
loadCommon(router, handler)
|
||||
loadNormal(router, handler)
|
||||
handler.NormalRouter = router
|
||||
router.Use(handler.queryArgValidator)
|
||||
|
||||
// Restricted router.
|
||||
router = mux.NewRouter()
|
||||
loadCommon(router, handler)
|
||||
loadRestricted(router, handler)
|
||||
handler.RestrictedRouter = router
|
||||
router.Use(handler.queryArgValidator)
|
||||
|
||||
handler.SetRestricted()
|
||||
}
|
||||
|
||||
func (h *Handler) populateValidators() {
|
||||
h.validators = map[string]*queryValidationSpec{}
|
||||
h.validators["GetFragmentNodes"] = queryValidationSpecRequired("slice", "index")
|
||||
|
|
@ -138,17 +118,9 @@ func (h *Handler) queryArgValidator(next http.Handler) http.Handler {
|
|||
})
|
||||
}
|
||||
|
||||
// SetNormal is a method of the SecurityManager interface which provides normal URI routing.
|
||||
func (h *Handler) SetNormal() {
|
||||
h.Router = h.NormalRouter
|
||||
}
|
||||
|
||||
// SetRestricted is a method of the SecurityManager interface which provides restricted URI routing.
|
||||
func (h *Handler) SetRestricted() {
|
||||
h.Router = h.RestrictedRouter
|
||||
}
|
||||
|
||||
func loadCommon(router *mux.Router, handler *Handler) {
|
||||
// NewRouter creates a new mux http router.
|
||||
func NewRouter(handler *Handler) *mux.Router {
|
||||
router := mux.NewRouter()
|
||||
router.HandleFunc("/", handler.handleWebUI).Methods("GET")
|
||||
router.HandleFunc("/assets/{file}", handler.handleWebUI).Methods("GET")
|
||||
router.HandleFunc("/cluster/message", handler.handlePostClusterMessage).Methods("POST")
|
||||
|
|
@ -162,16 +134,9 @@ func loadCommon(router *mux.Router, handler *Handler) {
|
|||
router.HandleFunc("/slices/max", handler.handleGetSlicesMax).Methods("GET") // TODO: deprecate, but it's being used by the client (for backups)
|
||||
router.HandleFunc("/status", handler.handleGetStatus).Methods("GET")
|
||||
router.HandleFunc("/version", handler.handleGetVersion).Methods("GET")
|
||||
router.Use(handler.queryArgValidator)
|
||||
}
|
||||
|
||||
func loadRestricted(router *mux.Router, handler *Handler) {
|
||||
router.HandleFunc("/cluster/resize/abort", handler.handlePostClusterResizeAbort).Methods("POST")
|
||||
router.NotFoundHandler = http.HandlerFunc(handler.reportRestricted)
|
||||
router.Use(handler.queryArgValidator)
|
||||
}
|
||||
|
||||
func loadNormal(router *mux.Router, handler *Handler) {
|
||||
router.HandleFunc("/cluster/resize/remove-node", handler.handlePostClusterResizeRemoveNode).Methods("POST")
|
||||
router.PathPrefix("/debug/pprof/").Handler(http.DefaultServeMux).Methods("GET")
|
||||
router.Handle("/debug/vars", expvar.Handler()).Methods("GET")
|
||||
|
|
@ -212,10 +177,8 @@ func loadNormal(router *mux.Router, handler *Handler) {
|
|||
// For now we just do it for the most commonly used handler, /query
|
||||
router.HandleFunc("/index/{index}/query", handler.methodNotAllowedHandler).Methods("GET")
|
||||
|
||||
}
|
||||
|
||||
func (h *Handler) reportRestricted(w http.ResponseWriter, r *http.Request) {
|
||||
http.Error(w, fmt.Sprintf("not allowed in cluster state %s", h.API.State()), http.StatusMethodNotAllowed)
|
||||
router.Use(handler.queryArgValidator)
|
||||
return router
|
||||
}
|
||||
|
||||
func (h *Handler) methodNotAllowedHandler(w http.ResponseWriter, r *http.Request) {
|
||||
|
|
@ -1167,7 +1130,11 @@ func (h *Handler) handleGetFragmentNodes(w http.ResponseWriter, r *http.Request)
|
|||
}
|
||||
|
||||
// Retrieve fragment owner nodes.
|
||||
nodes := h.API.SliceNodes(r.Context(), index, slice)
|
||||
nodes, err := h.API.SliceNodes(r.Context(), index, slice)
|
||||
if err != nil {
|
||||
http.Error(w, err.Error(), http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
// Write to response.
|
||||
if err := json.NewEncoder(w).Encode(nodes); err != nil {
|
||||
|
|
@ -1725,7 +1692,7 @@ func (h *Handler) handlePostClusterMessage(w http.ResponseWriter, r *http.Reques
|
|||
return
|
||||
}
|
||||
|
||||
err := h.API.PostClusterMessage(r.Context(), r.Body)
|
||||
err := h.API.ClusterMessage(r.Context(), r.Body)
|
||||
if err != nil {
|
||||
// TODO this was the previous behavior, but perhaps not everything is a bad request
|
||||
http.Error(w, err.Error(), http.StatusBadRequest)
|
||||
|
|
|
|||
|
|
@ -160,7 +160,7 @@ func TestHandler_ClusterResizeAbort(t *testing.T) {
|
|||
t.Run("No resize job", func(t *testing.T) {
|
||||
h := test.NewHandler()
|
||||
h.API.Cluster = test.NewCluster(1)
|
||||
h.SetRestricted()
|
||||
h.API.Cluster.SetState(pilosa.ClusterStateResizing)
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
h.ServeHTTP(w, test.MustNewHTTPRequest("POST", "/cluster/resize/abort", nil))
|
||||
|
|
|
|||
|
|
@ -79,6 +79,12 @@ var (
|
|||
ErrResizeNotRunning = errors.New("no resize job currently running")
|
||||
)
|
||||
|
||||
// ApiMethodNotAllowedError wraps an error value indicating that a particular
|
||||
// API method is not allowed in the current cluster state.
|
||||
type ApiMethodNotAllowedError struct {
|
||||
error
|
||||
}
|
||||
|
||||
// BadRequestError wraps an error value to signify that a request could not be
|
||||
// read, decoded, or parsed such that in an HTTP scenario, http.StatusBadRequest
|
||||
// would be returned.
|
||||
|
|
|
|||
|
|
@ -1,32 +0,0 @@
|
|||
// Copyright 2017 Pilosa Corp.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package pilosa
|
||||
|
||||
// SecurityManager provides the ability to limit access to restricted endpoints
|
||||
// during cluster configuration.
|
||||
type SecurityManager interface {
|
||||
SetRestricted()
|
||||
SetNormal()
|
||||
}
|
||||
|
||||
// NopSecurityManager provides a no-op implementation of the SecurityManager interface.
|
||||
type NopSecurityManager struct {
|
||||
}
|
||||
|
||||
// SetRestricted no-op.
|
||||
func (sdm *NopSecurityManager) SetRestricted() {}
|
||||
|
||||
// SetNormal no-op.
|
||||
func (sdm *NopSecurityManager) SetNormal() {}
|
||||
|
|
@ -169,10 +169,8 @@ func (s *Server) Open() error {
|
|||
s.Handler.API.StatusHandler = s
|
||||
s.Handler.API.URI = s.URI
|
||||
s.Handler.API.Cluster = s.Cluster
|
||||
s.Handler.Executor = e
|
||||
|
||||
s.Cluster.prefect = s.Handler
|
||||
s.Handler.API.Executor = e
|
||||
s.Handler.Executor = e
|
||||
|
||||
// Initialize Holder.
|
||||
s.Holder.Broadcaster = s.Broadcaster
|
||||
|
|
|
|||
|
|
@ -101,6 +101,9 @@ func TestMain_SendReceiveMessage(t *testing.T) {
|
|||
t.Fatal(err)
|
||||
}
|
||||
|
||||
m0.Server.Cluster.SetState(pilosa.ClusterStateNormal)
|
||||
m1.Server.Cluster.SetState(pilosa.ClusterStateNormal)
|
||||
|
||||
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
|
||||
|
||||
// Expected indexes and Frames
|
||||
|
|
|
|||
|
|
@ -50,6 +50,7 @@ func NewCluster(n int) *pilosa.Cluster {
|
|||
|
||||
c.Node = c.Nodes[0]
|
||||
c.Coordinator = c.Nodes[0].ID
|
||||
c.SetState(pilosa.ClusterStateNormal)
|
||||
|
||||
return c
|
||||
}
|
||||
|
|
|
|||
|
|
@ -47,8 +47,6 @@ func NewHandler() *Handler {
|
|||
// Handler test messages can no-op.
|
||||
h.API.Broadcaster = pilosa.NopBroadcaster
|
||||
|
||||
h.SetNormal()
|
||||
|
||||
return h
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue