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

Merge pull request #908 from travisturner/update-status-schema

Browse source 
Update status schema
This commit is contained in:
Travis Turner 2017-10-30 14:39:52 -05:00 • committed by GitHub
commit fa61824975
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
18 changed files with 677 additions and 653 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
client.go
View file

@ -100,9 +100,6 @@ func (c *Client) MaxInverseSliceByIndex(ctx context.Context) (map[string]uint64,
func (c *Client) maxSliceByIndex(ctx context.Context, inverse bool) (map[string]uint64, error) {
// Execute request against the host.
u := uriPathToURL(c.host, "/slices/max")
u.RawQuery = (&url.Values{
"inverse": {strconv.FormatBool(inverse)},
}).Encode()
// Build request.
req, err := http.NewRequest("GET", u.String(), nil)
@ -119,14 +116,17 @@ func (c *Client) maxSliceByIndex(ctx context.Context, inverse bool) (map[string]
}
defer resp.Body.Close()
var rsp sliceMaxResponse
var rsp getSlicesMaxResponse
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("http: status=%d", resp.StatusCode)
} else if err := json.NewDecoder(resp.Body).Decode(&rsp); err != nil {
return nil, fmt.Errorf("json decode: %s", err)
}
return rsp.MaxSlices, nil
if inverse {
return rsp.Inverse, nil
}
return rsp.Standard, nil
}
// Schema returns all index and frame schema information.

39
cluster.go
View file

