From 68d9d1543ff1a27a83dd92f04cd3f9fc049b8cc2 Mon Sep 17 00:00:00 2001 From: Travis Date: Wed, 19 Apr 2017 16:07:30 -0500 Subject: [PATCH 1/6] rename messenger_test to align with broadcast --- messenger_test.go => broadcast_test.go | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename messenger_test.go => broadcast_test.go (100%) diff --git a/messenger_test.go b/broadcast_test.go similarity index 100% rename from messenger_test.go rename to broadcast_test.go From f3189889d422777e056c9ee56dba391413bc1331 Mon Sep 17 00:00:00 2001 From: Travis Date: Wed, 19 Apr 2017 16:59:56 -0500 Subject: [PATCH 2/6] add support for CreateDB and CreateFrame without POST data (i.e. use defaults) --- handler.go | 33 +++++++++++++++++++++++++++------ 1 file changed, 27 insertions(+), 6 deletions(-) diff --git a/handler.go b/handler.go index a6309e44a..b94a9a0ed 100644 --- a/handler.go +++ b/handler.go @@ -309,6 +309,15 @@ func (h *Handler) handleDeleteDB(w http.ResponseWriter, r *http.Request) { return } + // Send the delete 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) @@ -323,13 +332,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 @@ -340,11 +353,15 @@ func (h *Handler) handlePostDB(w http.ResponseWriter, r *http.Request) { // Send the delete message to all nodes. err = h.Broadcaster.SendSync( - &internal.DeleteDBMessage{ + &internal.CreateDBMessage{ DB: dbName, + Meta: &internal.DBMeta{ + ColumnLabel: req.Options.ColumnLabel, + TimeQuantum: string(req.Options.TimeQuantum), + }, }) if err != nil { - h.logger().Printf("problem sending DeleteDB message: %s", err) + h.logger().Printf("problem sending CreateDB message: %s", err) } // Encode response. @@ -460,7 +477,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 } @@ -473,7 +494,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 From c101f647c819fb76450769376dd7c5632ca19ad0 Mon Sep 17 00:00:00 2001 From: Travis Date: Wed, 19 Apr 2017 17:08:10 -0500 Subject: [PATCH 3/6] adjust some handler comments. remove debugging line --- handler.go | 8 ++++---- server.go | 1 - 2 files changed, 4 insertions(+), 5 deletions(-) diff --git a/handler.go b/handler.go index b94a9a0ed..d9c79eeba 100644 --- a/handler.go +++ b/handler.go @@ -309,7 +309,7 @@ func (h *Handler) handleDeleteDB(w http.ResponseWriter, r *http.Request) { return } - // Send the delete message to all nodes. + // Send the delete database message to all nodes. err := h.Broadcaster.SendSync( &internal.DeleteDBMessage{ DB: dbName, @@ -351,7 +351,7 @@ func (h *Handler) handlePostDB(w http.ResponseWriter, r *http.Request) { return } - // Send the delete message to all nodes. + // Send the create database message to all nodes. err = h.Broadcaster.SendSync( &internal.CreateDBMessage{ DB: dbName, @@ -503,7 +503,7 @@ func (h *Handler) handlePostFrame(w http.ResponseWriter, r *http.Request) { return } - // Send the create message to all nodes. + // Send the create frame message to all nodes. err = h.Broadcaster.SendSync( &internal.CreateFrameMessage{ DB: dbName, @@ -575,7 +575,7 @@ func (h *Handler) handleDeleteFrame(w http.ResponseWriter, r *http.Request) { return } - // Send the delete message to all nodes. + // Send the delete frame message to all nodes. err := h.Broadcaster.SendSync( &internal.DeleteFrameMessage{ DB: dbName, diff --git a/server.go b/server.go index 229214805..e93b8e111 100644 --- a/server.go +++ b/server.go @@ -265,7 +265,6 @@ func (s *Server) ReceiveMessage(pb proto.Message) error { return err } case *internal.DeleteDBMessage: - fmt.Println("DELETE:", obj.DB) if err := s.Index.DeleteDB(obj.DB); err != nil { return err } From 749ffdcf65ad0898c1eca0258b10a498bc2da07a Mon Sep 17 00:00:00 2001 From: Travis Date: Thu, 20 Apr 2017 10:41:17 -0500 Subject: [PATCH 4/6] adjusted DBOptions/FrameOptions logic and removed some unused tests --- db.go | 8 +++ frame.go | 23 +++++-- handler.go | 12 +--- handler_test.go | 54 --------------- index_test.go | 28 -------- internal/private.pb.go | 148 ++++++++++++++++++++--------------------- internal/private.proto | 4 +- 7 files changed, 105 insertions(+), 172 deletions(-) diff --git a/db.go b/db.go index d2ceaf13a..8bf33300e 100644 --- a/db.go +++ b/db.go @@ -536,6 +536,14 @@ type DBOptions struct { 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. func hasTime(a []*time.Time) bool { for _, t := range a { diff --git a/frame.go b/frame.go index a497e6906..30fd41406 100644 --- a/frame.go +++ b/frame.go @@ -302,11 +302,11 @@ func (f *Frame) loadMeta() error { func (f *Frame) saveMeta() error { // Marshal metadata. buf, err := proto.Marshal(&internal.FrameMeta{ - TimeQuantum: string(f.timeQuantum), RowLabel: f.rowLabel, - CacheType: f.cacheType, InverseEnabled: f.inverseEnabled, + CacheType: f.cacheType, CacheSize: f.cacheSize, + TimeQuantum: string(f.timeQuantum), }) if err != nil { return err @@ -593,9 +593,11 @@ func encodeFrame(f *Frame) *internal.Frame { return &internal.Frame{ Name: f.name, Meta: &internal.FrameMeta{ - TimeQuantum: string(f.timeQuantum), - RowLabel: f.rowLabel, - CacheSize: f.cacheSize, + RowLabel: f.rowLabel, + InverseEnabled: f.inverseEnabled, + CacheType: f.cacheType, + CacheSize: f.cacheSize, + TimeQuantum: string(f.timeQuantum), }, } } @@ -627,6 +629,17 @@ type FrameOptions struct { 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. // This is used to sort data during import. type importBitSet struct { diff --git a/handler.go b/handler.go index d9c79eeba..8b79267bf 100644 --- a/handler.go +++ b/handler.go @@ -354,11 +354,8 @@ func (h *Handler) handlePostDB(w http.ResponseWriter, r *http.Request) { // Send the create database message to all nodes. err = h.Broadcaster.SendSync( &internal.CreateDBMessage{ - DB: dbName, - Meta: &internal.DBMeta{ - ColumnLabel: req.Options.ColumnLabel, - TimeQuantum: string(req.Options.TimeQuantum), - }, + DB: dbName, + Meta: req.Options.Encode(), }) if err != nil { h.logger().Printf("problem sending CreateDB message: %s", err) @@ -508,10 +505,7 @@ func (h *Handler) handlePostFrame(w http.ResponseWriter, r *http.Request) { &internal.CreateFrameMessage{ DB: dbName, Frame: frameName, - Meta: &internal.FrameMeta{ - RowLabel: req.Options.RowLabel, - TimeQuantum: string(req.Options.TimeQuantum), - }, + Meta: req.Options.Encode(), }) if err != nil { h.logger().Printf("problem sending CreateFrame message: %s", err) diff --git a/handler_test.go b/handler_test.go index 478720e23..049fdc036 100644 --- a/handler_test.go +++ b/handler_test.go @@ -874,57 +874,3 @@ func MustReadAll(r io.Reader) []byte { } return buf } - -/* -// TODO: move this test to messenger.go (with NewServer()) - -// Ensure that an HTTP message sent to the cluster reaches all nodes. -func TestHTTPNodeSet_Base(t *testing.T) { - - // servers - s1 := NewServer() - s1.Messenger = pilosa.NewMessenger() - n1 := NewHTTPMessageBroker() - n1.messenger = s1.Messenger - s1.Messenger.Broker = n1 - - s2 := NewServer() - s2.Messenger = pilosa.NewMessenger() - n2 := NewHTTPMessageBroker() - n2.messenger = s2.Messenger - s2.Messenger.Broker = n2 - - s3 := NewServer() - s3.Messenger = pilosa.NewMessenger() - n3 := NewHTTPMessageBroker() - n3.messenger = s3.Messenger - s3.Messenger.Broker = n3 - - nodes := []*pilosa.Node{ - {Host: s1.Host()}, - {Host: s2.Host()}, - {Host: s3.Host()}, - } - - // message - msg := &internal.CreateSliceMessage{ - DB: "d", - Slice: 8, - } - - // send message - if err := s1.Messenger.SendMessage(msg, ""); err != nil { - t.Fatalf("failure sending message: %s", err) - } - - if !reflect.DeepEqual(mb1.messageReceived, msg) { - t.Fatalf("unexpected message received by node1: %s", mb1.messageReceived) - } - if !reflect.DeepEqual(mb2.messageReceived, msg) { - t.Fatalf("unexpected message received by node2: %s", mb2.messageReceived) - } - if !reflect.DeepEqual(mb3.messageReceived, msg) { - t.Fatalf("unexpected message received by node3: %s", mb3.messageReceived) - } -} -*/ diff --git a/index_test.go b/index_test.go index c09fe69f5..4a1be4cf2 100644 --- a/index_test.go +++ b/index_test.go @@ -162,34 +162,6 @@ func TestIndexSyncer_SyncIndex(t *testing.T) { } } -/* TODO: move this to messenger.go -// Ensure index can handle Messenger messages. -func TestIndex_HandleMessage(t *testing.T) { - // Create a local index. - idx0 := MustOpenIndex() - defer idx0.Close() - - idx0.MustCreateDBIfNotExists("d", pilosa.DBOptions{}) - - msg0 := &internal.CreateSliceMessage{ - DB: "d", - Slice: 8, - } - idx0.HandleMessage(msg0) - if ms := idx0.MaxSlices(); !reflect.DeepEqual(ms, map[string]uint64{"d": 8}) { - t.Fatalf("unexpected max slice: %s", ms) - } - - msg1 := &internal.DeleteDBMessage{ - DB: "d", - } - idx0.HandleMessage(msg1) - if ms := idx0.MaxSlices(); !reflect.DeepEqual(ms, map[string]uint64{}) { - t.Fatalf("unexpected delete db: %s", ms) - } -} -*/ - // Index is a test wrapper for pilosa.Index. type Index struct { *pilosa.Index diff --git a/internal/private.pb.go b/internal/private.pb.go index 5ceb800c2..0423cfa08 100644 --- a/internal/private.pb.go +++ b/internal/private.pb.go @@ -45,8 +45,8 @@ var _ = math.Inf const _ = proto.ProtoPackageIsVersion2 // please upgrade the proto package type DBMeta 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"` + 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 *DBMeta) Reset() { *m = DBMeta{} } @@ -281,18 +281,18 @@ func (m *DBMeta) MarshalTo(dAtA []byte) (int, error) { _ = 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 } @@ -852,11 +852,11 @@ func encodeVarintPrivate(dAtA []byte, offset int, v uint64) 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)) } @@ -1137,35 +1137,6 @@ func (m *DBMeta) Unmarshal(dAtA []byte) error { } 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) } @@ -1194,6 +1165,35 @@ func (m *DBMeta) Unmarshal(dAtA []byte) error { } m.ColumnLabel = string(dAtA[iNdEx:postIndex]) iNdEx = postIndex + case 2: + 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:]) @@ -3146,43 +3146,43 @@ var ( func init() { proto.RegisterFile("private.proto", fileDescriptorPrivate) } var fileDescriptorPrivate = []byte{ - // 603 bytes of a gzipped FileDescriptorProto + // 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, 0xfa, 0xab, - 0xaf, 0xbe, 0xa9, 0x1e, 0x58, 0xcb, 0xf2, 0x70, 0xc2, 0x05, 0xee, 0x66, 0x79, 0x2a, 0x52, 0xb2, + 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, - 0xbd, 0x08, 0x63, 0x3c, 0x1f, 0xf3, 0x44, 0x8c, 0x63, 0xd7, 0xe8, 0x19, 0x3b, 0x0e, 0x6b, 0x43, - 0xb2, 0xe2, 0x20, 0x8d, 0xc6, 0x71, 0x72, 0xc2, 0x07, 0x18, 0xb9, 0xa6, 0xae, 0x68, 0x41, 0xf4, - 0xab, 0x01, 0xce, 0x8b, 0x9c, 0xc7, 0xa8, 0x18, 0x37, 0x61, 0x85, 0xa5, 0x57, 0xba, 0x58, 0xd3, - 0xd5, 0x31, 0x79, 0x0c, 0x9d, 0xe3, 0x64, 0x82, 0x79, 0x81, 0x87, 0x09, 0x1f, 0x44, 0x78, 0xa9, - 0xe8, 0x56, 0xd8, 0x1c, 0x4a, 0xb6, 0xc0, 0x39, 0xe0, 0xc3, 0xf7, 0x78, 0x31, 0xcd, 0xd0, 0xb5, - 0x14, 0x49, 0x03, 0xd4, 0xd9, 0x7e, 0xf8, 0x11, 0x5d, 0xbb, 0x67, 0xec, 0xac, 0xb1, 0x06, 0x98, - 0x9f, 0x68, 0xe9, 0xc6, 0x44, 0x94, 0x42, 0xe7, 0x38, 0xce, 0xd2, 0x5c, 0x30, 0x2c, 0xb2, 0x34, - 0x29, 0x90, 0x74, 0xc1, 0x3a, 0xcc, 0xf3, 0x52, 0xae, 0x3c, 0xd2, 0x6b, 0xe8, 0xfa, 0x51, 0x3a, - 0x1c, 0x05, 0x5c, 0x70, 0x86, 0x1f, 0xc6, 0x58, 0x08, 0xd2, 0x01, 0x33, 0xf0, 0xcb, 0x22, 0x33, - 0xf0, 0xc9, 0x7d, 0x58, 0x52, 0x63, 0x97, 0x9e, 0xe8, 0x40, 0xa2, 0xea, 0xa6, 0xd2, 0x6d, 0x33, - 0x1d, 0x48, 0xb4, 0x1f, 0x85, 0x43, 0xad, 0xd7, 0x66, 0x3a, 0x20, 0x04, 0xec, 0x37, 0x21, 0x5e, - 0x95, 0x22, 0xd5, 0x99, 0x9e, 0xc3, 0x46, 0xab, 0x73, 0x29, 0x70, 0x0b, 0x1c, 0x3f, 0x14, 0x31, - 0xcf, 0x8e, 0x83, 0xc2, 0x35, 0x7a, 0xd6, 0x8e, 0xcd, 0x1a, 0x80, 0x78, 0x00, 0xaf, 0xf2, 0xf4, - 0x6d, 0x18, 0xa1, 0x4c, 0x9b, 0x2a, 0xdd, 0x42, 0xe8, 0x23, 0x58, 0x52, 0xfe, 0xfc, 0x99, 0x86, - 0x7e, 0x31, 0x60, 0xe3, 0x94, 0x5f, 0x2b, 0x69, 0x45, 0xdd, 0xfa, 0x08, 0x9c, 0x1a, 0x54, 0x77, - 0x56, 0xf7, 0x9e, 0xec, 0x56, 0x9b, 0xb4, 0x7b, 0xa3, 0xbe, 0x41, 0x0e, 0x13, 0x91, 0x4f, 0x59, - 0x73, 0x79, 0xf3, 0x39, 0x74, 0x7e, 0x4f, 0x4a, 0xdf, 0x47, 0x38, 0xad, 0x7c, 0x1f, 0xe1, 0x54, - 0xfa, 0x34, 0xe1, 0xd1, 0x58, 0x7b, 0x6a, 0x33, 0x1d, 0xec, 0x9b, 0xcf, 0x0c, 0xba, 0x0f, 0xe4, - 0x20, 0x47, 0x2e, 0x50, 0x11, 0x9c, 0x62, 0x51, 0xf0, 0x77, 0xb8, 0xe8, 0x9b, 0x68, 0x9f, 0xcd, - 0x96, 0xcf, 0xf4, 0x01, 0xac, 0x07, 0x18, 0xa1, 0x40, 0xb9, 0xf5, 0x0b, 0x2f, 0xd2, 0x97, 0xb0, - 0xae, 0xe9, 0x6f, 0x2d, 0x21, 0x0f, 0xc1, 0x96, 0x1b, 0xae, 0xa8, 0x57, 0xf7, 0xba, 0x8d, 0x09, - 0xfa, 0x2d, 0x31, 0x95, 0xa5, 0xc3, 0x4a, 0x67, 0xf9, 0x24, 0x6e, 0xd5, 0xb9, 0x60, 0x77, 0xb6, - 0xcb, 0x0e, 0x96, 0xea, 0x70, 0xaf, 0xe9, 0x50, 0x3f, 0xaf, 0xb2, 0xc9, 0x3e, 0x10, 0x3d, 0xd0, - 0xff, 0x37, 0xa1, 0x41, 0x89, 0xca, 0xed, 0x3b, 0x93, 0x59, 0x7d, 0x41, 0x9d, 0x6b, 0x05, 0xe6, - 0xdf, 0x14, 0x7c, 0x32, 0x64, 0xb3, 0x85, 0x1c, 0xff, 0xe4, 0x93, 0xfc, 0x4f, 0x54, 0xdb, 0x50, - 0x3e, 0x95, 0x3a, 0x26, 0xdb, 0xb0, 0xac, 0xfa, 0x15, 0xae, 0xad, 0x16, 0x6e, 0x7d, 0x4e, 0x07, - 0x2b, 0xd3, 0xf4, 0x35, 0x38, 0x67, 0xe9, 0x25, 0xf6, 0x05, 0x17, 0x6a, 0x9e, 0xa3, 0xb4, 0x10, - 0x95, 0x16, 0x79, 0x56, 0xfb, 0x20, 0x93, 0x95, 0x05, 0xba, 0xd2, 0x03, 0x2b, 0xf0, 0x0b, 0xd7, - 0x52, 0xe4, 0x77, 0xdb, 0x02, 0x99, 0x4c, 0xf8, 0xdd, 0x6f, 0x33, 0xcf, 0xf8, 0x3e, 0xf3, 0x8c, - 0x1f, 0x33, 0xcf, 0xf8, 0xfc, 0xd3, 0xbb, 0x33, 0x58, 0x56, 0xbf, 0xd0, 0xa7, 0xbf, 0x02, 0x00, - 0x00, 0xff, 0xff, 0xc3, 0x9a, 0x29, 0xff, 0x53, 0x05, 0x00, 0x00, + 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 40e83eb6b..ec060e355 100644 --- a/internal/private.proto +++ b/internal/private.proto @@ -3,8 +3,8 @@ syntax = "proto3"; package internal; message DBMeta { - string TimeQuantum = 1; - string ColumnLabel = 2; + string ColumnLabel = 1; + string TimeQuantum = 2; } message FrameMeta { From 281e6e8e0fd40771d48459f64c6e783c92276b99 Mon Sep 17 00:00:00 2001 From: Travis Date: Thu, 20 Apr 2017 15:14:11 -0500 Subject: [PATCH 5/6] remove leftover Receive() method from gossip implementation --- gossip/gossip.go | 9 +-------- 1 file changed, 1 insertion(+), 8 deletions(-) diff --git a/gossip/gossip.go b/gossip/gossip.go index 050435441..772892926 100644 --- a/gossip/gossip.go +++ b/gossip/gossip.go @@ -154,13 +154,6 @@ func (g *GossipNodeSet) SendAsync(pb proto.Message) error { return nil } -func (g *GossipNodeSet) Receive(pb proto.Message) error { - if err := g.handler.ReceiveMessage(pb); err != nil { - return err - } - return nil -} - // implementation of the memberlist.Delegate interface func (g *GossipNodeSet) NodeMeta(limit int) []byte { return []byte{} @@ -172,7 +165,7 @@ func (g *GossipNodeSet) NotifyMsg(b []byte) { g.logger().Printf("unmarshal message error: %s", err) return } - if err := g.Receive(m); err != nil { + if err := g.handler.ReceiveMessage(m); err != nil { g.logger().Printf("receive message error: %s", err) return } From 3f3c40e2412d2e4fbd22a70ebd791e0f8fa9c258 Mon Sep 17 00:00:00 2001 From: Travis Date: Thu, 20 Apr 2017 15:45:00 -0500 Subject: [PATCH 6/6] group Server interface implementation function together --- server.go | 30 ++++++++++++++++-------------- 1 file changed, 16 insertions(+), 14 deletions(-) diff --git a/server.go b/server.go index e93b8e111..567461ccc 100644 --- a/server.go +++ b/server.go @@ -236,20 +236,7 @@ func (s *Server) monitorMaxSlices() { } } -// 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 -} - +// ReceiveMessage represents an implementation of BroadcastHandler. func (s *Server) ReceiveMessage(pb proto.Message) error { switch obj := pb.(type) { case *internal.CreateSliceMessage: @@ -284,6 +271,21 @@ func (s *Server) ReceiveMessage(pb proto.Message) error { 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))