From 87ba73fa163b494f70692a6b0d15e6a50ffe53fe Mon Sep 17 00:00:00 2001 From: Antonio Navarro Perez Date: Thu, 4 Feb 2021 17:30:14 +0100 Subject: [PATCH] Stop writes on DEGRADED state Signed-off-by: Antonio Navarro Perez --- api.go | 48 ++++++++++++++++++++++++++++-------- api_test.go | 6 ++++- apimethod_string.go | 24 +++++++++++------- cmd/random-query/main.go | 53 ++++++++++++++++++++-------------------- gossip/gossip.go | 7 +++++- http/handler.go | 13 ++++++++-- server/grpc.go | 24 +++++++++++++++--- server/grpc_test.go | 23 ++++++++++++++--- server/handler_test.go | 8 ++++-- server/server_test.go | 7 +++++- sql/show.go | 5 +++- 11 files changed, 157 insertions(+), 61 deletions(-) diff --git a/api.go b/api.go index 0d5309f09..4c8221170 100644 --- a/api.go +++ b/api.go @@ -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: {}, } diff --git a/api_test.go b/api_test.go index 3d452ed97..936e7b8c4 100644 --- a/api_test.go +++ b/api_test.go @@ -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") diff --git a/apimethod_string.go b/apimethod_string.go index b694fcb9b..8851ec725 100644 --- a/apimethod_string.go +++ b/apimethod_string.go @@ -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) { diff --git a/cmd/random-query/main.go b/cmd/random-query/main.go index 19ea4e5d1..c271f3e26 100644 --- a/cmd/random-query/main.go +++ b/cmd/random-query/main.go @@ -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 { diff --git a/gossip/gossip.go b/gossip/gossip.go index 0f4427d3c..2d3b413d0 100644 --- a/gossip/gossip.go +++ b/gossip/gossip.go @@ -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} diff --git a/http/handler.go b/http/handler.go index 9e674d497..23481bfa6 100644 --- a/http/handler.go +++ b/http/handler.go @@ -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 { diff --git a/server/grpc.go b/server/grpc.go index c2e68bb1f..213a9d467 100644 --- a/server/grpc.go +++ b/server/grpc.go @@ -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} diff --git a/server/grpc_test.go b/server/grpc_test.go index 3cc808bb2..126b48b12 100644 --- a/server/grpc_test.go +++ b/server/grpc_test.go @@ -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") } diff --git a/server/handler_test.go b/server/handler_test.go index 2f7486534..460a4eff3 100644 --- a/server/handler_test.go +++ b/server/handler_test.go @@ -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) } diff --git a/server/server_test.go b/server/server_test.go index ba93abc75..64f92ecac 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -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 diff --git a/sql/show.go b/sql/show.go index bdf99b054..c574ae4a5 100644 --- a/sql/show.go +++ b/sql/show.go @@ -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 {