diff --git a/broadcast.go b/broadcast.go new file mode 100644 index 000000000..a3f3cb6c9 --- /dev/null +++ b/broadcast.go @@ -0,0 +1,139 @@ +package pilosa + +import ( + "fmt" + "reflect" + + "github.com/gogo/protobuf/proto" + "github.com/pilosa/pilosa/internal" +) + +// NodeSet represents an interface for Node membership and inter-node communication. +type NodeSet interface { + // Returns a list of all Nodes in the cluster + Nodes() []*Node + + // Open starts any network activity implemented by the NodeSet + Open() error +} + +// StaticNodeSet represents a basic NodeSet for testing +type StaticNodeSet struct { + nodes []*Node +} + +func NewStaticNodeSet() *StaticNodeSet { + return &StaticNodeSet{} +} + +func (s *StaticNodeSet) Nodes() []*Node { + return s.nodes +} + +func (s *StaticNodeSet) Open() error { + return nil +} + +// Broadcaster is an interface for broadcasting messages. +type Broadcaster interface { + SendSync(pb proto.Message) error + SendAsync(pb proto.Message) error +} + +func init() { + NopBroadcaster = &nopBroadcaster{} +} + +var NopBroadcaster Broadcaster + +// nopBroadcaster represents a Broadcaster that doesn't do anything. +type nopBroadcaster struct{} + +// SendSync A no-op implemenetation of Broadcaster SendSync method. +func (c *nopBroadcaster) SendSync(pb proto.Message) error { + return nil +} + +// SendAsync A no-op implemenetation of Broadcaster SendAsync method. +func (c *nopBroadcaster) SendAsync(pb proto.Message) error { + return nil +} + +// BroadcastHandler is the interface for the pilosa object which knows how to +// handle broadcast messages. (Hint: this is implemented by pilosa.Server) +type BroadcastHandler interface { + ReceiveMessage(pb proto.Message) error +} + +// BroadcastReceiver is the interface for the object which will listen for and +// decode broadcast messages before passing them to pilosa to handle. The +// implementation of this could be an http server which listens for messages, +// gets the protobuf payload, and then passes it to +// BroadcastHandler.ReceiveMessage. +type BroadcastReceiver interface { + // Start starts listening for broadcast messages - it should return + // immediately, spawning a goroutine if necessary. + Start(BroadcastHandler) error +} + +type nopBroadcastReceiver struct{} + +func (n *nopBroadcastReceiver) Start(b BroadcastHandler) error { return nil } + +var NopBroadcastReceiver = &nopBroadcastReceiver{} + +const ( + MessageTypeCreateSlice = 1 + MessageTypeCreateDB = 2 + MessageTypeDeleteDB = 3 + MessageTypeCreateFrame = 4 + MessageTypeDeleteFrame = 5 +) + +func MarshalMessage(m proto.Message) ([]byte, error) { + var typ uint8 + switch obj := m.(type) { + case *internal.CreateSliceMessage: + typ = MessageTypeCreateSlice + case *internal.CreateDBMessage: + typ = MessageTypeCreateDB + case *internal.DeleteDBMessage: + typ = MessageTypeDeleteDB + case *internal.CreateFrameMessage: + typ = MessageTypeCreateFrame + case *internal.DeleteFrameMessage: + typ = MessageTypeDeleteFrame + default: + return nil, fmt.Errorf("message type not implemented for marshalling: %s", reflect.TypeOf(obj)) + } + buf, err := proto.Marshal(m) + if err != nil { + return nil, err + } + return append([]byte{typ}, buf...), nil +} + +func UnmarshalMessage(buf []byte) (proto.Message, error) { + typ, buf := buf[0], buf[1:] + + var m proto.Message + switch typ { + case MessageTypeCreateSlice: + m = &internal.CreateSliceMessage{} + case MessageTypeCreateDB: + m = &internal.CreateDBMessage{} + case MessageTypeDeleteDB: + m = &internal.DeleteDBMessage{} + case MessageTypeCreateFrame: + m = &internal.CreateFrameMessage{} + case MessageTypeDeleteFrame: + m = &internal.DeleteFrameMessage{} + default: + return nil, fmt.Errorf("invalid message type: %d", typ) + } + + if err := proto.Unmarshal(buf, m); err != nil { + return nil, err + } + return m, nil +} diff --git a/broadcast_test.go b/broadcast_test.go new file mode 100644 index 000000000..a2842a919 --- /dev/null +++ b/broadcast_test.go @@ -0,0 +1,91 @@ +package pilosa_test + +import ( + "reflect" + "testing" + + "github.com/gogo/protobuf/proto" + "github.com/pilosa/pilosa" + "github.com/pilosa/pilosa/internal" +) + +// Ensure a message can be marshaled and unmarshaled. +func TestMessage_Marshal(t *testing.T) { + + testMessageMarshal(t, &internal.CreateSliceMessage{ + DB: "d", + Slice: 8, + }) + + testMessageMarshal(t, &internal.DeleteDBMessage{ + DB: "d", + }) +} + +func testMessageMarshal(t *testing.T, m proto.Message) { + marshalled, err := pilosa.MarshalMessage(m) + if err != nil { + t.Fatal(err) + } + unmarshalled, err := pilosa.UnmarshalMessage(marshalled) + if err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(unmarshalled, m) { + t.Fatalf("unexpected message marshalling: %s", unmarshalled) + } +} + +// Ensure that BroadcastReceiver can register a BroadcastHandler. +func TestBroadcast_BroadcastReceiver(t *testing.T) { + + s := pilosa.NewServer() + + sbr := NewSimpleBroadcastReceiver() + sbh := NewSimpleBroadcastHandler() + + s.BroadcastReceiver = sbr + s.BroadcastReceiver.Start(sbh) + + msg := &internal.DeleteDBMessage{ + DB: "d", + } + + s.BroadcastReceiver.(*SimpleBroadcastReceiver).Receive(msg) + + // Make sure the message received is what was sentd + if !reflect.DeepEqual(sbh.receivedMessage, msg) { + t.Fatalf("unexpected message: %s", sbh.receivedMessage) + } +} + +type SimpleBroadcastReceiver struct { + broadcastHandler pilosa.BroadcastHandler +} + +func NewSimpleBroadcastReceiver() *SimpleBroadcastReceiver { + return &SimpleBroadcastReceiver{} +} + +func (r *SimpleBroadcastReceiver) Start(h pilosa.BroadcastHandler) error { + r.broadcastHandler = h + return nil +} + +func (r *SimpleBroadcastReceiver) Receive(pb proto.Message) error { + r.broadcastHandler.ReceiveMessage(pb) + return nil +} + +type SimpleBroadcastHandler struct { + receivedMessage proto.Message +} + +func NewSimpleBroadcastHandler() *SimpleBroadcastHandler { + return &SimpleBroadcastHandler{} +} + +func (h *SimpleBroadcastHandler) ReceiveMessage(pb proto.Message) error { + h.receivedMessage = pb.(proto.Message) + return nil +} diff --git a/cache.go b/cache.go index 3504806c6..7da28eb74 100644 --- a/cache.go +++ b/cache.go @@ -12,6 +12,11 @@ import ( "github.com/pilosa/pilosa/internal" ) +const ( + // ThresholdFactor is used to calculate the threshold for new items entering the cache + ThresholdFactor = 1.1 +) + // Cache represents a cache for bitmap counts. type Cache interface { Add(bitmapID uint64, n uint64) @@ -39,9 +44,9 @@ type LRUCache struct { } // NewLRUCache returns a new instance of LRUCache. -func NewLRUCache(maxEntries int) *LRUCache { +func NewLRUCache(maxEntries uint32) *LRUCache { c := &LRUCache{ - cache: lru.New(maxEntries), + cache: lru.New(int(maxEntries)), counts: make(map[uint64]uint64), } c.cache.OnEvicted = c.onEvicted @@ -111,15 +116,23 @@ type RankCache struct { updateN int updateTime time.Time - ThresholdLength int - ThresholdIndex int - ThresholdValue uint64 + // maxEntries is the user defined size of the cache + maxEntries uint32 + + // thresholdBuffer is used the calculate the lowest cached threshold value + // This threshold determines what new items are added to the cache + thresholdBuffer int + + // thresholdValue is the value of the last item in the cache + thresholdValue uint64 } // NewRankCache returns a new instance of RankCache. -func NewRankCache() *RankCache { +func NewRankCache(maxEntries uint32) *RankCache { return &RankCache{ - entries: make(map[uint64]uint64), + maxEntries: maxEntries, + thresholdBuffer: int(ThresholdFactor * float64(maxEntries)), + entries: make(map[uint64]uint64), } } @@ -128,7 +141,7 @@ func (c *RankCache) Add(bitmapID uint64, n uint64) { c.mu.Lock() defer c.mu.Unlock() // Ignore if the bit count on the bitmap is below the threshold. - if n < c.ThresholdValue { + if n < c.thresholdValue { return } @@ -141,7 +154,7 @@ func (c *RankCache) Add(bitmapID uint64, n uint64) { func (c *RankCache) BulkAdd(bitmapID uint64, n uint64) { c.mu.Lock() defer c.mu.Unlock() - if n < c.ThresholdValue { + if n < c.thresholdValue { return } @@ -209,19 +222,20 @@ func (c *RankCache) recalculate() { // Store the count of the item at the threshold index. c.rankings = rankings - if len(c.rankings) > c.ThresholdIndex { - c.ThresholdValue = rankings[c.ThresholdIndex].Count + if len(c.rankings) > int(c.maxEntries) { + c.thresholdValue = rankings[c.maxEntries].Count + c.rankings = c.rankings[0:c.maxEntries] } else { - c.ThresholdValue = 1 + c.thresholdValue = 1 } // Reset counters. c.updateTime, c.updateN = time.Now(), 0 // If size is larger than the threshold then trim it. - if len(c.entries) > c.ThresholdLength { + if len(c.entries) > c.thresholdBuffer { for id, cnt := range c.entries { - if cnt <= c.ThresholdValue { + if cnt <= c.thresholdValue { delete(c.entries, id) } } diff --git a/cluster.go b/cluster.go index beb33496b..f92b8c6e9 100644 --- a/cluster.go +++ b/cluster.go @@ -11,11 +11,16 @@ const ( // DefaultReplicaN is the default number of replicas per partition. DefaultReplicaN = 1 + + // HealthStatus is the return value of the /health endpoint for a node in the cluster. + HealthStatusUp = "UP" + HealthStatusDown = "DOWN" ) // Node represents a node in the cluster. type Node struct { - Host string `json:"host"` + Host string `json:"host"` + InternalHost string `json:"internalHost"` } // Nodes represents a list of nodes. @@ -81,7 +86,8 @@ func (a Nodes) Clone() []*Node { // Cluster represents a collection of nodes. type Cluster struct { - Nodes []*Node + Nodes []*Node + NodeSet NodeSet // Hashing algorithm used to assign partitions to nodes. Hasher Hasher @@ -102,6 +108,33 @@ func NewCluster() *Cluster { } } +// NodeSetHosts returns the list of host strings for NodeSet members +func (c *Cluster) NodeSetHosts() []string { + if c.NodeSet == nil { + return []string{} + } + a := make([]string, 0, len(c.NodeSet.Nodes())) + for _, m := range c.NodeSet.Nodes() { + a = append(a, m.Host) + } + return a +} + +// Health returns a map of nodes in the cluster with each node's state (UP/DOWN) as the value. +func (c *Cluster) Health() map[string]string { + h := make(map[string]string) + for _, n := range c.Nodes { + h[n.Host] = HealthStatusDown + } + // we are assuming that NodeSetHosts is a subset of c.Nodes + for _, m := range c.NodeSetHosts() { + if _, ok := h[m]; ok { + h[m] = HealthStatusUp + } + } + return h +} + // NodeByHost returns a node reference by host. func (c *Cluster) NodeByHost(host string) *Node { for _, n := range c.Nodes { diff --git a/cluster_test.go b/cluster_test.go index f5ff29edf..9dbaa420d 100644 --- a/cluster_test.go +++ b/cluster_test.go @@ -9,6 +9,7 @@ import ( "github.com/davecgh/go-spew/spew" "github.com/pilosa/pilosa" + "github.com/pilosa/pilosa/httpbroadcast" ) // Ensure the cluster can fairly distribute partitions across the nodes. @@ -77,6 +78,46 @@ func TestHasher(t *testing.T) { } } +// Ensure that an empty cluster returns a valid (empty) NodeSet +func TestCluster_NodeSetHosts(t *testing.T) { + + c := pilosa.Cluster{} + + if h := c.NodeSetHosts(); !reflect.DeepEqual(h, []string{}) { + t.Fatalf("unexpected slice of hosts: %s", h) + } +} + +// Ensure cluster can compare its Nodes and Members +func TestCluster_Health(t *testing.T) { + c := pilosa.Cluster{ + Nodes: []*pilosa.Node{ + {Host: "serverA:1000"}, + {Host: "serverB:1000"}, + {Host: "serverC:1000"}, + }, + NodeSet: &httpbroadcast.HTTPNodeSet{}, + } + + err := c.NodeSet.(*httpbroadcast.HTTPNodeSet).Join([]*pilosa.Node{ + &pilosa.Node{Host: "serverA:1000"}, + &pilosa.Node{Host: "serverC:1000"}, + &pilosa.Node{Host: "serverD:1000"}, + }) + if err != nil { + t.Fatalf("unexpected gossiper nodes: %s", err) + } + + // Verify a DOWN node is reported, and extraneous nodes are ignored + if a := c.Health(); !reflect.DeepEqual(a, map[string]string{ + "serverA:1000": pilosa.HealthStatusUp, + "serverB:1000": pilosa.HealthStatusDown, + "serverC:1000": pilosa.HealthStatusUp, + }) { + t.Fatalf("unexpected health: %s", spew.Sdump(a)) + } +} + // NewCluster returns a cluster with n nodes and uses a mod-based hasher. func NewCluster(n int) *pilosa.Cluster { c := pilosa.NewCluster() diff --git a/cmd/server.go b/cmd/server.go index 82363f849..358793af3 100644 --- a/cmd/server.go +++ b/cmd/server.go @@ -75,14 +75,18 @@ on the configured port.`, flags.StringVarP(&Server.Config.DataDir, "data-dir", "d", "~/.pilosa", "Directory to store pilosa data files.") flags.StringVarP(&Server.Config.Host, "bind", "b", ":10101", "Default URI on which pilosa should listen.") - flags.IntVarP(&Server.Config.Cluster.ReplicaN, "cluster.replicas", "", 1, "Number hosts each piece of data should be stored on.") - flags.StringSliceVarP(&Server.Config.Cluster.Nodes, "cluster.hosts", "", []string{}, "Comma separated list of hosts in cluster.") + flags.IntVarP(&Server.Config.Cluster.ReplicaN, "cluster.replicas", "", 1, "Number of hosts each piece of data should be stored on.") + flags.StringSliceVarP(&Server.Config.Cluster.Hosts, "cluster.hosts", "", []string{}, "Comma separated list of hosts in cluster.") + flags.StringSliceVarP(&Server.Config.Cluster.InternalHosts, "cluster.internal-hosts", "", []string{}, "Comma separated list of hosts in cluster used for internal communication.") flags.DurationVarP((*time.Duration)(&Server.Config.Cluster.PollingInterval), "cluster.poll-interval", "", time.Minute, "Polling interval for cluster.") // TODO what actually is this? flags.StringVarP(&Server.Config.Plugins.Path, "plugins.path", "", "", "Path to plugin directory.") flags.StringVar(&Server.Config.LogPath, "log-path", "", "Log path") flags.DurationVarP((*time.Duration)(&Server.Config.AntiEntropy.Interval), "anti-entropy.interval", "", time.Minute*10, "Interval at which to run anti-entropy routine.") flags.StringVarP(&Server.CPUProfile, "profile.cpu", "", "", "Where to store CPU profile.") flags.DurationVarP(&Server.CPUTime, "profile.cpu-time", "", 30*time.Second, "CPU profile duration.") + flags.StringVarP(&Server.Config.Cluster.Type, "cluster.type", "", "static", "Determine how the cluster handles membership and state sharing. Choose from [static, http, gossip]") + flags.StringVarP(&Server.Config.Cluster.GossipSeed, "cluster.gossip-seed", "", "", "Host with which to seed the gossip membership.") + flags.StringVarP(&Server.Config.Cluster.InternalPort, "cluster.internal-port", "", "", "Port to which pilosa should bind for internal state sharing.") return serveCmd } diff --git a/cmd/server_test.go b/cmd/server_test.go index 927ce8b66..e51825e60 100644 --- a/cmd/server_test.go +++ b/cmd/server_test.go @@ -47,7 +47,7 @@ bind = "localhost:0" v.Check(cmd.Server.Config.DataDir, actualDataDir) v.Check(cmd.Server.Config.Host, "localhost:0") v.Check(cmd.Server.Config.Cluster.ReplicaN, 2) - v.Check(cmd.Server.Config.Cluster.Nodes, []string{"example.com:10101", "example.com:10110"}) + v.Check(cmd.Server.Config.Cluster.Hosts, []string{"example.com:10101", "example.com:10110"}) v.Check(cmd.Server.Config.Cluster.PollingInterval, pilosa.Duration(time.Second*182)) return v.Error() }, @@ -68,7 +68,7 @@ data-dir = "` + actualDataDir + `" `, validation: func() error { v := validator{} - v.Check(cmd.Server.Config.Cluster.Nodes, []string{"example.com:1110", "example.com:1111"}) + v.Check(cmd.Server.Config.Cluster.Hosts, []string{"example.com:1110", "example.com:1111"}) v.Check(cmd.Server.Config.Plugins.Path, "/var/sloth") v.Check(cmd.Server.Config.AntiEntropy.Interval, pilosa.Duration(time.Minute*9)) return v.Error() @@ -94,7 +94,7 @@ data-dir = "` + actualDataDir + `" `, validation: func() error { v := validator{} - v.Check(cmd.Server.Config.Cluster.Nodes, []string{"localhost:19444"}) + v.Check(cmd.Server.Config.Cluster.Hosts, []string{"localhost:19444"}) v.Check(cmd.Server.Config.Cluster.PollingInterval, pilosa.Duration(time.Minute*2)) v.Check(cmd.Server.Config.AntiEntropy.Interval, pilosa.Duration(time.Minute*11)) v.Check(cmd.Server.CPUProfile, profFile.Name()) @@ -106,6 +106,8 @@ data-dir = "` + actualDataDir + `" // confirm log file was written info, err := logFile.Stat() if err != nil || info.Size() == 0 { + // NOTE: this test assumes that something is being written to the log + // currently, that is relying on log: "index sync monitor initializing" return errors.New("Log file was not written!") } return nil diff --git a/config.go b/config.go index 7d2efc5fe..d5d725ece 100644 --- a/config.go +++ b/config.go @@ -4,8 +4,10 @@ import "time" const ( // DefaultHost is the default hostname and port to use. - DefaultHost = "localhost" - DefaultPort = "10101" + DefaultHost = "localhost" + DefaultPort = "10101" + DefaultClusterType = "static" + DefaultInternalPort = "14000" ) // Config represents the configuration for the command. @@ -15,8 +17,12 @@ type Config struct { Cluster struct { ReplicaN int `toml:"replicas"` - Nodes []string `toml:"hosts"` + Type string `toml:"type"` + Hosts []string `toml:"hosts"` + InternalHosts []string `toml:"internal-hosts"` PollingInterval Duration `toml:"polling-interval"` + InternalPort string `toml:"internal-port"` + GossipSeed string `toml:"gossip-seed"` } `toml:"cluster"` Plugins struct { @@ -36,34 +42,14 @@ func NewConfig() *Config { Host: DefaultHost + ":" + DefaultPort, } c.Cluster.ReplicaN = DefaultReplicaN + c.Cluster.Type = DefaultClusterType c.Cluster.PollingInterval = Duration(DefaultPollingInterval) - c.Cluster.Nodes = []string{} + c.Cluster.Hosts = []string{} + c.Cluster.InternalHosts = []string{} c.AntiEntropy.Interval = Duration(DefaultAntiEntropyInterval) return c } -// NewConfigForHosts returns a Config object with Config.Cluster.Nodes already -// set up. -func NewConfigForHosts(hosts []string) *Config { - conf := NewConfig() - for _, hostport := range hosts { - conf.Cluster.Nodes = append(conf.Cluster.Nodes, hostport) - } - return conf -} - -// PilosaCluster returns a new instance of Cluster based on the config. -func (c *Config) PilosaCluster() *Cluster { - cluster := NewCluster() - cluster.ReplicaN = c.Cluster.ReplicaN - - for _, hostport := range c.Cluster.Nodes { - cluster.Nodes = append(cluster.Nodes, &Node{Host: hostport}) - } - - return cluster -} - // Duration is a TOML wrapper type for time.Duration. type Duration time.Duration diff --git a/ctl/import.go b/ctl/import.go index ad3281813..df8fb00eb 100644 --- a/ctl/import.go +++ b/ctl/import.go @@ -27,7 +27,7 @@ type ImportCommand struct { Paths []string `json:"paths"` // Size of buffer used to chunk import. - BufferSize int `json:"buffer-size"` + BufferSize int `json:"bufferSize"` // Reusable client. Client *pilosa.Client `json:"-"` diff --git a/db.go b/db.go index ef95a54bf..8bf33300e 100644 --- a/db.go +++ b/db.go @@ -43,7 +43,8 @@ type DB struct { // Profile attribute storage and cache profileAttrStore *AttrStore - stats StatsClient + broadcaster Broadcaster + stats StatsClient LogOutput io.Writer } @@ -171,7 +172,7 @@ func (db *DB) openFrames() error { // loadMeta reads meta data for the database, if any. func (db *DB) loadMeta() error { - var pb internal.DB + var pb internal.DBMeta // Read data from meta file. buf, err := ioutil.ReadFile(filepath.Join(db.path, ".meta")) @@ -197,7 +198,7 @@ func (db *DB) loadMeta() error { // saveMeta writes meta data for the database. func (db *DB) saveMeta() error { // Marshal metadata. - buf, err := proto.Marshal(&internal.DB{ + buf, err := proto.Marshal(&internal.DBMeta{ TimeQuantum: string(db.timeQuantum), ColumnLabel: db.columnLabel, }) @@ -249,10 +250,10 @@ func (db *DB) MaxSlice() uint64 { return max } -func (db *DB) SetRemoteMaxSlice(v uint64) { +func (db *DB) SetRemoteMaxSlice(newmax uint64) { db.mu.Lock() defer db.mu.Unlock() - db.remoteMaxSlice = v + db.remoteMaxSlice = newmax } // MaxInverseSlice returns the max inverse slice in the database according to this node. @@ -391,6 +392,10 @@ func (db *DB) createFrame(name string, opt FrameOptions) (*Frame, error) { if opt.RowLabel != "" { f.rowLabel = opt.RowLabel } + if opt.CacheSize != 0 { + f.cacheSize = opt.CacheSize + } + f.inverseEnabled = opt.InverseEnabled if err := f.saveMeta(); err != nil { f.Close() @@ -412,6 +417,7 @@ func (db *DB) newFrame(path, name string) (*Frame, error) { } f.LogOutput = db.LogOutput f.stats = db.stats.WithTags(fmt.Sprintf("frame:%s", name)) + f.broadcaster = db.broadcaster return f, nil } @@ -502,9 +508,40 @@ func MergeSchemas(a, b []*DBInfo) []*DBInfo { return dbs } +// encodeDBs converts a into its internal representation. +func encodeDBs(a []*DB) []*internal.DB { + other := make([]*internal.DB, len(a)) + for i := range a { + other[i] = encodeDB(a[i]) + } + return other +} + +// encodeDB converts d into its internal representation. +func encodeDB(d *DB) *internal.DB { + return &internal.DB{ + Name: d.name, + Meta: &internal.DBMeta{ + ColumnLabel: d.columnLabel, + TimeQuantum: string(d.timeQuantum), + }, + MaxSlice: d.remoteMaxSlice, + Frames: encodeFrames(d.Frames()), + } +} + // DBOptions represents options to set when initializing a db. type DBOptions struct { - ColumnLabel string `json:"columnLabel,omitempty"` + ColumnLabel string `json:"columnLabel,omitempty"` + TimeQuantum TimeQuantum `json:"timeQuantum,omitempty"` +} + +// Encode converts o into its internal representation. +func (o *DBOptions) Encode() *internal.DBMeta { + return &internal.DBMeta{ + ColumnLabel: o.ColumnLabel, + TimeQuantum: string(o.TimeQuantum), + } } // hasTime returns true if a contains a non-nil time. diff --git a/executor_test.go b/executor_test.go index be005e732..7a0c95622 100644 --- a/executor_test.go +++ b/executor_test.go @@ -470,7 +470,9 @@ func TestExecutor_Execute_Remote_Bitmap(t *testing.T) { t.Fatalf("unexpected db: %s", db) } else if query.String() != `Bitmap(frame="f", id=10)` { t.Fatalf("unexpected query: %s", query.String()) - } else if !reflect.DeepEqual(slices, []uint64{0}) { //TODO: this is incorrect because the calling node doesn't know about slice 2 + // NOTE: while the following is technically incorrect (it should be {0, 2}) because the calling node doesn't know about slice 2 yet, + // we are ok with this and assuming that the calling node will become aware of slice 2 via inter-node messaging + } else if !reflect.DeepEqual(slices, []uint64{0}) { t.Fatalf("unexpected slices: %+v", slices) } diff --git a/fragment.go b/fragment.go index 063b7e17c..9bee4288e 100644 --- a/fragment.go +++ b/fragment.go @@ -70,6 +70,7 @@ type Fragment struct { // Cache for bitmap counts. cacheType string // passed in by frame cache Cache + cacheSize uint32 // Cache containing full bitmaps (not just counts). bitmapCache BitmapCache @@ -101,6 +102,7 @@ func NewFragment(path, db, frame, view string, slice uint64) *Fragment { view: view, slice: slice, cacheType: DefaultCacheType, + cacheSize: DefaultCacheSize, LogOutput: ioutil.Discard, MaxOpN: DefaultFragmentMaxOpN, @@ -222,12 +224,9 @@ func (f *Fragment) openCache() error { // Determine cache type from frame name. switch f.cacheType { case CacheTypeRanked: - c := NewRankCache() - c.ThresholdLength = 50000 - c.ThresholdIndex = 45000 - f.cache = c + f.cache = NewRankCache(f.cacheSize) case CacheTypeLRU: - f.cache = NewLRUCache(50000) + f.cache = NewLRUCache(f.cacheSize) default: return ErrInvalidCacheType } diff --git a/fragment_test.go b/fragment_test.go index b915bd449..f54f5415b 100644 --- a/fragment_test.go +++ b/fragment_test.go @@ -277,6 +277,73 @@ func TestFragment_TopN_BitmapIDs(t *testing.T) { } } +// Ensure the fragment cache limit works +func TestFragment_TopN_CacheSize(t *testing.T) { + slice := uint64(0) + cacheSize := uint32(3) + + // Create DB. + db := MustOpenDB() + defer db.Close() + + // Create frame. + frame, err := db.CreateFrameIfNotExists("f", pilosa.FrameOptions{CacheType: pilosa.CacheTypeRanked, CacheSize: cacheSize}) + if err != nil { + t.Fatal(err) + } + + // Create view. + view, err := frame.CreateViewIfNotExists(pilosa.ViewStandard) + if err != nil { + t.Fatal(err) + } + + // Create fragment. + frag, err := view.CreateFragmentIfNotExists(slice) + if err != nil { + t.Fatal(err) + } + // Close the storage so we can re-open it without encountering a flock. + frag.Close() + + f := &Fragment{ + Fragment: frag, + BitmapAttrStore: MustOpenAttrStore(), + } + f.Fragment.BitmapAttrStore = f.BitmapAttrStore.AttrStore + if err := f.Open(); err != nil { + panic(err) + } + defer f.Close() + + // Set bits on various bitmaps. + f.MustSetBits(100, 1, 2, 3) + f.MustSetBits(101, 4, 5, 6, 7) + f.MustSetBits(102, 8, 9, 10, 11, 12) + f.MustSetBits(103, 8, 9, 10, 11, 12, 13) + f.MustSetBits(104, 8, 9, 10, 11, 12, 13, 14) + f.MustSetBits(105, 10, 11) + + f.RecalculateCache() + + p := []pilosa.Pair{ + {ID: 104, Count: 7}, + {ID: 103, Count: 6}, + {ID: 102, Count: 5}, + } + + // Retrieve top bitmaps. + if pairs, err := f.Top(pilosa.TopOptions{N: 5}); err != nil { + t.Fatal(err) + } else if len(pairs) > int(cacheSize) { + t.Fatalf("TopN count cannot exceed cache size: %d", cacheSize) + } else if pairs[0] != (pilosa.Pair{ID: 104, Count: 7}) { + t.Fatalf("unexpected pair(0): %v", pairs) + } else if !reflect.DeepEqual(pairs, p) { + t.Fatalf("Invalid TopN result set: %s", spew.Sdump(pairs)) + } +} + // Ensure fragment can return a checksum for its blocks. func TestFragment_Checksum(t *testing.T) { f := MustOpenFragment("d", "f", pilosa.ViewStandard, 0) diff --git a/frame.go b/frame.go index 414af55c6..684b2225d 100644 --- a/frame.go +++ b/frame.go @@ -20,6 +20,9 @@ const ( DefaultRowLabel = "id" DefaultCacheType = CacheTypeLRU DefaultInverseEnabled = false + + // Default ranked frame cache + DefaultCacheSize = 50000 ) // Frame represents a container for views. @@ -35,13 +38,17 @@ type Frame struct { // Bitmap attribute storage and cache bitmapAttrStore *AttrStore - stats StatsClient + broadcaster Broadcaster + stats StatsClient // Frame settings. rowLabel string cacheType string inverseEnabled bool + // Cache size for ranked frames + cacheSize uint32 + LogOutput io.Writer } @@ -63,8 +70,9 @@ func NewFrame(path, db, name string) (*Frame, error) { stats: NopStatsClient, rowLabel: DefaultRowLabel, - cacheType: DefaultCacheType, inverseEnabled: DefaultInverseEnabled, + cacheType: DefaultCacheType, + cacheSize: DefaultCacheSize, LogOutput: ioutil.Discard, }, nil @@ -149,13 +157,43 @@ func (f *Frame) InverseEnabled() bool { return f.inverseEnabled } +// SetCacheSize sets the cache size for ranked fames. Persists to meta file on update. +// defaults to DefaultCacheSize 50000 +func (f *Frame) SetCacheSize(v uint32) error { + f.mu.Lock() + defer f.mu.Unlock() + + // Ignore if no change occurred. + if v == 0 || f.cacheSize == v { + return nil + } + + // Persist meta data to disk on change. + f.cacheSize = v + if err := f.saveMeta(); err != nil { + return err + } + + return nil +} + +// CacheSize returns the ranked frame cache size. +func (f *Frame) CacheSize() uint32 { + f.mu.Lock() + v := f.cacheSize + f.mu.Unlock() + return v +} + // Options returns all options for this frame. func (f *Frame) Options() FrameOptions { f.mu.Lock() opt := FrameOptions{ RowLabel: f.rowLabel, - CacheType: f.cacheType, InverseEnabled: f.inverseEnabled, + CacheType: f.cacheType, + CacheSize: f.cacheSize, + TimeQuantum: f.timeQuantum, } f.mu.Unlock() return opt @@ -226,7 +264,7 @@ func (f *Frame) openViews() error { // loadMeta reads meta data for the frame, if any. func (f *Frame) loadMeta() error { - var pb internal.Frame + var pb internal.FrameMeta // Read data from meta file. buf, err := ioutil.ReadFile(filepath.Join(f.path, ".meta")) @@ -235,6 +273,7 @@ func (f *Frame) loadMeta() error { f.rowLabel = DefaultRowLabel f.cacheType = DefaultCacheType f.inverseEnabled = DefaultInverseEnabled + f.cacheSize = DefaultCacheSize return nil } else if err != nil { return err @@ -248,6 +287,7 @@ func (f *Frame) loadMeta() error { f.timeQuantum = TimeQuantum(pb.TimeQuantum) f.rowLabel = pb.RowLabel f.inverseEnabled = pb.InverseEnabled + f.cacheSize = pb.CacheSize // Copy cache type. f.cacheType = pb.CacheType @@ -261,11 +301,12 @@ func (f *Frame) loadMeta() error { // saveMeta writes meta data for the frame. func (f *Frame) saveMeta() error { // Marshal metadata. - buf, err := proto.Marshal(&internal.Frame{ - TimeQuantum: string(f.timeQuantum), + buf, err := proto.Marshal(&internal.FrameMeta{ RowLabel: f.rowLabel, - CacheType: f.cacheType, InverseEnabled: f.inverseEnabled, + CacheType: f.cacheType, + CacheSize: f.cacheSize, + TimeQuantum: string(f.timeQuantum), }) if err != nil { return err @@ -377,7 +418,7 @@ func (f *Frame) CreateViewIfNotExists(name string) (*View, error) { } func (f *Frame) newView(path, name string) *View { - view := NewView(path, f.db, f.name, name) + view := NewView(path, f.db, f.name, name, f.cacheSize) view.cacheType = f.cacheType view.LogOutput = f.LogOutput view.BitmapAttrStore = f.bitmapAttrStore @@ -545,6 +586,29 @@ func (f *Frame) Import(bitmapIDs, profileIDs []uint64, timestamps []*time.Time) return nil } +// encodeFrames converts a into its internal representation. +func encodeFrames(a []*Frame) []*internal.Frame { + other := make([]*internal.Frame, len(a)) + for i := range a { + other[i] = encodeFrame(a[i]) + } + return other +} + +// encodeFrame converts f into its internal representation. +func encodeFrame(f *Frame) *internal.Frame { + return &internal.Frame{ + Name: f.name, + Meta: &internal.FrameMeta{ + RowLabel: f.rowLabel, + InverseEnabled: f.inverseEnabled, + CacheType: f.cacheType, + CacheSize: f.cacheSize, + TimeQuantum: string(f.timeQuantum), + }, + } +} + type frameSlice []*Frame func (p frameSlice) Swap(i, j int) { p[i], p[j] = p[j], p[i] } @@ -565,9 +629,22 @@ func (p frameInfoSlice) Less(i, j int) bool { return p[i].Name < p[j].Name } // FrameOptions represents options to set when initializing a frame. type FrameOptions struct { - RowLabel string `json:"rowLabel,omitempty"` - CacheType string `json:"cacheType,omitempty"` - InverseEnabled bool `json:"inverseEnabled,omitempty"` + RowLabel string `json:"rowLabel,omitempty"` + InverseEnabled bool `json:"inverseEnabled,omitempty"` + CacheType string `json:"cacheType,omitempty"` + CacheSize uint32 `json:"cacheSize,omitempty"` + TimeQuantum TimeQuantum `json:"timeQuantum,omitempty"` +} + +// Encode converts o into its internal representation. +func (o *FrameOptions) Encode() *internal.FrameMeta { + return &internal.FrameMeta{ + RowLabel: o.RowLabel, + InverseEnabled: o.InverseEnabled, + CacheType: o.CacheType, + CacheSize: o.CacheSize, + TimeQuantum: string(o.TimeQuantum), + } } // importBitSet represents slices of row and column ids. diff --git a/frame_test.go b/frame_test.go index 5b659f826..4814ca67e 100644 --- a/frame_test.go +++ b/frame_test.go @@ -126,3 +126,24 @@ func (f *Frame) MustSetBit(view string, bitmapID, profileID uint64, t *time.Time } return changed } + +// Ensure frame can set its cache +func TestFrame_SetCacheSize(t *testing.T) { + f := MustOpenFrame() + defer f.Close() + cacheSize := uint32(100) + + // Set & retrieve frame cache size. + if err := f.SetCacheSize(cacheSize); err != nil { + t.Fatal(err) + } else if q := f.CacheSize(); q != cacheSize { + t.Fatalf("unexpected frame cache size: %d", q) + } + + // Reload frame and verify that it is persisted. + if err := f.Reopen(); err != nil { + t.Fatal(err) + } else if q := f.CacheSize(); q != cacheSize { + t.Fatalf("unexpected frame cache size (reopen): %d", q) + } +} diff --git a/glide.lock b/glide.lock index 7fffeca5e..c142737ff 100644 --- a/glide.lock +++ b/glide.lock @@ -1,6 +1,8 @@ -hash: 743e8f978eb4ad8f80a2ab71b05caebbf50b6769b71aa457bc4f144fef8c6595 -updated: 2017-04-18T15:33:39.035615802-05:00 +hash: 4bdea17c62dcd469584382515052e7a246fae6deefc2d62da70bf768e18e1f0c +updated: 2017-04-19T10:51:10.409094081-05:00 imports: +- name: github.com/armon/go-metrics + version: 97c69685293dce4c0a2d0b19535179bbc976e4d2 - name: github.com/boltdb/bolt version: 4b1ebc1869ad66568b313d0dc410e2be72670dda - name: github.com/BurntSushi/toml @@ -24,13 +26,21 @@ imports: subpackages: - lru - name: github.com/golang/protobuf - version: 888eb0692c857ec880338addf316bd662d5e630e + version: 8ee79997227bf9b34611aee7946ae64735e6fd93 subpackages: - proto - name: github.com/gorilla/context version: 08b5f424b9271eedf6f9f0ce86cb9396ed337a42 - name: github.com/gorilla/mux version: 392c28fe23e1c45ddba891b0320b3b5df220beea +- name: github.com/hashicorp/errwrap + version: 7554cd9344cec97297fa6649b055a8c98c2a1e55 +- name: github.com/hashicorp/go-msgpack + version: fa3f63826f7c23912c15263591e65d54d080b458 + subpackages: + - codec +- name: github.com/hashicorp/go-multierror + version: ed905158d87462226a13fe39ddf685ea65f1c11f - name: github.com/hashicorp/hcl version: 630949a3c5fa3c613328e1b8256052cbc2327c9b subpackages: @@ -42,10 +52,14 @@ imports: - json/parser - json/scanner - json/token +- name: github.com/hashicorp/memberlist + version: 9800c50ab79c002353852a9b1095e9591b161513 - name: github.com/inconshreveable/mousetrap version: 76626ae9c91c4f2a10f34cad8ce83ea42c93bb75 - name: github.com/magiconair/properties version: b3b15ef068fd0b17ddf408a23669f20811d194d2 +- name: github.com/miekg/dns + version: ca336a1f95a6b89be9c250df26c7a41742eb4a6f - name: github.com/mitchellh/mapstructure version: db1efb556f84b25a0a13a04aad883943538ad2e0 - name: github.com/pelletier/go-buffruneio @@ -68,6 +82,14 @@ imports: version: 9ff6c6923cfffbcd502984b8e0c80539a94968b7 - name: github.com/spf13/viper version: 7538d73b4eb9511d85a9f1dfef202eeb8ac260f4 +- name: golang.org/x/net + version: 60c41d1de8da134c05b7b40154a9a82bf5b7edb9 + subpackages: + - context +- name: golang.org/x/sync + version: 450f422ab23cf9881c94e2db30cac0eb1b7cf80c + subpackages: + - errgroup - name: golang.org/x/sys version: c200b10b5d5e122be351b67af224adc6128af5bf subpackages: diff --git a/glide.yaml b/glide.yaml index d4615b0ba..a7c0e98eb 100644 --- a/glide.yaml +++ b/glide.yaml @@ -31,3 +31,5 @@ import: - package: github.com/spf13/viper - package: github.com/gorilla/mux version: ^1.3.0 +- package: github.com/hashicorp/memberlist +- package: golang.org/x/sync diff --git a/gossip/gossip.go b/gossip/gossip.go new file mode 100644 index 000000000..772892926 --- /dev/null +++ b/gossip/gossip.go @@ -0,0 +1,225 @@ +package gossip + +import ( + "fmt" + "io" + "log" + "os" + + "golang.org/x/sync/errgroup" + + "github.com/gogo/protobuf/proto" + "github.com/hashicorp/memberlist" + "github.com/pilosa/pilosa" + "github.com/pilosa/pilosa/internal" +) + +// StateHandler specifies two methods which an object must implement to share +// state in the cluster. These are used by the GossipNodeSet to implement the +// LocalState and MergeRemoteState methods of memberlist.Delegate +type StateHandler interface { + LocalState() (proto.Message, error) + HandleRemoteState(proto.Message) error +} + +// GossipNodeSet represents a gossip implementation of NodeSet using memberlist +// GossipNodeSet also represents a gossip implementation of pilosa.Broadcaster +// GossipNodeSet also represents an implementation of memberlist.Delegate +type GossipNodeSet struct { + memberlist *memberlist.Memberlist + handler pilosa.BroadcastHandler + + broadcasts *memberlist.TransmitLimitedQueue + + stateHandler StateHandler + config *GossipConfig + + // The writer for any logging. + LogOutput io.Writer +} + +func (g *GossipNodeSet) Nodes() []*pilosa.Node { + a := make([]*pilosa.Node, 0, g.memberlist.NumMembers()) + for _, n := range g.memberlist.Members() { + a = append(a, &pilosa.Node{Host: n.Name}) + } + return a +} + +func (g *GossipNodeSet) Start(h pilosa.BroadcastHandler) error { + g.handler = h + return nil +} + +func (g *GossipNodeSet) Open() error { + if g.handler == nil { + return fmt.Errorf("opening GossipNodeSet: you must call Start(pilosa.BroadcastHandler) before calling Open()") + } + ml, err := memberlist.Create(g.config.memberlistConfig) + if err != nil { + return err + } + g.memberlist = ml + + // attach to gossip seed node + nodes := []*pilosa.Node{&pilosa.Node{Host: g.config.gossipSeed}} //TODO: support a list of seeds + _, err = g.memberlist.Join(pilosa.Nodes(nodes).Hosts()) + if err != nil { + return err + } + g.broadcasts = &memberlist.TransmitLimitedQueue{ + NumNodes: func() int { + return ml.NumMembers() + }, + RetransmitMult: 3, + } + return nil +} + +// logger returns a logger for the GossipNodeSet. +func (g *GossipNodeSet) logger() *log.Logger { + return log.New(g.LogOutput, "", log.LstdFlags) +} + +//////////////////////////////////////////////////////////////// + +type GossipConfig struct { + gossipSeed string + memberlistConfig *memberlist.Config +} + +// NewGossipNodeSet returns a new instance of GossipNodeSet. +func NewGossipNodeSet(name string, gossipHost string, gossipPort int, gossipSeed string, sh StateHandler) *GossipNodeSet { + g := &GossipNodeSet{ + LogOutput: os.Stderr, + } + + //TODO: pull memberlist config from pilosa.cfg file + g.config = &GossipConfig{ + memberlistConfig: memberlist.DefaultLocalConfig(), + gossipSeed: gossipSeed, + } + g.config.memberlistConfig.Name = name + g.config.memberlistConfig.BindAddr = gossipHost + g.config.memberlistConfig.BindPort = gossipPort + g.config.memberlistConfig.AdvertiseAddr = gossipHost + g.config.memberlistConfig.AdvertisePort = gossipPort + g.config.memberlistConfig.Delegate = g + + g.stateHandler = sh + + return g +} + +// SendSync implementation of the Broadcaster interface +func (g *GossipNodeSet) SendSync(pb proto.Message) error { + msg, err := pilosa.MarshalMessage(pb) + if err != nil { + return err + } + + mlist := g.memberlist + + // Direct sends the message directly to every node. + // An error from any node raises an error on the entire operation. + // + // Gossip uses the gossip protocol to eventually deliver the message + // to every node. + var eg errgroup.Group + for _, n := range mlist.Members() { + // Don't send the message to the local node. + if n == mlist.LocalNode() { + continue + } + node := n + eg.Go(func() error { + return mlist.SendToTCP(node, msg) + }) + } + return eg.Wait() +} + +// SendAsync implementation of the Broadcaster interface +func (g *GossipNodeSet) SendAsync(pb proto.Message) error { + msg, err := pilosa.MarshalMessage(pb) + if err != nil { + return err + } + + b := &broadcast{ + msg: msg, + notify: nil, + } + g.broadcasts.QueueBroadcast(b) + return nil +} + +// implementation of the memberlist.Delegate interface +func (g *GossipNodeSet) NodeMeta(limit int) []byte { + return []byte{} +} + +func (g *GossipNodeSet) NotifyMsg(b []byte) { + m, err := pilosa.UnmarshalMessage(b) + if err != nil { + g.logger().Printf("unmarshal message error: %s", err) + return + } + if err := g.handler.ReceiveMessage(m); err != nil { + g.logger().Printf("receive message error: %s", err) + return + } +} + +func (g *GossipNodeSet) GetBroadcasts(overhead, limit int) [][]byte { + return g.broadcasts.GetBroadcasts(overhead, limit) +} + +func (g *GossipNodeSet) LocalState(join bool) []byte { + pb, err := g.stateHandler.LocalState() + if err != nil { + g.logger().Printf("error getting local state, err=%s", err) + return []byte{} + } + + // Marshal nodestate data to bytes. + buf, err := proto.Marshal(pb) + if err != nil { + g.logger().Printf("error marshalling nodestate data, err=%s", err) + return []byte{} + } + return buf +} + +func (g *GossipNodeSet) MergeRemoteState(buf []byte, join bool) { + // Unmarshal nodestate data. + var pb internal.NodeState + if err := proto.Unmarshal(buf, &pb); err != nil { + g.logger().Printf("error unmarshalling nodestate data, err=%s", err) + return + } + err := g.stateHandler.HandleRemoteState(&pb) + if err != nil { + g.logger().Printf("merge state error: %s", err) + } +} + +// broadcast represents an implementation of memberlist.Broadcast +type broadcast struct { + msg []byte + notify chan<- struct{} +} + +func (b *broadcast) Invalidates(other memberlist.Broadcast) bool { + return false +} + +func (b *broadcast) Message() []byte { + return b.msg +} + +func (b *broadcast) Finished() { + if b.notify != nil { + close(b.notify) + } +} diff --git a/handler.go b/handler.go index 5af464764..ef5ad4f8d 100644 --- a/handler.go +++ b/handler.go @@ -26,7 +26,8 @@ import ( // Handler represents an HTTP handler. type Handler struct { - Index *Index + Index *Index + Broadcaster Broadcaster // Local hostname & cluster configuration. Host string @@ -112,10 +113,23 @@ 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) { + if err := json.NewEncoder(w).Encode(getStatusResponse{ + Health: h.Cluster.Health(), + }); err != nil { + h.logger().Printf("write status response error: %s", err) + } +} + type getSchemaResponse struct { DBs []*DBInfo `json:"dbs"` } +type getStatusResponse struct { + Health map[string]string `json:"health"` +} + // handlePostQuery handles /query requests. func (h *Handler) handlePostQuery(w http.ResponseWriter, r *http.Request) { dbName := mux.Vars(r)["db"] @@ -201,7 +215,7 @@ func (h *Handler) handleGetSliceMax(w http.ResponseWriter, r *http.Request) { } type sliceMaxResponse struct { - MaxSlices map[string]uint64 `json:"MaxSlices"` + MaxSlices map[string]uint64 `json:"maxSlices"` } // handleGetDBs handles GET /db request. @@ -303,6 +317,15 @@ func (h *Handler) handleDeleteDB(w http.ResponseWriter, r *http.Request) { return } + // Send the delete database message to all nodes. + err := h.Broadcaster.SendSync( + &internal.DeleteDBMessage{ + DB: dbName, + }) + if err != nil { + h.logger().Printf("problem sending DeleteDB message: %s", err) + } + // Encode response. if err := json.NewEncoder(w).Encode(deleteDBResponse{}); err != nil { h.logger().Printf("response encoding error: %s", err) @@ -317,13 +340,17 @@ func (h *Handler) handlePostDB(w http.ResponseWriter, r *http.Request) { // Decode request. var req postDBRequest - if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + err := json.NewDecoder(r.Body).Decode(&req) + if err == io.EOF { + // If no data was provided (EOF), we still create the database + // with default values. + } else if err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } // Create database. - _, err := h.Index.CreateDB(dbName, req.Options) + _, err = h.Index.CreateDB(dbName, req.Options) if err == ErrDatabaseExists { http.Error(w, err.Error(), http.StatusConflict) return @@ -332,6 +359,16 @@ func (h *Handler) handlePostDB(w http.ResponseWriter, r *http.Request) { return } + // Send the create database message to all nodes. + err = h.Broadcaster.SendSync( + &internal.CreateDBMessage{ + DB: dbName, + Meta: req.Options.Encode(), + }) + if err != nil { + h.logger().Printf("problem sending CreateDB message: %s", err) + } + // Encode response. if err := json.NewEncoder(w).Encode(postDBResponse{}); err != nil { h.logger().Printf("response encoding error: %s", err) @@ -376,7 +413,7 @@ func (h *Handler) handlePatchDBTimeQuantum(w http.ResponseWriter, r *http.Reques } type patchDBTimeQuantumRequest struct { - TimeQuantum string `json:"time_quantum"` + TimeQuantum string `json:"timeQuantum"` } type patchDBTimeQuantumResponse struct{} @@ -445,7 +482,11 @@ func (h *Handler) handlePostFrame(w http.ResponseWriter, r *http.Request) { // Decode request. var req postFrameRequest - if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + err := json.NewDecoder(r.Body).Decode(&req) + if err == io.EOF { + // If no data was provided (EOF), we still create the frame + // with default values. + } else if err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } @@ -458,7 +499,7 @@ func (h *Handler) handlePostFrame(w http.ResponseWriter, r *http.Request) { } // Create frame. - _, err := db.CreateFrame(frameName, req.Options) + _, err = db.CreateFrame(frameName, req.Options) if err == ErrFrameExists { http.Error(w, err.Error(), http.StatusConflict) return @@ -467,6 +508,17 @@ func (h *Handler) handlePostFrame(w http.ResponseWriter, r *http.Request) { return } + // Send the create frame message to all nodes. + err = h.Broadcaster.SendSync( + &internal.CreateFrameMessage{ + DB: dbName, + Frame: frameName, + Meta: req.Options.Encode(), + }) + if err != nil { + h.logger().Printf("problem sending CreateFrame message: %s", err) + } + // Encode response. if err := json.NewEncoder(w).Encode(postFrameResponse{}); err != nil { h.logger().Printf("response encoding error: %s", err) @@ -538,6 +590,16 @@ func (h *Handler) handleDeleteFrame(w http.ResponseWriter, r *http.Request) { return } + // Send the delete frame message to all nodes. + err := h.Broadcaster.SendSync( + &internal.DeleteFrameMessage{ + DB: dbName, + Frame: frameName, + }) + if err != nil { + h.logger().Printf("problem sending DeleteFrame message: %s", err) + } + // Encode response. if err := json.NewEncoder(w).Encode(deleteFrameResponse{}); err != nil { h.logger().Printf("response encoding error: %s", err) @@ -585,7 +647,7 @@ func (h *Handler) handlePatchFrameTimeQuantum(w http.ResponseWriter, r *http.Req } type patchFrameTimeQuantumRequest struct { - TimeQuantum string `json:"time_quantum"` + TimeQuantum string `json:"timeQuantum"` } type patchFrameTimeQuantumResponse struct{} diff --git a/handler_test.go b/handler_test.go index c19a83a93..049fdc036 100644 --- a/handler_test.go +++ b/handler_test.go @@ -83,7 +83,7 @@ func TestHandler_MaxSlices(t *testing.T) { h.ServeHTTP(w, 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":{"d0":3,"d1":0}}`+"\n" { + } else if body := w.Body.String(); body != `{"maxSlices":{"d0":3,"d1":0}}`+"\n" { t.Fatalf("unexpected body: %s", body) } } @@ -123,7 +123,7 @@ func TestHandler_MaxSlices_Inverse(t *testing.T) { h.ServeHTTP(w, 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":{"d0":3,"d1":0}}`+"\n" { + } else if body := w.Body.String(); body != `{"maxSlices":{"d0":3,"d1":0}}`+"\n" { t.Fatalf("unexpected body: %s", body) } } @@ -556,7 +556,7 @@ func TestHandler_SetDBTimeQuantum(t *testing.T) { h := NewHandler() h.Index = idx.Index w := httptest.NewRecorder() - h.ServeHTTP(w, MustNewHTTPRequest("PATCH", "/db/d0/time-quantum", strings.NewReader(`{"time_quantum":"ymdh"}`))) + h.ServeHTTP(w, MustNewHTTPRequest("PATCH", "/db/d0/time-quantum", strings.NewReader(`{"timeQuantum":"ymdh"}`))) if w.Code != http.StatusOK { t.Fatalf("unexpected status code: %d", w.Code) } else if body := w.Body.String(); body != `{}`+"\n" { @@ -579,7 +579,7 @@ func TestHandler_SetFrameTimeQuantum(t *testing.T) { h := NewHandler() h.Index = idx.Index w := httptest.NewRecorder() - h.ServeHTTP(w, MustNewHTTPRequest("PATCH", "/db/d0/frame/f1/time-quantum", strings.NewReader(`{"time_quantum":"ymdh"}`))) + h.ServeHTTP(w, MustNewHTTPRequest("PATCH", "/db/d0/frame/f1/time-quantum", strings.NewReader(`{"timeQuantum":"ymdh"}`))) if w.Code != http.StatusOK { t.Fatalf("unexpected status code: %d", w.Code) } else if body := w.Body.String(); body != `{}`+"\n" { @@ -763,7 +763,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 w.Body.String() != `[{"host":"host1"},{"host":"host2"}]`+"\n" { + } else if w.Body.String() != `[{"host":"host1","internalHost":""},{"host":"host2","internalHost":""}]`+"\n" { t.Fatalf("unexpected body: %q", w.Body.String()) } } @@ -792,6 +792,10 @@ func NewHandler() *Handler { } h.Handler.Executor = &h.Executor h.Handler.LogOutput = ioutil.Discard + + // Handler test messages can no-op. + h.Broadcaster = pilosa.NopBroadcaster + return h } @@ -823,6 +827,8 @@ func NewServer() *Server { // Update handler to use hostname. s.Handler.Host = s.Host() + // Handler test messages can no-op. + s.Handler.Broadcaster = pilosa.NopBroadcaster // Create a default cluster on the handler s.Handler.Cluster = NewCluster(1) s.Handler.Cluster.Nodes[0].Host = s.Host() diff --git a/httpbroadcast/messenger.go b/httpbroadcast/messenger.go new file mode 100644 index 000000000..ebd4b2e6a --- /dev/null +++ b/httpbroadcast/messenger.go @@ -0,0 +1,180 @@ +package httpbroadcast + +import ( + "bytes" + "errors" + "fmt" + "io" + "io/ioutil" + "net/http" + "net/url" + + "golang.org/x/sync/errgroup" + + "github.com/gogo/protobuf/proto" + "github.com/pilosa/pilosa" +) + +// HTTPBroadcaster represents a NodeSet that broadcasts messages over HTTP. +type HTTPBroadcaster struct { + server *pilosa.Server + internalPort string +} + +// NewHTTPBroadcaster returns a new instance of HTTPBroadcaster. +func NewHTTPBroadcaster(s *pilosa.Server, internalPort string) *HTTPBroadcaster { + return &HTTPBroadcaster{server: s, internalPort: internalPort} +} + +// SendSync sends a protobuf message to all nodes simultaneously. +// It waits for all nodes to respond before the function returns (and returns any errors). +func (h *HTTPBroadcaster) SendSync(pb proto.Message) error { + // Marshal the pb to []byte + buf, err := pilosa.MarshalMessage(pb) + if err != nil { + return err + } + + nodes, err := h.nodes() + if err != nil { + return err + } + + var g errgroup.Group + for _, n := range nodes { + // Don't send the message to the local node. + if n.Host == h.server.Host { + continue + } + node := n + g.Go(func() error { + return h.sendNodeMessage(node, buf) + }) + } + return g.Wait() +} + +// SendAsync exists to implement the Broadcaster interface, but just calls +// SendSync. +func (h *HTTPBroadcaster) SendAsync(pb proto.Message) error { + return h.SendSync(pb) +} + +func (h *HTTPBroadcaster) nodes() ([]*pilosa.Node, error) { + if h.server == nil { + return nil, errors.New("HTTPBroadcaster has no reference to Server.") + } + nodeset, ok := h.server.Cluster.NodeSet.(*HTTPNodeSet) + if !ok { + return nil, errors.New("NodeSet cannot be caste to HTTPNodeSet.") + } + return nodeset.Nodes(), nil +} + +func (h *HTTPBroadcaster) sendNodeMessage(node *pilosa.Node, msg []byte) error { + var client *http.Client + client = http.DefaultClient + + // Create HTTP request. + req, err := http.NewRequest("POST", (&url.URL{ + Scheme: "http", + Host: node.InternalHost, + }).String(), bytes.NewReader(msg)) + + // Require protobuf encoding. + req.Header.Set("Content-Type", "application/x-protobuf") + + // Send request to remote node. + resp, err := client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + + // Read response into buffer. + body, err := ioutil.ReadAll(resp.Body) + + if err != nil { + return err + } + + // Check status code. + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("invalid status: code=%d, err=%s", resp.StatusCode, body) + } + + return nil +} + +type HTTPBroadcastReceiver struct { + port string + handler pilosa.BroadcastHandler + logOutput io.Writer +} + +func NewHTTPBroadcastReceiver(port string, logOutput io.Writer) *HTTPBroadcastReceiver { + return &HTTPBroadcastReceiver{ + port: port, + logOutput: logOutput, + } +} + +func (rec *HTTPBroadcastReceiver) Start(b pilosa.BroadcastHandler) error { + rec.handler = b + go func() { + err := http.ListenAndServe(":"+rec.port, rec) + if err != nil { + fmt.Fprintf(rec.logOutput, "Error listening on %v for HTTPBroadcastReceiver: %v\n", ":"+rec.port, err) + } + }() + return nil +} + +func (rec *HTTPBroadcastReceiver) ServeHTTP(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Content-Type") != "application/x-protobuf" { + http.Error(w, "Unsupported media type", http.StatusUnsupportedMediaType) + return + } + + // Read entire body. + body, err := ioutil.ReadAll(r.Body) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + + // Unmarshal message to specific proto type. + m, err := pilosa.UnmarshalMessage(body) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + + if err := rec.handler.ReceiveMessage(m); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } +} + +// HTTPNodeSet represents a NodeSet that broadcasts messages over HTTP. +type HTTPNodeSet struct { + nodes []*pilosa.Node +} + +// NewHTTPNodeSet returns a new instance of HTTPNodeSet. +func NewHTTPNodeSet() *HTTPNodeSet { + return &HTTPNodeSet{} +} + +func (h *HTTPNodeSet) Nodes() []*pilosa.Node { + return h.nodes +} + +func (h *HTTPNodeSet) Open() error { + return nil +} + +func (h *HTTPNodeSet) Join(nodes []*pilosa.Node) error { + h.nodes = nodes + return nil +} diff --git a/index.go b/index.go index 7597984ba..eb8f4b204 100644 --- a/index.go +++ b/index.go @@ -23,6 +23,7 @@ type Index struct { // Databases by name. dbs map[string]*DB + Broadcaster Broadcaster // Close management wg sync.WaitGroup closing chan struct{} @@ -181,6 +182,7 @@ func (i *Index) DBs() []*DB { } // CreateDB creates a database. +// An error is returned if the database already exists. func (i *Index) CreateDB(name string, opt DBOptions) (*DB, error) { i.mu.Lock() defer i.mu.Unlock() @@ -198,7 +200,7 @@ func (i *Index) CreateDBIfNotExists(name string, opt DBOptions) (*DB, error) { i.mu.Lock() defer i.mu.Unlock() - // Find frame in cache first. + // Find database in cache first. if db := i.dbs[name]; db != nil { return db, nil } @@ -228,6 +230,7 @@ func (i *Index) createDB(name string, opt DBOptions) (*DB, error) { // Update options. db.SetColumnLabel(opt.ColumnLabel) + db.SetTimeQuantum(opt.TimeQuantum) i.dbs[db.Name()] = db @@ -243,6 +246,7 @@ func (i *Index) newDB(path, name string) (*DB, error) { } db.LogOutput = i.LogOutput db.stats = i.Stats.WithTags(fmt.Sprintf("db:%s", db.Name())) + db.broadcaster = i.Broadcaster return db, nil } diff --git a/internal/private.pb.go b/internal/private.pb.go index 1ffa98885..0423cfa08 100644 --- a/internal/private.pb.go +++ b/internal/private.pb.go @@ -9,13 +9,21 @@ private.proto It has these top-level messages: - DB - Frame + DBMeta + FrameMeta ImportResponse BlockDataRequest BlockDataResponse Cache MaxSlicesResponse + CreateSliceMessage + DeleteDBMessage + CreateDBMessage + CreateFrameMessage + DeleteFrameMessage + Frame + DB + NodeState */ package internal @@ -36,27 +44,28 @@ var _ = math.Inf // proto package needs to be updated. const _ = proto.ProtoPackageIsVersion2 // please upgrade the proto package -type DB struct { - TimeQuantum string `protobuf:"bytes,1,opt,name=TimeQuantum,proto3" json:"TimeQuantum,omitempty"` - ColumnLabel string `protobuf:"bytes,2,opt,name=ColumnLabel,proto3" json:"ColumnLabel,omitempty"` +type DBMeta struct { + ColumnLabel string `protobuf:"bytes,1,opt,name=ColumnLabel,proto3" json:"ColumnLabel,omitempty"` + TimeQuantum string `protobuf:"bytes,2,opt,name=TimeQuantum,proto3" json:"TimeQuantum,omitempty"` } -func (m *DB) Reset() { *m = DB{} } -func (m *DB) String() string { return proto.CompactTextString(m) } -func (*DB) ProtoMessage() {} -func (*DB) Descriptor() ([]byte, []int) { return fileDescriptorPrivate, []int{0} } +func (m *DBMeta) Reset() { *m = DBMeta{} } +func (m *DBMeta) String() string { return proto.CompactTextString(m) } +func (*DBMeta) ProtoMessage() {} +func (*DBMeta) Descriptor() ([]byte, []int) { return fileDescriptorPrivate, []int{0} } -type Frame struct { - TimeQuantum string `protobuf:"bytes,1,opt,name=TimeQuantum,proto3" json:"TimeQuantum,omitempty"` - RowLabel string `protobuf:"bytes,2,opt,name=RowLabel,proto3" json:"RowLabel,omitempty"` - InverseEnabled bool `protobuf:"varint,3,opt,name=InverseEnabled,proto3" json:"InverseEnabled,omitempty"` - CacheType string `protobuf:"bytes,4,opt,name=CacheType,proto3" json:"CacheType,omitempty"` +type FrameMeta struct { + RowLabel string `protobuf:"bytes,1,opt,name=RowLabel,proto3" json:"RowLabel,omitempty"` + InverseEnabled bool `protobuf:"varint,2,opt,name=InverseEnabled,proto3" json:"InverseEnabled,omitempty"` + CacheType string `protobuf:"bytes,3,opt,name=CacheType,proto3" json:"CacheType,omitempty"` + CacheSize uint32 `protobuf:"varint,4,opt,name=CacheSize,proto3" json:"CacheSize,omitempty"` + TimeQuantum string `protobuf:"bytes,5,opt,name=TimeQuantum,proto3" json:"TimeQuantum,omitempty"` } -func (m *Frame) Reset() { *m = Frame{} } -func (m *Frame) String() string { return proto.CompactTextString(m) } -func (*Frame) ProtoMessage() {} -func (*Frame) Descriptor() ([]byte, []int) { return fileDescriptorPrivate, []int{1} } +func (m *FrameMeta) Reset() { *m = FrameMeta{} } +func (m *FrameMeta) String() string { return proto.CompactTextString(m) } +func (*FrameMeta) ProtoMessage() {} +func (*FrameMeta) Descriptor() ([]byte, []int) { return fileDescriptorPrivate, []int{1} } type ImportResponse struct { Err string `protobuf:"bytes,1,opt,name=Err,proto3" json:"Err,omitempty"` @@ -115,16 +124,149 @@ func (m *MaxSlicesResponse) GetMaxSlices() map[string]uint64 { return nil } +type CreateSliceMessage struct { + DB string `protobuf:"bytes,1,opt,name=DB,proto3" json:"DB,omitempty"` + Slice uint64 `protobuf:"varint,2,opt,name=Slice,proto3" json:"Slice,omitempty"` +} + +func (m *CreateSliceMessage) Reset() { *m = CreateSliceMessage{} } +func (m *CreateSliceMessage) String() string { return proto.CompactTextString(m) } +func (*CreateSliceMessage) ProtoMessage() {} +func (*CreateSliceMessage) Descriptor() ([]byte, []int) { return fileDescriptorPrivate, []int{7} } + +type DeleteDBMessage struct { + DB string `protobuf:"bytes,1,opt,name=DB,proto3" json:"DB,omitempty"` +} + +func (m *DeleteDBMessage) Reset() { *m = DeleteDBMessage{} } +func (m *DeleteDBMessage) String() string { return proto.CompactTextString(m) } +func (*DeleteDBMessage) ProtoMessage() {} +func (*DeleteDBMessage) Descriptor() ([]byte, []int) { return fileDescriptorPrivate, []int{8} } + +type CreateDBMessage struct { + DB string `protobuf:"bytes,1,opt,name=DB,proto3" json:"DB,omitempty"` + Meta *DBMeta `protobuf:"bytes,2,opt,name=Meta" json:"Meta,omitempty"` +} + +func (m *CreateDBMessage) Reset() { *m = CreateDBMessage{} } +func (m *CreateDBMessage) String() string { return proto.CompactTextString(m) } +func (*CreateDBMessage) ProtoMessage() {} +func (*CreateDBMessage) Descriptor() ([]byte, []int) { return fileDescriptorPrivate, []int{9} } + +func (m *CreateDBMessage) GetMeta() *DBMeta { + if m != nil { + return m.Meta + } + return nil +} + +type CreateFrameMessage struct { + DB string `protobuf:"bytes,1,opt,name=DB,proto3" json:"DB,omitempty"` + Frame string `protobuf:"bytes,2,opt,name=Frame,proto3" json:"Frame,omitempty"` + Meta *FrameMeta `protobuf:"bytes,3,opt,name=Meta" json:"Meta,omitempty"` +} + +func (m *CreateFrameMessage) Reset() { *m = CreateFrameMessage{} } +func (m *CreateFrameMessage) String() string { return proto.CompactTextString(m) } +func (*CreateFrameMessage) ProtoMessage() {} +func (*CreateFrameMessage) Descriptor() ([]byte, []int) { return fileDescriptorPrivate, []int{10} } + +func (m *CreateFrameMessage) GetMeta() *FrameMeta { + if m != nil { + return m.Meta + } + return nil +} + +type DeleteFrameMessage struct { + DB string `protobuf:"bytes,1,opt,name=DB,proto3" json:"DB,omitempty"` + Frame string `protobuf:"bytes,2,opt,name=Frame,proto3" json:"Frame,omitempty"` +} + +func (m *DeleteFrameMessage) Reset() { *m = DeleteFrameMessage{} } +func (m *DeleteFrameMessage) String() string { return proto.CompactTextString(m) } +func (*DeleteFrameMessage) ProtoMessage() {} +func (*DeleteFrameMessage) Descriptor() ([]byte, []int) { return fileDescriptorPrivate, []int{11} } + +type Frame struct { + Name string `protobuf:"bytes,1,opt,name=Name,proto3" json:"Name,omitempty"` + Meta *FrameMeta `protobuf:"bytes,2,opt,name=Meta" json:"Meta,omitempty"` +} + +func (m *Frame) Reset() { *m = Frame{} } +func (m *Frame) String() string { return proto.CompactTextString(m) } +func (*Frame) ProtoMessage() {} +func (*Frame) Descriptor() ([]byte, []int) { return fileDescriptorPrivate, []int{12} } + +func (m *Frame) GetMeta() *FrameMeta { + if m != nil { + return m.Meta + } + return nil +} + +type DB struct { + Name string `protobuf:"bytes,1,opt,name=Name,proto3" json:"Name,omitempty"` + Meta *DBMeta `protobuf:"bytes,2,opt,name=Meta" json:"Meta,omitempty"` + MaxSlice uint64 `protobuf:"varint,3,opt,name=MaxSlice,proto3" json:"MaxSlice,omitempty"` + Frames []*Frame `protobuf:"bytes,4,rep,name=Frames" json:"Frames,omitempty"` +} + +func (m *DB) Reset() { *m = DB{} } +func (m *DB) String() string { return proto.CompactTextString(m) } +func (*DB) ProtoMessage() {} +func (*DB) Descriptor() ([]byte, []int) { return fileDescriptorPrivate, []int{13} } + +func (m *DB) GetMeta() *DBMeta { + if m != nil { + return m.Meta + } + return nil +} + +func (m *DB) GetFrames() []*Frame { + if m != nil { + return m.Frames + } + return nil +} + +type NodeState struct { + Host string `protobuf:"bytes,1,opt,name=Host,proto3" json:"Host,omitempty"` + State string `protobuf:"bytes,2,opt,name=State,proto3" json:"State,omitempty"` + DBs []*DB `protobuf:"bytes,3,rep,name=DBs" json:"DBs,omitempty"` +} + +func (m *NodeState) Reset() { *m = NodeState{} } +func (m *NodeState) String() string { return proto.CompactTextString(m) } +func (*NodeState) ProtoMessage() {} +func (*NodeState) Descriptor() ([]byte, []int) { return fileDescriptorPrivate, []int{14} } + +func (m *NodeState) GetDBs() []*DB { + if m != nil { + return m.DBs + } + return nil +} + func init() { - proto.RegisterType((*DB)(nil), "internal.DB") - proto.RegisterType((*Frame)(nil), "internal.Frame") + proto.RegisterType((*DBMeta)(nil), "internal.DBMeta") + proto.RegisterType((*FrameMeta)(nil), "internal.FrameMeta") proto.RegisterType((*ImportResponse)(nil), "internal.ImportResponse") proto.RegisterType((*BlockDataRequest)(nil), "internal.BlockDataRequest") proto.RegisterType((*BlockDataResponse)(nil), "internal.BlockDataResponse") proto.RegisterType((*Cache)(nil), "internal.Cache") proto.RegisterType((*MaxSlicesResponse)(nil), "internal.MaxSlicesResponse") + proto.RegisterType((*CreateSliceMessage)(nil), "internal.CreateSliceMessage") + proto.RegisterType((*DeleteDBMessage)(nil), "internal.DeleteDBMessage") + proto.RegisterType((*CreateDBMessage)(nil), "internal.CreateDBMessage") + proto.RegisterType((*CreateFrameMessage)(nil), "internal.CreateFrameMessage") + proto.RegisterType((*DeleteFrameMessage)(nil), "internal.DeleteFrameMessage") + proto.RegisterType((*Frame)(nil), "internal.Frame") + proto.RegisterType((*DB)(nil), "internal.DB") + proto.RegisterType((*NodeState)(nil), "internal.NodeState") } -func (m *DB) Marshal() (dAtA []byte, err error) { +func (m *DBMeta) Marshal() (dAtA []byte, err error) { size := m.Size() dAtA = make([]byte, size) n, err := m.MarshalTo(dAtA) @@ -134,27 +276,27 @@ func (m *DB) Marshal() (dAtA []byte, err error) { return dAtA[:n], nil } -func (m *DB) MarshalTo(dAtA []byte) (int, error) { +func (m *DBMeta) MarshalTo(dAtA []byte) (int, error) { var i int _ = i var l int _ = l - if len(m.TimeQuantum) > 0 { - dAtA[i] = 0xa - i++ - i = encodeVarintPrivate(dAtA, i, uint64(len(m.TimeQuantum))) - i += copy(dAtA[i:], m.TimeQuantum) - } if len(m.ColumnLabel) > 0 { - dAtA[i] = 0x12 + dAtA[i] = 0xa i++ i = encodeVarintPrivate(dAtA, i, uint64(len(m.ColumnLabel))) i += copy(dAtA[i:], m.ColumnLabel) } + if len(m.TimeQuantum) > 0 { + dAtA[i] = 0x12 + i++ + i = encodeVarintPrivate(dAtA, i, uint64(len(m.TimeQuantum))) + i += copy(dAtA[i:], m.TimeQuantum) + } return i, nil } -func (m *Frame) Marshal() (dAtA []byte, err error) { +func (m *FrameMeta) Marshal() (dAtA []byte, err error) { size := m.Size() dAtA = make([]byte, size) n, err := m.MarshalTo(dAtA) @@ -164,25 +306,19 @@ func (m *Frame) Marshal() (dAtA []byte, err error) { return dAtA[:n], nil } -func (m *Frame) MarshalTo(dAtA []byte) (int, error) { +func (m *FrameMeta) MarshalTo(dAtA []byte) (int, error) { var i int _ = i var l int _ = l - if len(m.TimeQuantum) > 0 { - dAtA[i] = 0xa - i++ - i = encodeVarintPrivate(dAtA, i, uint64(len(m.TimeQuantum))) - i += copy(dAtA[i:], m.TimeQuantum) - } if len(m.RowLabel) > 0 { - dAtA[i] = 0x12 + dAtA[i] = 0xa i++ i = encodeVarintPrivate(dAtA, i, uint64(len(m.RowLabel))) i += copy(dAtA[i:], m.RowLabel) } if m.InverseEnabled { - dAtA[i] = 0x18 + dAtA[i] = 0x10 i++ if m.InverseEnabled { dAtA[i] = 1 @@ -192,11 +328,22 @@ func (m *Frame) MarshalTo(dAtA []byte) (int, error) { i++ } if len(m.CacheType) > 0 { - dAtA[i] = 0x22 + dAtA[i] = 0x1a i++ i = encodeVarintPrivate(dAtA, i, uint64(len(m.CacheType))) i += copy(dAtA[i:], m.CacheType) } + if m.CacheSize != 0 { + dAtA[i] = 0x20 + i++ + i = encodeVarintPrivate(dAtA, i, uint64(m.CacheSize)) + } + if len(m.TimeQuantum) > 0 { + dAtA[i] = 0x2a + i++ + i = encodeVarintPrivate(dAtA, i, uint64(len(m.TimeQuantum))) + i += copy(dAtA[i:], m.TimeQuantum) + } return i, nil } @@ -391,6 +538,290 @@ func (m *MaxSlicesResponse) MarshalTo(dAtA []byte) (int, error) { return i, nil } +func (m *CreateSliceMessage) 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 *CreateSliceMessage) MarshalTo(dAtA []byte) (int, error) { + var i int + _ = i + var l int + _ = l + if len(m.DB) > 0 { + dAtA[i] = 0xa + i++ + i = encodeVarintPrivate(dAtA, i, uint64(len(m.DB))) + i += copy(dAtA[i:], m.DB) + } + if m.Slice != 0 { + dAtA[i] = 0x10 + i++ + i = encodeVarintPrivate(dAtA, i, uint64(m.Slice)) + } + return i, nil +} + +func (m *DeleteDBMessage) 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 *DeleteDBMessage) MarshalTo(dAtA []byte) (int, error) { + var i int + _ = i + var l int + _ = l + if len(m.DB) > 0 { + dAtA[i] = 0xa + i++ + i = encodeVarintPrivate(dAtA, i, uint64(len(m.DB))) + i += copy(dAtA[i:], m.DB) + } + return i, nil +} + +func (m *CreateDBMessage) 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 *CreateDBMessage) MarshalTo(dAtA []byte) (int, error) { + var i int + _ = i + var l int + _ = l + if len(m.DB) > 0 { + dAtA[i] = 0xa + i++ + i = encodeVarintPrivate(dAtA, i, uint64(len(m.DB))) + i += copy(dAtA[i:], m.DB) + } + if m.Meta != nil { + dAtA[i] = 0x12 + i++ + i = encodeVarintPrivate(dAtA, i, uint64(m.Meta.Size())) + n7, err := m.Meta.MarshalTo(dAtA[i:]) + if err != nil { + return 0, err + } + i += n7 + } + return i, nil +} + +func (m *CreateFrameMessage) 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 *CreateFrameMessage) MarshalTo(dAtA []byte) (int, error) { + var i int + _ = i + var l int + _ = l + if len(m.DB) > 0 { + dAtA[i] = 0xa + i++ + i = encodeVarintPrivate(dAtA, i, uint64(len(m.DB))) + i += copy(dAtA[i:], m.DB) + } + if len(m.Frame) > 0 { + dAtA[i] = 0x12 + i++ + i = encodeVarintPrivate(dAtA, i, uint64(len(m.Frame))) + i += copy(dAtA[i:], m.Frame) + } + if m.Meta != nil { + dAtA[i] = 0x1a + i++ + i = encodeVarintPrivate(dAtA, i, uint64(m.Meta.Size())) + n8, err := m.Meta.MarshalTo(dAtA[i:]) + if err != nil { + return 0, err + } + i += n8 + } + return i, nil +} + +func (m *DeleteFrameMessage) 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 *DeleteFrameMessage) MarshalTo(dAtA []byte) (int, error) { + var i int + _ = i + var l int + _ = l + if len(m.DB) > 0 { + dAtA[i] = 0xa + i++ + i = encodeVarintPrivate(dAtA, i, uint64(len(m.DB))) + i += copy(dAtA[i:], m.DB) + } + if len(m.Frame) > 0 { + dAtA[i] = 0x12 + i++ + i = encodeVarintPrivate(dAtA, i, uint64(len(m.Frame))) + i += copy(dAtA[i:], m.Frame) + } + return i, nil +} + +func (m *Frame) 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 *Frame) MarshalTo(dAtA []byte) (int, error) { + var i int + _ = i + var l int + _ = l + if len(m.Name) > 0 { + dAtA[i] = 0xa + i++ + i = encodeVarintPrivate(dAtA, i, uint64(len(m.Name))) + i += copy(dAtA[i:], m.Name) + } + if m.Meta != nil { + dAtA[i] = 0x12 + i++ + i = encodeVarintPrivate(dAtA, i, uint64(m.Meta.Size())) + n9, err := m.Meta.MarshalTo(dAtA[i:]) + if err != nil { + return 0, err + } + i += n9 + } + return i, nil +} + +func (m *DB) 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 *DB) MarshalTo(dAtA []byte) (int, error) { + var i int + _ = i + var l int + _ = l + if len(m.Name) > 0 { + dAtA[i] = 0xa + i++ + i = encodeVarintPrivate(dAtA, i, uint64(len(m.Name))) + i += copy(dAtA[i:], m.Name) + } + if m.Meta != nil { + dAtA[i] = 0x12 + i++ + i = encodeVarintPrivate(dAtA, i, uint64(m.Meta.Size())) + n10, err := m.Meta.MarshalTo(dAtA[i:]) + if err != nil { + return 0, err + } + i += n10 + } + if m.MaxSlice != 0 { + dAtA[i] = 0x18 + i++ + i = encodeVarintPrivate(dAtA, i, uint64(m.MaxSlice)) + } + if len(m.Frames) > 0 { + for _, msg := range m.Frames { + dAtA[i] = 0x22 + i++ + i = encodeVarintPrivate(dAtA, i, uint64(msg.Size())) + n, err := msg.MarshalTo(dAtA[i:]) + if err != nil { + return 0, err + } + i += n + } + } + return i, nil +} + +func (m *NodeState) 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 *NodeState) MarshalTo(dAtA []byte) (int, error) { + var i int + _ = i + var l int + _ = l + if len(m.Host) > 0 { + dAtA[i] = 0xa + i++ + i = encodeVarintPrivate(dAtA, i, uint64(len(m.Host))) + i += copy(dAtA[i:], m.Host) + } + if len(m.State) > 0 { + dAtA[i] = 0x12 + i++ + i = encodeVarintPrivate(dAtA, i, uint64(len(m.State))) + i += copy(dAtA[i:], m.State) + } + if len(m.DBs) > 0 { + for _, msg := range m.DBs { + dAtA[i] = 0x1a + i++ + i = encodeVarintPrivate(dAtA, i, uint64(msg.Size())) + n, err := msg.MarshalTo(dAtA[i:]) + if err != nil { + return 0, err + } + i += n + } + } + return i, nil +} + func encodeFixed64Private(dAtA []byte, offset int, v uint64) int { dAtA[offset] = uint8(v) dAtA[offset+1] = uint8(v >> 8) @@ -418,27 +849,23 @@ func encodeVarintPrivate(dAtA []byte, offset int, v uint64) int { dAtA[offset] = uint8(v) return offset + 1 } -func (m *DB) Size() (n int) { +func (m *DBMeta) Size() (n int) { var l int _ = l - l = len(m.TimeQuantum) + l = len(m.ColumnLabel) if l > 0 { n += 1 + l + sovPrivate(uint64(l)) } - l = len(m.ColumnLabel) + l = len(m.TimeQuantum) if l > 0 { n += 1 + l + sovPrivate(uint64(l)) } return n } -func (m *Frame) Size() (n int) { +func (m *FrameMeta) Size() (n int) { var l int _ = l - l = len(m.TimeQuantum) - if l > 0 { - n += 1 + l + sovPrivate(uint64(l)) - } l = len(m.RowLabel) if l > 0 { n += 1 + l + sovPrivate(uint64(l)) @@ -450,6 +877,13 @@ func (m *Frame) Size() (n int) { if l > 0 { n += 1 + l + sovPrivate(uint64(l)) } + if m.CacheSize != 0 { + n += 1 + sovPrivate(uint64(m.CacheSize)) + } + l = len(m.TimeQuantum) + if l > 0 { + n += 1 + l + sovPrivate(uint64(l)) + } return n } @@ -534,6 +968,132 @@ func (m *MaxSlicesResponse) Size() (n int) { return n } +func (m *CreateSliceMessage) Size() (n int) { + var l int + _ = l + l = len(m.DB) + if l > 0 { + n += 1 + l + sovPrivate(uint64(l)) + } + if m.Slice != 0 { + n += 1 + sovPrivate(uint64(m.Slice)) + } + return n +} + +func (m *DeleteDBMessage) Size() (n int) { + var l int + _ = l + l = len(m.DB) + if l > 0 { + n += 1 + l + sovPrivate(uint64(l)) + } + return n +} + +func (m *CreateDBMessage) Size() (n int) { + var l int + _ = l + l = len(m.DB) + if l > 0 { + n += 1 + l + sovPrivate(uint64(l)) + } + if m.Meta != nil { + l = m.Meta.Size() + n += 1 + l + sovPrivate(uint64(l)) + } + return n +} + +func (m *CreateFrameMessage) Size() (n int) { + var l int + _ = l + l = len(m.DB) + if l > 0 { + n += 1 + l + sovPrivate(uint64(l)) + } + l = len(m.Frame) + if l > 0 { + n += 1 + l + sovPrivate(uint64(l)) + } + if m.Meta != nil { + l = m.Meta.Size() + n += 1 + l + sovPrivate(uint64(l)) + } + return n +} + +func (m *DeleteFrameMessage) Size() (n int) { + var l int + _ = l + l = len(m.DB) + if l > 0 { + n += 1 + l + sovPrivate(uint64(l)) + } + l = len(m.Frame) + if l > 0 { + n += 1 + l + sovPrivate(uint64(l)) + } + return n +} + +func (m *Frame) Size() (n int) { + var l int + _ = l + l = len(m.Name) + if l > 0 { + n += 1 + l + sovPrivate(uint64(l)) + } + if m.Meta != nil { + l = m.Meta.Size() + n += 1 + l + sovPrivate(uint64(l)) + } + return n +} + +func (m *DB) Size() (n int) { + var l int + _ = l + l = len(m.Name) + if l > 0 { + n += 1 + l + sovPrivate(uint64(l)) + } + if m.Meta != nil { + l = m.Meta.Size() + n += 1 + l + sovPrivate(uint64(l)) + } + if m.MaxSlice != 0 { + n += 1 + sovPrivate(uint64(m.MaxSlice)) + } + if len(m.Frames) > 0 { + for _, e := range m.Frames { + l = e.Size() + n += 1 + l + sovPrivate(uint64(l)) + } + } + return n +} + +func (m *NodeState) Size() (n int) { + var l int + _ = l + l = len(m.Host) + if l > 0 { + n += 1 + l + sovPrivate(uint64(l)) + } + l = len(m.State) + if l > 0 { + n += 1 + l + sovPrivate(uint64(l)) + } + if len(m.DBs) > 0 { + for _, e := range m.DBs { + l = e.Size() + n += 1 + l + sovPrivate(uint64(l)) + } + } + return n +} + func sovPrivate(x uint64) (n int) { for { n++ @@ -547,7 +1107,7 @@ func sovPrivate(x uint64) (n int) { func sozPrivate(x uint64) (n int) { return sovPrivate(uint64((x << 1) ^ uint64((int64(x) >> 63)))) } -func (m *DB) Unmarshal(dAtA []byte) error { +func (m *DBMeta) Unmarshal(dAtA []byte) error { l := len(dAtA) iNdEx := 0 for iNdEx < l { @@ -570,42 +1130,13 @@ func (m *DB) Unmarshal(dAtA []byte) error { fieldNum := int32(wire >> 3) wireType := int(wire & 0x7) if wireType == 4 { - return fmt.Errorf("proto: DB: wiretype end group for non-group") + return fmt.Errorf("proto: DBMeta: wiretype end group for non-group") } if fieldNum <= 0 { - return fmt.Errorf("proto: DB: illegal tag %d (wire type %d)", fieldNum, wire) + return fmt.Errorf("proto: DBMeta: illegal tag %d (wire type %d)", fieldNum, wire) } switch fieldNum { case 1: - if wireType != 2 { - return fmt.Errorf("proto: wrong wireType = %d for field TimeQuantum", wireType) - } - var stringLen uint64 - for shift := uint(0); ; shift += 7 { - if shift >= 64 { - return ErrIntOverflowPrivate - } - if iNdEx >= l { - return io.ErrUnexpectedEOF - } - b := dAtA[iNdEx] - iNdEx++ - stringLen |= (uint64(b) & 0x7F) << shift - if b < 0x80 { - break - } - } - intStringLen := int(stringLen) - if intStringLen < 0 { - return ErrInvalidLengthPrivate - } - postIndex := iNdEx + intStringLen - if postIndex > l { - return io.ErrUnexpectedEOF - } - m.TimeQuantum = string(dAtA[iNdEx:postIndex]) - iNdEx = postIndex - case 2: if wireType != 2 { return fmt.Errorf("proto: wrong wireType = %d for field ColumnLabel", wireType) } @@ -634,57 +1165,7 @@ func (m *DB) Unmarshal(dAtA []byte) error { } m.ColumnLabel = string(dAtA[iNdEx:postIndex]) 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 *Frame) 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: Frame: wiretype end group for non-group") - } - if fieldNum <= 0 { - return fmt.Errorf("proto: Frame: illegal tag %d (wire type %d)", fieldNum, wire) - } - switch fieldNum { - case 1: + case 2: if wireType != 2 { return fmt.Errorf("proto: wrong wireType = %d for field TimeQuantum", wireType) } @@ -713,7 +1194,57 @@ func (m *Frame) Unmarshal(dAtA []byte) error { } m.TimeQuantum = string(dAtA[iNdEx:postIndex]) iNdEx = postIndex - case 2: + 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 *FrameMeta) 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: FrameMeta: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: FrameMeta: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: if wireType != 2 { return fmt.Errorf("proto: wrong wireType = %d for field RowLabel", wireType) } @@ -742,7 +1273,7 @@ func (m *Frame) Unmarshal(dAtA []byte) error { } m.RowLabel = string(dAtA[iNdEx:postIndex]) iNdEx = postIndex - case 3: + case 2: if wireType != 0 { return fmt.Errorf("proto: wrong wireType = %d for field InverseEnabled", wireType) } @@ -762,7 +1293,7 @@ func (m *Frame) Unmarshal(dAtA []byte) error { } } m.InverseEnabled = bool(v != 0) - case 4: + case 3: if wireType != 2 { return fmt.Errorf("proto: wrong wireType = %d for field CacheType", wireType) } @@ -791,6 +1322,54 @@ func (m *Frame) Unmarshal(dAtA []byte) error { } m.CacheType = string(dAtA[iNdEx:postIndex]) iNdEx = postIndex + case 4: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field CacheSize", wireType) + } + m.CacheSize = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.CacheSize |= (uint32(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + case 5: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field TimeQuantum", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthPrivate + } + postIndex := iNdEx + intStringLen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.TimeQuantum = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex default: iNdEx = preIndex skippy, err := skipPrivate(dAtA[iNdEx:]) @@ -1508,6 +2087,957 @@ func (m *MaxSlicesResponse) Unmarshal(dAtA []byte) error { } return nil } +func (m *CreateSliceMessage) 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: CreateSliceMessage: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: CreateSliceMessage: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field DB", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthPrivate + } + postIndex := iNdEx + intStringLen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.DB = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex + case 2: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field Slice", wireType) + } + m.Slice = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.Slice |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + 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 *DeleteDBMessage) 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: DeleteDBMessage: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: DeleteDBMessage: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field DB", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthPrivate + } + postIndex := iNdEx + intStringLen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.DB = string(dAtA[iNdEx:postIndex]) + 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 *CreateDBMessage) 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: CreateDBMessage: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: CreateDBMessage: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field DB", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthPrivate + } + postIndex := iNdEx + intStringLen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.DB = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex + case 2: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Meta", 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.Meta == nil { + m.Meta = &DBMeta{} + } + if err := m.Meta.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 *CreateFrameMessage) 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: CreateFrameMessage: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: CreateFrameMessage: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field DB", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthPrivate + } + postIndex := iNdEx + intStringLen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.DB = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex + case 2: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Frame", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthPrivate + } + postIndex := iNdEx + intStringLen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.Frame = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex + case 3: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Meta", 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.Meta == nil { + m.Meta = &FrameMeta{} + } + if err := m.Meta.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 *DeleteFrameMessage) 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: DeleteFrameMessage: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: DeleteFrameMessage: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field DB", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthPrivate + } + postIndex := iNdEx + intStringLen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.DB = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex + case 2: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Frame", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthPrivate + } + postIndex := iNdEx + intStringLen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.Frame = string(dAtA[iNdEx:postIndex]) + 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 *Frame) 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: Frame: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: Frame: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Name", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthPrivate + } + postIndex := iNdEx + intStringLen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.Name = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex + case 2: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Meta", 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.Meta == nil { + m.Meta = &FrameMeta{} + } + if err := m.Meta.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 *DB) 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: DB: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: DB: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Name", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthPrivate + } + postIndex := iNdEx + intStringLen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.Name = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex + case 2: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Meta", 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.Meta == nil { + m.Meta = &DBMeta{} + } + if err := m.Meta.Unmarshal(dAtA[iNdEx:postIndex]); err != nil { + return err + } + iNdEx = postIndex + case 3: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field MaxSlice", wireType) + } + m.MaxSlice = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.MaxSlice |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + case 4: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Frames", 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 + } + m.Frames = append(m.Frames, &Frame{}) + if err := m.Frames[len(m.Frames)-1].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 *NodeState) 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: NodeState: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: NodeState: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Host", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthPrivate + } + postIndex := iNdEx + intStringLen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.Host = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex + case 2: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field State", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthPrivate + } + postIndex := iNdEx + intStringLen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.State = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex + case 3: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field DBs", 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 + } + m.DBs = append(m.DBs, &DB{}) + if err := m.DBs[len(m.DBs)-1].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 skipPrivate(dAtA []byte) (n int, err error) { l := len(dAtA) iNdEx := 0 @@ -1616,31 +3146,43 @@ var ( func init() { proto.RegisterFile("private.proto", fileDescriptorPrivate) } var fileDescriptorPrivate = []byte{ - // 406 bytes of a gzipped FileDescriptorProto - 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x09, 0x6e, 0x88, 0x02, 0xff, 0x8c, 0x92, 0xd1, 0x8a, 0xd3, 0x40, - 0x14, 0x86, 0x9d, 0x36, 0x95, 0xf6, 0x2c, 0x86, 0xee, 0xb0, 0x17, 0x61, 0x59, 0x42, 0x18, 0x50, - 0x8a, 0x17, 0xbd, 0xd0, 0x1b, 0x11, 0xaf, 0xb2, 0xad, 0x6c, 0x41, 0xc1, 0x1d, 0x17, 0xef, 0xa7, - 0xf5, 0x88, 0xa1, 0x93, 0x99, 0x38, 0x99, 0x74, 0x37, 0xcf, 0xe0, 0x0b, 0x08, 0xbe, 0x90, 0x97, - 0x3e, 0x82, 0xd4, 0x17, 0x91, 0x99, 0xa4, 0x4d, 0x54, 0x10, 0xef, 0xe6, 0x7c, 0x73, 0xce, 0x3f, - 0xff, 0xf9, 0x19, 0x78, 0x50, 0x98, 0x6c, 0x27, 0x2c, 0xce, 0x0b, 0xa3, 0xad, 0xa6, 0xe3, 0x4c, - 0x59, 0x34, 0x4a, 0x48, 0x76, 0x05, 0x83, 0x45, 0x4a, 0x13, 0x38, 0xb9, 0xc9, 0x72, 0xbc, 0xae, - 0x84, 0xb2, 0x55, 0x1e, 0x91, 0x84, 0xcc, 0x26, 0xbc, 0x8f, 0x5c, 0xc7, 0xa5, 0x96, 0x55, 0xae, - 0x5e, 0x89, 0x35, 0xca, 0x68, 0xd0, 0x74, 0xf4, 0x10, 0xfb, 0x4c, 0x60, 0xf4, 0xd2, 0x88, 0x1c, - 0xff, 0x43, 0xed, 0x1c, 0xc6, 0x5c, 0xdf, 0xf6, 0xa5, 0x8e, 0x35, 0x7d, 0x04, 0xe1, 0x4a, 0xed, - 0xd0, 0x94, 0xb8, 0x54, 0x62, 0x2d, 0xf1, 0x7d, 0x34, 0x4c, 0xc8, 0x6c, 0xcc, 0xff, 0xa0, 0xf4, - 0x02, 0x26, 0x97, 0x62, 0xf3, 0x11, 0x6f, 0xea, 0x02, 0xa3, 0xc0, 0x8b, 0x74, 0x80, 0x31, 0x08, - 0x57, 0x79, 0xa1, 0x8d, 0xe5, 0x58, 0x16, 0x5a, 0x95, 0x48, 0xa7, 0x30, 0x5c, 0x1a, 0xd3, 0xba, - 0x71, 0x47, 0x76, 0x07, 0xd3, 0x54, 0xea, 0xcd, 0x76, 0x21, 0xac, 0xe0, 0xf8, 0xa9, 0xc2, 0xd2, - 0xd2, 0xd0, 0xe5, 0xd1, 0x36, 0xb9, 0x64, 0xce, 0xda, 0xa5, 0x5a, 0x9b, 0xed, 0x86, 0x67, 0x30, - 0xf2, 0x93, 0xde, 0x5a, 0xc0, 0x9b, 0xc2, 0xd1, 0xb7, 0x32, 0xdb, 0x34, 0x6e, 0x02, 0xde, 0x14, - 0x94, 0x42, 0xf0, 0x2e, 0xc3, 0xdb, 0x68, 0xe4, 0x05, 0xfc, 0x99, 0x5d, 0xc3, 0x69, 0xef, 0xe5, - 0xd6, 0xe0, 0x05, 0x4c, 0xd2, 0xcc, 0xe6, 0xa2, 0x58, 0x2d, 0xca, 0x88, 0x24, 0xc3, 0x59, 0xc0, - 0x3b, 0x40, 0x63, 0x80, 0x37, 0x46, 0x7f, 0xc8, 0x24, 0xba, 0xeb, 0x81, 0xbf, 0xee, 0x11, 0xf6, - 0x10, 0x46, 0x7e, 0xfb, 0x7f, 0xcb, 0xb0, 0xaf, 0x04, 0x4e, 0x5f, 0x8b, 0x3b, 0x6f, 0xad, 0x3c, - 0x3e, 0x7d, 0x05, 0x93, 0x23, 0xf4, 0x33, 0x27, 0x4f, 0x1e, 0xcf, 0x0f, 0x7f, 0x64, 0xfe, 0x57, - 0x7f, 0x47, 0x96, 0xca, 0x9a, 0x9a, 0x77, 0xc3, 0xe7, 0x2f, 0x20, 0xfc, 0xfd, 0xd2, 0xe5, 0xbe, - 0xc5, 0xfa, 0x90, 0xfb, 0x16, 0x6b, 0x97, 0xd3, 0x4e, 0xc8, 0xaa, 0xc9, 0x34, 0xe0, 0x4d, 0xf1, - 0x7c, 0xf0, 0x8c, 0xa4, 0xd3, 0x6f, 0xfb, 0x98, 0x7c, 0xdf, 0xc7, 0xe4, 0xc7, 0x3e, 0x26, 0x5f, - 0x7e, 0xc6, 0xf7, 0xd6, 0xf7, 0xfd, 0x87, 0x7d, 0xfa, 0x2b, 0x00, 0x00, 0xff, 0xff, 0x99, 0x45, - 0xbe, 0x6d, 0xc1, 0x02, 0x00, 0x00, + // 600 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x09, 0x6e, 0x88, 0x02, 0xff, 0x94, 0x54, 0xcd, 0x4e, 0x14, 0x41, + 0x10, 0x76, 0x7e, 0x20, 0x4c, 0x21, 0xcb, 0xd2, 0x7a, 0x98, 0x10, 0x32, 0x59, 0x3b, 0x2a, 0xc4, + 0x03, 0x07, 0xbc, 0x18, 0xe2, 0x69, 0x18, 0x14, 0x12, 0x20, 0xd2, 0x8b, 0xde, 0x7b, 0x97, 0x52, + 0x27, 0x3b, 0x7f, 0xce, 0xf4, 0x2e, 0xac, 0x57, 0x5f, 0xc2, 0xc4, 0x67, 0xf0, 0x3d, 0x3c, 0xfa, + 0x08, 0x66, 0x7d, 0x11, 0xd3, 0xdd, 0xf3, 0xe7, 0xb2, 0xf8, 0x73, 0xeb, 0xfa, 0xaa, 0xea, 0xab, + 0xaf, 0xbf, 0xa9, 0x1e, 0x58, 0xcb, 0xf2, 0x70, 0xc2, 0x05, 0xee, 0x66, 0x79, 0x2a, 0x52, 0xb2, + 0x12, 0x26, 0x02, 0xf3, 0x84, 0x47, 0xf4, 0x04, 0x96, 0x03, 0xff, 0x14, 0x05, 0x27, 0x3d, 0x58, + 0x3d, 0x48, 0xa3, 0x71, 0x9c, 0x9c, 0xf0, 0x01, 0x46, 0xae, 0xd1, 0x33, 0x76, 0x1c, 0xd6, 0x86, + 0x64, 0xc5, 0x45, 0x18, 0xe3, 0xf9, 0x98, 0x27, 0x62, 0x1c, 0xbb, 0xa6, 0xae, 0x68, 0x41, 0xf4, + 0xab, 0x01, 0xce, 0x8b, 0x9c, 0xc7, 0xa8, 0x18, 0x37, 0x61, 0x85, 0xa5, 0x57, 0x6d, 0xba, 0x3a, + 0x26, 0x8f, 0xa1, 0x73, 0x9c, 0x4c, 0x30, 0x2f, 0xf0, 0x30, 0xe1, 0x83, 0x08, 0x2f, 0x15, 0xdd, + 0x0a, 0x9b, 0x43, 0xc9, 0x16, 0x38, 0x07, 0x7c, 0xf8, 0x1e, 0x2f, 0xa6, 0x19, 0xba, 0x96, 0x22, + 0x69, 0x80, 0x3a, 0xdb, 0x0f, 0x3f, 0xa2, 0x6b, 0xf7, 0x8c, 0x9d, 0x35, 0xd6, 0x00, 0xf3, 0x7a, + 0x97, 0x6e, 0xea, 0xa5, 0xd0, 0x39, 0x8e, 0xb3, 0x34, 0x17, 0x0c, 0x8b, 0x2c, 0x4d, 0x0a, 0x24, + 0x5d, 0xb0, 0x0e, 0xf3, 0xbc, 0x94, 0x2b, 0x8f, 0xf4, 0x1a, 0xba, 0x7e, 0x94, 0x0e, 0x47, 0x01, + 0x17, 0x9c, 0xe1, 0x87, 0x31, 0x16, 0x82, 0x74, 0xc0, 0x0c, 0xfc, 0xb2, 0xc8, 0x0c, 0x7c, 0x72, + 0x1f, 0x96, 0xd4, 0xb5, 0x4b, 0x4f, 0x74, 0x20, 0x51, 0xd5, 0xa9, 0x74, 0xdb, 0x4c, 0x07, 0x12, + 0xed, 0x47, 0xe1, 0x50, 0xeb, 0xb5, 0x99, 0x0e, 0x08, 0x01, 0xfb, 0x4d, 0x88, 0x57, 0xa5, 0x48, + 0x75, 0xa6, 0xe7, 0xb0, 0xd1, 0x9a, 0x5c, 0x0a, 0xdc, 0x02, 0xc7, 0x0f, 0x45, 0xcc, 0xb3, 0xe3, + 0xa0, 0x70, 0x8d, 0x9e, 0xb5, 0x63, 0xb3, 0x06, 0x20, 0x1e, 0xc0, 0xab, 0x3c, 0x7d, 0x1b, 0x46, + 0x28, 0xd3, 0xa6, 0x4a, 0xb7, 0x10, 0xfa, 0x08, 0x96, 0x94, 0x3f, 0x7f, 0xa6, 0xa1, 0x5f, 0x0c, + 0xd8, 0x38, 0xe5, 0xd7, 0x4a, 0x5a, 0x51, 0x8f, 0x3e, 0x02, 0xa7, 0x06, 0x55, 0xcf, 0xea, 0xde, + 0x93, 0xdd, 0x6a, 0x93, 0x76, 0x6f, 0xd4, 0x37, 0xc8, 0x61, 0x22, 0xf2, 0x29, 0x6b, 0x9a, 0x37, + 0x9f, 0x43, 0xe7, 0xf7, 0xa4, 0xf4, 0x7d, 0x84, 0xd3, 0xca, 0xf7, 0x11, 0x4e, 0xa5, 0x4f, 0x13, + 0x1e, 0x8d, 0xb5, 0xa7, 0x36, 0xd3, 0xc1, 0xbe, 0xf9, 0xcc, 0xa0, 0xfb, 0x40, 0x0e, 0x72, 0xe4, + 0x02, 0x15, 0xc1, 0x29, 0x16, 0x05, 0x7f, 0x87, 0x8b, 0xbe, 0x89, 0xf6, 0xd9, 0x6c, 0xf9, 0x4c, + 0x1f, 0xc0, 0x7a, 0x80, 0x11, 0x0a, 0x94, 0x5b, 0xbf, 0xb0, 0x91, 0xbe, 0x84, 0x75, 0x4d, 0x7f, + 0x6b, 0x09, 0x79, 0x08, 0xb6, 0xdc, 0x70, 0x45, 0xbd, 0xba, 0xd7, 0x6d, 0x4c, 0xd0, 0x6f, 0x89, + 0xa9, 0x2c, 0x1d, 0x56, 0x3a, 0xcb, 0x27, 0x71, 0xab, 0xce, 0x05, 0xbb, 0xb3, 0x5d, 0x4e, 0xb0, + 0xd4, 0x84, 0x7b, 0xcd, 0x84, 0xfa, 0x79, 0x95, 0x43, 0xf6, 0x81, 0xe8, 0x0b, 0xfd, 0xff, 0x10, + 0x1a, 0x94, 0xa8, 0xdc, 0xbe, 0x33, 0x99, 0xd5, 0x0d, 0xea, 0x5c, 0x2b, 0x30, 0xff, 0xa6, 0xe0, + 0x93, 0x21, 0x87, 0x2d, 0xe4, 0xf8, 0x27, 0x9f, 0xe4, 0x7f, 0xa2, 0xda, 0x86, 0xf2, 0xa9, 0xd4, + 0x31, 0xd9, 0x86, 0x65, 0x35, 0xaf, 0x70, 0x6d, 0xb5, 0x70, 0xeb, 0x73, 0x3a, 0x58, 0x99, 0xa6, + 0xaf, 0xc1, 0x39, 0x4b, 0x2f, 0xb1, 0x2f, 0xb8, 0x50, 0xf7, 0x39, 0x4a, 0x0b, 0x51, 0x69, 0x91, + 0x67, 0xb5, 0x0f, 0x32, 0x59, 0x59, 0xa0, 0x2b, 0x3d, 0xb0, 0x02, 0xbf, 0x70, 0x2d, 0x45, 0x7e, + 0xb7, 0x2d, 0x90, 0xc9, 0x84, 0xdf, 0xfd, 0x36, 0xf3, 0x8c, 0xef, 0x33, 0xcf, 0xf8, 0x31, 0xf3, + 0x8c, 0xcf, 0x3f, 0xbd, 0x3b, 0x83, 0x65, 0xf5, 0x0b, 0x7d, 0xfa, 0x2b, 0x00, 0x00, 0xff, 0xff, + 0x3a, 0x23, 0x0f, 0xb4, 0x53, 0x05, 0x00, 0x00, } diff --git a/internal/private.proto b/internal/private.proto index acd9ec43c..ec060e355 100644 --- a/internal/private.proto +++ b/internal/private.proto @@ -2,16 +2,17 @@ syntax = "proto3"; package internal; -message DB { - string TimeQuantum = 1; - string ColumnLabel = 2; +message DBMeta { + string ColumnLabel = 1; + string TimeQuantum = 2; } -message Frame { - string TimeQuantum = 1; - string RowLabel = 2; - bool InverseEnabled = 3; - string CacheType = 4; +message FrameMeta { + string RowLabel = 1; + bool InverseEnabled = 2; + string CacheType = 3; + uint32 CacheSize = 4; + string TimeQuantum = 5; } message ImportResponse { @@ -39,3 +40,45 @@ message MaxSlicesResponse { map MaxSlices = 1; } +message CreateSliceMessage { + string DB = 1; + uint64 Slice = 2; +} + +message DeleteDBMessage { + string DB = 1; +} + +message CreateDBMessage { + string DB = 1; + DBMeta Meta = 2; +} + +message CreateFrameMessage { + string DB = 1; + string Frame = 2; + FrameMeta Meta = 3; +} + +message DeleteFrameMessage { + string DB = 1; + string Frame = 2; +} + +message Frame { + string Name = 1; + FrameMeta Meta = 2; +} + +message DB { + string Name = 1; + DBMeta Meta = 2; + uint64 MaxSlice = 3; + repeated Frame Frames = 4; +} + +message NodeState { + string Host = 1; + string State = 2; + repeated DB DBs = 3; +} diff --git a/server.go b/server.go index 55702acf9..567461ccc 100644 --- a/server.go +++ b/server.go @@ -1,6 +1,7 @@ package pilosa import ( + "errors" "fmt" "io" "io/ioutil" @@ -32,8 +33,10 @@ type Server struct { closing chan struct{} // Data storage and HTTP interface. - Index *Index - Handler *Handler + Index *Index + Handler *Handler + Broadcaster Broadcaster + BroadcastReceiver BroadcastReceiver // Cluster configuration. // Host is replaced with actual host after opening if port is ":0". @@ -52,8 +55,10 @@ func NewServer() *Server { s := &Server{ closing: make(chan struct{}), - Index: NewIndex(), - Handler: NewHandler(), + Index: NewIndex(), + Handler: NewHandler(), + Broadcaster: NopBroadcaster, + BroadcastReceiver: NopBroadcastReceiver, AntiEntropyInterval: DefaultAntiEntropyInterval, PollingInterval: DefaultPollingInterval, @@ -96,6 +101,15 @@ func (s *Server) Open() error { return err } + if err := s.BroadcastReceiver.Start(s); err != nil { + return err + } + + // Open NodeSet communication + if err := s.Cluster.NodeSet.Open(); err != nil { + return err + } + // Create executor for executing queries. e := NewExecutor() e.Index = s.Index @@ -103,10 +117,14 @@ func (s *Server) Open() error { e.Cluster = s.Cluster // Initialize HTTP handler. + s.Handler.Broadcaster = s.Broadcaster s.Handler.Host = s.Host s.Handler.Cluster = s.Cluster s.Handler.Executor = e s.Handler.LogOutput = s.LogOutput + + // Initialize Index. + s.Index.Broadcaster = s.Broadcaster s.Index.LogOutput = s.LogOutput // Serve HTTP. @@ -202,21 +220,15 @@ func (s *Server) monitorMaxSlices() { if s.Host != node.Host { maxSlices, _ := checkMaxSlices(node.Host) for db, newmax := range maxSlices { - // if we don't know about a db locally, create it - // so that the /schema endpoint can report it + // if we don't know about a db locally, log an error because + // db's should be created and synced prior to slice creation if localdb := s.Index.DB(db); localdb != nil { if newmax > oldmaxslices[db] { oldmaxslices[db] = newmax localdb.SetRemoteMaxSlice(newmax) } } else { - d := s.Index.DB(db) - if d == nil { - s.logger().Printf("Local DB not found: %s", db) - return - } - oldmaxslices[db] = newmax - d.SetRemoteMaxSlice(newmax) + s.logger().Printf("Local DB not found: %s", db) } } } @@ -224,6 +236,91 @@ func (s *Server) monitorMaxSlices() { } } +// ReceiveMessage represents an implementation of BroadcastHandler. +func (s *Server) ReceiveMessage(pb proto.Message) error { + switch obj := pb.(type) { + case *internal.CreateSliceMessage: + d := s.Index.DB(obj.DB) + if d == nil { + return fmt.Errorf("Local DB not found: %s", obj.DB) + } + d.SetRemoteMaxSlice(obj.Slice) + case *internal.CreateDBMessage: + opt := DBOptions{ColumnLabel: obj.Meta.ColumnLabel} + _, err := s.Index.CreateDB(obj.DB, opt) + if err != nil { + return err + } + case *internal.DeleteDBMessage: + if err := s.Index.DeleteDB(obj.DB); err != nil { + return err + } + case *internal.CreateFrameMessage: + db := s.Index.DB(obj.DB) + opt := FrameOptions{RowLabel: obj.Meta.RowLabel} + _, err := db.CreateFrame(obj.Frame, opt) + if err != nil { + return err + } + case *internal.DeleteFrameMessage: + db := s.Index.DB(obj.DB) + if err := db.DeleteFrame(obj.Frame); err != nil { + return err + } + } + return nil +} + +// Server implements gossip.StateHandler. +// LocalState returns the state of the local node as well as the +// index (dbs/frames) according to the local node. +// In a gossip implementation, memberlist.Delegate.LocalState() uses this. +func (s *Server) LocalState() (proto.Message, error) { + if s.Index == nil { + return nil, errors.New("Server.Index is nil.") + } + return &internal.NodeState{ + Host: s.Host, + State: "OK", // TODO: make this work, pull from s.Cluster.Node + DBs: encodeDBs(s.Index.DBs()), + }, nil +} + +// HandleRemoteState receives incoming NodeState from remote nodes. +func (s *Server) HandleRemoteState(pb proto.Message) error { + return s.mergeRemoteState(pb.(*internal.NodeState)) +} + +func (s *Server) mergeRemoteState(ns *internal.NodeState) error { + // TODO: update some node state value in the cluster (it should be in cluster.node i guess) + + // Create databases that don't exist. + for _, db := range ns.DBs { + opt := DBOptions{ + ColumnLabel: db.Meta.ColumnLabel, + TimeQuantum: TimeQuantum(db.Meta.TimeQuantum), + } + d, err := s.Index.CreateDBIfNotExists(db.Name, opt) + if err != nil { + return err + } + // Create frames that don't exist. + for _, f := range db.Frames { + opt := FrameOptions{ + RowLabel: f.Meta.RowLabel, + TimeQuantum: TimeQuantum(f.Meta.TimeQuantum), + CacheSize: f.Meta.CacheSize, + } + _, err := d.CreateFrameIfNotExists(f.Name, opt) + if err != nil { + return err + } + } + } + + return nil +} + func checkMaxSlices(hostport string) (map[string]uint64, error) { // Create HTTP request. req, err := http.NewRequest("GET", (&url.URL{ diff --git a/server/server.go b/server/server.go index 251a054aa..ec2b55c7c 100644 --- a/server/server.go +++ b/server/server.go @@ -9,12 +9,16 @@ import ( "fmt" "io" "math/rand" + "net" "os" "path/filepath" + "strconv" "strings" "time" "github.com/pilosa/pilosa" + "github.com/pilosa/pilosa/gossip" + "github.com/pilosa/pilosa/httpbroadcast" ) func init() { @@ -71,6 +75,35 @@ func (m *Command) Run(args ...string) (err error) { m.Config.DataDir = filepath.Join(HomeDir, strings.TrimPrefix(m.Config.DataDir, prefix)) } + // SetupServer + err = m.SetupServer() + if err != nil { + return err + } + + // Initialize server. + if err = m.Server.Open(); err != nil { + return fmt.Errorf("server.Open: %v", err) + } + fmt.Fprintf(m.Stderr, "Listening as http://%s\n", m.Server.Host) + return nil +} + +func (m *Command) SetupServer() error { + cluster := pilosa.NewCluster() + cluster.ReplicaN = m.Config.Cluster.ReplicaN + + for _, hostport := range m.Config.Cluster.Hosts { + cluster.Nodes = append(cluster.Nodes, &pilosa.Node{Host: hostport}) + } + // TODO: if InternalHosts is not provided then pilosa.Node.InternalHost is empty. + // This will throw an error when trying to Broadcast messages over HTTP. + // One option may be to fall back to using host from hostport + config.InternalPort. + for i, internalhostport := range m.Config.Cluster.InternalHosts { + cluster.Nodes[i].InternalHost = internalhostport + } + m.Server.Cluster = cluster + // Setup logging output. if m.Config.LogPath == "" { m.Server.LogOutput = m.Stderr @@ -87,21 +120,55 @@ func (m *Command) Run(args ...string) (err error) { m.Server.Index.Path = m.Config.DataDir m.Server.Index.Stats = pilosa.NewExpvarStatsClient() - // Build cluster from config file. + var err error m.Server.Host, err = normalizeHost(m.Config.Host) if err != nil { return err } - m.Server.Cluster = m.Config.PilosaCluster() + + // Set internal port (string). + internalPortStr := pilosa.DefaultInternalPort + if m.Config.Cluster.InternalPort != "" { + internalPortStr = m.Config.Cluster.InternalPort + } + + switch m.Config.Cluster.Type { + case "http": + m.Server.Broadcaster = httpbroadcast.NewHTTPBroadcaster(m.Server, internalPortStr) + m.Server.BroadcastReceiver = httpbroadcast.NewHTTPBroadcastReceiver(internalPortStr, m.Stderr) + m.Server.Cluster.NodeSet = httpbroadcast.NewHTTPNodeSet() + err := m.Server.Cluster.NodeSet.(*httpbroadcast.HTTPNodeSet).Join(m.Server.Cluster.Nodes) + if err != nil { + return err + } + case "gossip": + gossipPort, err := strconv.Atoi(internalPortStr) + if err != nil { + return err + } + gossipSeed := pilosa.DefaultHost + if m.Config.Cluster.GossipSeed != "" { + gossipSeed = m.Config.Cluster.GossipSeed + } + // get the host portion of addr to use for binding + gossipHost, _, err := net.SplitHostPort(m.Config.Host) + if err != nil { + gossipHost = m.Config.Host + } + gossipNodeSet := gossip.NewGossipNodeSet(m.Config.Host, gossipHost, gossipPort, gossipSeed, m.Server) + m.Server.Cluster.NodeSet = gossipNodeSet + m.Server.Broadcaster = gossipNodeSet + m.Server.BroadcastReceiver = gossipNodeSet + case "static", "": + m.Server.Broadcaster = pilosa.NopBroadcaster + m.Server.Cluster.NodeSet = pilosa.NewStaticNodeSet() + m.Server.BroadcastReceiver = pilosa.NopBroadcastReceiver + default: + return fmt.Errorf("'%v' is not a supported value for broadcaster type.", m.Config.Cluster.Type) + } // Set configuration options. m.Server.AntiEntropyInterval = time.Duration(m.Config.AntiEntropy.Interval) - - // Initialize server. - if err = m.Server.Open(); err != nil { - return fmt.Errorf("server.Open: %v", err) - } - fmt.Fprintf(m.Stderr, "Listening as http://%s\n", m.Server.Host) return nil } diff --git a/view.go b/view.go index 54dda6f56..c7ec92733 100644 --- a/view.go +++ b/view.go @@ -30,6 +30,8 @@ type View struct { frame string name string + cacheSize uint32 + // Fragments by slice. cacheType string // passed in by frame fragments map[uint64]*Fragment @@ -41,12 +43,13 @@ type View struct { } // NewView returns a new instance of View. -func NewView(path, db, frame, name string) *View { +func NewView(path, db, frame, name string, cacheSize uint32) *View { return &View{ - path: path, - db: db, - frame: frame, - name: name, + path: path, + db: db, + frame: frame, + name: name, + cacheSize: cacheSize, cacheType: DefaultCacheType, fragments: make(map[uint64]*Fragment), @@ -215,6 +218,7 @@ func (v *View) createFragmentIfNotExists(slice uint64) (*Fragment, error) { func (v *View) newFragment(path string, slice uint64) *Fragment { frag := NewFragment(path, v.db, v.frame, v.name, slice) frag.cacheType = v.cacheType + frag.cacheSize = v.cacheSize frag.LogOutput = v.LogOutput frag.stats = v.stats.WithTags(fmt.Sprintf("slice:%d", slice)) return frag diff --git a/view_test.go b/view_test.go index 87b3e5826..d10c90936 100644 --- a/view_test.go +++ b/view_test.go @@ -22,7 +22,7 @@ func NewView(db, frame, name string) *View { file.Close() v := &View{ - View: pilosa.NewView(file.Name(), db, frame, name), + View: pilosa.NewView(file.Name(), db, frame, name, pilosa.DefaultCacheSize), BitmapAttrStore: MustOpenAttrStore(), } v.View.BitmapAttrStore = v.BitmapAttrStore.AttrStore @@ -52,7 +52,7 @@ func (v *View) Reopen() error { return err } - v.View = pilosa.NewView(path, v.DB(), v.Frame(), v.Name()) + v.View = pilosa.NewView(path, v.DB(), v.Frame(), v.Name(), pilosa.DefaultCacheSize) v.View.BitmapAttrStore = v.BitmapAttrStore.AttrStore if err := v.Open(); err != nil { return err