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

Merge branch 'cluster-resize' into webui-interface

Browse source
This commit is contained in:
Cody Soyland 2018-03-19 14:02:01 -05:00
commit 4e1ff7c86f
24 changed files with 1315 additions and 648 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
Gopkg.lock generated
View file

@ -205,12 +205,6 @@
packages = ["."]
revision = "bb4de0191aa41b5507caa14b0650cdbddcd9280b"
[[projects]]
name = "github.com/sony/gobreaker"
packages = ["."]
revision = "e9556a45379ef1da12e54847edb2fb3d7d566f36"
version = "0.3.0"
[[projects]]
branch = "master"
name = "github.com/spf13/afero"

30
broadcast.go
View file

@ -35,8 +35,10 @@ type StaticMemberSet struct {
}
// NewStaticMemberSet creates a statically defined MemberSet.
func NewStaticMemberSet() *StaticMemberSet {
return &StaticMemberSet{}
func NewStaticMemberSet(nodes []*Node) *StaticMemberSet {
return &StaticMemberSet{
nodes: nodes,
}
}
// Open implements the MemberSet interface to start network activity, but for a static MemberSet it does nothing.
@ -44,12 +46,6 @@ func (s *StaticMemberSet) Open(n *Node) error {
return nil
}
// Join sets the MemberSet nodes to the slice of Nodes passed in.
func (s *StaticMemberSet) Join(nodes []*Node) error {
s.nodes = nodes
return nil
}
// Broadcaster is an interface for broadcasting messages.
type Broadcaster interface {
SendSync(pb proto.Message) error
@ -67,17 +63,17 @@ var NopBroadcaster Broadcaster
type nopBroadcaster struct{}
// SendSync A no-op implemenetation of Broadcaster SendSync method.
// SendSync A no-op implementation of Broadcaster SendSync method.
func (n *nopBroadcaster) SendSync(pb proto.Message) error {
return nil
}
// SendAsync A no-op implemenetation of Broadcaster SendAsync method.
// SendAsync A no-op implementation of Broadcaster SendAsync method.
func (n *nopBroadcaster) SendAsync(pb proto.Message) error {
return nil
}
// SendTo is a no-op implemenetation of Broadcaster SendTo method.
// SendTo is a no-op implementation of Broadcaster SendTo method.
func (c *nopBroadcaster) SendTo(to *Node, pb proto.Message) error {
return nil
}
@ -116,7 +112,7 @@ var NopGossiper Gossiper
type nopGossiper struct{}
// SendAsync A no-op implemenetation of Gossiper SendAsync method.
// SendAsync A no-op implementation of Gossiper SendAsync method.
func (n *nopGossiper) SendAsync(pb proto.Message) error {
return nil
}
@ -138,8 +134,10 @@ const (
MessageTypeResizeInstruction
MessageTypeResizeInstructionComplete
MessageTypeSetCoordinator
MessageTypeUpdateCoordinator
MessageTypeNodeState
MessageTypeRecalculateCaches
MessageTypeNodeEvent
)
// MarshalMessage encodes the protobuf message into a byte slice.
@ -176,10 +174,14 @@ func MarshalMessage(m proto.Message) ([]byte, error) {
typ = MessageTypeResizeInstructionComplete
case *internal.SetCoordinatorMessage:
typ = MessageTypeSetCoordinator
case *internal.UpdateCoordinatorMessage:
typ = MessageTypeUpdateCoordinator
case *internal.NodeStateMessage:
typ = MessageTypeNodeState
case *internal.RecalculateCaches:
typ = MessageTypeRecalculateCaches
case *internal.NodeEventMessage:
typ = MessageTypeNodeEvent
default:
return nil, fmt.Errorf("message type not implemented for marshalling: %s", reflect.TypeOf(obj))
}
@ -226,10 +228,14 @@ func UnmarshalMessage(buf []byte) (proto.Message, error) {
m = &internal.ResizeInstructionComplete{}
case MessageTypeSetCoordinator:
m = &internal.SetCoordinatorMessage{}
case MessageTypeUpdateCoordinator:
m = &internal.UpdateCoordinatorMessage{}
case MessageTypeNodeState:
m = &internal.NodeStateMessage{}
case MessageTypeRecalculateCaches:
m = &internal.RecalculateCaches{}
case MessageTypeNodeEvent:
m = &internal.NodeEventMessage{}
default:
return nil, fmt.Errorf("invalid message type: %d", typ)
}

143
cluster.go
View file

@ -66,15 +66,16 @@ const (
// Node represents a node in the cluster.
type Node struct {
ID string `json:"id"`
URI URI `json:"uri"`
ID string `json:"id"`
URI URI `json:"uri"`
IsCoordinator bool `json:"isCoordinator"`
}
func (n Node) String() string {
return fmt.Sprintf("Node: %s", n.ID)
}
// EncodeNodes converts a into its internal representation.
// EncodeNodes converts a slice of Nodes into its internal representation.
func EncodeNodes(a []*Node) []*internal.Node {
other := make([]*internal.Node, len(a))
for i := range a {
@ -83,14 +84,16 @@ func EncodeNodes(a []*Node) []*internal.Node {
return other
}
// EncodeNode converts n into its internal representation.
// EncodeNode converts a Node into its internal representation.
func EncodeNode(n *Node) *internal.Node {
return &internal.Node{
ID: n.ID,
URI: n.URI.Encode(),
ID: n.ID,
URI: n.URI.Encode(),
IsCoordinator: n.IsCoordinator,
}
}
// DecodeNodes converts a proto message into a slice of Nodes.
func DecodeNodes(a []*internal.Node) []*Node {
if len(a) == 0 {
return nil
@ -102,10 +105,19 @@ func DecodeNodes(a []*internal.Node) []*Node {
return other
}
// DecodeNode converts a proto message into a Node.
func DecodeNode(node *internal.Node) *Node {
return &Node{
ID: node.ID,
URI: decodeURI(node.URI),
ID: node.ID,
URI: decodeURI(node.URI),
IsCoordinator: node.IsCoordinator,
}
}
func DecodeNodeEvent(ne *internal.NodeEventMessage) *NodeEvent {
return &NodeEvent{
Event: NodeEventType(ne.Event),
Node: DecodeNode(ne.Node),
}
}
@ -236,7 +248,7 @@ type Cluster struct {
// Required for cluster Resize.
Static bool // Static is primarily used for testing in a non-gossip environment.
state string
Coordinator URI
Coordinator string
Holder *Holder
Broadcaster Broadcaster
@ -289,28 +301,59 @@ func (c *Cluster) logger() *log.Logger {
// Coordinator returns the coordinator node.
func (c *Cluster) CoordinatorNode() *Node {
return c.nodeByURI(c.Coordinator)
return c.nodeByID(c.Coordinator)
}
// IsCoordinator is true if this node is the coordinator.
func (c *Cluster) IsCoordinator() bool {
return c.Static || c.Coordinator == c.Node.URI
return c.Coordinator == c.Node.ID
}
// SetCoordinator updates the Coordinator to n.
// Returns true if the Coordinator changed.
func (c *Cluster) SetCoordinator(n *Node) bool {
// Get new node.
newNode := c.nodeByID(n.ID)
if newNode == nil {
return false
// SetCoordinator tells the current node to become the
// Coordinator. In response to this, the current node
// will consider itself coordinator and update the other
// nodes with its version of Cluster.Status.
func (c *Cluster) SetCoordinator(n *Node) error {
// Verify that the new Coordinator value matches
// this node.
if c.Node.ID != n.ID {
return fmt.Errorf("coordinator node does not match this node")
}
if c.Coordinator != newNode.URI {
c.Coordinator = newNode.URI
return true
// Update IsCoordinator on all nodes (locally).
_ = c.UpdateCoordinator(n)
// Send the update coordinator message to all nodes.
err := c.Broadcaster.SendSync(
&internal.UpdateCoordinatorMessage{
New: EncodeNode(n),
})
if err != nil {
return fmt.Errorf("problem sending UpdateCoordinator message: %v", err)
}
return false
// Broadcast cluster status.
return c.Broadcaster.SendSync(c.Status())
}
// UpdateCoordinator updates this nodes Coordinator value as well as
// changing the corresponding node's IsCoordinator value
// to true, and sets all other nodes to false. Returns true if the value
// changed.
func (c *Cluster) UpdateCoordinator(n *Node) bool {
var changed bool
if c.Coordinator != n.ID {
c.Coordinator = n.ID
changed = true
}
for _, node := range c.Nodes {
if node.ID == n.ID {
node.IsCoordinator = true
} else {
node.IsCoordinator = false
}
}
return changed
}
// AddNode adds a node to the Cluster and updates and saves the
@ -318,6 +361,11 @@ func (c *Cluster) SetCoordinator(n *Node) bool {
func (c *Cluster) AddNode(node *Node) error {
c.logger().Printf("add node %s to cluster on %s", node, c.Node)
// If the node being added is the coordinator, set it for this node.
if node.IsCoordinator {
c.Coordinator = node.ID
}
// add to cluster
if !c.addNodeBasicSorted(node) {
return nil
@ -737,7 +785,7 @@ func (c *Cluster) fragSources(to *Cluster, idx *Index) (map[string][]*internal.R
// the fragment.
srcNodeID, ok := srcNodesByFrag[frag]
if !ok {
return nil, errors.New("not enough data to perform resize")
return nil, errors.New("not enough data to perform resize (replica factor may need to be increased)")
}
src := &internal.ResizeSource{
@ -887,6 +935,22 @@ func (c *Cluster) Open() error {
// If not coordinator then wait for ClusterStatus from coordinator.
if !c.IsCoordinator() {
// In the case where a node has been restarted and memberlist has
// not had enough time to determine the node went down/up, then
// the coorninator needs to be alerted that this node is back up
// (and now in a state of STARTING) so that it can be put to the correct
// cluster state.
// TODO: Because the normal code path already sends a NodeJoin event (via
// memberlist), this it a bit redundant in most cases. Perhaps determine
// that the node has been restarted and don't do this step.
msg := &internal.NodeEventMessage{
Event: uint32(NodeJoin),
Node: EncodeNode(c.Node),
}
if err := c.Broadcaster.SendAsync(msg); err != nil {
return fmt.Errorf("sending restart NodeJoin: %v", err)
}
c.logger().Printf("wait for joining to complete")
<-c.joining
c.logger().Printf("joining has completed")
@ -937,6 +1001,10 @@ func (c *Cluster) allNodesReady() bool {
func (c *Cluster) handleNodeAction(nodeAction nodeAction) error {
j, err := c.generateResizeJob(nodeAction)
if err != nil {
c.logger().Printf("generateResizeJob error: err=%s", err)
if err := c.setStateAndBroadcast(ClusterStateNormal); err != nil {
c.logger().Printf("setStateAndBroadcast error: err=%s", err)
}
return err
}
@ -998,7 +1066,13 @@ func (c *Cluster) ListenForJoins() {
}
func (c *Cluster) listenForJoins() {
var uriJoined bool
// When a cluster starts, the state is STARTING.
// We first want to wait for at least one node to join.
// Then we want to clear out the joiningLeavingNodes queue (buffered channel).
// Then we want to set the cluster state to NORMAL and resume processing of joiningLeavingNodes events.
// We use a bool `setNormal` to indicate when at least one node has joined.
var setNormal bool
for {
@ -1010,13 +1084,13 @@ func (c *Cluster) listenForJoins() {
c.logger().Printf("handleNodeAction error: err=%s", err)
continue
}
uriJoined = true
setNormal = true
continue
default:
}
// Only change state to NORMAL if we have successfully added at least one host.
if uriJoined {
if setNormal {
// Put the cluster back to state NORMAL and broadcast.
if err := c.setStateAndBroadcast(ClusterStateNormal); err != nil {
c.logger().Printf("setStateAndBroadcast error: err=%s", err)
@ -1033,7 +1107,7 @@ func (c *Cluster) listenForJoins() {
c.logger().Printf("handleNodeAction error: err=%s", err)
continue
}
uriJoined = true
setNormal = true
continue
}
}
@ -1658,9 +1732,11 @@ func (c *Cluster) nodeJoin(node *Node) error {
return nil
}
// Don't do anything else if the cluster already contains the node.
if c.nodeByID(node.ID) != nil {
return nil
// If the cluster already contains the node, just send it the cluster status.
// This is useful in the case where a node is restarted or temporarily leaves
// the cluster.
if node := c.nodeByID(node.ID); node != nil {
return c.sendTo(node, c.Status())
}
// If the holder does not yet contain data, go ahead and add the node.
@ -1702,6 +1778,13 @@ func (c *Cluster) NodeLeave(node *Node) error {
return fmt.Errorf("The coordinator node cannot be removed. First, make a different node the new coordinator.")
}
// See if resize job can be generated
_, err := c.generateResizeJobByAction(nodeAction{c.nodeByID(node.ID), ResizeJobActionRemove})
if err != nil {
return err
}
return c.nodeLeave(node)
}

16
cluster_internal_test.go
View file

@ -180,8 +180,8 @@ func TestFragSources(t *testing.T) {
"node0": []*internal.ResizeSource{},
"node1": []*internal.ResizeSource{},
"node2": []*internal.ResizeSource{
{&internal.Node{"node0", &internal.URI{"http", "host0", 10101}}, "i", "f", "standard", uint64(0)},
{&internal.Node{"node1", &internal.URI{"http", "host1", 10101}}, "i", "f", "standard", uint64(2)},
{&internal.Node{"node0", &internal.URI{"http", "host0", 10101}, false}, "i", "f", "standard", uint64(0)},
{&internal.Node{"node1", &internal.URI{"http", "host1", 10101}, false}, "i", "f", "standard", uint64(2)},
},
},
err: "",
@ -192,11 +192,11 @@ func TestFragSources(t *testing.T) {
idx: idx,
expected: map[string][]*internal.ResizeSource{
"node0": []*internal.ResizeSource{
{&internal.Node{"node1", &internal.URI{"http", "host1", 10101}}, "i", "f", "standard", uint64(1)},
{&internal.Node{"node1", &internal.URI{"http", "host1", 10101}, false}, "i", "f", "standard", uint64(1)},
},
"node1": []*internal.ResizeSource{
{&internal.Node{"node0", &internal.URI{"http", "host0", 10101}}, "i", "f", "standard", uint64(0)},
{&internal.Node{"node0", &internal.URI{"http", "host0", 10101}}, "i", "f", "standard", uint64(2)},
{&internal.Node{"node0", &internal.URI{"http", "host0", 10101}, false}, "i", "f", "standard", uint64(0)},
{&internal.Node{"node0", &internal.URI{"http", "host0", 10101}, false}, "i", "f", "standard", uint64(2)},
},
},
err: "",
@ -207,11 +207,11 @@ func TestFragSources(t *testing.T) {
idx: idx,
expected: map[string][]*internal.ResizeSource{
"node0": []*internal.ResizeSource{
{&internal.Node{"node2", &internal.URI{"http", "host2", 10101}}, "i", "f", "standard", uint64(0)},
{&internal.Node{"node2", &internal.URI{"http", "host2", 10101}}, "i", "f", "standard", uint64(2)},
{&internal.Node{"node2", &internal.URI{"http", "host2", 10101}, false}, "i", "f", "standard", uint64(0)},
{&internal.Node{"node2", &internal.URI{"http", "host2", 10101}, false}, "i", "f", "standard", uint64(2)},
},
"node1": []*internal.ResizeSource{
{&internal.Node{"node0", &internal.URI{"http", "host0", 10101}}, "i", "f", "standard", uint64(3)},
{&internal.Node{"node0", &internal.URI{"http", "host0", 10101}, false}, "i", "f", "standard", uint64(3)},
},
"node2": []*internal.ResizeSource{},
},

20
cluster_test.go
View file

@ -180,10 +180,10 @@ func TestCluster_Coordinator(t *testing.T) {
c1 := *pilosa.NewCluster()
c1.Node = node1
c1.Coordinator = node1.URI
c1.Coordinator = node1.ID
c2 := *pilosa.NewCluster()
c2.Node = node2
c2.Coordinator = node1.URI
c2.Coordinator = node1.ID
t.Run("IsCoordinator", func(t *testing.T) {
if !c1.IsCoordinator() {
@ -509,24 +509,24 @@ func TestCluster_ResizeStates(t *testing.T) {
}
// Ensures that coordinator can be changed.
func TestCluster_SetCoordinator(t *testing.T) {
t.Run("SetCoordinator", func(t *testing.T) {
func TestCluster_UpdateCoordinator(t *testing.T) {
t.Run("UpdateCoordinator", func(t *testing.T) {
c := test.NewCluster(2)
oldNode := c.Nodes[0]
newNode := c.Nodes[1]
// Set coordinator to the same value.
if c.SetCoordinator(oldNode) {
// Update coordinator to the same value.
if c.UpdateCoordinator(oldNode) {
t.Errorf("did not expect coordinator to change")
} else if c.Coordinator != oldNode.URI {
} else if c.Coordinator != oldNode.ID {
t.Errorf("expected coordinator: %s, but got: %s", c.Coordinator, oldNode.URI)
}
// Set coordinator to a new value.
if !c.SetCoordinator(newNode) {
// Update coordinator to a new value.
if !c.UpdateCoordinator(newNode) {
t.Errorf("expected coordinator to change")
} else if c.Coordinator != newNode.URI {
} else if c.Coordinator != newNode.ID {
t.Errorf("expected coordinator: %s, but got: %s", c.Coordinator, newNode.URI)
}
})

4
config.go
View file

@ -139,7 +139,7 @@ type Config struct {
Cluster struct {
Disabled bool `toml:"disabled"`
Coordinator string `toml:"coordinator"`
Coordinator bool `toml:"coordinator"`
ReplicaN int `toml:"replicas"`
Hosts []string `toml:"hosts"`
LongQueryTime Duration `toml:"long-query-time"`
@ -183,7 +183,7 @@ func NewConfig() *Config {
// Cluster config.
c.Cluster.Disabled = DefaultClusterDisabled
// c.Cluster.Coordinator = ""
// c.Cluster.Coordinator = false
c.Cluster.ReplicaN = DefaultReplicaN
c.Cluster.Hosts = []string{}
c.Cluster.LongQueryTime = Duration(time.Minute)

4
ctl/server.go
View file

@ -34,9 +34,9 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) {
// Cluster
flags.BoolVarP(&srv.Config.Cluster.Disabled, "cluster.disabled", "", srv.Config.Cluster.Disabled, "Disabled multi-node cluster communication (used for testing)")
flags.StringVarP(&srv.Config.Cluster.Coordinator, "cluster.coordinator", "", "", "Host that will act as cluster coordinator during startup and resizing.")
flags.BoolVarP(&srv.Config.Cluster.Coordinator, "cluster.coordinator", "", srv.Config.Cluster.Coordinator, "Host that will act as cluster coordinator during startup and resizing.")
flags.IntVarP(&srv.Config.Cluster.ReplicaN, "cluster.replicas", "", 1, "Number of hosts each piece of data should be stored on.")
flags.StringSliceVarP(&srv.Config.Cluster.Hosts, "cluster.hosts", "", []string{}, "Comma separated list of hosts in cluster.")
flags.StringSliceVarP(&srv.Config.Cluster.Hosts, "cluster.hosts", "", []string{}, "Comma separated list of hosts in cluster. Only used for testing.")
flags.DurationVarP((*time.Duration)(&srv.Config.Cluster.LongQueryTime), "cluster.long-query-time", "", time.Minute, "Duration that will trigger log and stat messages for slow queries.")
// Gossip

324
diagnostics.go Normal file
View file

@ -0,0 +1,324 @@
// Copyright 2017 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package pilosa
import (
"bytes"
"encoding/json"
"fmt"
"io"
"io/ioutil"
"log"
"net/http"
"strconv"
"strings"
"sync"
"time"
)
// Default version check URL.
const (
defaultVersionCheckURL = "https://diagnostics.pilosa.com/v0/version"
)
type versionResponse struct {
Version string `json:"version"`
Message string `json:"message"`
}
// DiagnosticsCollector represents a collector/sender of diagnostics data.
type DiagnosticsCollector struct {
mu sync.Mutex
host string
VersionURL string
version string
lastVersion string
startTime int64
start time.Time
metrics map[string]interface{}
client *http.Client
logOutput io.Writer
server *Server
}
// NewDiagnosticsCollector returns a new DiagnosticsCollector given an addr in the format "hostname:port".
func NewDiagnosticsCollector(host string) *DiagnosticsCollector {
return &DiagnosticsCollector{
host: host,
VersionURL: defaultVersionCheckURL,
startTime: time.Now().Unix(),
start: time.Now(),
client: &http.Client{Timeout: 10 * time.Second},
metrics: make(map[string]interface{}),
logOutput: ioutil.Discard,
}
}
// SetVersion of locally running Pilosa Cluster to check against master.
func (d *DiagnosticsCollector) SetVersion(v string) {
d.version = v
d.Set("Version", v)
}
// Flush sends the current metrics.
func (d *DiagnosticsCollector) Flush() error {
d.mu.Lock()
defer d.mu.Unlock()
d.metrics["Uptime"] = (time.Now().Unix() - d.startTime)
buf, err := d.encode()
if err != nil {
return err
}
req, err := http.NewRequest("POST", d.host, bytes.NewReader(buf))
req.Header.Set("Content-Type", "application/json")
resp, err := d.client.Do(req)
if err != nil {
return err
}
// Intentionally ignoring response body, as user does not need to be notified of error.
defer resp.Body.Close()
return nil
}
// CheckVersion of the local build against Pilosa master.
func (d *DiagnosticsCollector) CheckVersion() error {
var rsp versionResponse
req, err := http.NewRequest("GET", d.VersionURL, nil)
resp, err := d.client.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return fmt.Errorf("http: status=%d", resp.StatusCode)
} else if err := json.NewDecoder(resp.Body).Decode(&rsp); err != nil {
return fmt.Errorf("json decode: %s", err)
}
// If version has not changed since the last check, return
if rsp.Version == d.lastVersion {
return nil
}
d.lastVersion = rsp.Version
if err := d.compareVersion(rsp.Version); err != nil {
d.logger().Printf("%s\n", err.Error())
}
return nil
}
// compareVersion check version strings.
func (d *DiagnosticsCollector) compareVersion(value string) error {
currentVersion := versionSegments(value)
localVersion := versionSegments(d.version)
if localVersion[0] < currentVersion[0] { //Major
return fmt.Errorf("Warning: You are running Pilosa %s. A newer version (%s) is available: https://github.com/pilosa/pilosa/releases", d.version, value)
} else if localVersion[1] < currentVersion[1] && localVersion[0] == currentVersion[0] { // Minor
return fmt.Errorf("Warning: You are running Pilosa %s. The latest Minor release is %s: https://github.com/pilosa/pilosa/releases", d.version, value)
} else if localVersion[2] < currentVersion[2] && localVersion[0] == currentVersion[0] && localVersion[1] == currentVersion[1] { // Patch
return fmt.Errorf("There is a new patch release of Pilosa available: %s: https://github.com/pilosa/pilosa/releases", value)
}
return nil
}
// Encode metrics maps into the json message format.
func (d *DiagnosticsCollector) encode() ([]byte, error) {
return json.Marshal(d.metrics)
}
// Set adds a key value metric.
func (d *DiagnosticsCollector) Set(name string, value interface{}) {
switch v := value.(type) {
case string:
if v == "" {
// Do not set empty string
return
}
}
d.mu.Lock()
defer d.mu.Unlock()
d.metrics[name] = value
}
// SetLogger Set the logger output type.
func (d *DiagnosticsCollector) SetLogger(logger io.Writer) {
d.logOutput = logger
}
// logger returns a logger that writes to LogOutput.
func (d *DiagnosticsCollector) logger() *log.Logger {
return log.New(d.logOutput, "", log.LstdFlags)
}
// logErr logs the error and returns true if an error exists
func (d *DiagnosticsCollector) logErr(err error) bool {
if err != nil {
d.logOutput.Write([]byte(err.Error()))
return true
}
return false
}
// EnrichWithOSInfo adds OS information to the diagnostics payload.
func (d *DiagnosticsCollector) EnrichWithOSInfo() {
uptime, err := d.server.SystemInfo.Uptime()
if !d.logErr(err) {
d.Set("HostUptime", uptime)
}
platform, err := d.server.SystemInfo.Platform()
if !d.logErr(err) {
d.Set("OSPlatform", platform)
}
family, err := d.server.SystemInfo.Family()
if !d.logErr(err) {
d.Set("OSFamily", family)
}
version, err := d.server.SystemInfo.OSVersion()
if !d.logErr(err) {
d.Set("OSVersion", version)
}
kernelVersion, err := d.server.SystemInfo.KernelVersion()
if !d.logErr(err) {
d.Set("OSKernelVersion", kernelVersion)
}
}
// EnrichWithMemoryInfo adds memory information to the diagnostics payload.
func (d *DiagnosticsCollector) EnrichWithMemoryInfo() {
memFree, err := d.server.SystemInfo.MemFree()
if !d.logErr(err) {
d.Set("MemFree", memFree)
}
memTotal, err := d.server.SystemInfo.MemTotal()
if !d.logErr(err) {
d.Set("MemTotal", memTotal)
}
memUsed, err := d.server.SystemInfo.MemUsed()
if !d.logErr(err) {
d.Set("MemUsed", memUsed)
}
}
// EnrichWithSchemaProperties adds schema info to the diagnostics payload.
func (d *DiagnosticsCollector) EnrichWithSchemaProperties() {
var numSlices uint64
numFrames := 0
numIndexes := 0
bsiFieldCount := 0
timeQuantumEnabled := false
for _, index := range d.server.Holder.Indexes() {
numSlices += index.MaxSlice() + 1
numIndexes += 1
for _, frame := range index.Frames() {
numFrames += 1
if frame.rangeEnabled {
if fields, err := frame.GetFields(); err == nil {
bsiFieldCount += len(fields)
}
}
if frame.TimeQuantum() != "" {
timeQuantumEnabled = true
}
}
}
d.Set("NumIndexes", numIndexes)
d.Set("NumFrames", numFrames)
d.Set("NumSlices", numSlices)
d.Set("BSIFieldCount", bsiFieldCount)
d.Set("TimeQuantumEnabled", timeQuantumEnabled)
}
// versionSegments returns the numeric segments of the version as a slice of ints.
func versionSegments(segments string) []int {
segments = strings.Trim(segments, "v")
segments = strings.Split(segments, "-")[0]
s := strings.Split(segments, ".")
segmentSlice := make([]int, len(s))
for i, v := range s {
segmentSlice[i], _ = strconv.Atoi(v)
}
return segmentSlice
}
// SystemInfo collects information about the host OS.
type SystemInfo interface {
Uptime() (uint64, error)
Platform() (string, error)
Family() (string, error)
OSVersion() (string, error)
KernelVersion() (string, error)
MemFree() (uint64, error)
MemTotal() (uint64, error)
MemUsed() (uint64, error)
}
// NewNopSystemInfo creates a no-op implementation of SystemInfo.
func NewNopSystemInfo() *NopSystemInfo {
return &NopSystemInfo{}
}
// NopSystemInfo is a no-op implementation of SystemInfo.
type NopSystemInfo struct {
}
// Uptime is a no-op implementation of SystemInfo.Uptime.
func (n *NopSystemInfo) Uptime() (uint64, error) {
return 0, nil
}
// Platform is a no-op implementation of SystemInfo.Platform.
func (n *NopSystemInfo) Platform() (string, error) {
return "", nil
}
// Family is a no-op implementation of SystemInfo.Family.
func (n *NopSystemInfo) Family() (string, error) {
return "", nil
}
// OSVersion is a no-op implementation of SystemInfo.OSVersion.
func (n *NopSystemInfo) OSVersion() (string, error) {
return "", nil
}
// KernelVersion is a no-op implementation of SystemInfo.KernelVersion.
func (n *NopSystemInfo) KernelVersion() (string, error) {
return "", nil
}
// MemFree is a no-op implementation of SystemInfo.MemFree.
func (n *NopSystemInfo) MemFree() (uint64, error) {
return 0, nil
}
// MemTotal is a no-op implementation of SystemInfo.MemTotal.
func (n *NopSystemInfo) MemTotal() (uint64, error) {
return 0, nil
}
// MemUsed is a no-op implementation of SystemInfo.MemUsed.
func (n *NopSystemInfo) MemUsed() (uint64, error) {
return 0, nil
}

267
diagnostics/diagnostics.go
View file

@ -1,267 +0,0 @@
// Copyright 2017 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package diagnostics
import (
"bytes"
"encoding/json"
"fmt"
"io"
"io/ioutil"
"log"
"net/http"
"strconv"
"strings"
"sync"
"time"
"github.com/shirou/gopsutil/host"
"github.com/shirou/gopsutil/mem"
"github.com/sony/gobreaker"
)
// TODO: unique Cluster ID
// Default version check URL.
const (
DefaultVersionCheckURL = "https://diagnostics.pilosa.com/v0/version"
)
type versionResponse struct {
Version string `json:"version"`
Message string `json:"message"`
}
// Diagnostics represents a client to the Pilosa cluster.
type Diagnostics struct {
mu sync.Mutex
wg sync.WaitGroup
closing chan struct{}
host string
VersionURL string
version string
lastVersion string
startTime int64
start time.Time
metrics map[string]interface{}
client *http.Client
interval time.Duration
cb *gobreaker.CircuitBreaker
logOutput io.Writer
}
// New returns a pointer to a new Diagnostics Client given an addr in the format "hostname:port".
func New(host string) *Diagnostics {
return &Diagnostics{
closing: make(chan struct{}),
host: host,
VersionURL: DefaultVersionCheckURL,
startTime: time.Now().Unix(),
start: time.Now(),
client: http.DefaultClient,
metrics: make(map[string]interface{}),
logOutput: ioutil.Discard,
}
}
// SetVersion of locally running Pilosa Cluster to check against master.
func (d *Diagnostics) SetVersion(v string) {
d.version = v
d.Set("Version", v)
}
// SetInterval of the diagnostic go routine and match with the circuit breaker timeout.
func (d *Diagnostics) SetInterval(i time.Duration) {
d.interval = i
}
// schedule start the diagnostics service ticker.
func (d *Diagnostics) schedule() {
ticker := time.NewTicker(d.interval)
defer ticker.Stop()
for {
select {
case <-d.closing:
return
case <-ticker.C:
d.CheckVersion()
d.Flush()
}
}
}
// Flush sends the current metrics.
func (d *Diagnostics) Flush() error {
d.mu.Lock()
d.metrics["Uptime"] = (time.Now().Unix() - d.startTime)
buf, _ := d.Encode()
d.mu.Unlock()
_, err := d.cb.Execute(func() (interface{}, error) {
req, err := http.NewRequest("POST", d.host, bytes.NewReader(buf))
req.Header.Set("Content-Type", "application/json")
resp, err := d.client.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
// TODO verify response
body, err := ioutil.ReadAll(resp.Body)
if err != nil {
return nil, err
}
return body, nil
})
return err
}
// Open configures the circuit breaker used by the HTTP client.
func (d *Diagnostics) Open() {
var st gobreaker.Settings
if d.interval > 0 {
st.Timeout = d.interval * 2
}
d.cb = gobreaker.NewCircuitBreaker(st)
d.logger().Printf("Pilosa is currently configured to send small diagnostics reports to our team every hour. More information here: https://www.pilosa.com/docs/latest/administration/#diagnostics")
}
// Close notify goroutine to stop.
func (d *Diagnostics) Close() error {
close(d.closing)
d.wg.Wait()
return nil
}
// CheckVersion of the local build against Pilosa master.
func (d *Diagnostics) CheckVersion() error {
var rsp versionResponse
req, err := http.NewRequest("GET", d.VersionURL, nil)
resp, err := d.client.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return fmt.Errorf("http: status=%d", resp.StatusCode)
} else if err := json.NewDecoder(resp.Body).Decode(&rsp); err != nil {
return fmt.Errorf("json decode: %s", err)
}
// Same a version as last test
if rsp.Version == d.lastVersion {
return nil
}
d.lastVersion = rsp.Version
if err := d.CompareVersion(rsp.Version); err != nil {
d.logger().Printf("%s\n", err.Error())
}
return nil
}
// CompareVersion check version strings.
func (d *Diagnostics) CompareVersion(value string) error {
currentVersion := VersionSegments(value)
localVersion := VersionSegments(d.version)
if localVersion[0] < currentVersion[0] { //Major
return fmt.Errorf("Warning: You are running Pilosa %s. A newer version (%s) is available: https://github.com/pilosa/pilosa/releases", d.version, value)
} else if localVersion[1] < currentVersion[1] && localVersion[0] == currentVersion[0] { // Minor
return fmt.Errorf("Warning: You are running Pilosa %s. The latest Minor release is %s: https://github.com/pilosa/pilosa/releases", d.version, value)
} else if localVersion[2] < currentVersion[2] && localVersion[0] == currentVersion[0] && localVersion[1] == currentVersion[1] { // Patch
return fmt.Errorf("There is a new patch release of Pilosa available: %s: https://github.com/pilosa/pilosa/releases", value)
}
return nil
}
// Encode metrics maps into the json message format.
func (d *Diagnostics) Encode() ([]byte, error) {
return json.Marshal(d.metrics)
}
// Set adds a key value metric.
func (d *Diagnostics) Set(name string, value interface{}) {
d.mu.Lock()
defer d.mu.Unlock()
d.metrics[name] = value
}
// SetLogger Set the logger output type.
func (d *Diagnostics) SetLogger(logger io.Writer) {
d.logOutput = logger
}
// logger returns a logger that writes to LogOutput.
func (d *Diagnostics) logger() *log.Logger {
return log.New(d.logOutput, "", log.LstdFlags)
}
// EnrichWithOSInfo adds OS information to the diagnostics payload.
func (d *Diagnostics) EnrichWithOSInfo() {
osInfo, err := host.Info()
if err != nil {
d.logOutput.Write([]byte(err.Error()))
}
d.Set("HostUptime", osInfo.Uptime)
platform, family, version, err := host.PlatformInformation()
if err != nil {
d.logOutput.Write([]byte(err.Error()))
}
d.Set("OSPlatform", platform)
d.Set("OSFamily", family)
d.Set("OSVersion", version)
kernelVersion, err := host.KernelVersion()
if err != nil {
d.logOutput.Write([]byte(err.Error()))
}
d.Set("OSKernelVersion", kernelVersion)
}
// EnrichWithMemoryInfo adds memory information to the diagnostics payload.
func (d *Diagnostics) EnrichWithMemoryInfo() {
memory, err := mem.VirtualMemory()
if err != nil {
d.logOutput.Write([]byte(err.Error()))
}
d.Set("MemFree", memory.Free)
d.Set("MemTotal", memory.Total)
d.Set("MemUsed", memory.Used)
}
// VersionSegments returns the numeric segments of the version as a slice of ints.
func VersionSegments(segments string) []int {
segments = strings.Trim(segments, "v")
segments = strings.Split(segments, "-")[0]
s := strings.Split(segments, ".")
segmentSlice := make([]int, len(s))
for i, v := range s {
segmentSlice[i], _ = strconv.Atoi(v)
}
return segmentSlice
}

43
diagnostics/diagnostics_test.go → diagnostics_internal_test.go
View file

@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
package diagnostics_test
package pilosa
import (
"encoding/json"
@ -23,25 +23,20 @@ import (
"runtime"
"strings"
"testing"
"github.com/pilosa/pilosa/diagnostics"
)
func TestDiagnosticsClient(t *testing.T) {
// Mock server.
server := httptest.NewServer(nil)
defer server.Close()
// Create a new client.
d := diagnostics.New(server.URL)
d := NewDiagnosticsCollector(server.URL)
d.SetLogger(ioutil.Discard)
d.Open()
defer d.Close()
d.Set("gg", 10)
d.Set("ss", "ss")
data, err := d.Encode()
data, err := d.encode()
if err != nil {
t.Fatal(err)
}
@ -58,7 +53,7 @@ func TestDiagnosticsClient(t *testing.T) {
// Test the metrics after a flush.
d.Flush()
data, err = d.Encode()
data, err = d.encode()
if err != nil {
t.Fatal(err)
}
@ -74,7 +69,7 @@ func TestDiagnosticsClient(t *testing.T) {
func TestDiagnosticsVersion_Parse(t *testing.T) {
version := "0.1.1"
vs := diagnostics.VersionSegments(version)
vs := versionSegments(version)
output := []int{0, 1, 1}
if !reflect.DeepEqual(vs, output) {
@ -83,35 +78,33 @@ func TestDiagnosticsVersion_Parse(t *testing.T) {
}
func TestDiagnosticsVersion_Compare(t *testing.T) {
d := diagnostics.New("localhost:10101")
d.Open()
defer d.Close()
d := NewDiagnosticsCollector("localhost:10101")
version := "v0.1.1"
d.SetVersion(version)
err := d.CompareVersion("v1.7.0")
err := d.compareVersion("v1.7.0")
if !strings.Contains(err.Error(), "A newer version") {
t.Fatalf("Expected a newer version is available, actual error: %s", err)
}
err = d.CompareVersion("1.7.0")
err = d.compareVersion("1.7.0")
if !strings.Contains(err.Error(), "A newer version") {
t.Fatalf("Expected a newer version is available, actual error: %s", err)
}
err = d.CompareVersion("0.7.0")
err = d.compareVersion("0.7.0")
if !strings.Contains(err.Error(), "The latest Minor release is") {
t.Fatalf("Expected Minor Version Missmatch, actual error: %s", err)
}
err = d.CompareVersion("0.1.2")
err = d.compareVersion("0.1.2")
if !strings.Contains(err.Error(), "There is a new patch release of Pilosa") {
t.Fatalf("Expected Patch Version Missmatch, actual error: %s", err)
}
err = d.CompareVersion("0.1.1")
err = d.compareVersion("0.1.1")
if err != nil {
t.Fatalf("Versions should match")
}
d.SetVersion("v1.7.0")
err = d.CompareVersion("0.7.2")
err = d.compareVersion("0.7.2")
if err != nil {
t.Fatalf("Local version is greater")
}
@ -125,11 +118,9 @@ func TestDiagnosticsVersion_Check(t *testing.T) {
Version: "1.1.1",
})
}))
defer server.Close()
// Create a new client.
d := diagnostics.New("localhost:10101")
defer d.Close()
d := NewDiagnosticsCollector("localhost:10101")
version := "0.1.1"
d.SetVersion(version)
@ -138,10 +129,6 @@ func TestDiagnosticsVersion_Check(t *testing.T) {
d.CheckVersion()
}
type versionResponse struct {
Version string `json:"version"`
}
func compareJSON(a, b []byte) (bool, error) {
var j1, j2 interface{}
if err := json.Unmarshal(a, &j1); err != nil {
@ -156,12 +143,10 @@ func compareJSON(a, b []byte) (bool, error) {
func BenchmarkDiagnostics(b *testing.B) {
// Mock server.
server := httptest.NewServer(nil)
defer server.Close()
// Create a new client.
d := diagnostics.New(server.URL)
d := NewDiagnosticsCollector(server.URL)
d.SetLogger(ioutil.Discard)
defer d.Close()
prev := runtime.GOMAXPROCS(4)
defer runtime.GOMAXPROCS(prev)

7
gc.go
View file

@ -14,6 +14,9 @@
package pilosa
// Ensure nopGCNotifier implements interface.
var _ GCNotifier = &nopGCNotifier{}
// GCNotifier represents an interface for garbage collection notificationss.
type GCNotifier interface {
Close()
@ -29,10 +32,10 @@ var NopGCNotifier GCNotifier
type nopGCNotifier struct{}
// Close is a no-op implemenetation of GCNotifier Close method.
// Close is a no-op implementation of GCNotifier Close method.
func (n *nopGCNotifier) Close() {}
// AfterGC is a no-op implemenetation of GCNotifier AfterGC method.
// AfterGC is a no-op implementation of GCNotifier AfterGC method.
func (c *nopGCNotifier) AfterGC() <-chan struct{} {
return nil
}

115
gopsutil/systeminfo.go Normal file
View file

@ -0,0 +1,115 @@
// Copyright 2017 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package gopsutil
import (
"github.com/pilosa/pilosa"
"github.com/shirou/gopsutil/host"
"github.com/shirou/gopsutil/mem"
)
var _ pilosa.SystemInfo = NewSystemInfo()
// SystemInfo is an implementation of pilosa.SystemInfo that uses gopsutil to collect information about the host OS.
type SystemInfo struct {
platform string
family string
osVersion string
}
// Uptime returns the system uptime in seconds.
func (s *SystemInfo) Uptime() (uptime uint64, err error) {
hostInfo, err := host.Info()
if err != nil {
return 0, err
}
return hostInfo.Uptime, nil
}
// collectPlatformInfo fetches and caches system platform information.
func (s *SystemInfo) collectPlatformInfo() error {
var err error
if s.platform == "" {
s.platform, s.family, s.osVersion, err = host.PlatformInformation()
if err != nil {
return err
}
}
return nil
}
// Platform returns the system platform.
func (s *SystemInfo) Platform() (string, error) {
err := s.collectPlatformInfo()
if err != nil {
return "", err
}
return s.platform, nil
}
// Family returns the system family.
func (s *SystemInfo) Family() (string, error) {
err := s.collectPlatformInfo()
if err != nil {
return "", err
}
return s.family, err
}
// OSVersion returns the OS Version.
func (s *SystemInfo) OSVersion() (string, error) {
err := s.collectPlatformInfo()
if err != nil {
return "", err
}
return s.osVersion, err
}
// MemFree returns the amount of free memory in bytes.
func (s *SystemInfo) MemFree() (uint64, error) {
memInfo, err := mem.VirtualMemory()
if err != nil {
return 0, err
}
return memInfo.Free, err
}
// MemTotal returns the amount of total memory in bytes.
func (s *SystemInfo) MemTotal() (uint64, error) {
memInfo, err := mem.VirtualMemory()
if err != nil {
return 0, err
}
return memInfo.Total, err
}
// MemUsed returns the amount of used memory in bytes.
func (s *SystemInfo) MemUsed() (uint64, error) {
memInfo, err := mem.VirtualMemory()
if err != nil {
return 0, err
}
return memInfo.Used, err
}
// KernelVersion returns the kernel version as a string.
func (s *SystemInfo) KernelVersion() (string, error) {
return host.KernelVersion()
}
// NewSystemInfo is a constructor for the gopsutil implementation of SystemInfo.
func NewSystemInfo() *SystemInfo {
return &SystemInfo{}
}

77
gopsutil/systeminfo_test.go Normal file
View file

@ -0,0 +1,77 @@
// Copyright 2017 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package gopsutil_test
import (
"log"
"testing"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/gopsutil"
)
func TestSystemInfo(t *testing.T) {
var systemInfo pilosa.SystemInfo = gopsutil.NewSystemInfo()
// Uptime()(uint64, error)
// Platform()(string, error)
// Family()(string, error)
// OSVersion()(string, error)
// KernelVersion()(string, error)
// MemFree()(uint64, error)
// MemTotal()(uint64, error)
// MemUsed()(uint64, error)
//
uptime, err := systemInfo.Uptime()
if err != nil || uptime == 0 {
t.Fatalf("Error collecting uptime (error: %v)", err)
}
platform, err := systemInfo.Platform()
if err != nil {
t.Fatalf("Error getting platform. (platform: %v, error: %v)", platform, err)
}
family, err := systemInfo.Family()
if err != nil {
t.Fatalf("Error getting OS family. (family: %v, error: %v)", family, err)
}
osversion, err := systemInfo.OSVersion()
if err != nil {
t.Fatalf("Error getting OS version. (osversion: %v, error: %v)", osversion, err)
}
kernelversion, err := systemInfo.KernelVersion()
if err != nil {
t.Fatalf("Error getting kernel version. (kernelversion: %v, error: %v)", kernelversion, err)
}
memfree, err := systemInfo.MemFree()
if err != nil {
t.Fatalf("Error getting memfree. (memfree: %v, error: %v)", memfree, err)
}
memused, err := systemInfo.MemUsed()
if err != nil {
t.Fatalf("Error getting memused. (memused: %v, error: %v)", memused, err)
}
memtotal, err := systemInfo.MemTotal()
log.Println(memtotal)
if err != nil {
t.Fatalf("Error getting memtotal. (memtotal: %v, error: %v)", memtotal, err)
}
}

15
handler.go
View file

@ -2027,6 +2027,7 @@ func (h *Handler) handlePostClusterResizeSetCoordinator(w http.ResponseWriter, r
return
}
oldNode := h.Cluster.nodeByID(h.Cluster.Coordinator)
newNode := h.Cluster.nodeByID(req.ID)
if newNode == nil {
http.Error(w, "Node with provided ID does not exist", http.StatusBadRequest)
@ -2034,8 +2035,14 @@ func (h *Handler) handlePostClusterResizeSetCoordinator(w http.ResponseWriter, r
}
if err := func() error {
// Send the set-coordinator message to all nodes.
err := h.Broadcaster.SendSync(
// If the new coordinator is this node, do the SetCoordinator directly.
if newNode.ID == h.Node.ID {
return h.Cluster.SetCoordinator(newNode)
}
// Send the set-coordinator message to new node.
err := h.Broadcaster.SendTo(
newNode,
&internal.SetCoordinatorMessage{
New: EncodeNode(newNode),
})
@ -2043,9 +2050,6 @@ func (h *Handler) handlePostClusterResizeSetCoordinator(w http.ResponseWriter, r
return fmt.Errorf("problem sending SetCoordinator message: %s", err)
}
// Set Coordinator on local node.
_ = h.Cluster.SetCoordinator(newNode)
return nil
}(); err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
@ -2054,6 +2058,7 @@ func (h *Handler) handlePostClusterResizeSetCoordinator(w http.ResponseWriter, r
// Encode response.
if err := json.NewEncoder(w).Encode(setCoordinatorResponse{
Old: oldNode,
New: newNode,
}); err != nil {
h.logger().Printf("response encoding error: %s", err)

4
handler_test.go
View file

@ -148,7 +148,7 @@ func TestHandler_Status(t *testing.T) {
h.ServeHTTP(w, test.MustNewHTTPRequest("GET", "/status", nil))
if w.Code != http.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
} else if body := w.Body.String(); body != `{"state":"NORMAL","nodes":[{"id":"test-node","uri":{"scheme":"http","host":"localhost","port":10101}}]}`+"\n" {
} else if body := w.Body.String(); body != `{"state":"NORMAL","nodes":[{"id":"test-node","uri":{"scheme":"http","host":"localhost","port":10101},"isCoordinator":false}]}`+"\n" {
t.Fatalf("unexpected body: %s", body)
}
}
@ -1213,7 +1213,7 @@ func TestHandler_Fragment_Nodes(t *testing.T) {
h.ServeHTTP(w, r)
if w.Code != http.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
} else if body := w.Body.String(); body != `[{"id":"node2","uri":{"scheme":"http","host":"host2"}},{"id":"node0","uri":{"scheme":"http","host":"host0"}}]`+"\n" {
} else if body := w.Body.String(); body != `[{"id":"node2","uri":{"scheme":"http","host":"host2"},"isCoordinator":false},{"id":"node0","uri":{"scheme":"http","host":"host0"},"isCoordinator":false}]`+"\n" {
t.Fatalf("unexpected body: %q", body)
}

31
holder_test.go
View file

@ -338,37 +338,6 @@ func TestHolder_HasData(t *testing.T) {
})
}
/*
func TestHolder_Schema(t *testing.T) {
t.Run("Schema", func(t *testing.T) {
h := test.MustOpenHolder()
defer h.Close()
if idx, err := h.CreateIndex("i", pilosa.IndexOptions{}); err != nil {
t.Fatal(err)
} else if frame, err := idx.CreateFrame("f", pilosa.FrameOptions{}); err != nil {
t.Fatal(err)
} else if view, err := frame.CreateViewIfNotExists(pilosa.ViewStandard); err != nil {
t.Fatal(err)
} else if _, err := view.SetBit(0, 0); err != nil {
t.Fatal(err)
} else if err := h.Holder.Close(); err != nil {
t.Fatal(err)
} else if err := os.Chmod(filepath.Join(h.Path, "i", "f", "views", "standard", "fragments", "0"), 0000); err != nil {
t.Fatal(err)
}
fmt.Printf("%v\n", h.Schema())
defer os.Chmod(filepath.Join(h.Path, "i", "f", "views", "standard", "fragments", "0"), 0666)
if err := h.Reopen(); err == nil || !strings.Contains(err.Error(), "permission denied") {
t.Fatalf("unexpected error: %s", err)
}
t.Fatalf("STOPPER")
})
}
*/
// Ensure holder can delete an index and its underlying files.
func TestHolder_DeleteIndex(t *testing.T) {
hldr := test.MustOpenHolder()

605
internal/private.pb.go
View file

@ -34,6 +34,7 @@
URI
Node
NodeStateMessage
NodeEventMessage
NodeStatus
ClusterStatus
Field
@ -43,6 +44,7 @@
ResizeSource
ResizeInstructionComplete
SetCoordinatorMessage
UpdateCoordinatorMessage
Topology
RecalculateCaches
*/
@ -742,8 +744,9 @@ func (m *URI) GetPort() uint32 {
}
type Node struct {
ID string `protobuf:"bytes,1,opt,name=ID,proto3" json:"ID,omitempty"`
URI *URI `protobuf:"bytes,2,opt,name=URI" json:"URI,omitempty"`
ID string `protobuf:"bytes,1,opt,name=ID,proto3" json:"ID,omitempty"`
URI *URI `protobuf:"bytes,2,opt,name=URI" json:"URI,omitempty"`
IsCoordinator bool `protobuf:"varint,3,opt,name=IsCoordinator,proto3" json:"IsCoordinator,omitempty"`
}
func (m *Node) Reset() { *m = Node{} }
@ -765,6 +768,13 @@ func (m *Node) GetURI() *URI {
return nil
}
func (m *Node) GetIsCoordinator() bool {
if m != nil {
return m.IsCoordinator
}
return false
}
type NodeStateMessage struct {
NodeID string `protobuf:"bytes,1,opt,name=NodeID,proto3" json:"NodeID,omitempty"`
State string `protobuf:"bytes,2,opt,name=State,proto3" json:"State,omitempty"`
@ -789,6 +799,30 @@ func (m *NodeStateMessage) GetState() string {
return ""
}
type NodeEventMessage struct {
Event uint32 `protobuf:"varint,1,opt,name=Event,proto3" json:"Event,omitempty"`
Node *Node `protobuf:"bytes,2,opt,name=Node" json:"Node,omitempty"`
}
func (m *NodeEventMessage) Reset() { *m = NodeEventMessage{} }
func (m *NodeEventMessage) String() string { return proto.CompactTextString(m) }
func (*NodeEventMessage) ProtoMessage() {}
func (*NodeEventMessage) Descriptor() ([]byte, []int) { return fileDescriptorPrivate, []int{25} }
func (m *NodeEventMessage) GetEvent() uint32 {
if m != nil {
return m.Event
}
return 0
}
func (m *NodeEventMessage) GetNode() *Node {
if m != nil {
return m.Node
}
return nil
}
type NodeStatus struct {
Node *Node `protobuf:"bytes,1,opt,name=Node" json:"Node,omitempty"`
MaxSlices *MaxSlices `protobuf:"bytes,2,opt,name=MaxSlices" json:"MaxSlices,omitempty"`
@ -798,7 +832,7 @@ type NodeStatus struct {
func (m *NodeStatus) Reset() { *m = NodeStatus{} }
func (m *NodeStatus) String() string { return proto.CompactTextString(m) }
func (*NodeStatus) ProtoMessage() {}
func (*NodeStatus) Descriptor() ([]byte, []int) { return fileDescriptorPrivate, []int{25} }
func (*NodeStatus) Descriptor() ([]byte, []int) { return fileDescriptorPrivate, []int{26} }
func (m *NodeStatus) GetNode() *Node {
if m != nil {
@ -830,7 +864,7 @@ type ClusterStatus struct {
func (m *ClusterStatus) Reset() { *m = ClusterStatus{} }
func (m *ClusterStatus) String() string { return proto.CompactTextString(m) }
func (*ClusterStatus) ProtoMessage() {}
func (*ClusterStatus) Descriptor() ([]byte, []int) { return fileDescriptorPrivate, []int{26} }
func (*ClusterStatus) Descriptor() ([]byte, []int) { return fileDescriptorPrivate, []int{27} }
func (m *ClusterStatus) GetClusterID() string {
if m != nil {
@ -863,7 +897,7 @@ type Field struct {
func (m *Field) Reset() { *m = Field{} }
func (m *Field) String() string { return proto.CompactTextString(m) }
func (*Field) ProtoMessage() {}
func (*Field) Descriptor() ([]byte, []int) { return fileDescriptorPrivate, []int{27} }
func (*Field) Descriptor() ([]byte, []int) { return fileDescriptorPrivate, []int{28} }
func (m *Field) GetName() string {
if m != nil {
@ -902,7 +936,7 @@ type CreateViewMessage struct {
func (m *CreateViewMessage) Reset() { *m = CreateViewMessage{} }
func (m *CreateViewMessage) String() string { return proto.CompactTextString(m) }
func (*CreateViewMessage) ProtoMessage() {}
func (*CreateViewMessage) Descriptor() ([]byte, []int) { return fileDescriptorPrivate, []int{28} }
func (*CreateViewMessage) Descriptor() ([]byte, []int) { return fileDescriptorPrivate, []int{29} }
func (m *CreateViewMessage) GetIndex() string {
if m != nil {
@ -934,7 +968,7 @@ type DeleteViewMessage struct {
func (m *DeleteViewMessage) Reset() { *m = DeleteViewMessage{} }
func (m *DeleteViewMessage) String() string { return proto.CompactTextString(m) }
func (*DeleteViewMessage) ProtoMessage() {}
func (*DeleteViewMessage) Descriptor() ([]byte, []int) { return fileDescriptorPrivate, []int{29} }
func (*DeleteViewMessage) Descriptor() ([]byte, []int) { return fileDescriptorPrivate, []int{30} }
func (m *DeleteViewMessage) GetIndex() string {
if m != nil {
@ -969,7 +1003,7 @@ type ResizeInstruction struct {
func (m *ResizeInstruction) Reset() { *m = ResizeInstruction{} }
func (m *ResizeInstruction) String() string { return proto.CompactTextString(m) }
func (*ResizeInstruction) ProtoMessage() {}
func (*ResizeInstruction) Descriptor() ([]byte, []int) { return fileDescriptorPrivate, []int{30} }
func (*ResizeInstruction) Descriptor() ([]byte, []int) { return fileDescriptorPrivate, []int{31} }
func (m *ResizeInstruction) GetJobID() int64 {
if m != nil {
@ -1024,7 +1058,7 @@ type ResizeSource struct {
func (m *ResizeSource) Reset() { *m = ResizeSource{} }
func (m *ResizeSource) String() string { return proto.CompactTextString(m) }
func (*ResizeSource) ProtoMessage() {}
func (*ResizeSource) Descriptor() ([]byte, []int) { return fileDescriptorPrivate, []int{31} }
func (*ResizeSource) Descriptor() ([]byte, []int) { return fileDescriptorPrivate, []int{32} }
func (m *ResizeSource) GetNode() *Node {
if m != nil {
@ -1071,7 +1105,7 @@ func (m *ResizeInstructionComplete) Reset() { *m = ResizeInstructionComp
func (m *ResizeInstructionComplete) String() string { return proto.CompactTextString(m) }
func (*ResizeInstructionComplete) ProtoMessage() {}
func (*ResizeInstructionComplete) Descriptor() ([]byte, []int) {
return fileDescriptorPrivate, []int{32}
return fileDescriptorPrivate, []int{33}
}
func (m *ResizeInstructionComplete) GetJobID() int64 {
@ -1102,7 +1136,7 @@ type SetCoordinatorMessage struct {
func (m *SetCoordinatorMessage) Reset() { *m = SetCoordinatorMessage{} }
func (m *SetCoordinatorMessage) String() string { return proto.CompactTextString(m) }
func (*SetCoordinatorMessage) ProtoMessage() {}
func (*SetCoordinatorMessage) Descriptor() ([]byte, []int) { return fileDescriptorPrivate, []int{33} }
func (*SetCoordinatorMessage) Descriptor() ([]byte, []int) { return fileDescriptorPrivate, []int{34} }
func (m *SetCoordinatorMessage) GetNew() *Node {
if m != nil {
@ -1111,6 +1145,22 @@ func (m *SetCoordinatorMessage) GetNew() *Node {
return nil
}
type UpdateCoordinatorMessage struct {
New *Node `protobuf:"bytes,1,opt,name=New" json:"New,omitempty"`
}
func (m *UpdateCoordinatorMessage) Reset() { *m = UpdateCoordinatorMessage{} }
func (m *UpdateCoordinatorMessage) String() string { return proto.CompactTextString(m) }
func (*UpdateCoordinatorMessage) ProtoMessage() {}
func (*UpdateCoordinatorMessage) Descriptor() ([]byte, []int) { return fileDescriptorPrivate, []int{35} }
func (m *UpdateCoordinatorMessage) GetNew() *Node {
if m != nil {
return m.New
}
return nil
}
type Topology struct {
ClusterID string `protobuf:"bytes,1,opt,name=ClusterID,proto3" json:"ClusterID,omitempty"`
NodeIDs []string `protobuf:"bytes,2,rep,name=NodeIDs" json:"NodeIDs,omitempty"`
@ -1119,7 +1169,7 @@ type Topology struct {
func (m *Topology) Reset() { *m = Topology{} }
func (m *Topology) String() string { return proto.CompactTextString(m) }
func (*Topology) ProtoMessage() {}
func (*Topology) Descriptor() ([]byte, []int) { return fileDescriptorPrivate, []int{34} }
func (*Topology) Descriptor() ([]byte, []int) { return fileDescriptorPrivate, []int{36} }
func (m *Topology) GetClusterID() string {
if m != nil {
@ -1141,7 +1191,7 @@ type RecalculateCaches struct {
func (m *RecalculateCaches) Reset() { *m = RecalculateCaches{} }
func (m *RecalculateCaches) String() string { return proto.CompactTextString(m) }
func (*RecalculateCaches) ProtoMessage() {}
func (*RecalculateCaches) Descriptor() ([]byte, []int) { return fileDescriptorPrivate, []int{35} }
func (*RecalculateCaches) Descriptor() ([]byte, []int) { return fileDescriptorPrivate, []int{37} }
func init() {
proto.RegisterType((*IndexMeta)(nil), "internal.IndexMeta")
@ -1169,6 +1219,7 @@ func init() {
proto.RegisterType((*URI)(nil), "internal.URI")
proto.RegisterType((*Node)(nil), "internal.Node")
proto.RegisterType((*NodeStateMessage)(nil), "internal.NodeStateMessage")
proto.RegisterType((*NodeEventMessage)(nil), "internal.NodeEventMessage")
proto.RegisterType((*NodeStatus)(nil), "internal.NodeStatus")
proto.RegisterType((*ClusterStatus)(nil), "internal.ClusterStatus")
proto.RegisterType((*Field)(nil), "internal.Field")
@ -1178,6 +1229,7 @@ func init() {
proto.RegisterType((*ResizeSource)(nil), "internal.ResizeSource")
proto.RegisterType((*ResizeInstructionComplete)(nil), "internal.ResizeInstructionComplete")
proto.RegisterType((*SetCoordinatorMessage)(nil), "internal.SetCoordinatorMessage")
proto.RegisterType((*UpdateCoordinatorMessage)(nil), "internal.UpdateCoordinatorMessage")
proto.RegisterType((*Topology)(nil), "internal.Topology")
proto.RegisterType((*RecalculateCaches)(nil), "internal.RecalculateCaches")
}
@ -2136,6 +2188,16 @@ func (m *Node) MarshalTo(dAtA []byte) (int, error) {
}
i += n12
}
if m.IsCoordinator {
dAtA[i] = 0x18
i++
if m.IsCoordinator {
dAtA[i] = 1
} else {
dAtA[i] = 0
}
i++
}
return i, nil
}
@ -2169,6 +2231,39 @@ func (m *NodeStateMessage) MarshalTo(dAtA []byte) (int, error) {
return i, nil
}
func (m *NodeEventMessage) Marshal() (dAtA []byte, err error) {
size := m.Size()
dAtA = make([]byte, size)
n, err := m.MarshalTo(dAtA)
if err != nil {
return nil, err
}
return dAtA[:n], nil
}
func (m *NodeEventMessage) MarshalTo(dAtA []byte) (int, error) {
var i int
_ = i
var l int
_ = l
if m.Event != 0 {
dAtA[i] = 0x8
i++
i = encodeVarintPrivate(dAtA, i, uint64(m.Event))
}
if m.Node != nil {
dAtA[i] = 0x12
i++
i = encodeVarintPrivate(dAtA, i, uint64(m.Node.Size()))
n13, err := m.Node.MarshalTo(dAtA[i:])
if err != nil {
return 0, err
}
i += n13
}
return i, nil
}
func (m *NodeStatus) Marshal() (dAtA []byte, err error) {
size := m.Size()
dAtA = make([]byte, size)
@ -2188,32 +2283,32 @@ func (m *NodeStatus) MarshalTo(dAtA []byte) (int, error) {
dAtA[i] = 0xa
i++
i = encodeVarintPrivate(dAtA, i, uint64(m.Node.Size()))
n13, err := m.Node.MarshalTo(dAtA[i:])
if err != nil {
return 0, err
}
i += n13
}
if m.MaxSlices != nil {
dAtA[i] = 0x12
i++
i = encodeVarintPrivate(dAtA, i, uint64(m.MaxSlices.Size()))
n14, err := m.MaxSlices.MarshalTo(dAtA[i:])
n14, err := m.Node.MarshalTo(dAtA[i:])
if err != nil {
return 0, err
}
i += n14
}
if m.Schema != nil {
dAtA[i] = 0x1a
if m.MaxSlices != nil {
dAtA[i] = 0x12
i++
i = encodeVarintPrivate(dAtA, i, uint64(m.Schema.Size()))
n15, err := m.Schema.MarshalTo(dAtA[i:])
i = encodeVarintPrivate(dAtA, i, uint64(m.MaxSlices.Size()))
n15, err := m.MaxSlices.MarshalTo(dAtA[i:])
if err != nil {
return 0, err
}
i += n15
}
if m.Schema != nil {
dAtA[i] = 0x1a
i++
i = encodeVarintPrivate(dAtA, i, uint64(m.Schema.Size()))
n16, err := m.Schema.MarshalTo(dAtA[i:])
if err != nil {
return 0, err
}
i += n16
}
return i, nil
}
@ -2395,21 +2490,21 @@ func (m *ResizeInstruction) MarshalTo(dAtA []byte) (int, error) {
dAtA[i] = 0x12
i++
i = encodeVarintPrivate(dAtA, i, uint64(m.Node.Size()))
n16, err := m.Node.MarshalTo(dAtA[i:])
n17, err := m.Node.MarshalTo(dAtA[i:])
if err != nil {
return 0, err
}
i += n16
i += n17
}
if m.Coordinator != nil {
dAtA[i] = 0x1a
i++
i = encodeVarintPrivate(dAtA, i, uint64(m.Coordinator.Size()))
n17, err := m.Coordinator.MarshalTo(dAtA[i:])
n18, err := m.Coordinator.MarshalTo(dAtA[i:])
if err != nil {
return 0, err
}
i += n17
i += n18
}
if len(m.Sources) > 0 {
for _, msg := range m.Sources {
@ -2427,21 +2522,21 @@ func (m *ResizeInstruction) MarshalTo(dAtA []byte) (int, error) {
dAtA[i] = 0x2a
i++
i = encodeVarintPrivate(dAtA, i, uint64(m.Schema.Size()))
n18, err := m.Schema.MarshalTo(dAtA[i:])
n19, err := m.Schema.MarshalTo(dAtA[i:])
if err != nil {
return 0, err
}
i += n18
i += n19
}
if m.ClusterStatus != nil {
dAtA[i] = 0x32
i++
i = encodeVarintPrivate(dAtA, i, uint64(m.ClusterStatus.Size()))
n19, err := m.ClusterStatus.MarshalTo(dAtA[i:])
n20, err := m.ClusterStatus.MarshalTo(dAtA[i:])
if err != nil {
return 0, err
}
i += n19
i += n20
}
return i, nil
}
@ -2465,11 +2560,11 @@ func (m *ResizeSource) MarshalTo(dAtA []byte) (int, error) {
dAtA[i] = 0xa
i++
i = encodeVarintPrivate(dAtA, i, uint64(m.Node.Size()))
n20, err := m.Node.MarshalTo(dAtA[i:])
n21, err := m.Node.MarshalTo(dAtA[i:])
if err != nil {
return 0, err
}
i += n20
i += n21
}
if len(m.Index) > 0 {
dAtA[i] = 0x12
@ -2521,11 +2616,11 @@ func (m *ResizeInstructionComplete) MarshalTo(dAtA []byte) (int, error) {
dAtA[i] = 0x12
i++
i = encodeVarintPrivate(dAtA, i, uint64(m.Node.Size()))
n21, err := m.Node.MarshalTo(dAtA[i:])
n22, err := m.Node.MarshalTo(dAtA[i:])
if err != nil {
return 0, err
}
i += n21
i += n22
}
if len(m.Error) > 0 {
dAtA[i] = 0x1a
@ -2555,11 +2650,39 @@ func (m *SetCoordinatorMessage) MarshalTo(dAtA []byte) (int, error) {
dAtA[i] = 0xa
i++
i = encodeVarintPrivate(dAtA, i, uint64(m.New.Size()))
n22, err := m.New.MarshalTo(dAtA[i:])
n23, err := m.New.MarshalTo(dAtA[i:])
if err != nil {
return 0, err
}
i += n22
i += n23
}
return i, nil
}
func (m *UpdateCoordinatorMessage) Marshal() (dAtA []byte, err error) {
size := m.Size()
dAtA = make([]byte, size)
n, err := m.MarshalTo(dAtA)
if err != nil {
return nil, err
}
return dAtA[:n], nil
}
func (m *UpdateCoordinatorMessage) MarshalTo(dAtA []byte) (int, error) {
var i int
_ = i
var l int
_ = l
if m.New != nil {
dAtA[i] = 0xa
i++
i = encodeVarintPrivate(dAtA, i, uint64(m.New.Size()))
n24, err := m.New.MarshalTo(dAtA[i:])
if err != nil {
return 0, err
}
i += n24
}
return i, nil
}
@ -3068,6 +3191,9 @@ func (m *Node) Size() (n int) {
l = m.URI.Size()
n += 1 + l + sovPrivate(uint64(l))
}
if m.IsCoordinator {
n += 2
}
return n
}
@ -3085,6 +3211,19 @@ func (m *NodeStateMessage) Size() (n int) {
return n
}
func (m *NodeEventMessage) Size() (n int) {
var l int
_ = l
if m.Event != 0 {
n += 1 + sovPrivate(uint64(m.Event))
}
if m.Node != nil {
l = m.Node.Size()
n += 1 + l + sovPrivate(uint64(l))
}
return n
}
func (m *NodeStatus) Size() (n int) {
var l int
_ = l
@ -3262,6 +3401,16 @@ func (m *SetCoordinatorMessage) Size() (n int) {
return n
}
func (m *UpdateCoordinatorMessage) Size() (n int) {
var l int
_ = l
if m.New != nil {
l = m.New.Size()
n += 1 + l + sovPrivate(uint64(l))
}
return n
}
func (m *Topology) Size() (n int) {
var l int
_ = l
@ -6575,6 +6724,26 @@ func (m *Node) Unmarshal(dAtA []byte) error {
return err
}
iNdEx = postIndex
case 3:
if wireType != 0 {
return fmt.Errorf("proto: wrong wireType = %d for field IsCoordinator", wireType)
}
var v int
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPrivate
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
v |= (int(b) & 0x7F) << shift
if b < 0x80 {
break
}
}
m.IsCoordinator = bool(v != 0)
default:
iNdEx = preIndex
skippy, err := skipPrivate(dAtA[iNdEx:])
@ -6704,6 +6873,108 @@ func (m *NodeStateMessage) Unmarshal(dAtA []byte) error {
}
return nil
}
func (m *NodeEventMessage) Unmarshal(dAtA []byte) error {
l := len(dAtA)
iNdEx := 0
for iNdEx < l {
preIndex := iNdEx
var wire uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPrivate
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
wire |= (uint64(b) & 0x7F) << shift
if b < 0x80 {
break
}
}
fieldNum := int32(wire >> 3)
wireType := int(wire & 0x7)
if wireType == 4 {
return fmt.Errorf("proto: NodeEventMessage: wiretype end group for non-group")
}
if fieldNum <= 0 {
return fmt.Errorf("proto: NodeEventMessage: illegal tag %d (wire type %d)", fieldNum, wire)
}
switch fieldNum {
case 1:
if wireType != 0 {
return fmt.Errorf("proto: wrong wireType = %d for field Event", wireType)
}
m.Event = 0
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPrivate
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
m.Event |= (uint32(b) & 0x7F) << shift
if b < 0x80 {
break
}
}
case 2:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field Node", wireType)
}
var msglen int
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPrivate
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
msglen |= (int(b) & 0x7F) << shift
if b < 0x80 {
break
}
}
if msglen < 0 {
return ErrInvalidLengthPrivate
}
postIndex := iNdEx + msglen
if postIndex > l {
return io.ErrUnexpectedEOF
}
if m.Node == nil {
m.Node = &Node{}
}
if err := m.Node.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
return err
}
iNdEx = postIndex
default:
iNdEx = preIndex
skippy, err := skipPrivate(dAtA[iNdEx:])
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPrivate
}
if (iNdEx + skippy) > l {
return io.ErrUnexpectedEOF
}
iNdEx += skippy
}
}
if iNdEx > l {
return io.ErrUnexpectedEOF
}
return nil
}
func (m *NodeStatus) Unmarshal(dAtA []byte) error {
l := len(dAtA)
iNdEx := 0
@ -8047,6 +8318,89 @@ func (m *SetCoordinatorMessage) Unmarshal(dAtA []byte) error {
}
return nil
}
func (m *UpdateCoordinatorMessage) Unmarshal(dAtA []byte) error {
l := len(dAtA)
iNdEx := 0
for iNdEx < l {
preIndex := iNdEx
var wire uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPrivate
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
wire |= (uint64(b) & 0x7F) << shift
if b < 0x80 {
break
}
}
fieldNum := int32(wire >> 3)
wireType := int(wire & 0x7)
if wireType == 4 {
return fmt.Errorf("proto: UpdateCoordinatorMessage: wiretype end group for non-group")
}
if fieldNum <= 0 {
return fmt.Errorf("proto: UpdateCoordinatorMessage: illegal tag %d (wire type %d)", fieldNum, wire)
}
switch fieldNum {
case 1:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field New", wireType)
}
var msglen int
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPrivate
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
msglen |= (int(b) & 0x7F) << shift
if b < 0x80 {
break
}
}
if msglen < 0 {
return ErrInvalidLengthPrivate
}
postIndex := iNdEx + msglen
if postIndex > l {
return io.ErrUnexpectedEOF
}
if m.New == nil {
m.New = &Node{}
}
if err := m.New.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
return err
}
iNdEx = postIndex
default:
iNdEx = preIndex
skippy, err := skipPrivate(dAtA[iNdEx:])
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPrivate
}
if (iNdEx + skippy) > l {
return io.ErrUnexpectedEOF
}
iNdEx += skippy
}
}
if iNdEx > l {
return io.ErrUnexpectedEOF
}
return nil
}
func (m *Topology) Unmarshal(dAtA []byte) error {
l := len(dAtA)
iNdEx := 0
@ -8313,86 +8667,89 @@ var (
func init() { proto.RegisterFile("private.proto", fileDescriptorPrivate) }
var fileDescriptorPrivate = []byte{
// 1296 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x57, 0x4f, 0x6f, 0x1b, 0x45,
0x14, 0x67, 0xbd, 0xb6, 0x63, 0x3f, 0xc7, 0x89, 0x33, 0x4d, 0x83, 0x13, 0x45, 0xae, 0x19, 0x01,
0x0d, 0x95, 0x88, 0x8a, 0x2b, 0x01, 0x0d, 0xaa, 0x54, 0x12, 0xbb, 0xea, 0x02, 0x09, 0x65, 0x9c,
0x06, 0x89, 0x03, 0xd2, 0xc4, 0x1e, 0xd2, 0x55, 0xd6, 0xbb, 0x66, 0x77, 0x9c, 0xc4, 0x3d, 0x70,
0x44, 0x48, 0x88, 0x3b, 0xe2, 0xca, 0x97, 0xe1, 0xc8, 0x47, 0x40, 0xe1, 0x43, 0x20, 0x71, 0x01,
0xcd, 0xbf, 0xdd, 0xf5, 0xdf, 0x90, 0xd0, 0xdb, 0xbe, 0xdf, 0xfb, 0x33, 0xbf, 0x79, 0xef, 0xcd,
0x9b, 0x59, 0x28, 0xf7, 0x43, 0xf7, 0x8c, 0x72, 0xb6, 0xdd, 0x0f, 0x03, 0x1e, 0xa0, 0x82, 0xeb,
0x73, 0x16, 0xfa, 0xd4, 0xc3, 0x9f, 0x43, 0xd1, 0xf1, 0xbb, 0xec, 0x62, 0x9f, 0x71, 0x8a, 0xea,
0x50, 0xda, 0x0b, 0xbc, 0x41, 0xcf, 0xff, 0x8c, 0x1e, 0x33, 0xaf, 0x6a, 0xd5, 0xad, 0xad, 0x22,
0x49, 0x43, 0xc2, 0xe2, 0xd0, 0xed, 0xb1, 0x2f, 0x06, 0xd4, 0xe7, 0x83, 0x5e, 0x35, 0xa3, 0x2c,
0x52, 0x10, 0xfe, 0xdb, 0x82, 0xe2, 0x93, 0x90, 0xf6, 0x98, 0x8c, 0xb8, 0x01, 0x05, 0x12, 0x9c,
0xa7, 0xc3, 0xc5, 0x32, 0x7a, 0x1b, 0x96, 0x1c, 0xff, 0x8c, 0x85, 0x11, 0x6b, 0xf9, 0xf4, 0xd8,
0x63, 0x5d, 0x19, 0xae, 0x40, 0xc6, 0x50, 0xb4, 0x09, 0xc5, 0x3d, 0xda, 0x79, 0xc1, 0x0e, 0x87,
0x7d, 0x56, 0xb5, 0x65, 0x90, 0x04, 0x88, 0xb5, 0x6d, 0xf7, 0x25, 0xab, 0x66, 0xeb, 0xd6, 0x56,
0x99, 0x24, 0xc0, 0x38, 0xdf, 0xdc, 0x04, 0x5f, 0x84, 0x61, 0x91, 0x50, 0xff, 0x24, 0xe6, 0x90,
0x97, 0x1c, 0x46, 0x30, 0x74, 0x17, 0xf2, 0x4f, 0x5c, 0xe6, 0x75, 0xa3, 0xea, 0x42, 0xdd, 0xde,
0x2a, 0x35, 0x96, 0xb7, 0x4d, 0xfe, 0xb6, 0x25, 0x4e, 0xb4, 0x1a, 0x63, 0x58, 0x72, 0x7a, 0xfd,
0x20, 0xe4, 0x84, 0x45, 0xfd, 0xc0, 0x8f, 0x18, 0xaa, 0x80, 0xdd, 0x0a, 0x43, 0xbd, 0x77, 0xf1,
0x89, 0xbf, 0x83, 0xca, 0xae, 0x17, 0x74, 0x4e, 0x9b, 0x94, 0x53, 0xc2, 0xbe, 0x1d, 0xb0, 0x88,
0xa3, 0x55, 0xc8, 0xc9, 0x2a, 0x68, 0x3b, 0x25, 0x08, 0x54, 0x66, 0x52, 0xa7, 0x59, 0x09, 0x02,
0x95, 0xfe, 0x32, 0x15, 0x59, 0xa2, 0x04, 0x81, 0xb6, 0x3d, 0xb7, 0xa3, 0x52, 0x90, 0x25, 0x4a,
0x40, 0x08, 0xb2, 0x47, 0x2e, 0x3b, 0xd7, 0xfb, 0x96, 0xdf, 0xd8, 0x81, 0x95, 0xd4, 0xfa, 0x9a,
0xe6, 0x1a, 0xe4, 0x49, 0x70, 0xee, 0x34, 0xa3, 0xaa, 0x55, 0xb7, 0xb7, 0xb2, 0x44, 0x4b, 0x32,
0xbb, 0xb2, 0xfc, 0x42, 0x95, 0x91, 0xaa, 0x04, 0xc0, 0xeb, 0x90, 0x93, 0xa9, 0x16, 0xbb, 0x4c,
0x7c, 0xc5, 0x27, 0xfe, 0xc7, 0x82, 0xe2, 0x3e, 0xbd, 0x90, 0x34, 0x22, 0xf4, 0x08, 0x0a, 0x6d,
0x4e, 0xfd, 0x2e, 0x0d, 0xbb, 0xd2, 0xa8, 0xd4, 0x78, 0x23, 0x49, 0x61, 0x6c, 0xb6, 0x6d, 0x6c,
0x5a, 0x3e, 0x0f, 0x87, 0x24, 0x76, 0x41, 0x3b, 0xb0, 0xa0, 0x7b, 0x42, 0x72, 0x28, 0x35, 0xea,
0xd3, 0xbc, 0xe3, 0xb6, 0x11, 0xce, 0xc6, 0x61, 0xe3, 0x23, 0x28, 0x8f, 0x84, 0x15, 0x5c, 0x4f,
0xd9, 0xd0, 0x54, 0xe4, 0x94, 0x0d, 0x45, 0xee, 0xce, 0xa8, 0x37, 0x50, 0x79, 0xce, 0x12, 0x25,
0xec, 0x64, 0x3e, 0xb4, 0x36, 0x76, 0x60, 0x31, 0x1d, 0xf5, 0x3a, 0xbe, 0xf8, 0x6b, 0x40, 0x7b,
0x21, 0xa3, 0x9c, 0x49, 0x7a, 0xfb, 0x2c, 0x8a, 0xe8, 0x09, 0x9b, 0x5d, 0x69, 0x55, 0xbd, 0x4c,
0xba, 0x7a, 0x9b, 0x50, 0x74, 0x22, 0xb3, 0x71, 0x5b, 0xf6, 0x65, 0x02, 0xe0, 0x7b, 0x80, 0x9a,
0xcc, 0x63, 0x9c, 0xe9, 0xf3, 0x3b, 0x27, 0x3e, 0x6e, 0x1b, 0x2e, 0x57, 0xdb, 0xa2, 0xbb, 0x90,
0x15, 0x47, 0x57, 0x52, 0x29, 0x35, 0x6e, 0x25, 0x99, 0x8e, 0xe7, 0x04, 0x91, 0x06, 0xd8, 0x35,
0x41, 0xf5, 0x71, 0xbf, 0x62, 0x83, 0x53, 0x5a, 0xd9, 0x2c, 0x65, 0x8f, 0x2f, 0x15, 0x0f, 0x10,
0xbd, 0xd4, 0x63, 0xb3, 0xd7, 0x9b, 0x2e, 0x85, 0x4f, 0x62, 0xb2, 0xe2, 0xa4, 0xde, 0x84, 0xec,
0x5b, 0x90, 0x93, 0xbe, 0x9a, 0xed, 0xc4, 0x0c, 0x50, 0x5a, 0x7c, 0x14, 0x53, 0xbd, 0xe9, 0x42,
0xab, 0xe9, 0x85, 0x8a, 0x26, 0xee, 0x57, 0xda, 0x56, 0x9c, 0xe9, 0x03, 0xe1, 0xa3, 0x22, 0xc9,
0xef, 0xd9, 0x35, 0x1b, 0x4b, 0xa4, 0x88, 0x2d, 0x86, 0x40, 0x54, 0xb5, 0xeb, 0xb6, 0x88, 0x2d,
0x05, 0xfc, 0x00, 0xf2, 0xed, 0xce, 0x0b, 0xd6, 0xa3, 0xe8, 0x1d, 0x71, 0xd2, 0xba, 0xec, 0x82,
0x45, 0xfa, 0x9c, 0x2e, 0x8f, 0xd5, 0x9f, 0x18, 0x3d, 0xfe, 0xd1, 0xd2, 0x7b, 0x9a, 0xc1, 0x28,
0x2f, 0xd7, 0x8e, 0xaa, 0xd9, 0x89, 0x91, 0x29, 0x70, 0xa2, 0xd5, 0xa8, 0x05, 0x15, 0xc7, 0xef,
0x0f, 0x78, 0x93, 0x7d, 0xe3, 0xfa, 0x2e, 0x77, 0x03, 0x3f, 0xaa, 0xe6, 0xa5, 0xcb, 0x7a, 0x7a,
0xe9, 0x11, 0x0b, 0x32, 0xe1, 0x82, 0xbf, 0xb7, 0x60, 0x79, 0x0c, 0xbc, 0x82, 0x57, 0x66, 0x3e,
0xaf, 0xf7, 0xe3, 0x99, 0x6f, 0x4b, 0xc3, 0xda, 0x4c, 0x36, 0xa3, 0x57, 0xc0, 0xaf, 0x16, 0xac,
0x4e, 0x33, 0x98, 0xca, 0xa6, 0x06, 0xf0, 0x2c, 0x74, 0x7b, 0x34, 0x1c, 0x7e, 0xca, 0x86, 0xfa,
0xfa, 0x4b, 0x21, 0xe8, 0x4b, 0x58, 0x1b, 0x8b, 0xf5, 0x71, 0x47, 0xa5, 0x48, 0x91, 0xba, 0x33,
0x93, 0x94, 0xb2, 0x23, 0x33, 0xdc, 0xf1, 0x5f, 0x16, 0xdc, 0x9e, 0xaa, 0x4a, 0x7a, 0xd2, 0x4a,
0xf7, 0xe4, 0x3d, 0xa8, 0x1c, 0x89, 0xc9, 0xd6, 0x64, 0x11, 0x77, 0x7d, 0x2a, 0x2c, 0x75, 0xd3,
0x4e, 0xe0, 0xc8, 0x81, 0x82, 0xc4, 0xf6, 0x69, 0x5f, 0xd3, 0x7c, 0xf7, 0x0a, 0x9a, 0xdb, 0xc6,
0x5e, 0x0f, 0x7e, 0x23, 0x0a, 0x32, 0xf2, 0x22, 0x32, 0xb7, 0x9a, 0x14, 0xc4, 0x48, 0x1f, 0x71,
0xb8, 0xd6, 0x58, 0x0e, 0x60, 0xd3, 0x8c, 0xc2, 0x11, 0x26, 0xf3, 0x4f, 0xea, 0x43, 0x80, 0xc4,
0x54, 0x4f, 0x80, 0x39, 0xfd, 0x99, 0x32, 0xc6, 0x4f, 0x61, 0xd3, 0xcc, 0xe9, 0x6b, 0x2c, 0x68,
0xba, 0x25, 0x93, 0x74, 0x0b, 0x6e, 0x81, 0xfd, 0x9c, 0x38, 0xe2, 0xae, 0x96, 0xa7, 0xd5, 0x94,
0x48, 0x4b, 0xc2, 0xe5, 0x69, 0x10, 0x71, 0xe3, 0x22, 0xbe, 0x05, 0xf6, 0x2c, 0x08, 0xb9, 0x64,
0x5c, 0x26, 0xf2, 0x1b, 0x7f, 0x00, 0xd9, 0x83, 0xa0, 0xcb, 0xd0, 0x12, 0x64, 0x9c, 0xa6, 0x8e,
0x91, 0x71, 0x9a, 0xe8, 0x8e, 0x0c, 0xaf, 0x67, 0x48, 0x39, 0xd9, 0xdc, 0x73, 0xe2, 0x10, 0xa1,
0xc1, 0x8f, 0xa1, 0x22, 0x1c, 0xdb, 0x9c, 0xf2, 0x78, 0x06, 0xaf, 0x41, 0x5e, 0x60, 0x71, 0x20,
0x2d, 0xc9, 0x1b, 0x4d, 0xd8, 0x99, 0xd1, 0x26, 0x05, 0xfc, 0x93, 0x05, 0x60, 0x42, 0x0c, 0x22,
0x84, 0x15, 0x13, 0xe9, 0x5a, 0x6a, 0x2c, 0x25, 0x4b, 0x0a, 0x94, 0x28, 0x96, 0xef, 0xa5, 0xde,
0x11, 0x93, 0xf3, 0x2d, 0x56, 0x91, 0xd4, 0x6b, 0x63, 0xcb, 0x8c, 0x33, 0x5d, 0xa8, 0x4a, 0x62,
0xaf, 0x70, 0x9d, 0x32, 0x71, 0x85, 0x95, 0xf7, 0xbc, 0x41, 0xc4, 0x59, 0xa8, 0x19, 0x89, 0xf7,
0x8e, 0x02, 0xe2, 0x1d, 0x25, 0xc0, 0xf4, 0x4d, 0xa1, 0x37, 0x21, 0x27, 0x98, 0x9a, 0x33, 0x39,
0xbe, 0x0d, 0xa5, 0xc4, 0x6d, 0x3d, 0xd5, 0xa7, 0xce, 0x01, 0x04, 0x59, 0xf9, 0xba, 0xd5, 0xa5,
0x93, 0x0f, 0xdb, 0x0a, 0xd8, 0xfb, 0xae, 0xea, 0x35, 0x9b, 0x88, 0x4f, 0x89, 0xd0, 0x0b, 0x79,
0x16, 0x04, 0x42, 0xc5, 0xbd, 0xbe, 0xa2, 0x9a, 0x59, 0xcc, 0xf1, 0x9b, 0xdc, 0x35, 0xe6, 0x81,
0x68, 0xa7, 0x1e, 0x88, 0x6d, 0x58, 0x51, 0x0d, 0xfb, 0x2a, 0x83, 0xfe, 0x92, 0x81, 0x15, 0xc2,
0x22, 0xf7, 0x25, 0x73, 0xfc, 0x88, 0x87, 0x83, 0x78, 0xd8, 0x7c, 0x12, 0x1c, 0xeb, 0x54, 0xdb,
0x44, 0x09, 0x71, 0x5b, 0x64, 0xe6, 0xb4, 0xc5, 0x7d, 0xf1, 0xab, 0x12, 0x84, 0x5d, 0x31, 0x74,
0x82, 0x50, 0x17, 0x7a, 0xdc, 0x34, 0x6d, 0x82, 0xee, 0xc3, 0x42, 0x3b, 0x18, 0x84, 0x9d, 0xf8,
0x4a, 0x5a, 0x4b, 0xac, 0x15, 0x33, 0xa5, 0x26, 0xc6, 0x2c, 0xd5, 0x47, 0xb9, 0xf9, 0x7d, 0x84,
0x1e, 0x8d, 0xf5, 0x91, 0xfc, 0x8b, 0x28, 0x35, 0x5e, 0x4f, 0x1c, 0x46, 0xd4, 0x64, 0xd4, 0x1a,
0xff, 0x60, 0xc1, 0x62, 0x9a, 0xc2, 0x7f, 0x3a, 0x18, 0x71, 0x45, 0x32, 0x53, 0x2b, 0x62, 0x4f,
0xab, 0x48, 0x36, 0xa9, 0x48, 0xf2, 0xe6, 0xcc, 0xa5, 0xde, 0x9c, 0xf8, 0x14, 0xd6, 0x27, 0xca,
0xb4, 0x17, 0xf4, 0xfa, 0xa2, 0x1f, 0xfe, 0x47, 0xb9, 0x56, 0x21, 0xd7, 0x0a, 0x43, 0x5d, 0xa8,
0x22, 0x51, 0x02, 0x7e, 0x08, 0xb7, 0xdb, 0x8c, 0xa7, 0x8a, 0x64, 0xba, 0xad, 0x0e, 0xf6, 0x01,
0x3b, 0x9f, 0xb1, 0x7d, 0xa1, 0xc2, 0xbb, 0x50, 0x38, 0x0c, 0xfa, 0x81, 0x17, 0x9c, 0x0c, 0xaf,
0x38, 0xb4, 0x55, 0x58, 0x50, 0x33, 0x49, 0x5d, 0xf9, 0x45, 0x62, 0x44, 0x7c, 0x4b, 0xb4, 0x64,
0x87, 0x7a, 0x9d, 0x81, 0x47, 0x39, 0x93, 0x7f, 0x32, 0xd1, 0x6e, 0xe5, 0xb7, 0xcb, 0x9a, 0xf5,
0xfb, 0x65, 0xcd, 0xfa, 0xe3, 0xb2, 0x66, 0xfd, 0xfc, 0x67, 0xed, 0xb5, 0xe3, 0xbc, 0xfc, 0x67,
0x7e, 0xf0, 0x6f, 0x00, 0x00, 0x00, 0xff, 0xff, 0x14, 0x23, 0x92, 0x89, 0x44, 0x0f, 0x00, 0x00,
// 1334 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x57, 0x5d, 0x6f, 0x1b, 0x45,
0x17, 0x7e, 0xd7, 0x6b, 0x3b, 0xf6, 0x71, 0x9c, 0x38, 0xd3, 0x34, 0xaf, 0x13, 0x45, 0xae, 0x19,
0x15, 0x1a, 0x2a, 0x11, 0x95, 0x54, 0x42, 0x34, 0x50, 0xa9, 0xc4, 0x76, 0xd5, 0x85, 0x26, 0x94,
0x71, 0x12, 0x24, 0x24, 0x90, 0x26, 0xf6, 0x90, 0xae, 0xb2, 0xde, 0x35, 0xbb, 0xe3, 0x24, 0xee,
0x05, 0x97, 0x08, 0x09, 0x71, 0x8f, 0xb8, 0xe5, 0xcf, 0x70, 0xc9, 0x4f, 0x40, 0xe1, 0x47, 0x20,
0x71, 0x03, 0x9a, 0xaf, 0xdd, 0xf5, 0x57, 0xd2, 0x04, 0xee, 0xf6, 0x3c, 0x73, 0xce, 0x99, 0x67,
0xce, 0xd7, 0xcc, 0x42, 0xb9, 0x1f, 0xba, 0xa7, 0x94, 0xb3, 0xcd, 0x7e, 0x18, 0xf0, 0x00, 0x15,
0x5c, 0x9f, 0xb3, 0xd0, 0xa7, 0x1e, 0xfe, 0x14, 0x8a, 0x8e, 0xdf, 0x65, 0xe7, 0xbb, 0x8c, 0x53,
0x54, 0x87, 0x52, 0x23, 0xf0, 0x06, 0x3d, 0xff, 0x39, 0x3d, 0x62, 0x5e, 0xd5, 0xaa, 0x5b, 0x1b,
0x45, 0x92, 0x86, 0x84, 0xc6, 0xbe, 0xdb, 0x63, 0x9f, 0x0d, 0xa8, 0xcf, 0x07, 0xbd, 0x6a, 0x46,
0x69, 0xa4, 0x20, 0xfc, 0x97, 0x05, 0xc5, 0xa7, 0x21, 0xed, 0x31, 0xe9, 0x71, 0x0d, 0x0a, 0x24,
0x38, 0x4b, 0xbb, 0x8b, 0x65, 0xf4, 0x16, 0x2c, 0x38, 0xfe, 0x29, 0x0b, 0x23, 0xd6, 0xf2, 0xe9,
0x91, 0xc7, 0xba, 0xd2, 0x5d, 0x81, 0x8c, 0xa1, 0x68, 0x1d, 0x8a, 0x0d, 0xda, 0x79, 0xc9, 0xf6,
0x87, 0x7d, 0x56, 0xb5, 0xa5, 0x93, 0x04, 0x88, 0x57, 0xdb, 0xee, 0x2b, 0x56, 0xcd, 0xd6, 0xad,
0x8d, 0x32, 0x49, 0x80, 0x71, 0xbe, 0xb9, 0x09, 0xbe, 0x08, 0xc3, 0x3c, 0xa1, 0xfe, 0x71, 0xcc,
0x21, 0x2f, 0x39, 0x8c, 0x60, 0xe8, 0x1e, 0xe4, 0x9f, 0xba, 0xcc, 0xeb, 0x46, 0xd5, 0xb9, 0xba,
0xbd, 0x51, 0xda, 0x5a, 0xdc, 0x34, 0xf1, 0xdb, 0x94, 0x38, 0xd1, 0xcb, 0x18, 0xc3, 0x82, 0xd3,
0xeb, 0x07, 0x21, 0x27, 0x2c, 0xea, 0x07, 0x7e, 0xc4, 0x50, 0x05, 0xec, 0x56, 0x18, 0xea, 0xb3,
0x8b, 0x4f, 0xfc, 0x2d, 0x54, 0x76, 0xbc, 0xa0, 0x73, 0xd2, 0xa4, 0x9c, 0x12, 0xf6, 0xcd, 0x80,
0x45, 0x1c, 0x2d, 0x43, 0x4e, 0x66, 0x41, 0xeb, 0x29, 0x41, 0xa0, 0x32, 0x92, 0x3a, 0xcc, 0x4a,
0x10, 0xa8, 0xb4, 0x97, 0xa1, 0xc8, 0x12, 0x25, 0x08, 0xb4, 0xed, 0xb9, 0x1d, 0x15, 0x82, 0x2c,
0x51, 0x02, 0x42, 0x90, 0x3d, 0x74, 0xd9, 0x99, 0x3e, 0xb7, 0xfc, 0xc6, 0x0e, 0x2c, 0xa5, 0xf6,
0xd7, 0x34, 0x57, 0x20, 0x4f, 0x82, 0x33, 0xa7, 0x19, 0x55, 0xad, 0xba, 0xbd, 0x91, 0x25, 0x5a,
0x92, 0xd1, 0x95, 0xe9, 0x17, 0x4b, 0x19, 0xb9, 0x94, 0x00, 0x78, 0x15, 0x72, 0x32, 0xd4, 0xe2,
0x94, 0x89, 0xad, 0xf8, 0xc4, 0x7f, 0x5b, 0x50, 0xdc, 0xa5, 0xe7, 0x92, 0x46, 0x84, 0x1e, 0x43,
0xa1, 0xcd, 0xa9, 0xdf, 0xa5, 0x61, 0x57, 0x2a, 0x95, 0xb6, 0xde, 0x48, 0x42, 0x18, 0xab, 0x6d,
0x1a, 0x9d, 0x96, 0xcf, 0xc3, 0x21, 0x89, 0x4d, 0xd0, 0x36, 0xcc, 0xe9, 0x9a, 0x90, 0x1c, 0x4a,
0x5b, 0xf5, 0x69, 0xd6, 0x71, 0xd9, 0x08, 0x63, 0x63, 0xb0, 0xf6, 0x01, 0x94, 0x47, 0xdc, 0x0a,
0xae, 0x27, 0x6c, 0x68, 0x32, 0x72, 0xc2, 0x86, 0x22, 0x76, 0xa7, 0xd4, 0x1b, 0xa8, 0x38, 0x67,
0x89, 0x12, 0xb6, 0x33, 0xef, 0x5b, 0x6b, 0xdb, 0x30, 0x9f, 0xf6, 0x7a, 0x1d, 0x5b, 0xfc, 0x15,
0xa0, 0x46, 0xc8, 0x28, 0x67, 0x92, 0xde, 0x2e, 0x8b, 0x22, 0x7a, 0xcc, 0x66, 0x67, 0x5a, 0x65,
0x2f, 0x93, 0xce, 0xde, 0x3a, 0x14, 0x9d, 0xc8, 0x1c, 0xdc, 0x96, 0x75, 0x99, 0x00, 0xf8, 0x3e,
0xa0, 0x26, 0xf3, 0x18, 0x67, 0xba, 0x7f, 0x2f, 0xf1, 0x8f, 0xdb, 0x86, 0xcb, 0xd5, 0xba, 0xe8,
0x1e, 0x64, 0x45, 0xeb, 0x4a, 0x2a, 0xa5, 0xad, 0x5b, 0x49, 0xa4, 0xe3, 0x39, 0x41, 0xa4, 0x02,
0x76, 0x8d, 0x53, 0xdd, 0xee, 0x57, 0x1c, 0x70, 0x4a, 0x29, 0x9b, 0xad, 0xec, 0xf1, 0xad, 0xe2,
0x01, 0xa2, 0xb7, 0x7a, 0x62, 0xce, 0x7a, 0xd3, 0xad, 0xf0, 0x71, 0x4c, 0x56, 0x74, 0xea, 0x4d,
0xc8, 0xbe, 0x09, 0x39, 0x69, 0xab, 0xd9, 0x4e, 0xcc, 0x00, 0xb5, 0x8a, 0x0f, 0x63, 0xaa, 0x37,
0xdd, 0x68, 0x39, 0xbd, 0x51, 0xd1, 0xf8, 0xfd, 0x42, 0xeb, 0x8a, 0x9e, 0xde, 0x13, 0x36, 0xca,
0x93, 0xfc, 0x9e, 0x9d, 0xb3, 0xb1, 0x40, 0x0a, 0xdf, 0x62, 0x08, 0x44, 0x55, 0xbb, 0x6e, 0x0b,
0xdf, 0x52, 0xc0, 0x0f, 0x21, 0xdf, 0xee, 0xbc, 0x64, 0x3d, 0x8a, 0xde, 0x16, 0x9d, 0xd6, 0x65,
0xe7, 0x2c, 0xd2, 0x7d, 0xba, 0x38, 0x96, 0x7f, 0x62, 0xd6, 0xf1, 0x0f, 0x96, 0x3e, 0xd3, 0x0c,
0x46, 0x79, 0xb9, 0x77, 0x54, 0xcd, 0x4e, 0x8c, 0x4c, 0x81, 0x13, 0xbd, 0x8c, 0x5a, 0x50, 0x71,
0xfc, 0xfe, 0x80, 0x37, 0xd9, 0xd7, 0xae, 0xef, 0x72, 0x37, 0xf0, 0xa3, 0x6a, 0x5e, 0x9a, 0xac,
0xa6, 0xb7, 0x1e, 0xd1, 0x20, 0x13, 0x26, 0xf8, 0x3b, 0x0b, 0x16, 0xc7, 0xc0, 0x2b, 0x78, 0x65,
0x2e, 0xe7, 0xf5, 0x5e, 0x3c, 0xf3, 0x6d, 0xa9, 0x58, 0x9b, 0xc9, 0x66, 0xf4, 0x0a, 0xf8, 0xc5,
0x82, 0xe5, 0x69, 0x0a, 0x53, 0xd9, 0xd4, 0x00, 0x5e, 0x84, 0x6e, 0x8f, 0x86, 0xc3, 0x4f, 0xd8,
0x50, 0x5f, 0x7f, 0x29, 0x04, 0x7d, 0x0e, 0x2b, 0x63, 0xbe, 0x3e, 0xea, 0xa8, 0x10, 0x29, 0x52,
0x77, 0x66, 0x92, 0x52, 0x7a, 0x64, 0x86, 0x39, 0xfe, 0xd3, 0x82, 0xdb, 0x53, 0x97, 0x92, 0x9a,
0xb4, 0xd2, 0x35, 0x79, 0x1f, 0x2a, 0x87, 0x62, 0xb2, 0x35, 0x59, 0xc4, 0x5d, 0x9f, 0x0a, 0x4d,
0x5d, 0xb4, 0x13, 0x38, 0x72, 0xa0, 0x20, 0xb1, 0x5d, 0xda, 0xd7, 0x34, 0xdf, 0xb9, 0x82, 0xe6,
0xa6, 0xd1, 0xd7, 0x83, 0xdf, 0x88, 0x82, 0x8c, 0xbc, 0x88, 0xcc, 0xad, 0x26, 0x05, 0x31, 0xd2,
0x47, 0x0c, 0xae, 0x35, 0x96, 0x03, 0x58, 0x37, 0xa3, 0x70, 0x84, 0xc9, 0xe5, 0x9d, 0xfa, 0x08,
0x20, 0x51, 0xd5, 0x13, 0xe0, 0x92, 0xfa, 0x4c, 0x29, 0xe3, 0x67, 0xb0, 0x6e, 0xe6, 0xf4, 0x35,
0x36, 0x34, 0xd5, 0x92, 0x49, 0xaa, 0x05, 0xb7, 0xc0, 0x3e, 0x20, 0x8e, 0xb8, 0xab, 0x65, 0xb7,
0x9a, 0x14, 0x69, 0x49, 0x98, 0x3c, 0x0b, 0x22, 0x6e, 0x4c, 0xc4, 0xb7, 0xc0, 0x5e, 0x04, 0x21,
0x97, 0x8c, 0xcb, 0x44, 0x7e, 0xe3, 0x2f, 0x21, 0xbb, 0x17, 0x74, 0x19, 0x5a, 0x80, 0x8c, 0xd3,
0xd4, 0x3e, 0x32, 0x4e, 0x13, 0xdd, 0x91, 0xee, 0xf5, 0x0c, 0x29, 0x27, 0x87, 0x3b, 0x20, 0x0e,
0x91, 0x1b, 0xdf, 0x85, 0xb2, 0x13, 0x35, 0x82, 0x20, 0xec, 0x8a, 0x54, 0x07, 0xa1, 0xbe, 0x93,
0x46, 0x41, 0xfc, 0x04, 0x2a, 0xc2, 0x7d, 0x9b, 0x53, 0x1e, 0x4f, 0xea, 0x15, 0xc8, 0x0b, 0x2c,
0xde, 0x4e, 0x4b, 0xf2, 0xde, 0x13, 0x7a, 0x66, 0x00, 0x4a, 0x01, 0x3f, 0x57, 0x1e, 0x5a, 0xa7,
0xcc, 0xe7, 0xa9, 0x28, 0x49, 0x59, 0x3a, 0x28, 0x13, 0x25, 0x20, 0xac, 0x8e, 0xa2, 0x39, 0x2f,
0x24, 0x9c, 0x05, 0x4a, 0xe4, 0x1a, 0xfe, 0xd1, 0x02, 0x30, 0x84, 0x06, 0x51, 0x6c, 0x62, 0xcd,
0x36, 0x41, 0xef, 0xa6, 0xde, 0x2e, 0x93, 0x33, 0x35, 0x5e, 0x22, 0xa9, 0x17, 0xce, 0x86, 0x19,
0xa1, 0xba, 0x38, 0x2a, 0x89, 0xbe, 0xc2, 0x75, 0x9a, 0xc4, 0xb5, 0x59, 0x6e, 0x78, 0x83, 0x88,
0xb3, 0x50, 0x33, 0x12, 0x6f, 0x2c, 0x05, 0xc4, 0xf1, 0x49, 0x80, 0xe9, 0x21, 0x42, 0x77, 0x21,
0x27, 0x98, 0x9a, 0x39, 0x30, 0x7e, 0x0c, 0xb5, 0x88, 0xdb, 0xfa, 0x26, 0x99, 0x3a, 0x7b, 0x10,
0x64, 0xe5, 0x8b, 0x5a, 0x97, 0x8b, 0x7c, 0x4c, 0x57, 0xc0, 0xde, 0x75, 0x55, 0x7d, 0xdb, 0x44,
0x7c, 0x4a, 0x84, 0x9e, 0xcb, 0xfe, 0x13, 0x08, 0x15, 0x6f, 0x89, 0x25, 0xd5, 0x40, 0xe2, 0xee,
0xb8, 0xc9, 0xfd, 0x66, 0x1e, 0xa5, 0x76, 0xea, 0x51, 0xda, 0x86, 0x25, 0xd5, 0x24, 0xff, 0xa5,
0xd3, 0x9f, 0x33, 0xb0, 0x44, 0x58, 0xe4, 0xbe, 0x62, 0x8e, 0x1f, 0xf1, 0x70, 0x10, 0x0f, 0xb8,
0x8f, 0x83, 0x23, 0x1d, 0x6a, 0x9b, 0x28, 0xe1, 0x75, 0x2a, 0x09, 0x3d, 0x10, 0xbf, 0x47, 0xa3,
0xd5, 0x3f, 0xa9, 0x9a, 0x56, 0x41, 0x0f, 0x60, 0xae, 0x1d, 0x0c, 0xc2, 0x4e, 0x7c, 0x0d, 0xae,
0x24, 0xda, 0x8a, 0x99, 0x5a, 0x26, 0x46, 0x2d, 0x55, 0x47, 0xb9, 0xcb, 0xeb, 0x08, 0x3d, 0x1e,
0xab, 0x23, 0xf9, 0xe7, 0x52, 0xda, 0xfa, 0x7f, 0x62, 0x30, 0xb2, 0x4c, 0x46, 0xb5, 0xf1, 0xf7,
0x16, 0xcc, 0xa7, 0x29, 0xbc, 0x56, 0x63, 0xc4, 0x19, 0xc9, 0x4c, 0xcd, 0x88, 0x3d, 0x2d, 0x23,
0xd9, 0x24, 0x23, 0xc9, 0x3b, 0x37, 0x97, 0x7a, 0xe7, 0xe2, 0x13, 0x58, 0x9d, 0x48, 0x53, 0x23,
0xe8, 0xf5, 0x45, 0x3d, 0xfc, 0x8b, 0x74, 0x89, 0x91, 0x11, 0x86, 0x3a, 0x51, 0x45, 0xa2, 0x04,
0xfc, 0x08, 0x6e, 0xb7, 0x19, 0x4f, 0x25, 0xc9, 0x54, 0x5b, 0x1d, 0xec, 0x3d, 0x76, 0x36, 0xe3,
0xf8, 0x62, 0x09, 0x7f, 0x08, 0xd5, 0x83, 0x7e, 0x97, 0x72, 0x76, 0x23, 0xeb, 0x1d, 0x28, 0xec,
0x07, 0xfd, 0xc0, 0x0b, 0x8e, 0x87, 0x57, 0xb4, 0x7c, 0x15, 0xe6, 0xd4, 0x7c, 0x54, 0x8f, 0x94,
0x22, 0x31, 0x22, 0xbe, 0x25, 0x0a, 0xba, 0x43, 0xbd, 0xce, 0xc0, 0x13, 0x34, 0xc4, 0xbf, 0x57,
0xb4, 0x53, 0xf9, 0xf5, 0xa2, 0x66, 0xfd, 0x76, 0x51, 0xb3, 0x7e, 0xbf, 0xa8, 0x59, 0x3f, 0xfd,
0x51, 0xfb, 0xdf, 0x51, 0x5e, 0xfe, 0xe5, 0x3f, 0xfc, 0x27, 0x00, 0x00, 0xff, 0xff, 0x66, 0x19,
0x3d, 0xd2, 0xf6, 0x0f, 0x00, 0x00,
}

10
internal/private.proto
View file

@ -136,6 +136,7 @@ message URI {
message Node {
string ID = 1;
URI URI = 2;
bool IsCoordinator = 3;
}
message NodeStateMessage {
@ -143,6 +144,11 @@ message NodeStateMessage {
string State = 2;
}
message NodeEventMessage {
uint32 Event = 1;
Node Node = 2;
}
message NodeStatus {
Node Node = 1;
MaxSlices MaxSlices = 2;
@ -201,6 +207,10 @@ message SetCoordinatorMessage {
Node New = 1;
}
message UpdateCoordinatorMessage {
Node New = 1;
}
message Topology {
string ClusterID = 1;
repeated string NodeIDs = 2;

14
security_manager.go
View file

@ -1,3 +1,17 @@
// Copyright 2017 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package pilosa
// SecurityManager provides the ability to limit access to restricted endpoints

73
server.go
View file

@ -32,7 +32,6 @@ import (
"time"
"github.com/gogo/protobuf/proto"
"github.com/pilosa/pilosa/diagnostics"
"github.com/pilosa/pilosa/internal"
"golang.org/x/sync/errgroup"
@ -70,7 +69,8 @@ type Server struct {
NodeID string
URI URI
Cluster *Cluster
diagnostics *diagnostics.Diagnostics
diagnostics *DiagnosticsCollector
SystemInfo SystemInfo
GCNotifier GCNotifier
@ -100,7 +100,8 @@ func NewServer() *Server {
Handler: NewHandler(),
Broadcaster: NopBroadcaster,
BroadcastReceiver: NopBroadcastReceiver,
diagnostics: diagnostics.New(DefaultDiagnosticServer),
diagnostics: NewDiagnosticsCollector(DefaultDiagnosticServer),
SystemInfo: NewNopSystemInfo(),
Network: "tcp",
@ -115,6 +116,7 @@ func NewServer() *Server {
s.logger = log.New(s.LogOutput, "", log.LstdFlags)
s.Handler.Holder = s.Holder
s.diagnostics.server = s
return s
}
@ -132,7 +134,11 @@ func (s *Server) Open() error {
s.NodeID = s.LoadNodeID()
// Set Cluster Node.
node := &Node{ID: s.NodeID, URI: s.URI}
node := &Node{
ID: s.NodeID,
URI: s.URI,
IsCoordinator: s.Cluster.Coordinator == s.NodeID,
}
s.Cluster.Node = node
// Append the NodeID tag to stats.
@ -185,11 +191,6 @@ func (s *Server) Open() error {
return fmt.Errorf("starting BroadcastReceiver: %v", err)
}
// If a Coordinator is not specified, then default to s.URI.
if s.Cluster.Coordinator.Port() == 0 {
s.Cluster.Coordinator = s.URI
}
// Open Cluster management.
if err := s.Cluster.Open(); err != nil {
return fmt.Errorf("opening Cluster: %v", err)
@ -457,6 +458,8 @@ func (s *Server) ReceiveMessage(pb proto.Message) error {
}
case *internal.SetCoordinatorMessage:
s.Cluster.SetCoordinator(DecodeNode(obj.New))
case *internal.UpdateCoordinatorMessage:
s.Cluster.UpdateCoordinator(DecodeNode(obj.New))
case *internal.NodeStateMessage:
err := s.Cluster.ReceiveNodeState(obj.NodeID, obj.State)
if err != nil {
@ -464,6 +467,8 @@ func (s *Server) ReceiveMessage(pb proto.Message) error {
}
case *internal.RecalculateCaches:
s.Holder.RecalculateCaches()
case *internal.NodeEventMessage:
s.Cluster.ReceiveEvent(DecodeNodeEvent(obj))
}
return nil
@ -600,15 +605,16 @@ func (s *Server) mergeRemoteStatus(ns *internal.NodeStatus) error {
// monitorDiagnostics periodically polls the Pilosa Indexes for cluster info.
func (s *Server) monitorDiagnostics() {
if s.DiagnosticInterval <= 0 {
// Do not send more than once a minute
if s.DiagnosticInterval < time.Minute {
s.Logger().Printf("diagnostics disabled")
return
} else {
s.Logger().Printf("Pilosa is currently configured to send small diagnostics reports to our team every %v. More information here: https://www.pilosa.com/docs/latest/administration/#diagnostics", s.DiagnosticInterval)
}
s.diagnostics.SetLogger(s.LogOutput)
s.diagnostics.SetVersion(Version)
s.diagnostics.SetInterval(s.DiagnosticInterval)
s.diagnostics.Open()
s.diagnostics.Set("Host", s.URI.host)
s.diagnostics.Set("Cluster", strings.Join(s.Cluster.NodeIDs(), ","))
s.diagnostics.Set("NumNodes", len(s.Cluster.Nodes))
@ -619,15 +625,18 @@ func (s *Server) monitorDiagnostics() {
// Flush the diagnostics metrics at startup, then on each tick interval
flush := func() {
enrichDiagnosticsWithSchemaProperties(s.diagnostics, s.Holder)
openFiles, err := CountOpenFiles()
if err == nil {
s.diagnostics.Set("OpenFiles", openFiles)
}
s.diagnostics.Set("GoRoutines", runtime.NumGoroutine())
s.diagnostics.EnrichWithMemoryInfo()
s.diagnostics.EnrichWithSchemaProperties()
s.diagnostics.CheckVersion()
s.diagnostics.Flush()
err = s.diagnostics.Flush()
if err != nil {
s.Logger().Printf("Diagnostics error: %s", err)
}
}
ticker := time.NewTicker(s.DiagnosticInterval)
@ -722,39 +731,3 @@ type StatusHandler interface {
ClusterStatus() (proto.Message, error)
HandleRemoteStatus(proto.Message) error
}
type diagnosticsFrameProperties struct {
BSIFieldCount int
TimeQuantumEnabled bool
}
func enrichDiagnosticsWithSchemaProperties(d *diagnostics.Diagnostics, holder *Holder) {
// NOTE: this function is not in the diagnostics package, since circular imports are not allowed.
var numSlices uint64
numFrames := 0
numIndexes := 0
bsiFieldCount := 0
timeQuantumEnabled := false
for _, index := range holder.Indexes() {
numSlices += index.MaxSlice() + 1
numIndexes += 1
for _, frame := range index.Frames() {
numFrames += 1
if frame.rangeEnabled {
if fields, err := frame.GetFields(); err == nil {
bsiFieldCount += len(fields)
}
}
if frame.TimeQuantum() != "" {
timeQuantumEnabled = true
}
}
}
d.Set("NumIndexes", numIndexes)
d.Set("NumFrames", numFrames)
d.Set("NumSlices", numSlices)
d.Set("BSIFieldCount", bsiFieldCount)
d.Set("TimeQuantumEnabled", timeQuantumEnabled)
}

89
server/cluster_test.go
View file

@ -52,7 +52,7 @@ func TestMain_SendReceiveMessage(t *testing.T) {
m0.Config.Gossip.Port = "0"
m0.Config.Gossip.Seeds = []string{}
m0.Server.Cluster.Coordinator = m0.Server.URI
m0.Server.Cluster.Coordinator = m0.Server.NodeID
m0.Server.Cluster.Topology = &pilosa.Topology{NodeIDs: []string{m0.Server.NodeID, m1.Server.NodeID}}
m0.Server.Cluster.EventReceiver = gossip.NewGossipEventReceiver(m0.Server.LogOutput)
gossipMemberSet0, err := gossip.NewGossipMemberSet(m0.Server.URI.HostPort(), m0.Config, m0.Server)
@ -80,7 +80,7 @@ func TestMain_SendReceiveMessage(t *testing.T) {
m1.Config.Gossip.Port = "0"
m1.Config.Gossip.Seeds = gossipMemberSet0.Seeds()
m1.Server.Cluster.Coordinator = m0.Server.URI
m1.Server.Cluster.Coordinator = m0.Server.NodeID
m1.Server.Cluster.EventReceiver = gossip.NewGossipEventReceiver(m1.Server.LogOutput)
gossipMemberSet1, err := gossip.NewGossipMemberSet(m1.Server.URI.HostPort(), m1.Config, m1.Server)
if err != nil {
@ -220,21 +220,21 @@ func TestClusterResize_EmptyNode(t *testing.T) {
// Ensure that a cluster of empty nodes comes up in a NORMAL state.
func TestClusterResize_EmptyNodes(t *testing.T) {
// Configure node0
m0 := test.NewMainWithCluster()
m0 := test.NewMainWithCluster(true)
defer m0.Close()
gossipHost := "localhost"
gossipPort := 0
seed, coord, err := m0.RunWithTransport(gossipHost, gossipPort, []string{}, pilosa.URI{})
seed, err := m0.RunWithTransport(gossipHost, gossipPort, []string{})
if err != nil {
t.Fatal(err)
}
// Configure node1
m1 := test.NewMainWithCluster()
m1 := test.NewMainWithCluster(false)
defer m1.Close()
seed, coord, err = m1.RunWithTransport(gossipHost, gossipPort, []string{seed}, coord)
seed, err = m1.RunWithTransport(gossipHost, gossipPort, []string{seed})
if err != nil {
t.Fatal(err)
}
@ -250,21 +250,21 @@ func TestClusterResize_EmptyNodes(t *testing.T) {
func TestClusterResize_AddNode(t *testing.T) {
t.Run("NoData", func(t *testing.T) {
// Configure node0
m0 := test.NewMainWithCluster()
m0 := test.NewMainWithCluster(true)
defer m0.Close()
seed, coord, err := m0.RunWithTransport("localhost", 0, []string{}, pilosa.URI{})
seed, err := m0.RunWithTransport("localhost", 0, []string{})
if err != nil {
t.Fatal(err)
}
// Configure node1
m1 := test.NewMainWithCluster()
m1 := test.NewMainWithCluster(false)
defer m1.Close()
var eg errgroup.Group
eg.Go(func() error {
_, _, err = m1.RunWithTransport("localhost", 0, []string{seed}, coord)
_, err = m1.RunWithTransport("localhost", 0, []string{seed})
if err != nil {
return err
}
@ -284,10 +284,10 @@ func TestClusterResize_AddNode(t *testing.T) {
})
t.Run("WithIndex", func(t *testing.T) {
// Configure node0
m0 := test.NewMainWithCluster()
m0 := test.NewMainWithCluster(true)
defer m0.Close()
seed, coord, err := m0.RunWithTransport("localhost", 0, []string{}, pilosa.URI{})
seed, err := m0.RunWithTransport("localhost", 0, []string{})
if err != nil {
t.Fatal(err)
}
@ -303,12 +303,12 @@ func TestClusterResize_AddNode(t *testing.T) {
}
// Configure node1
m1 := test.NewMainWithCluster()
m1 := test.NewMainWithCluster(false)
defer m1.Close()
var eg errgroup.Group
eg.Go(func() error {
_, _, err = m1.RunWithTransport("localhost", 0, []string{seed}, coord)
_, err = m1.RunWithTransport("localhost", 0, []string{seed})
if err != nil {
return err
}
@ -330,10 +330,10 @@ func TestClusterResize_AddNode(t *testing.T) {
t.Run("ContinuousSlices", func(t *testing.T) {
// Configure node0
m0 := test.NewMainWithCluster()
m0 := test.NewMainWithCluster(true)
defer m0.Close()
seed, coord, err := m0.RunWithTransport("localhost", 0, []string{}, pilosa.URI{})
seed, err := m0.RunWithTransport("localhost", 0, []string{})
if err != nil {
t.Fatal(err)
}
@ -358,12 +358,12 @@ func TestClusterResize_AddNode(t *testing.T) {
}
// Configure node1
m1 := test.NewMainWithCluster()
m1 := test.NewMainWithCluster(false)
defer m1.Close()
var eg errgroup.Group
eg.Go(func() error {
_, _, err = m1.RunWithTransport("localhost", 0, []string{seed}, coord)
_, err = m1.RunWithTransport("localhost", 0, []string{seed})
if err != nil {
return err
}
@ -385,10 +385,10 @@ func TestClusterResize_AddNode(t *testing.T) {
t.Run("SkippedSlice", func(t *testing.T) {
// Configure node0
m0 := test.NewMainWithCluster()
m0 := test.NewMainWithCluster(true)
defer m0.Close()
seed, coord, err := m0.RunWithTransport("localhost", 0, []string{}, pilosa.URI{})
seed, err := m0.RunWithTransport("localhost", 0, []string{})
if err != nil {
t.Fatal(err)
}
@ -413,12 +413,12 @@ func TestClusterResize_AddNode(t *testing.T) {
}
// Configure node1
m1 := test.NewMainWithCluster()
m1 := test.NewMainWithCluster(false)
defer m1.Close()
var eg errgroup.Group
eg.Go(func() error {
_, _, err = m1.RunWithTransport("localhost", 0, []string{seed}, coord)
_, err = m1.RunWithTransport("localhost", 0, []string{seed})
if err != nil {
return err
}
@ -443,22 +443,22 @@ func TestClusterResize_AddNode(t *testing.T) {
func TestCluster_GossipMembership(t *testing.T) {
t.Run("Node0Down", func(t *testing.T) {
// Configure node0
m0 := test.NewMainWithCluster()
m0 := test.NewMainWithCluster(true)
defer m0.Close()
seed, coord, err := m0.RunWithTransport("localhost", 0, []string{}, pilosa.URI{})
seed, err := m0.RunWithTransport("localhost", 0, []string{})
if err != nil {
t.Fatal(err)
}
// Configure node1
m1 := test.NewMainWithCluster()
m1 := test.NewMainWithCluster(false)
defer m1.Close()
var eg errgroup.Group
eg.Go(func() error {
// Pass invalid seed as first in list
_, _, err = m1.RunWithTransport("localhost", 0, []string{"http://localhost:8765", seed}, coord)
_, err = m1.RunWithTransport("localhost", 0, []string{"http://localhost:8765", seed})
if err != nil {
return err
}
@ -466,12 +466,12 @@ func TestCluster_GossipMembership(t *testing.T) {
})
// Configure node2
m2 := test.NewMainWithCluster()
m2 := test.NewMainWithCluster(false)
defer m2.Close()
eg.Go(func() error {
// Pass invalid seed as last in list
_, _, err = m2.RunWithTransport("localhost", 0, []string{seed, "http://localhost:8765"}, coord)
_, err = m2.RunWithTransport("localhost", 0, []string{seed, "http://localhost:8765"})
if err != nil {
return err
}
@ -545,4 +545,37 @@ func TestClusterResize_RemoveNode(t *testing.T) {
t.Fatalf("expected Body '%s' but got '%s'", expBody, strings.TrimSpace(resp.Body))
}
})
t.Run("ErrorRemoveWithoutReplicas", func(t *testing.T) {
client0 := m0.Client()
// Create indexes and frames on one node.
if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists {
t.Fatal(err)
} else if err := client0.CreateFrame(context.Background(), "i", "f", pilosa.FrameOptions{}); err != nil {
t.Fatal(err)
}
// This is an attempt to ensure there is data on both nodes, but is not guaranteed.
// TODO: Deterministic node IDs would ensure consistent results
setBits := ""
for i := 0; i < 20; i++ {
setBits += fmt.Sprintf("SetBit(rowID=1, frame=\"f\", columnID=%d) ", i*pilosa.SliceWidth)
}
if _, err := m0.Query("i", "", setBits); err != nil {
t.Fatal(err)
}
resp := test.MustDo("GET", m1.URL()+fmt.Sprintf("/id"), "")
nodeID := resp.Body
resp = test.MustDo("POST", m0.URL()+fmt.Sprintf("/cluster/resize/remove-node"), fmt.Sprintf(`{"id": "%s"}`, nodeID))
expBody := "not enough data to perform resize"
if resp.StatusCode != http.StatusInternalServerError {
t.Fatalf("expected StatusCode %d but got %d", http.StatusInternalServerError, resp.StatusCode)
} else if !strings.Contains(resp.Body, expBody) {
t.Fatalf("expected to contain '%s' but got '%s'", expBody, strings.TrimSpace(resp.Body))
}
})
}

35
server/server.go
View file

@ -34,6 +34,7 @@ import (
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/gcnotify"
"github.com/pilosa/pilosa/gopsutil"
"github.com/pilosa/pilosa/gossip"
"github.com/pilosa/pilosa/statik"
"github.com/pilosa/pilosa/statsd"
@ -153,6 +154,7 @@ func (m *Command) SetupServer() error {
if m.Config.Metric.Diagnostics {
m.Server.DiagnosticInterval = time.Duration(DefaultDiagnosticsInterval)
}
m.Server.SystemInfo = gopsutil.NewSystemInfo()
m.Server.GCNotifier = gcnotify.NewActiveGCNotifier()
m.Server.Holder.Stats, err = NewStatsClient(m.Config.Metric.Service, m.Config.Metric.Host)
if err != nil {
@ -182,10 +184,7 @@ func (m *Command) SetupServer() error {
InsecureSkipVerify: m.Config.TLS.SkipVerify,
}
// TODO Review this location
TLSConfig = m.Server.TLS
}
c := pilosa.GetHTTPClient(TLSConfig)
m.Server.RemoteClient = c
@ -195,21 +194,6 @@ func (m *Command) SetupServer() error {
// Statik file system.
m.Server.Handler.FileSystem = &statik.FileSystem{}
// Default coordintor to port 0 when not specified so that coordinator
// can be set to the value of server.URI after server binds to a port.
// This would only be useful in a one-node cluster.
coord := m.Config.Cluster.Coordinator
if coord == "" {
coord = ":0"
}
// Set the coordinator node.
curi, err := pilosa.AddressWithDefaults(coord)
if err != nil {
return err
}
m.Server.Cluster.Coordinator = *curi
// Set configuration options.
m.Server.AntiEntropyInterval = time.Duration(m.Config.AntiEntropy.Interval)
m.Server.Cluster.LongQueryTime = time.Duration(m.Config.Cluster.LongQueryTime)
@ -218,8 +202,12 @@ func (m *Command) SetupServer() error {
// SetupNetworking sets up internode communication based on the configuration.
func (m *Command) SetupNetworking() error {
m.Server.NodeID = m.Server.LoadNodeID()
if m.Config.Cluster.Disabled {
m.Server.Cluster.Static = true
m.Server.Cluster.Coordinator = m.Server.NodeID
for _, address := range m.Config.Cluster.Hosts {
uri, err := pilosa.NewURIFromAddress(address)
if err != nil {
@ -231,13 +219,9 @@ func (m *Command) SetupNetworking() error {
}
m.Server.Broadcaster = pilosa.NopBroadcaster
m.Server.Cluster.MemberSet = pilosa.NewStaticMemberSet()
m.Server.Cluster.MemberSet = pilosa.NewStaticMemberSet(m.Server.Cluster.Nodes)
m.Server.BroadcastReceiver = pilosa.NopBroadcastReceiver
m.Server.Gossiper = pilosa.NopGossiper
err := m.Server.Cluster.MemberSet.(*pilosa.StaticMemberSet).Join(m.Server.Cluster.Nodes)
if err != nil {
return err
}
return nil
}
@ -264,7 +248,10 @@ func (m *Command) SetupNetworking() error {
}
}
m.Server.NodeID = m.Server.LoadNodeID()
// Set Coordinator.
if m.Config.Cluster.Coordinator || len(m.Config.Gossip.Seeds) == 0 {
m.Server.Cluster.Coordinator = m.Server.NodeID
}
m.Server.Cluster.EventReceiver = gossip.NewGossipEventReceiver(m.Server.LogOutput)
gossipMemberSet, err := gossip.NewGossipMemberSetWithTransport(m.Server.NodeID, m.Config, transport, m.Server)

6
test/cluster.go
View file

@ -49,7 +49,7 @@ func NewCluster(n int) *pilosa.Cluster {
}
c.Node = c.Nodes[0]
c.Coordinator = c.Nodes[0].URI
c.Coordinator = c.Nodes[0].ID
return c
}
@ -260,9 +260,9 @@ func (t *TestCluster) addCluster(i int, saveTopology bool) (*pilosa.Cluster, err
c.Path = path
c.Topology = pilosa.NewTopology()
c.Holder = h
c.MemberSet = pilosa.NewStaticMemberSet()
c.MemberSet = pilosa.NewStaticMemberSet(c.Nodes)
c.Node = node
c.Coordinator = t.common.Nodes[0].URI // the first node is the coordinator
c.Coordinator = t.common.Nodes[0].ID // the first node is the coordinator
c.Broadcaster = t
// add nodes

25
test/pilosa.go
View file

@ -65,9 +65,10 @@ func NewMain() *Main {
}
// NewMainWithCluster returns a new instance of Main with clustering enabled.
func NewMainWithCluster() *Main {
func NewMainWithCluster(isCoordinator bool) *Main {
m := NewMain()
m.Config.Cluster.Disabled = false
m.Config.Cluster.Coordinator = isCoordinator
return m
}
@ -94,12 +95,11 @@ func runMainWithCluster(size int) ([]*Main, error) {
gossipPort := 0
var err error
var gossipSeeds = make([]string, size)
var coordinator pilosa.URI
for i := 0; i < size; i++ {
m := NewMainWithCluster()
m := NewMainWithCluster(i == 0)
gossipSeeds[i], coordinator, err = m.RunWithTransport(gossipHost, gossipPort, gossipSeeds[:i], coordinator)
gossipSeeds[i], err = m.RunWithTransport(gossipHost, gossipPort, gossipSeeds[:i])
if err != nil {
return nil, errors.Wrap(err, "RunWithTransport")
}
@ -146,7 +146,7 @@ func (m *Main) Reopen() error {
}
// RunWithTransport runs Main and returns the dynamically allocated gossip port.
func (m *Main) RunWithTransport(host string, bindPort int, joinSeeds []string, coordinator pilosa.URI) (seed string, coord pilosa.URI, err error) {
func (m *Main) RunWithTransport(host string, bindPort int, joinSeeds []string) (seed string, err error) {
defer close(m.Started)
/*
@ -166,19 +166,19 @@ func (m *Main) RunWithTransport(host string, bindPort int, joinSeeds []string, c
// SetupServer
err = m.SetupServer()
if err != nil {
return seed, coord, err
return seed, err
}
// Open server listener.
err = m.Server.OpenListener()
if err != nil {
return seed, coord, err
return seed, err
}
// Open gossip transport to use in SetupServer.
transport, err := gossip.NewTransport(host, bindPort)
if err != nil {
return seed, coord, err
return seed, err
}
m.GossipTransport = transport
@ -193,23 +193,22 @@ func (m *Main) RunWithTransport(host string, bindPort int, joinSeeds []string, c
// SetupNetworking
err = m.SetupNetworking()
if err != nil {
return seed, coord, err
return seed, err
}
if err = m.Server.BroadcastReceiver.Start(m.Server); err != nil {
return seed, coord, err
return seed, err
}
m.Server.Cluster.Coordinator = coordinator
m.Server.Cluster.Static = false
// Initialize server.
err = m.Server.Open()
if err != nil {
return seed, coord, err
return seed, err
}
return seed, m.Server.Cluster.Coordinator, nil
return seed, nil
}
// URL returns the base URL string for accessing the running program.