@ -53,24 +53,7 @@ const (
// Node represents a node in the cluster.
type Node struct {
//Scheme string `json:"scheme"`
//Host string `json:"host"` // HostPort
URI URI `json:"uri"`
status *internal.NodeStatus `json:"status"`
}
// SetStatus sets the NodeStatus.
func (n *Node) SetStatus(s *internal.NodeStatus) {
n.status = s
}
// SetState sets the Node.status.state.
func (n *Node) SetState(s string) {
if n.status == nil {
n.status = &internal.NodeStatus{}
}
n.status.State = s
}
// Nodes represents a list of nodes.
@ -242,13 +225,12 @@ func (c *Cluster) URISet() []URI {
func (c *Cluster) setState(state string) {
c.State = state
localNode := c.localNode()
localNode.SetState(state)
}
func (c *Cluster) localNode() *Node {
return c.NodeByURI(c.URI)
}
// localNode is not being used.
//func (c *Cluster) localNode() *Node {
// return c.NodeByURI(c.URI)
//}
// Status returns the internal ClusterStatus representation.
func (c *Cluster) Status() *internal.ClusterStatus {
@ -258,17 +240,6 @@ func (c *Cluster) Status() *internal.ClusterStatus {
}
}
/*
// encodeNodeStatuses converts a into its internal representation.
func encodeNodeStatuses(a []*Node) []*internal.NodeStatus {
other := make([]*internal.NodeStatus, len(a))
for i := range a {
other[i] = a[i].status
}
return other
}
*/
// NodeByURI returns a node reference by uri.
func (c *Cluster) NodeByURI(uri URI) *Node {
for _, n := range c.Nodes {
@ -609,7 +580,7 @@ func (c *Cluster) handleJoiningHost(uri URI) error {
func (c *Cluster) setStateAndBroadcast(state string) error {
c.setState(state)
// Broadcast status changes to the cluster.
// Broadcast cluster status changes to the cluster.
return c.Broadcaster.SendSync(c.Status())
}

1
cmd/root_test.go
View file

@ -181,7 +181,6 @@ func TestRootCommand_Config(t *testing.T) {
bind = "127.0.0.1:10101"
[cluster]
poll-interval = "2m0s"
replicas = 2
partitions = 128
hosts = [

4
cmd/server_test.go
View file

@ -51,7 +51,6 @@ func TestServerConfig(t *testing.T) {
bind = "localhost:0"
[cluster]
poll-interval = "45s"
type = "static"
replicas = 2
hosts = [
@ -64,7 +63,6 @@ func TestServerConfig(t *testing.T) {
v.Check(cmd.Server.Config.Bind, "localhost:10111")
v.Check(cmd.Server.Config.Cluster.ReplicaN, 2)
v.Check(cmd.Server.Config.Cluster.Hosts, []string{"localhost:10111", "localhost:10110"})
v.Check(cmd.Server.Config.Cluster.PollInterval, pilosa.Duration(time.Second*182))
return v.Error()
},
},
@ -99,7 +97,6 @@ func TestServerConfig(t *testing.T) {
bind = "localhost:19444"
data-dir = "` + actualDataDir + `"
[cluster]
poll-interval = "2m0s"
hosts = [
"localhost:19444",
]
@ -115,7 +112,6 @@ func TestServerConfig(t *testing.T) {
validation: func() error {
v := validator{}
v.Check(cmd.Server.Config.Cluster.Hosts, []string{"localhost:19444"})
v.Check(cmd.Server.Config.Cluster.PollInterval, pilosa.Duration(time.Minute*2))
v.Check(cmd.Server.Config.AntiEntropy.Interval, pilosa.Duration(time.Minute*11))
v.Check(cmd.Server.CPUProfile, profFile.Name())
v.Check(cmd.Server.CPUTime, time.Minute)

2
config.go
View file

@ -78,7 +78,6 @@ type Config struct {
ReplicaN int `toml:"replicas"`
Type string `toml:"type"`
Hosts []string `toml:"hosts"`
PollInterval Duration `toml:"poll-interval"`
LongQueryTime Duration `toml:"long-query-time"`
} `toml:"cluster"`
@ -113,7 +112,6 @@ func NewConfig() *Config {
}
c.Cluster.ReplicaN = DefaultReplicaN
c.Cluster.Type = DefaultClusterType
c.Cluster.PollInterval = Duration(DefaultPollingInterval)
c.Cluster.Hosts = []string{}
c.AntiEntropy.Interval = Duration(DefaultAntiEntropyInterval)
c.Metric.Service = DefaultMetrics

1
ctl/generate_config.go
View file

@ -43,7 +43,6 @@ bind = "localhost:10101"
max-writes-per-request = 5000
[cluster]
poll-interval = "2m0s"
replicas = 1
hosts = [
"localhost:10101",

1
ctl/server.go
View file

@ -35,7 +35,6 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) {
flags.IntVarP(&srv.Config.MaxWritesPerRequest, "max-writes-per-request", "", srv.Config.MaxWritesPerRequest, "Number of write commands per request.")
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.DurationVarP((*time.Duration)(&srv.Config.Cluster.PollInterval), "cluster.poll-interval", "", time.Minute, "Polling interval for cluster.") // TODO what actually is this?
flags.DurationVarP((*time.Duration)(&srv.Config.Cluster.LongQueryTime), "cluster.long-query-time", "", time.Minute, "Long Query Time.")
flags.StringVarP(&srv.Config.Plugins.Path, "plugins.path", "", "", "Path to plugin directory.")
flags.StringVar(&srv.Config.LogPath, "log-path", "", "Log path")

15
docs/configuration.md
View file

@ -27,10 +27,9 @@ Every command line flag has a corresponding environment variable. The environmen
### Config file
The config file is in the [toml format](https://github.com/toml-lang/toml) and has exactly the same options available as the flags and environment variables. Any flag which contains a dot (".") denotes nesting within the config file, so the two flags `--cluster.poll-interval=2m0s` and `--cluster.replicas=1` look like this in the config file:
The config file is in the [toml format](https://github.com/toml-lang/toml) and has exactly the same options available as the flags and environment variables. Any flag which contains a dot (".") denotes nesting within the config file, so the two flag `--cluster.replicas=1` looks like this in the config file:
```toml
[cluster]
poll-interval = "2m0s"
replicas = 1
```
@ -123,18 +122,6 @@ Any flag that has a value that is a comma separated list on the command line bec
hosts = ["localhost:10101"]
```
#### Cluster Poll Interval
* Description: Polling interval for cluster.
* Flag: `cluster.poll-interval="1m0s"`
* Env: `PILOSA_CLUSTER_POLL_INTERVAL="1m0s"`
* Config:
```toml
[cluster]
poll-interval = "1m0s"
```
#### Cluster Replicas
* Description: Number of hosts each piece of data should be stored on.

3
gossip/gossip.go
View file

@ -148,8 +148,7 @@ func NewGossipNodeSet(name string, gossipHost string, gossipPort int, gossipSeed
g.config.memberlistConfig.BindPort = gossipPort
g.config.memberlistConfig.AdvertiseAddr = pilosa.HostToIP(gossipHost)
g.config.memberlistConfig.AdvertisePort = gossipPort
// TODO travis: pause node status (remove this next line)
g.config.memberlistConfig.PushPullInterval = 0 * time.Millisecond
//g.config.memberlistConfig.PushPullInterval = 15 * time.Second // Default is 15s in DefaultLocalConfig.
g.config.memberlistConfig.Delegate = g
g.config.memberlistConfig.SecretKey = secretKey
g.config.memberlistConfig.Events = server.Cluster.EventReceiver.(memberlist.EventDelegate)

44
handler.go
View file

@ -132,7 +132,7 @@ func NewRouter(handler *Handler) *mux.Router {
router.HandleFunc("/index/{index}/time-quantum", handler.handlePatchIndexTimeQuantum).Methods("PATCH")
router.HandleFunc("/hosts", handler.handleGetHosts).Methods("GET")
router.HandleFunc("/schema", handler.handleGetSchema).Methods("GET")
router.HandleFunc("/slices/max", handler.handleGetSliceMax).Methods("GET")
router.HandleFunc("/slices/max", handler.handleGetSlicesMax).Methods("GET") // TODO: deprecate, but it's being used by the client (for backups)
router.HandleFunc("/status", handler.handleGetStatus).Methods("GET")
router.HandleFunc("/version", handler.handleGetVersion).Methods("GET")
router.HandleFunc("/recalculate-caches", handler.handleRecalculateCaches).Methods("POST")
@ -216,13 +216,16 @@ func (h *Handler) handleGetSchema(w http.ResponseWriter, r *http.Request) {
// handleGetStatus handles GET /status requests.
func (h *Handler) handleGetStatus(w http.ResponseWriter, r *http.Request) {
status, err := h.StatusHandler.ClusterStatus()
pb, err := h.StatusHandler.ClusterStatus()
if err != nil {
h.logger().Printf("cluster status error: %s", err)
return
}
cs := pb.(*internal.ClusterStatus)
if err := json.NewEncoder(w).Encode(getStatusResponse{
Status: status,
State: cs.State,
URISet: decodeURIs(cs.URISet),
}); err != nil {
h.logger().Printf("write status response error: %s", err)
}
@ -233,7 +236,8 @@ type getSchemaResponse struct {
}
type getStatusResponse struct {
Status proto.Message `json:"status"`
State string `json:"state"`
URISet []URI `json:"uri-set"`
}
// handlePostQuery handles /query requests.
@ -305,31 +309,19 @@ func (h *Handler) handlePostQuery(w http.ResponseWriter, r *http.Request) {
}
}
func (h *Handler) handleGetSliceMax(w http.ResponseWriter, r *http.Request) {
var ms map[string]uint64
if inverse, _ := strconv.ParseBool(r.URL.Query().Get("inverse")); inverse {
ms = h.Holder.MaxInverseSlices()
} else {
ms = h.Holder.MaxSlices()
// handleGetSlicesMax handles GET /schema requests.
func (h *Handler) handleGetSlicesMax(w http.ResponseWriter, r *http.Request) {
if err := json.NewEncoder(w).Encode(getSlicesMaxResponse{
Standard: h.Holder.MaxSlices(),
Inverse: h.Holder.MaxInverseSlices(),
}); err != nil {
h.logger().Printf("write slices-max response error: %s", err)
}
if strings.Contains(r.Header.Get("Accept"), "application/x-protobuf") {
pb := &internal.MaxSlicesResponse{
MaxSlices: ms,
}
if buf, err := proto.Marshal(pb); err != nil {
h.logger().Printf("protobuf marshal error: %s", err)
} else if _, err := w.Write(buf); err != nil {
h.logger().Printf("stream write error: %s", err)
}
return
}
json.NewEncoder(w).Encode(sliceMaxResponse{
MaxSlices: ms,
})
}
type sliceMaxResponse struct {
MaxSlices map[string]uint64 `json:"maxSlices"`
type getSlicesMaxResponse struct {
Standard map[string]uint64 `json:"standard"`
Inverse map[string]uint64 `json:"inverse"`
}
// handleGetIndexes handles GET /index request.

6
handler_test.go
View file

@ -147,7 +147,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 != `{"status":{"State":"NORMAL","URISet":[{"Scheme":"http","Host":"localhost","Port":10101}]}}`+"\n" {
} else if body := w.Body.String(); body != `{"state":"NORMAL","uri-set":[{"scheme":"http","host":"localhost","port":10101}]}`+"\n" {
t.Fatalf("unexpected body: %s", body)
}
}
@ -172,7 +172,7 @@ func TestHandler_MaxSlices(t *testing.T) {
h.ServeHTTP(w, test.MustNewHTTPRequest("GET", "/slices/max", nil))
if w.Code != http.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
} else if body := w.Body.String(); body != `{"maxSlices":{"i0":3,"i1":0}}`+"\n" {
} else if body := w.Body.String(); body != `{"standard":{"i0":3,"i1":0},"inverse":{"i0":0,"i1":0}}`+"\n" {
t.Fatalf("unexpected body: %s", body)
}
}
@ -213,7 +213,7 @@ func TestHandler_MaxSlices_Inverse(t *testing.T) {
h.ServeHTTP(w, test.MustNewHTTPRequest("GET", "/slices/max?inverse=true", nil))
if w.Code != http.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
} else if body := w.Body.String(); body != `{"maxSlices":{"i0":3,"i1":0}}`+"\n" {
} else if body := w.Body.String(); body != `{"standard":{"i0":0,"i1":0},"inverse":{"i0":3,"i1":0}}`+"\n" {
t.Fatalf("unexpected body: %s", body)
}
}

17
holder.go
View file

@ -26,6 +26,8 @@ import (
"sync"
"syscall"
"time"
"github.com/pilosa/pilosa/internal"
)
const (
@ -185,6 +187,21 @@ func (h *Holder) Schema() []*IndexInfo {
return a
}
// EncodeMaxSlices creates and internal representation of max slices.
func (h *Holder) EncodeMaxSlices() *internal.MaxSlices {
return &internal.MaxSlices{
Standard: h.MaxSlices(),
Inverse: h.MaxInverseSlices(),
}
}
// EncodeSchema creates and internal representation of schema.
func (h *Holder) EncodeSchema() *internal.Schema {
return &internal.Schema{
Indexes: EncodeIndexes(h.Indexes()),
}
}
// IndexPath returns the path where a given index is stored.
func (h *Holder) IndexPath(name string) string { return filepath.Join(h.Path, name) }

23
index.go
View file

@ -392,6 +392,20 @@ func (i *Index) Frames() []*Frame {
return a
}
// InputDefinitions returns a list of all inputDefinitions in the index.
func (i *Index) InputDefinitions() []*InputDefinition {
i.mu.RLock()
defer i.mu.RUnlock()
a := make([]*InputDefinition, 0, len(i.inputDefinitions))
for _, d := range i.inputDefinitions {
a = append(a, d)
}
//sort.Sort(inputDefintionSlice(a)) // TODO
return a
}
// RecalculateCaches recalculates caches on every frame in the index.
func (i *Index) RecalculateCaches() {
for _, frame := range i.Frames() {
@ -617,12 +631,10 @@ func EncodeIndexes(a []*Index) []*internal.Index {
// encodeIndex converts d into its internal representation.
func encodeIndex(d *Index) *internal.Index {
io := d.options()
return &internal.Index{
Name: d.name,
Meta: io.Encode(),
MaxSlice: d.MaxSlice(),
Frames: encodeFrames(d.Frames()),
Name: d.name,
Frames: encodeFrames(d.Frames()),
InputDefinitions: encodeInputDefinitions(d.InputDefinitions()),
}
}
@ -716,7 +728,6 @@ func (i *Index) newInputDefinition(name string) (*InputDefinition, error) {
if err != nil {
return nil, err
}
inputDef.broadcaster = i.broadcaster
return inputDef, nil
}

59
input_definition.go
View file

@ -36,12 +36,11 @@ var validValueDestination = []string{InputMapping, InputValueToRow, InputSingleR
// InputDefinition represents a container for the data input definition.
type InputDefinition struct {
name string
path string
index string
broadcaster Broadcaster
frames []InputFrame
fields []InputDefinitionField
name string
path string
index string
frames []InputFrame
fields []InputDefinitionField
}
// NewInputDefinition returns a new instance of InputDefinition.
@ -86,16 +85,9 @@ func (i *InputDefinition) LoadDefinition(pb *internal.InputDefinition) error {
// Copy metadata fields.
i.name = pb.Name
for _, fr := range pb.Frames {
frameMeta := fr.Meta
inputFrame := InputFrame{
Name: fr.Name,
Options: FrameOptions{
RowLabel: frameMeta.RowLabel,
InverseEnabled: frameMeta.InverseEnabled,
CacheSize: frameMeta.CacheSize,
CacheType: frameMeta.CacheType,
TimeQuantum: TimeQuantum(frameMeta.TimeQuantum),
},
Name: fr.Name,
Options: *decodeFrameOptions(fr.Meta),
}
i.frames = append(i.frames, inputFrame)
}
@ -327,6 +319,43 @@ func (i *InputDefinitionInfo) Encode() *internal.InputDefinition {
return &def
}
// encodeInputDefinitions converts a into its internal representation.
func encodeInputDefinitions(a []*InputDefinition) []*internal.InputDefinition {
other := make([]*internal.InputDefinition, len(a))
for i := range a {
other[i] = encodeInputDefinition(a[i])
}
return other
}
// encodeInputDefinition converts i into its internal representation.
func encodeInputDefinition(i *InputDefinition) *internal.InputDefinition {
//fo := f.options()
return &internal.InputDefinition{
Name: i.name,
Frames: encodeInputFrames(i.frames),
Fields: encodeInputDefinitionFields(i.fields),
}
}
// encodeInputFrames converts a into its internal representation.
func encodeInputFrames(a []InputFrame) []*internal.Frame {
other := make([]*internal.Frame, len(a))
for i := range a {
other[i] = a[i].Encode()
}
return other
}
// encodeInputDefinitionFields converts a into its internal representation.
func encodeInputDefinitionFields(a []InputDefinitionField) []*internal.InputDefinitionField {
other := make([]*internal.InputDefinitionField, len(a))
for i := range a {
other[i] = a[i].Encode()
}
return other
}
// AddFrame manually add frame to input definition.
func (i *InputDefinition) AddFrame(frame InputFrame) error {
i.frames = append(i.frames, frame)

905
internal/private.pb.go
View file

File diff suppressed because it is too large Load diff

18
internal/private.proto
View file

@ -38,8 +38,9 @@ message Cache {
repeated uint64 IDs = 1;
}
message MaxSlicesResponse {
map<string, uint64> MaxSlices = 1;
message MaxSlices {
map<string, uint64> Standard = 1;
map<string, uint64> Inverse = 2;
}
message CreateSliceMessage {
@ -71,14 +72,16 @@ message DeleteFrameMessage {
message Frame {
string Name = 1;
FrameMeta Meta = 2;
repeated string Views = 3;
}
message Schema {
repeated Index Indexes = 1;
}
message Index {
string Name = 1;
IndexMeta Meta = 2;
uint64 MaxSlice = 3;
repeated Frame Frames = 4;
repeated uint64 Slices = 5;
repeated InputDefinition InputDefinitions = 6;
}
@ -120,9 +123,8 @@ message URI {
message NodeStatus {
URI URI = 1;
string State = 2;
repeated Index Indexes = 3;
repeated URI URISet = 4;
MaxSlices MaxSlices = 2;
Schema Schema = 3;
}
message ClusterStatus {

169
server.go
View file

@ -19,11 +19,9 @@ import (
"errors"
"fmt"
"io"
"io/ioutil"
"log"
"net"
"net/http"
"net/url"
"os"
"os/exec"
"runtime"
@ -40,7 +38,6 @@ import (
// Default server settings.
const (
DefaultAntiEntropyInterval = 10 * time.Minute
DefaultPollingInterval = 60 * time.Second
)
// Server represents a holder wrapped by a running HTTP server.
@ -65,7 +62,6 @@ type Server struct {
// Background monitoring intervals.
AntiEntropyInterval time.Duration
PollingInterval time.Duration
MetricInterval time.Duration
// TLS configuration
@ -92,7 +88,6 @@ func NewServer() *Server {
Network: "tcp",
AntiEntropyInterval: DefaultAntiEntropyInterval,
PollingInterval: DefaultPollingInterval,
MetricInterval: 0,
LogOutput: os.Stderr,
@ -198,9 +193,8 @@ func (s *Server) Open() error {
/*
// Start background monitoring.
s.wg.Add(3)
s.wg.Add(2)
go func() { defer s.wg.Done(); s.monitorAntiEntropy() }()
go func() { defer s.wg.Done(); s.monitorMaxSlices() }()
go func() { defer s.wg.Done(); s.monitorRuntime() }()
*/
@ -275,39 +269,6 @@ func (s *Server) monitorAntiEntropy() {
s.Holder.Stats.Histogram("AntiEntropyDuration", float64(dif), 1.0)
}
// monitorMaxSlices periodically pulls the highest slice from each node in the cluster.
func (s *Server) monitorMaxSlices() {
ticker := time.NewTicker(s.PollingInterval)
defer ticker.Stop()
for {
select {
case <-s.closing:
return
case <-ticker.C:
}
oldmaxslices := s.Holder.MaxSlices()
for _, node := range s.Cluster.Nodes {
if s.URI != node.URI {
maxSlices, _ := s.checkMaxSlices(node.URI)
for index, newmax := range maxSlices {
// if we don't know about an index locally, log an error because
// indexes should be created and synced prior to slice creation
if localIndex := s.Holder.Index(index); localIndex != nil {
if newmax > oldmaxslices[index] {
oldmaxslices[index] = newmax
localIndex.SetRemoteMaxSlice(newmax)
}
} else {
s.Logger().Printf("Local Index not found: %s", index)
}
}
}
}
}
}
// ReceiveMessage represents an implementation of BroadcastHandler.
func (s *Server) ReceiveMessage(pb proto.Message) error {
switch obj := pb.(type) {
@ -392,10 +353,16 @@ func (s *Server) State() string {
return s.Cluster.State
}
// LocalStatus returns the state of the local node as well as the
// holder (indexes/frames) according to the local node.
// In a gossip implementation, memberlist.Delegate.LocalState() uses this.
// Server implements StatusHandler.
// LocalStatus is used to periodically sync information
// between nodes. Under normal conditions, nodes should
// remain in sync through Broadcast messages. For cases
// where a node fails to receive a Broadcast message, or
// when a new (empty) node needs to get in sync with the
// rest of the cluster, two things are shared via gossip:
// - MaxSlice/MaxInverseSlice by Index
// - Schema
// In a gossip implementation, memberlist.Delegate.LocalState() uses this.
func (s *Server) LocalStatus() (proto.Message, error) {
if s.Cluster == nil {
return nil, errors.New("Server.Cluster is nil")
@ -405,61 +372,66 @@ func (s *Server) LocalStatus() (proto.Message, error) {
}
ns := internal.NodeStatus{
URI: encodeURI(s.URI),
State: s.State(),
Indexes: EncodeIndexes(s.Holder.Indexes()),
URISet: encodeURIs(s.Cluster.URISet()),
}
// TODO: get rid of this
// Append Slice list per this Node's indexes
for _, index := range ns.Indexes {
index.Slices = s.Cluster.OwnsSlices(index.Name, index.MaxSlice, s.URI)
URI: encodeURI(s.URI),
MaxSlices: s.Holder.EncodeMaxSlices(),
Schema: s.Holder.EncodeSchema(),
}
return &ns, nil
}
// ClusterStatus returns the NodeState for all nodes in the cluster.
// ClusterStatus returns the ClusterState and URISet for the cluster.
func (s *Server) ClusterStatus() (proto.Message, error) {
// Update local Node.state.
ns, err := s.LocalStatus()
if err != nil {
return nil, err
}
localNode := s.Cluster.localNode()
localNode.SetStatus(ns.(*internal.NodeStatus))
return s.Cluster.Status(), nil
}
// HandleRemoteStatus receives incoming NodeState from remote nodes.
// HandleRemoteStatus receives incoming NodeStatus from remote nodes.
func (s *Server) HandleRemoteStatus(pb proto.Message) error {
return s.mergeRemoteStatus(pb.(*internal.NodeStatus))
}
func (s *Server) mergeRemoteStatus(ns *internal.NodeStatus) error {
// Ignore status updates from self.
if s.URI == decodeURI(ns.URI) {
return nil
}
fmt.Printf("mergeRemoteStatus on (%s) from (%s)\n", s.URI, ns.URI)
// Update Node.state.
// Node can be nil if a merge occurs (via gossip) before the coordinator has
// a chance to broadcast the existence of the node.
uri := decodeURI(ns.URI)
if node := s.Cluster.NodeByURI(uri); node != nil {
node.SetStatus(ns)
// Sync maxSlices (standard).
oldmaxslices := s.Holder.MaxSlices()
for index, newMax := range ns.MaxSlices.Standard {
localIndex := s.Holder.Index(index)
// if we don't know about an index locally, log an error because
// indexes should be created and synced prior to slice creation
if localIndex == nil {
s.Logger().Printf("Local Index not found: %s", index)
continue
}
if newMax > oldmaxslices[index] {
oldmaxslices[index] = newMax
localIndex.SetRemoteMaxSlice(newMax)
}
}
// Create indexes that don't exist.
for _, index := range ns.Indexes {
opt := IndexOptions{
ColumnLabel: index.Meta.ColumnLabel,
TimeQuantum: TimeQuantum(index.Meta.TimeQuantum),
// Sync maxSlices (inverse).
oldMaxInverseSlices := s.Holder.MaxInverseSlices()
for index, newMaxInverse := range ns.MaxSlices.Inverse {
localIndex := s.Holder.Index(index)
// if we don't know about an index locally, log an error because
// indexes should be created and synced prior to slice creation
if localIndex == nil {
s.Logger().Printf("Local Index not found: %s", index)
continue
}
if newMaxInverse > oldMaxInverseSlices[index] {
oldMaxInverseSlices[index] = newMaxInverse
localIndex.SetRemoteMaxSlice(newMaxInverse)
}
}
// Sync schema.
// Create indexes that don't exist.
for _, index := range ns.Schema.Indexes {
opt := IndexOptions{}
idx, err := s.Holder.CreateIndexIfNotExists(index.Name, opt)
if err != nil {
return err
@ -472,55 +444,12 @@ func (s *Server) mergeRemoteStatus(ns *internal.NodeStatus) error {
return err
}
}
// TODO: Create inputDefinitions that don't exist.
}
return nil
}
func (s *Server) checkMaxSlices(uri URI) (map[string]uint64, error) {
// Create HTTP request.
req, err := http.NewRequest("GET", (&url.URL{
Scheme: uri.Scheme(),
Host: uri.HostPort(),
Path: "/slices/max",
}).String(), nil)
if err != nil {
return nil, err
}
// Require protobuf encoding.
req.Header.Set("Accept", "application/x-protobuf")
req.Header.Set("Content-Type", "application/x-protobuf")
req.Header.Set("User-Agent", "pilosa/"+Version)
resp, err := s.defaultClient.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
// Read response into buffer.
body, err := ioutil.ReadAll(resp.Body)
if err != nil {
return nil, err
}
// Check status code.
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("invalid status checkMaxSlices: code=%d, err=%s, req=%v", resp.StatusCode, body, req)
}
// Decode response object.
pb := internal.MaxSlicesResponse{}
if err = proto.Unmarshal(body, &pb); err != nil {
return nil, err
}
return pb.MaxSlices, nil
}
// monitorRuntime periodically polls the Go runtime metrics.
func (s *Server) monitorRuntime() {
// Disable metrics when poll interval is zero

13
server/server.go
View file

@ -124,19 +124,6 @@ func (m *Command) SetupServer() error {
cluster.ReplicaN = m.Config.Cluster.ReplicaN
cluster.IndexReporter = m.Server.Holder
/*
// TODO travis: get rid of this URI code
for _, address := range m.Config.Cluster.Hosts {
uri, err := pilosa.NewURIFromAddress(address)
if err != nil {
return err
}
cluster.Nodes = append(cluster.Nodes, &pilosa.Node{
Scheme: uri.Scheme(),
Host: uri.HostPort(),
})
}
*/
m.Server.Cluster = cluster
// Setup logging output.