From 65760496fe5db8204d522b21743b3e9c27084ef5 Mon Sep 17 00:00:00 2001
From: Travis <travis@pilosa.com>
Date: Mon, 8 May 2017 10:49:41 -0500
Subject: [PATCH] Add Cluster.UnavailableNodes support to skip reads on certain
 nodes.

---
 cluster.go  |  3 +++
 config.go   | 11 ++++++++---
 executor.go | 10 ++++++++++
 3 files changed, 21 insertions(+), 3 deletions(-)

diff --git a/cluster.go b/cluster.go
index beb33496b..a75673801 100644
--- a/cluster.go
+++ b/cluster.go
@@ -83,6 +83,9 @@ func (a Nodes) Clone() []*Node {
 type Cluster struct {
 	Nodes []*Node
 
+	// UnavailableNodes do not receive read queries.
+	UnavailableNodes []*Node
+
 	// Hashing algorithm used to assign partitions to nodes.
 	Hasher Hasher
 
diff --git a/config.go b/config.go
index 3b8924c96..f82c9c730 100644
--- a/config.go
+++ b/config.go
@@ -13,9 +13,10 @@ type Config struct {
 	Host    string `toml:"host"`
 
 	Cluster struct {
-		ReplicaN        int           `toml:"replicas"`
-		Nodes           []*ConfigNode `toml:"node"`
-		PollingInterval Duration      `toml:"polling-interval"`
+		ReplicaN         int           `toml:"replicas"`
+		Nodes            []*ConfigNode `toml:"node"`
+		UnavailableNodes []*ConfigNode `toml:"unavailable-node"`
+		PollingInterval  Duration      `toml:"polling-interval"`
 	} `toml:"cluster"`
 
 	Plugins struct {
@@ -59,6 +60,10 @@ func (c *Config) PilosaCluster() *Cluster {
 		cluster.Nodes = append(cluster.Nodes, &Node{Host: n.Host})
 	}
 
+	for _, u := range c.Cluster.UnavailableNodes {
+		cluster.UnavailableNodes = append(cluster.UnavailableNodes, &Node{Host: u.Host})
+	}
+
 	return cluster
 }
 
diff --git a/executor.go b/executor.go
index 75c55a74d..42f32ee12 100644
--- a/executor.go
+++ b/executor.go
@@ -718,6 +718,16 @@ func (e *Executor) mapReduce(ctx context.Context, db string, slices []uint64, c
 	var nodes []*Node
 	if !opt.Remote {
 		nodes = Nodes(e.Cluster.Nodes).Clone()
+
+		// If this is a read operation, don't send the query to nodes
+		// marked as unavailable in the config file.
+		switch c.(type) {
+		case pql.BitmapCall, *pql.TopN, *pql.Count:
+			for _, u := range e.Cluster.UnavailableNodes {
+				nodes = Nodes(nodes).FilterHost(u.Host)
+			}
+		}
+
 	} else {
 		nodes = []*Node{e.Cluster.NodeByHost(e.Host)}
 	}