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

Merge pull request #1831 from molecula/tlt/ignore-down

Browse source 
Stop blocking API called when cluster is DOWN or DEGRADED
This commit is contained in:
Travis Turner 2021-12-28 14:14:29 -06:00 • committed by GitHub
commit 56b9e2aba7
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 91 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
api.go
View file

@ -131,9 +131,16 @@ func (api *API) SetAPIOptions(opts ...apiOption) error {
var validAPIMethods = map[disco.ClusterState]map[apiMethod]struct{}{
disco.ClusterStateStarting: methodsCommon,
disco.ClusterStateNormal: appendMap(methodsCommon, methodsNormal),
disco.ClusterStateDegraded: appendMap(methodsCommon, methodsDegraded),
// Ideally, this would be just `appendMap(methodsCommon, methodsDegraded)`,
// but in an attempt to reduce the influence that state (determined by etcd)
// has on a node under load, this is set to effectively allow all requests
// in a DEGRADED state.
disco.ClusterStateDegraded: appendMap(methodsCommon, methodsNormal),
disco.ClusterStateResizing: appendMap(methodsCommon, methodsResizing),
disco.ClusterStateDown: methodsCommon,
// Ideally, this would be just `methodsCommon`, but in an attempt to reduce
// the influence that state (determined by etcd) has on a node under load,
// this is set to effectively allow all requests in a DOWN state.
disco.ClusterStateDown: appendMap(methodsCommon, methodsNormal),
}
func appendMap(a, b map[apiMethod]struct{}) map[apiMethod]struct{} {

86
server/server_test.go
View file

@ -17,7 +17,7 @@ import (
"testing"
"time"
"github.com/molecula/featurebase/v2"
pilosa "github.com/molecula/featurebase/v2"
"github.com/molecula/featurebase/v2/disco"
"github.com/molecula/featurebase/v2/http"
"github.com/molecula/featurebase/v2/pql"
@ -26,6 +26,7 @@ import (
"github.com/molecula/featurebase/v2/test"
"github.com/molecula/featurebase/v2/testhook"
"github.com/pkg/errors"
"github.com/stretchr/testify/require"
"golang.org/x/sync/errgroup"
)
@ -504,6 +505,30 @@ func TestClusteringNodesReplica1(t *testing.T) {
t.Fatalf("starting cluster: %v", err)
}
indexName := "idx"
fieldName := "fld"
// Create the schema.
if _, err := cluster.GetPrimary().API.CreateIndex(context.Background(), indexName, pilosa.IndexOptions{}); err != nil {
t.Fatalf("creating index: %v", err)
}
if _, err := cluster.GetPrimary().API.CreateField(context.Background(), indexName, fieldName); err != nil {
t.Fatalf("creating field: %v", err)
}
// Set some columns across shards to ensure that the Row query will require
// data from all nodes.
data := []string{}
for rowID := 1; rowID < 2; rowID++ {
for columnID := 1; columnID < 10; columnID++ {
data = append(data, fmt.Sprintf(`Set(%d, %s=%d)`, columnID*pilosa.ShardWidth, fieldName, rowID))
}
}
if _, err := cluster.GetPrimary().Query(t, indexName, "", strings.Join(data, "")); err != nil {
t.Fatalf("setting columns: %v", err)
}
// Shut down a node.
if err := cluster.GetNonPrimary().Command.Close(); err != nil {
t.Fatalf("closing third node: %v", err)
}
@ -513,7 +538,12 @@ func TestClusteringNodesReplica1(t *testing.T) {
}
// confirm that cluster stops accepting queries after one node closes
if _, err := cluster.GetPrimary().API.Query(context.Background(), &pilosa.QueryRequest{}); !strings.Contains(err.Error(), "not allowed in state DOWN") {
qry := &pilosa.QueryRequest{
Index: "idx",
Query: fmt.Sprintf("Row(%s=1)", fieldName),
}
if _, err := cluster.GetPrimary().API.Query(context.Background(), qry); !strings.Contains(err.Error(), "shard unavailable") {
t.Fatalf("got unexpected error querying an incomplete cluster: %v", err)
}
}
@ -540,8 +570,34 @@ func TestClusteringNodesReplica2(t *testing.T) {
}
defer cluster.Close()
indexName := "idx"
fieldName := "fld"
coord, others := cluster.GetPrimary(), cluster.GetNonPrimaries()
// Create the schema.
if _, err := coord.API.CreateIndex(context.Background(), indexName, pilosa.IndexOptions{}); err != nil {
t.Fatalf("creating index: %v", err)
}
if _, err := coord.API.CreateField(context.Background(), indexName, fieldName); err != nil {
t.Fatalf("creating field: %v", err)
}
// Set some columns across shards to ensure that the Row query will require
// data from all nodes.
data := []string{}
cols := []uint64{}
for rowID := 1; rowID < 2; rowID++ {
for columnID := 1; columnID < 30; columnID++ {
col := uint64(columnID * pilosa.ShardWidth)
cols = append(cols, col)
data = append(data, fmt.Sprintf(`Set(%d, %s=%d)`, col, fieldName, rowID))
}
}
if _, err := coord.Query(t, indexName, "", strings.Join(data, "")); err != nil {
t.Fatalf("setting columns: %v", err)
}
if err := others[0].Close(); err != nil {
t.Fatalf("closing third node: %v", err)
}
@ -569,8 +625,30 @@ func TestClusteringNodesReplica2(t *testing.T) {
t.Fatalf("after closing second server: %v", err)
}
if _, err := coord.API.Query(context.Background(), &pilosa.QueryRequest{}); !strings.Contains(err.Error(), "not allowed in state DOWN") {
t.Fatalf("got unexpected error querying an incomplete cluster: %v", err)
qry := &pilosa.QueryRequest{
Index: "idx",
Query: fmt.Sprintf("Row(%s=1)", fieldName),
}
// Because we no longer block queries when the cluster is in state DOWN,
// there are cases where a DOWN cluster can still respond to a query. In
// that case, we want the test to pass. But if the unavailable node(s) cause
// the query to result in an error, we check that it's the error we expect.
resp, err := coord.API.Query(context.Background(), qry)
if err != nil {
if !strings.Contains(err.Error(), "shard unavailable") {
t.Fatalf("got unexpected error querying an incomplete cluster: %v", err)
}
} else {
if len(resp.Results) == 0 {
t.Fatal("got no results")
}
row, ok := resp.Results[0].(*pilosa.Row)
if !ok {
t.Fatalf("expected a *pilosa.Row, but got %T", resp.Results[0])
}
require.Equal(t, row.Columns(), cols)
}
}