scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-07 17:15:56 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Stop writes on DEGRADED state

Browse source 
Signed-off-by: Antonio Navarro Perez <antnavper@gmail.com>
This commit is contained in:
Antonio Navarro Perez 2021-02-04 17:30:14 +01:00
parent 39eacfb9a3
commit 87ba73fa16
11 changed files with 157 additions and 61 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

48
api.go
View file

@ -980,10 +980,14 @@ func (err MessageProcessingError) Unwrap() error {
// Schema returns information about each index in Pilosa including which fields
// they contain.
func (api *API) Schema(ctx context.Context) []*IndexInfo {
func (api *API) Schema(ctx context.Context) ([]*IndexInfo, error) {
if err := api.validate(apiSchema); err != nil {
return nil, errors.Wrap(err, "validating api method")
}
span, _ := tracing.StartSpanFromContext(ctx, "API.Schema")
defer span.Finish()
return api.holder.limitedSchema()
return api.holder.limitedSchema(), nil
}
// ApplySchema takes the given schema and applies it across the
@ -1777,6 +1781,10 @@ func (api *API) ResizeAbort() error {
// "STARTING", "RESIZING", or potentially others. See cluster.go for more
// details.
func (api *API) State() (string, error) {
if err := api.validate(apiState); err != nil {
return "", errors.Wrap(err, "validating api method")
}
return api.cluster.State()
}
@ -2211,10 +2219,9 @@ const (
apiRecalculateCaches
apiRemoveNode
apiResizeAbort
//apiSchema // not implemented
apiSetCoordinator
apiSchema
apiShardNodes
//apiState // not implemented
apiState
//apiStatsWithTags // not implemented
//apiVersion // not implemented
apiViews
@ -2232,13 +2239,36 @@ const (
var methodsCommon = map[apiMethod]struct{}{
apiClusterMessage: {},
apiSetCoordinator: {},
}
var methodsResizing = map[apiMethod]struct{}{
apiFragmentData: {},
apiTranslateData: {},
apiResizeAbort: {},
apiSchema: {},
apiState: {},
}
var methodsDegraded = map[apiMethod]struct{}{
apiExportCSV: {},
apiFragmentBlockData: {},
apiFragmentBlocks: {},
apiField: {},
apiFieldAttrDiff: {},
apiIndex: {},
apiIndexAttrDiff: {},
apiQuery: {},
apiRecalculateCaches: {},
apiRemoveNode: {},
apiShardNodes: {},
apiSchema: {},
apiState: {},
apiViews: {},
apiStartTransaction: {},
apiFinishTransaction: {},
apiTransactions: {},
apiGetTransaction: {},
apiActiveQueries: {},
}
var methodsNormal = map[apiMethod]struct{}{
@ -2261,6 +2291,8 @@ var methodsNormal = map[apiMethod]struct{}{
apiRecalculateCaches: {},
apiRemoveNode: {},
apiShardNodes: {},
apiSchema: {},
apiState: {},
apiViews: {},
apiApplySchema: {},
apiStartTransaction: {},
@ -2268,8 +2300,4 @@ var methodsNormal = map[apiMethod]struct{}{
apiTransactions: {},
apiGetTransaction: {},
apiActiveQueries: {},
apiPastQueries: {},
apiIDReserve: {},
apiIDCommit: {},
apiIDReset: {},
}

6
api_test.go
View file

@ -269,7 +269,11 @@ func TestAPI_Import(t *testing.T) {
// Relies on the previous test creating an index with TrackExistence and
// adding some data.
t.Run("SchemaHasNoExists", func(t *testing.T) {
schema := m1.API.Schema(context.Background())
schema, err := m1.API.Schema(context.Background())
if err != nil {
t.Fatal(err)
}
for _, f := range schema[0].Fields {
if f.Name == "_exists" {
t.Fatalf("found _exists field in schema")

24
apimethod_string.go
View file

@ -30,19 +30,25 @@ func _() {
_ = x[apiRecalculateCaches-19]
_ = x[apiRemoveNode-20]
_ = x[apiResizeAbort-21]
_ = x[apiSetCoordinator-22]
_ = x[apiSchema-22]
_ = x[apiShardNodes-23]
_ = x[apiViews-24]
_ = x[apiApplySchema-25]
_ = x[apiStartTransaction-26]
_ = x[apiFinishTransaction-27]
_ = x[apiTransactions-28]
_ = x[apiGetTransaction-29]
_ = x[apiState-24]
_ = x[apiViews-25]
_ = x[apiApplySchema-26]
_ = x[apiStartTransaction-27]
_ = x[apiFinishTransaction-28]
_ = x[apiTransactions-29]
_ = x[apiGetTransaction-30]
_ = x[apiActiveQueries-31]
_ = x[apiPastQueries-32]
_ = x[apiIDReserve-33]
_ = x[apiIDCommit-34]
_ = x[apiIDReset-35]
}
const _apiMethod_name = "apiClusterMessageapiCreateFieldapiCreateIndexapiDeleteFieldapiDeleteAvailableShardapiDeleteIndexapiDeleteViewapiExportCSVapiFragmentBlockDataapiFragmentBlocksapiFragmentDataapiTranslateDataapiFieldapiFieldAttrDiffapiImportapiImportValueapiIndexapiIndexAttrDiffapiQueryapiRecalculateCachesapiRemoveNodeapiResizeAbortapiSetCoordinatorapiShardNodesapiViewsapiApplySchemaapiStartTransactionapiFinishTransactionapiTransactionsapiGetTransaction"
const _apiMethod_name = "apiClusterMessageapiCreateFieldapiCreateIndexapiDeleteFieldapiDeleteAvailableShardapiDeleteIndexapiDeleteViewapiExportCSVapiFragmentBlockDataapiFragmentBlocksapiFragmentDataapiTranslateDataapiFieldapiFieldAttrDiffapiImportapiImportValueapiIndexapiIndexAttrDiffapiQueryapiRecalculateCachesapiRemoveNodeapiResizeAbortapiSchemaapiShardNodesapiStateapiViewsapiApplySchemaapiStartTransactionapiFinishTransactionapiTransactionsapiGetTransactionapiActiveQueriesapiPastQueriesapiIDReserveapiIDCommitapiIDReset"
var _apiMethod_index = [...]uint16{0, 17, 31, 45, 59, 82, 96, 109, 121, 141, 158, 173, 189, 197, 213, 222, 236, 244, 260, 268, 288, 301, 315, 332, 345, 353, 367, 386, 406, 421, 438}
var _apiMethod_index = [...]uint16{0, 17, 31, 45, 59, 82, 96, 109, 121, 141, 158, 173, 189, 197, 213, 222, 236, 244, 260, 268, 288, 301, 315, 324, 337, 345, 353, 367, 386, 406, 421, 438, 454, 468, 480, 491, 501}
func (i apiMethod) String() string {
if i < 0 || i >= apiMethod(len(_apiMethod_index)-1) {

53
cmd/random-query/main.go
View file

@ -27,24 +27,24 @@ import (
"time"
"github.com/pilosa/pilosa/v2"
"github.com/pilosa/pilosa/v2/pql"
"github.com/pilosa/pilosa/v2/http"
"github.com/pilosa/pilosa/v2/pql"
)
// RandomQueryConfig
type RandomQueryConfig struct {
// user facing flags
HostPort string // -hostport
TreeDepth int // -d
QueryCount int // -n
Verbose bool // -v
VeryVerbose bool // -V
TimeFromArg string // --time.from
TimeToArg string // --time.to
TimeFrom time.Time // parsed time
TimeTo time.Time // parsed time
TimeRange int64 // hours between parsed times
HostPort string // -hostport
TreeDepth int // -d
QueryCount int // -n
Verbose bool // -v
VeryVerbose bool // -V
TimeFromArg string // --time.from
TimeToArg string // --time.to
TimeFrom time.Time // parsed time
TimeTo time.Time // parsed time
TimeRange int64 // hours between parsed times
IndexMap map[string]*Features
@ -73,7 +73,7 @@ type wrapper struct {
}
func (w *wrapper) Schema(ctx context.Context) ([]*pilosa.IndexInfo, error) {
return w.api.Schema(ctx), nil
return w.api.Schema(ctx)
}
func (w *wrapper) Query(ctx context.Context, index string, queryRequest *pilosa.QueryRequest) (*pilosa.QueryResponse, error) {
@ -234,11 +234,11 @@ NewSetup:
}
type Features struct {
Slc []IndexFieldRow
Ranges []IndexFieldRange
Slc []IndexFieldRow
Ranges []IndexFieldRange
Distinctables []IndexFieldRange
SlcWeight int
RangeWeight int
SlcWeight int
RangeWeight int
}
// Pick either a feature entry or a random query on a range, weighted
@ -274,7 +274,7 @@ func (fea *IndexFieldRow) Query(cfg *RandomQueryConfig) *Tree {
// anyway.
if fea.HasTime && cfg.Rnd.Int63n(20) != 0 {
startHours := (cfg.Rnd.Int63n(cfg.TimeRange - 1))
endHours := cfg.Rnd.Int63n(cfg.TimeRange - startHours) + 1 + startHours
endHours := cfg.Rnd.Int63n(cfg.TimeRange-startHours) + 1 + startHours
startTime := cfg.TimeFrom.Add(time.Duration(startHours) * time.Hour)
endTime := cfg.TimeFrom.Add(time.Duration(endHours) * time.Hour)
fromTo = fmt.Sprintf(", from=%s, to=%s",
@ -288,11 +288,11 @@ func (fea *IndexFieldRow) Query(cfg *RandomQueryConfig) *Tree {
}
type IndexFieldRange struct {
Index string
Field string
Index string
Field string
Min, Max, Scale int64
ScaleDiv float64
Range uint64
ScaleDiv float64
Range uint64
}
// We want to pick one of (1) a single-operation filter, (2) a
@ -316,8 +316,8 @@ func (i *IndexFieldRange) Query(cfg *RandomQueryConfig) *Tree {
v2 = v2 + uint64(i.Min)
var v1s, v2s string
if i.Scale != 0 {
v1s = fmt.Sprintf("%.*f", i.Scale, float64(int64(v1)) / i.ScaleDiv)
v2s = fmt.Sprintf("%.*f", i.Scale, float64(int64(v2)) / i.ScaleDiv)
v1s = fmt.Sprintf("%.*f", i.Scale, float64(int64(v1))/i.ScaleDiv)
v2s = fmt.Sprintf("%.*f", i.Scale, float64(int64(v2))/i.ScaleDiv)
} else {
v1s = strconv.FormatInt(int64(v1), 10)
v2s = strconv.FormatInt(int64(v2), 10)
@ -332,7 +332,7 @@ func (i *IndexFieldRange) Query(cfg *RandomQueryConfig) *Tree {
if cfg.Rnd.Int63n(2) == 1 {
v1s = v2s
}
return &Tree{S: fmt.Sprintf("Row(%s %s %s)", i.Field, binaryOps[r - 4], v1s)}
return &Tree{S: fmt.Sprintf("Row(%s %s %s)", i.Field, binaryOps[r-4], v1s)}
}
}
@ -463,8 +463,8 @@ func (cfg *RandomQueryConfig) GenQuery(index string) (pql string, err error) {
}
type Tree struct {
Chd []*Tree
S string
Chd []*Tree
S string
Args []string // Extra args to pass after children, such as a field for Distinct.
}
@ -496,6 +496,7 @@ func (tr *Tree) StringIndent(ind int) (s string) {
}
const pilosaTimeFmt = "2006-01-02T15:04"
func (cfg *RandomQueryConfig) GenTree(index string, depth int) (tr *Tree) {
features := cfg.IndexMap[index]
if depth == 0 {

7
gossip/gossip.go
View file

@ -324,9 +324,14 @@ func (g *memberSet) GetBroadcasts(overhead, limit int) [][]byte {
// LocalState implementation of the memberlist.Delegate interface
// sends this Node's state data.
func (g *memberSet) LocalState(join bool) []byte {
schema, err := g.papi.Schema(context.Background())
if err != nil {
// just panic, this code will be removed soon
panic(err)
}
m := &pilosa.NodeStatus{
Node: g.papi.Node(),
Schema: &pilosa.Schema{Indexes: g.papi.Schema(context.Background())},
Schema: &pilosa.Schema{Indexes: schema},
}
for _, idx := range m.Schema.Indexes {
is := &pilosa.IndexStatus{Name: idx.Name, CreatedAt: idx.CreatedAt}

13
http/handler.go
View file

@ -668,7 +668,11 @@ func (h *Handler) handleGetSchema(w http.ResponseWriter, r *http.Request) {
}
w.Header().Set("Content-Type", "application/json")
schema := h.api.Schema(r.Context())
schema, err := h.api.Schema(r.Context())
if err != nil {
h.logger.Printf("getting schema error: %s", err)
}
if err := json.NewEncoder(w).Encode(pilosa.Schema{Indexes: schema}); err != nil {
h.logger.Printf("write schema response error: %s", err)
}
@ -977,7 +981,12 @@ func (h *Handler) handleGetIndex(w http.ResponseWriter, r *http.Request) {
return
}
indexName := mux.Vars(r)["index"]
for _, idx := range h.api.Schema(r.Context()) {
schema, err := h.api.Schema(r.Context())
if err != nil {
h.logger.Printf("getting schema error: %s", err)
}
for _, idx := range schema {
if idx.Name == indexName {
w.Header().Set("Content-Type", "application/json")
if err := json.NewEncoder(w).Encode(idx); err != nil {

24
server/grpc.go
View file

@ -284,7 +284,11 @@ func (h *GRPCHandler) CreateIndex(ctx context.Context, req *pb.CreateIndexReques
// GetIndex returns a single Index given a name
func (h *GRPCHandler) GetIndex(ctx context.Context, req *pb.GetIndexRequest) (*pb.GetIndexResponse, error) {
schema := h.api.Schema(ctx)
schema, err := h.api.Schema(ctx)
if err != nil {
return nil, errToStatusError(err)
}
for _, index := range schema {
if req.Name == index.Name {
return &pb.GetIndexResponse{Index: &pb.Index{Name: index.Name}}, nil
@ -295,7 +299,11 @@ func (h *GRPCHandler) GetIndex(ctx context.Context, req *pb.GetIndexRequest) (*p
// GetIndexes returns a list of all Indexes
func (h *GRPCHandler) GetIndexes(ctx context.Context, req *pb.GetIndexesRequest) (*pb.GetIndexesResponse, error) {
schema := h.api.Schema(ctx)
schema, err := h.api.Schema(ctx)
if err != nil {
return nil, errToStatusError(err)
}
indexes := make([]*pb.Index, len(schema))
for i, index := range schema {
indexes[i] = &pb.Index{Name: index.Name}
@ -341,7 +349,11 @@ func (h *VDSMGRPCHandler) GetVDS(ctx context.Context, req *vdsm_pb.GetVDSRequest
case *vdsm_pb.GetVDSRequest_Id:
return nil, status.Error(codes.InvalidArgument, "VDS IDs are no longer supported")
case *vdsm_pb.GetVDSRequest_Name:
schema := h.api.Schema(ctx)
schema, err := h.api.Schema(ctx)
if err != nil {
return nil, errToStatusError(err)
}
for _, index := range schema {
if idOrName.Name == index.Name {
return &vdsm_pb.GetVDSResponse{Vds: &vdsm_pb.VDS{Name: index.Name}}, nil
@ -355,7 +367,11 @@ func (h *VDSMGRPCHandler) GetVDS(ctx context.Context, req *vdsm_pb.GetVDSRequest
// GetVDSs returns a list of all VDSs
func (h *VDSMGRPCHandler) GetVDSs(ctx context.Context, req *vdsm_pb.GetVDSsRequest) (*vdsm_pb.GetVDSsResponse, error) {
schema := h.api.Schema(ctx)
schema, err := h.api.Schema(ctx)
if err != nil {
return nil, errToStatusError(err)
}
vdss := make([]*vdsm_pb.VDS, len(schema))
for i, index := range schema {
vdss[i] = &vdsm_pb.VDS{Name: index.Name}

23
server/grpc_test.go
View file

@ -1009,7 +1009,10 @@ func TestCRUDIndexes(t *testing.T) {
t.Fatal(err)
}
schema := m.API.Schema(ctx)
schema, err := m.API.Schema(ctx)
if err != nil {
t.Fatal("Getting schema error", err)
}
if len(schema) != 1 {
t.Fatal("Schema should include one index")
}
@ -1029,14 +1032,22 @@ func TestCRUDIndexes(t *testing.T) {
t.Fatal(err)
}
schema = m.API.Schema(ctx)
schema, err = m.API.Schema(ctx)
if err != nil {
t.Fatal("Getting schema error", err)
}
if len(schema) != 2 {
t.Fatal("Schema should include two indexes")
}
_ = m.API.DeleteIndex(ctx, "testindex1")
schema = m.API.Schema(ctx)
schema, err = m.API.Schema(ctx)
if err != nil {
t.Fatal("Getting schema error", err)
}
if len(schema) != 1 {
t.Fatal("Schema should include one index")
}
@ -1146,7 +1157,11 @@ func TestCRUDIndexes(t *testing.T) {
t.Fatal(err)
}
schema := m.API.Schema(ctx)
schema, err := m.API.Schema(ctx)
if err != nil {
t.Fatal("Getting schema error", err)
}
if len(schema) != 0 {
t.Fatal("Schema should include no index")
}

8
server/handler_test.go
View file

@ -226,8 +226,12 @@ func TestHandler_Endpoints(t *testing.T) {
})
t.Run("Import", func(t *testing.T) {
indexInfo := cmd.API.Schema(context.Background())
err := cmd.API.ApplySchema(context.Background(), &pilosa.Schema{Indexes: indexInfo}, false)
indexInfo, err := cmd.API.Schema(context.Background())
if err != nil {
t.Fatalf("getting schema: %v", err)
}
err = cmd.API.ApplySchema(context.Background(), &pilosa.Schema{Indexes: indexInfo}, false)
if err != nil {
t.Fatalf("applying schema: %v", err)
}

7
server/server_test.go
View file

@ -1253,7 +1253,12 @@ func TestClusterCreatedAtRace(t *testing.T) {
schemas := make([]*pilosa.IndexInfo, len(cluster.Nodes))
for i, cmd := range cluster.Nodes {
schemas[i] = cmd.API.Schema(context.Background())[0]
s, err := cmd.API.Schema(context.Background())
if err != nil {
t.Fatalf("getting schema: %v", err)
}
schemas[i] = s[0]
}
createdAtField := schemas[0].Fields[0].CreatedAt

5
sql/show.go
View file

@ -54,7 +54,10 @@ func (s *ShowHandler) Handle(ctx context.Context, mapped *MappedSQL) (pproto.ToR
}
func (s *ShowHandler) execShowTables(ctx context.Context, showStmt *sqlparser.Show) (pproto.ToRowser, error) {
indexInfo := s.api.Schema(ctx)
indexInfo, err := s.api.Schema(ctx)
if err != nil {
return nil, errors.Wrap(err, "getting schema")
}
result := make(pproto.ConstRowser, len(indexInfo))
for i, ii := range indexInfo {