Stop blocking API called when cluster is DOWN or DEGRADED

This commit effectively removes the API-level validation that was
blocking certain API methods when the cluster was in a particular state
(namely DOWN and DEGRADED). The thinking is that we shouldn't be
blocking these requests at the API level, but rather should let them
pass through and allow the fact that a node is ACTUALLY down dictate the
behavior.

With this change, two tests were modified. They were previously
expecting the error message from the API validation on DOWN, but now
they check for a "shard unavailable" error, which is what gets returned
for a particular query when the cluster is in an unhealthy state.

api.go

@@ -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{} {

server/server_test.go

@@ -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)
 	}
 }