You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@servicecomb.apache.org by ze...@apache.org on 2019/06/21 01:51:52 UTC
[servicecomb-service-center] branch master updated (c5a0f9f ->
71b9cca)
This is an automated email from the ASF dual-hosted git repository.
zenlin pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/servicecomb-service-center.git.
from c5a0f9f Merge pull request #558 from zenlint/clustermode
new 3c2ca33 Correct nodeName to clusterName as appropriate
new 71b9cca Change nodeName to clusterName in storage
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
syncer/etcd/storage.go | 20 ++---
syncer/pkg/mock/mocksotrage/storage.go | 16 ++--
syncer/proto/event.pb.go | 119 -----------------------------
syncer/proto/event.proto | 15 ----
syncer/proto/syncer.pb.go | 84 ++++++++++++--------
syncer/proto/syncer.proto | 2 +-
syncer/serf/config.go | 4 +-
syncer/server/handler.go | 35 +++------
syncer/servicecenter/servicecenter.go | 16 ++--
syncer/servicecenter/servicecenter_test.go | 14 ++--
10 files changed, 101 insertions(+), 224 deletions(-)
delete mode 100644 syncer/proto/event.pb.go
delete mode 100644 syncer/proto/event.proto
[servicecomb-service-center] 01/02: Correct nodeName to clusterName
as appropriate
Posted by ze...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zenlin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/servicecomb-service-center.git
commit 3c2ca3392058360796f88004f531dd7eaaa02a4c
Author: chinx <c5...@126.com>
AuthorDate: Thu Jun 20 18:11:55 2019 +0800
Correct nodeName to clusterName as appropriate
---
syncer/etcd/storage.go | 20 ++---
syncer/pkg/mock/mocksotrage/storage.go | 16 ++--
syncer/proto/event.pb.go | 119 -----------------------------
syncer/proto/event.proto | 15 ----
syncer/proto/syncer.pb.go | 84 ++++++++++++--------
syncer/proto/syncer.proto | 2 +-
syncer/serf/config.go | 4 +-
syncer/server/handler.go | 35 +++------
syncer/servicecenter/servicecenter.go | 14 ++--
syncer/servicecenter/servicecenter_test.go | 14 ++--
10 files changed, 100 insertions(+), 223 deletions(-)
diff --git a/syncer/etcd/storage.go b/syncer/etcd/storage.go
index 5d149f2..b1e4f7f 100644
--- a/syncer/etcd/storage.go
+++ b/syncer/etcd/storage.go
@@ -87,18 +87,18 @@ next:
continue next
}
}
- key := mappingsKey + "/" + entry.NodeName + "/" + entry.OrgInstanceID
+ key := mappingsKey + "/" + entry.ClusterName + "/" + entry.OrgInstanceID
if _, err := s.client.Delete(context.Background(), key); err != nil {
- log.Errorf(err, "Delete instance nodeName=%s instanceID=%s failed", entry.NodeName, entry.OrgInstanceID)
+ log.Errorf(err, "Delete instance clusterName=%s instanceID=%s failed", entry.ClusterName, entry.OrgInstanceID)
}
}
}
-// UpdateMapByNode update map to storage by nodeName of other node
-func (s *storage) UpdateMapByNode(nodeName string, mapping pb.SyncMapping) {
+// UpdateMapByNode update map to storage by clusterName of other node
+func (s *storage) UpdateMapByNode(clusterName string, mapping pb.SyncMapping) {
newMaps := make(pb.SyncMapping, 0, len(mapping))
for _, val := range mapping {
- key := mappingsKey + "/" + nodeName + "/" + val.OrgInstanceID
+ key := mappingsKey + "/" + clusterName + "/" + val.OrgInstanceID
data, err := proto.Marshal(val)
if err != nil {
log.Errorf(err, "Proto marshal failed: %s", err)
@@ -110,13 +110,13 @@ func (s *storage) UpdateMapByNode(nodeName string, mapping pb.SyncMapping) {
}
newMaps = append(newMaps, val)
}
- s.cleanExpired(s.GetMapByNode(nodeName), newMaps)
+ s.cleanExpired(s.GetMapByNode(clusterName), newMaps)
}
-// GetMapByNode get map by nodeName of other node
-func (s *storage) GetMapByNode(nodeName string) (mapping pb.SyncMapping) {
+// GetMapByNode get map by clusterName of other node
+func (s *storage) GetMapByNode(clusterName string) (mapping pb.SyncMapping) {
maps := make(pb.SyncMapping, 0, 10)
- s.getPrefixKey(mappingsKey+"/"+nodeName, func(key, val []byte) (next bool) {
+ s.getPrefixKey(mappingsKey+"/"+clusterName, func(key, val []byte) (next bool) {
next = true
item := &pb.MappingEntry{}
if err := proto.Unmarshal(val, item); err != nil {
@@ -135,7 +135,7 @@ func (s *storage) UpdateMaps(maps pb.SyncMapping) {
srcMaps := s.GetMaps()
mappings := make(pb.SyncMapping, 0, len(maps))
for _, val := range maps {
- key := mappingsKey + "/" + val.NodeName + "/" + val.OrgInstanceID
+ key := mappingsKey + "/" + val.ClusterName + "/" + val.OrgInstanceID
data, err := proto.Marshal(val)
if err != nil {
log.Errorf(err, "Proto marshal failed: %s", err)
diff --git a/syncer/pkg/mock/mocksotrage/storage.go b/syncer/pkg/mock/mocksotrage/storage.go
index e72366c..36a1685 100644
--- a/syncer/pkg/mock/mocksotrage/storage.go
+++ b/syncer/pkg/mock/mocksotrage/storage.go
@@ -57,17 +57,17 @@ func (s *Storage) GetData() (data *pb.SyncData) {
return
}
-// UpdateMapByNode update map to storage by nodeName of other node
-func (s *Storage) UpdateMapByNode(nodeName string, mapping pb.SyncMapping) {
+// UpdateMapByNode update map to storage by clusterName of other node
+func (s *Storage) UpdateMapByNode(clusterName string, mapping pb.SyncMapping) {
s.lock.Lock()
- s.maps[nodeName] = mapping
+ s.maps[clusterName] = mapping
s.lock.Unlock()
}
-// GetMapByNode get map by nodeName of other node
-func (s *Storage) GetMapByNode(nodeName string) (mapping pb.SyncMapping) {
+// GetMapByNode get map by clusterName of other node
+func (s *Storage) GetMapByNode(clusterName string) (mapping pb.SyncMapping) {
s.lock.RLock()
- data, ok := s.maps[nodeName]
+ data, ok := s.maps[clusterName]
if !ok {
data = defaultMapping
}
@@ -79,12 +79,12 @@ func (s *Storage) UpdateMaps(maps pb.SyncMapping) {
s.lock.Lock()
mappings := make(map[string]pb.SyncMapping)
for _, item := range maps {
- mapping, ok := mappings[item.NodeName]
+ mapping, ok := mappings[item.ClusterName]
if !ok {
mapping = make(pb.SyncMapping, 0, 10)
}
mapping = append(mapping, item)
- mappings[item.NodeName] = mapping
+ mappings[item.ClusterName] = mapping
}
s.maps = mappings
s.lock.Unlock()
diff --git a/syncer/proto/event.pb.go b/syncer/proto/event.pb.go
deleted file mode 100644
index 8b44a52..0000000
--- a/syncer/proto/event.pb.go
+++ /dev/null
@@ -1,119 +0,0 @@
-// Code generated by protoc-gen-go. DO NOT EDIT.
-// source: event.proto
-
-/*
-Package proto is a generated protocol buffer package.
-
-It is generated from these files:
- event.proto
- syncer.proto
-
-It has these top-level messages:
- Member
- Discover
- PullRequest
- SyncService
- SyncData
- MapEntry
-*/
-package proto
-
-import proto1 "github.com/golang/protobuf/proto"
-import fmt "fmt"
-import math "math"
-
-// Reference imports to suppress errors if they are not otherwise used.
-var _ = proto1.Marshal
-var _ = fmt.Errorf
-var _ = math.Inf
-
-// This is a compile-time assertion to ensure that this generated file
-// is compatible with the proto package it is being compiled against.
-// A compilation error at this line likely means your copy of the
-// proto package needs to be updated.
-const _ = proto1.ProtoPackageIsVersion2 // please upgrade the proto package
-
-type Member struct {
- NodeName string `protobuf:"bytes,1,opt,name=nodeName" json:"nodeName,omitempty"`
- RPCPort int32 `protobuf:"varint,2,opt,name=RPCPort" json:"RPCPort,omitempty"`
- Time string `protobuf:"bytes,3,opt,name=time" json:"time,omitempty"`
-}
-
-func (m *Member) Reset() { *m = Member{} }
-func (m *Member) String() string { return proto1.CompactTextString(m) }
-func (*Member) ProtoMessage() {}
-func (*Member) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{0} }
-
-func (m *Member) GetNodeName() string {
- if m != nil {
- return m.NodeName
- }
- return ""
-}
-
-func (m *Member) GetRPCPort() int32 {
- if m != nil {
- return m.RPCPort
- }
- return 0
-}
-
-func (m *Member) GetTime() string {
- if m != nil {
- return m.Time
- }
- return ""
-}
-
-type Discover struct {
- ServiceName string `protobuf:"bytes,1,opt,name=serviceName" json:"serviceName,omitempty"`
- Options string `protobuf:"bytes,2,opt,name=options" json:"options,omitempty"`
- Time string `protobuf:"bytes,3,opt,name=time" json:"time,omitempty"`
-}
-
-func (m *Discover) Reset() { *m = Discover{} }
-func (m *Discover) String() string { return proto1.CompactTextString(m) }
-func (*Discover) ProtoMessage() {}
-func (*Discover) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{1} }
-
-func (m *Discover) GetServiceName() string {
- if m != nil {
- return m.ServiceName
- }
- return ""
-}
-
-func (m *Discover) GetOptions() string {
- if m != nil {
- return m.Options
- }
- return ""
-}
-
-func (m *Discover) GetTime() string {
- if m != nil {
- return m.Time
- }
- return ""
-}
-
-func init() {
- proto1.RegisterType((*Member)(nil), "proto.Member")
- proto1.RegisterType((*Discover)(nil), "proto.Discover")
-}
-
-func init() { proto1.RegisterFile("event.proto", fileDescriptor0) }
-
-var fileDescriptor0 = []byte{
- // 154 bytes of a gzipped FileDescriptorProto
- 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xe2, 0xe2, 0x4e, 0x2d, 0x4b, 0xcd,
- 0x2b, 0xd1, 0x2b, 0x28, 0xca, 0x2f, 0xc9, 0x17, 0x62, 0x05, 0x53, 0x4a, 0x41, 0x5c, 0x6c, 0xbe,
- 0xa9, 0xb9, 0x49, 0xa9, 0x45, 0x42, 0x52, 0x5c, 0x1c, 0x79, 0xf9, 0x29, 0xa9, 0x7e, 0x89, 0xb9,
- 0xa9, 0x12, 0x8c, 0x0a, 0x8c, 0x1a, 0x9c, 0x41, 0x70, 0xbe, 0x90, 0x04, 0x17, 0x7b, 0x50, 0x80,
- 0x73, 0x40, 0x7e, 0x51, 0x89, 0x04, 0x93, 0x02, 0xa3, 0x06, 0x6b, 0x10, 0x8c, 0x2b, 0x24, 0xc4,
- 0xc5, 0x52, 0x92, 0x99, 0x9b, 0x2a, 0xc1, 0x0c, 0xd6, 0x01, 0x66, 0x2b, 0x45, 0x71, 0x71, 0xb8,
- 0x64, 0x16, 0x27, 0xe7, 0x97, 0xa5, 0x16, 0x09, 0x29, 0x70, 0x71, 0x17, 0xa7, 0x16, 0x95, 0x65,
- 0x26, 0x23, 0x1b, 0x8c, 0x2c, 0x04, 0x32, 0x3b, 0xbf, 0xa0, 0x24, 0x33, 0x3f, 0xaf, 0x18, 0x6c,
- 0x36, 0x67, 0x10, 0x8c, 0x8b, 0xcd, 0xec, 0x24, 0x36, 0xb0, 0xb3, 0x8d, 0x01, 0x01, 0x00, 0x00,
- 0xff, 0xff, 0x00, 0xd3, 0xab, 0xa6, 0xcc, 0x00, 0x00, 0x00,
-}
diff --git a/syncer/proto/event.proto b/syncer/proto/event.proto
deleted file mode 100644
index 65aed26..0000000
--- a/syncer/proto/event.proto
+++ /dev/null
@@ -1,15 +0,0 @@
-syntax = "proto3";
-
-package proto;
-
-message Member {
- string nodeName = 1;
- int32 RPCPort = 2;
- string time = 3;
-}
-
-message Discover {
- string serviceName = 1;
- string options = 2;
- string time = 3;
-}
\ No newline at end of file
diff --git a/syncer/proto/syncer.pb.go b/syncer/proto/syncer.pb.go
index 58fe17a..7d502a2 100644
--- a/syncer/proto/syncer.pb.go
+++ b/syncer/proto/syncer.pb.go
@@ -1,6 +1,20 @@
// Code generated by protoc-gen-go. DO NOT EDIT.
// source: syncer.proto
+/*
+Package proto is a generated protocol buffer package.
+
+import "../../server/core/proto/services.proto";
+
+It is generated from these files:
+ syncer.proto
+
+It has these top-level messages:
+ PullRequest
+ SyncService
+ SyncData
+ MappingEntry
+*/
package proto
import proto1 "github.com/golang/protobuf/proto"
@@ -18,6 +32,12 @@ var _ = proto1.Marshal
var _ = fmt.Errorf
var _ = math.Inf
+// This is a compile-time assertion to ensure that this generated file
+// is compatible with the proto package it is being compiled against.
+// A compilation error at this line likely means your copy of the
+// proto package needs to be updated.
+const _ = proto1.ProtoPackageIsVersion2 // please upgrade the proto package
+
type PullRequest struct {
ServiceName string `protobuf:"bytes,1,opt,name=serviceName" json:"serviceName,omitempty"`
Options string `protobuf:"bytes,2,opt,name=options" json:"options,omitempty"`
@@ -27,7 +47,7 @@ type PullRequest struct {
func (m *PullRequest) Reset() { *m = PullRequest{} }
func (m *PullRequest) String() string { return proto1.CompactTextString(m) }
func (*PullRequest) ProtoMessage() {}
-func (*PullRequest) Descriptor() ([]byte, []int) { return fileDescriptor1, []int{0} }
+func (*PullRequest) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{0} }
func (m *PullRequest) GetServiceName() string {
if m != nil {
@@ -59,7 +79,7 @@ type SyncService struct {
func (m *SyncService) Reset() { *m = SyncService{} }
func (m *SyncService) String() string { return proto1.CompactTextString(m) }
func (*SyncService) ProtoMessage() {}
-func (*SyncService) Descriptor() ([]byte, []int) { return fileDescriptor1, []int{1} }
+func (*SyncService) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{1} }
func (m *SyncService) GetDomainProject() string {
if m != nil {
@@ -89,7 +109,7 @@ type SyncData struct {
func (m *SyncData) Reset() { *m = SyncData{} }
func (m *SyncData) String() string { return proto1.CompactTextString(m) }
func (*SyncData) ProtoMessage() {}
-func (*SyncData) Descriptor() ([]byte, []int) { return fileDescriptor1, []int{2} }
+func (*SyncData) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{2} }
func (m *SyncData) GetServices() []*SyncService {
if m != nil {
@@ -99,7 +119,7 @@ func (m *SyncData) GetServices() []*SyncService {
}
type MappingEntry struct {
- NodeName string `protobuf:"bytes,1,opt,name=nodeName" json:"nodeName,omitempty"`
+ ClusterName string `protobuf:"bytes,1,opt,name=clusterName" json:"clusterName,omitempty"`
DomainProject string `protobuf:"bytes,2,opt,name=domainProject" json:"domainProject,omitempty"`
OrgServiceID string `protobuf:"bytes,3,opt,name=orgServiceID" json:"orgServiceID,omitempty"`
OrgInstanceID string `protobuf:"bytes,4,opt,name=orgInstanceID" json:"orgInstanceID,omitempty"`
@@ -110,11 +130,11 @@ type MappingEntry struct {
func (m *MappingEntry) Reset() { *m = MappingEntry{} }
func (m *MappingEntry) String() string { return proto1.CompactTextString(m) }
func (*MappingEntry) ProtoMessage() {}
-func (*MappingEntry) Descriptor() ([]byte, []int) { return fileDescriptor1, []int{3} }
+func (*MappingEntry) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{3} }
-func (m *MappingEntry) GetNodeName() string {
+func (m *MappingEntry) GetClusterName() string {
if m != nil {
- return m.NodeName
+ return m.ClusterName
}
return ""
}
@@ -233,29 +253,29 @@ var _Sync_serviceDesc = grpc.ServiceDesc{
Metadata: "syncer.proto",
}
-func init() { proto1.RegisterFile("syncer.proto", fileDescriptor1) }
-
-var fileDescriptor1 = []byte{
- // 335 bytes of a gzipped FileDescriptorProto
- 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x6c, 0x92, 0xbd, 0x6e, 0xc2, 0x30,
- 0x14, 0x85, 0x1b, 0x08, 0x7f, 0x37, 0xb4, 0x95, 0x6e, 0x17, 0x8b, 0x2e, 0x28, 0xea, 0xc0, 0x02,
- 0x03, 0x55, 0x87, 0x76, 0xa6, 0x03, 0x03, 0x15, 0x0a, 0x73, 0x87, 0xd4, 0x58, 0xc8, 0x15, 0xd8,
- 0xa9, 0xed, 0x54, 0xe2, 0x61, 0xfa, 0x76, 0x7d, 0x90, 0xca, 0x8e, 0x03, 0x8e, 0xca, 0x94, 0xf8,
- 0xe4, 0xe4, 0xbb, 0xf7, 0x1c, 0x19, 0x86, 0xfa, 0x28, 0x28, 0x53, 0xb3, 0x42, 0x49, 0x23, 0xb1,
- 0xe3, 0x1e, 0xa3, 0x1b, 0xcd, 0xd4, 0x37, 0xa7, 0x4c, 0x57, 0x72, 0xfa, 0x0e, 0xc9, 0xba, 0xdc,
- 0xef, 0x33, 0xf6, 0x55, 0x32, 0x6d, 0x70, 0x0c, 0x89, 0x37, 0xbc, 0xe5, 0x07, 0x46, 0xa2, 0x71,
- 0x34, 0x19, 0x64, 0xa1, 0x84, 0x04, 0x7a, 0xb2, 0x30, 0x5c, 0x0a, 0x4d, 0x5a, 0xee, 0x6b, 0x7d,
- 0x44, 0x84, 0xd8, 0xf0, 0x03, 0x23, 0x6d, 0x27, 0xbb, 0xf7, 0xf4, 0x27, 0x82, 0x64, 0x73, 0x14,
- 0x74, 0x53, 0x11, 0xf0, 0x01, 0xae, 0xb7, 0xf2, 0x90, 0x73, 0xb1, 0x56, 0xf2, 0x93, 0x51, 0xe3,
- 0x27, 0x34, 0x45, 0x9c, 0x42, 0xcf, 0x8f, 0x74, 0x33, 0x92, 0xf9, 0x5d, 0xb5, 0xed, 0x6c, 0xc5,
- 0xa9, 0x92, 0x9e, 0x95, 0xd5, 0x1e, 0x7c, 0x86, 0x01, 0x17, 0xda, 0xe4, 0x82, 0x32, 0x4d, 0xda,
- 0xe3, 0xf6, 0x24, 0x99, 0xdf, 0x5f, 0xf8, 0x61, 0xe9, 0x3d, 0xd9, 0xd9, 0x9d, 0xbe, 0x40, 0xdf,
- 0xae, 0xb7, 0xc8, 0x4d, 0x8e, 0x33, 0xe8, 0xd7, 0xe5, 0x90, 0xc8, 0x51, 0xd0, 0x53, 0x82, 0x04,
- 0xd9, 0xc9, 0x93, 0xfe, 0x46, 0x30, 0x5c, 0xe5, 0x45, 0xc1, 0xc5, 0xee, 0x55, 0x18, 0x75, 0xc4,
- 0x11, 0xf4, 0x85, 0xdc, 0x86, 0xcd, 0x9d, 0xce, 0xff, 0x83, 0xb7, 0x2e, 0x05, 0x4f, 0x61, 0x28,
- 0xd5, 0xae, 0xde, 0x77, 0xe1, 0xab, 0x6c, 0x68, 0x96, 0x24, 0xd5, 0xae, 0x0e, 0xb3, 0x5c, 0x90,
- 0xb8, 0x22, 0x35, 0x44, 0x4b, 0xa2, 0xa5, 0x3a, 0x93, 0x3a, 0x15, 0x29, 0xd4, 0x2c, 0x89, 0x96,
- 0x2a, 0x20, 0x75, 0x2b, 0x52, 0x43, 0x9c, 0x3f, 0x41, 0x6c, 0xf3, 0xe3, 0x14, 0x62, 0x7b, 0x53,
- 0xb0, 0x2e, 0x25, 0xb8, 0x36, 0xa3, 0xdb, 0xa0, 0x28, 0xdb, 0x65, 0x7a, 0xf5, 0xd1, 0x75, 0xca,
- 0xe3, 0x5f, 0x00, 0x00, 0x00, 0xff, 0xff, 0xf7, 0xdc, 0x31, 0x80, 0x86, 0x02, 0x00, 0x00,
+func init() { proto1.RegisterFile("syncer.proto", fileDescriptor0) }
+
+var fileDescriptor0 = []byte{
+ // 336 bytes of a gzipped FileDescriptorProto
+ 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x6c, 0x92, 0xbf, 0x4e, 0xc3, 0x30,
+ 0x10, 0x87, 0x49, 0x9b, 0xfe, 0xbb, 0x14, 0x90, 0x8e, 0xc5, 0x2a, 0x4b, 0x15, 0x31, 0x74, 0x69,
+ 0x87, 0x22, 0x06, 0x98, 0xcb, 0xd0, 0xa1, 0xa8, 0x4a, 0x67, 0x86, 0x60, 0xac, 0xca, 0xa8, 0xb5,
+ 0x83, 0xed, 0x20, 0xe5, 0x61, 0x78, 0x41, 0x9e, 0x02, 0xd9, 0x71, 0xa8, 0x23, 0x3a, 0x25, 0xf9,
+ 0x72, 0xfe, 0xee, 0xee, 0x27, 0xc3, 0x58, 0x57, 0x82, 0x32, 0xb5, 0x28, 0x94, 0x34, 0x12, 0x7b,
+ 0xee, 0x31, 0xb9, 0xd2, 0x4c, 0x7d, 0x71, 0xca, 0x74, 0x8d, 0xd3, 0x57, 0x48, 0xb6, 0xe5, 0xe1,
+ 0x90, 0xb1, 0xcf, 0x92, 0x69, 0x83, 0x53, 0x48, 0x7c, 0xc1, 0x4b, 0x7e, 0x64, 0x24, 0x9a, 0x46,
+ 0xb3, 0x51, 0x16, 0x22, 0x24, 0x30, 0x90, 0x85, 0xe1, 0x52, 0x68, 0xd2, 0x71, 0x7f, 0x9b, 0x4f,
+ 0x44, 0x88, 0x0d, 0x3f, 0x32, 0xd2, 0x75, 0xd8, 0xbd, 0xa7, 0xdf, 0x11, 0x24, 0xbb, 0x4a, 0xd0,
+ 0x5d, 0x6d, 0xc0, 0x3b, 0xb8, 0x7c, 0x97, 0xc7, 0x9c, 0x8b, 0xad, 0x92, 0x1f, 0x8c, 0x1a, 0xdf,
+ 0xa1, 0x0d, 0x71, 0x0e, 0x03, 0xdf, 0xd2, 0xf5, 0x48, 0x96, 0x37, 0xf5, 0xb4, 0x8b, 0x0d, 0xa7,
+ 0x4a, 0x7a, 0x57, 0xd6, 0xd4, 0xe0, 0x23, 0x8c, 0xb8, 0xd0, 0x26, 0x17, 0x94, 0x69, 0xd2, 0x9d,
+ 0x76, 0x67, 0xc9, 0xf2, 0xf6, 0xcc, 0x81, 0xb5, 0xaf, 0xc9, 0x4e, 0xd5, 0xe9, 0x13, 0x0c, 0xed,
+ 0x78, 0xab, 0xdc, 0xe4, 0xb8, 0x80, 0x61, 0x13, 0x0e, 0x89, 0x9c, 0x05, 0xbd, 0x25, 0xd8, 0x20,
+ 0xfb, 0xab, 0x49, 0x7f, 0x22, 0x18, 0x6f, 0xf2, 0xa2, 0xe0, 0x62, 0xff, 0x2c, 0x8c, 0xaa, 0x6c,
+ 0x78, 0xf4, 0x50, 0x6a, 0xc3, 0x54, 0x18, 0x5e, 0x80, 0xfe, 0xaf, 0xdf, 0x39, 0xb7, 0x7e, 0x0a,
+ 0x63, 0xa9, 0xf6, 0xcd, 0xd4, 0x2b, 0x1f, 0x68, 0x8b, 0x59, 0x93, 0x54, 0xfb, 0x66, 0xa5, 0xf5,
+ 0x8a, 0xc4, 0xb5, 0xa9, 0x05, 0xad, 0x89, 0x96, 0xea, 0x64, 0xea, 0xd5, 0xa6, 0x90, 0x59, 0x13,
+ 0x2d, 0x55, 0x60, 0xea, 0xd7, 0xa6, 0x16, 0x5c, 0x3e, 0x40, 0x6c, 0x53, 0xc0, 0x39, 0xc4, 0xf6,
+ 0xbe, 0x60, 0x13, 0x4d, 0x70, 0x79, 0x26, 0xd7, 0x41, 0x5c, 0x36, 0xd1, 0xf4, 0xe2, 0xad, 0xef,
+ 0xc8, 0xfd, 0x6f, 0x00, 0x00, 0x00, 0xff, 0xff, 0xfa, 0x2b, 0x2b, 0x6b, 0x8c, 0x02, 0x00, 0x00,
}
diff --git a/syncer/proto/syncer.proto b/syncer/proto/syncer.proto
index 18d5be3..103e82c 100644
--- a/syncer/proto/syncer.proto
+++ b/syncer/proto/syncer.proto
@@ -24,7 +24,7 @@ message SyncData {
}
message MappingEntry {
- string nodeName = 1;
+ string clusterName = 1;
string domainProject = 2;
string orgServiceID = 3;
string orgInstanceID = 4;
diff --git a/syncer/serf/config.go b/syncer/serf/config.go
index 0ca8c70..8a25908 100644
--- a/syncer/serf/config.go
+++ b/syncer/serf/config.go
@@ -36,6 +36,7 @@ const (
groupExpect = 3
tagKeyClusterName = "syncer-cluster-name"
TagKeyClusterPort = "syncer-cluster-port"
+ TagKeyRPCPort = "syncer-rpc-port"
)
// DefaultConfig default config
@@ -95,9 +96,10 @@ func (c *Config) convertToSerf() (*serf.Config, error) {
serfConf.MemberlistConfig.BindAddr = bindIP
serfConf.MemberlistConfig.BindPort = bindPort
serfConf.NodeName = c.NodeName
+ serfConf.Tags = map[string]string{TagKeyRPCPort: strconv.Itoa(c.RPCPort)}
if c.ClusterName != "" {
- serfConf.Tags = map[string]string{tagKeyClusterName: c.ClusterName}
+ serfConf.Tags[tagKeyClusterName] = c.ClusterName
serfConf.Tags[TagKeyClusterPort] = strconv.Itoa(c.ClusterPort)
}
diff --git a/syncer/server/handler.go b/syncer/server/handler.go
index 0aa0c92..8760a83 100644
--- a/syncer/server/handler.go
+++ b/syncer/server/handler.go
@@ -19,13 +19,13 @@ package server
import (
"context"
"fmt"
- "time"
"github.com/apache/servicecomb-service-center/pkg/log"
+ "github.com/apache/servicecomb-service-center/pkg/util"
"github.com/apache/servicecomb-service-center/syncer/grpc"
pb "github.com/apache/servicecomb-service-center/syncer/proto"
- "github.com/gogo/protobuf/proto"
"github.com/hashicorp/serf/serf"
+ myserf "github.com/apache/servicecomb-service-center/syncer/serf"
)
const (
@@ -42,14 +42,8 @@ func (s *Server) tickHandler(ctx context.Context) {
// Flush data to the storage of servicecenter
s.servicecenter.FlushData()
- event, _ := proto.Marshal(&pb.Member{
- NodeName: s.conf.ClusterName,
- RPCPort: int32(s.conf.RPCPort),
- Time: fmt.Sprintf("%d", time.Now().UTC().Second()),
- })
-
// sends a UserEvent on Serf, the event will be broadcast between members
- err := s.agent.UserEvent(EventDiscovered, event, true)
+ err := s.agent.UserEvent(EventDiscovered, util.StringToBytesWithNoCopy(s.conf.ClusterName), true)
if err != nil {
log.Errorf(err, "Syncer send user event failed")
}
@@ -81,36 +75,31 @@ func (s *Server) HandleEvent(event serf.Event) {
// userEvent Handles "EventUser" notification events, no response required
func (s *Server) userEvent(event serf.UserEvent) {
log.Debug("Receive serf user event")
- m := &pb.Member{}
- err := proto.Unmarshal(event.Payload, m)
- if err != nil {
- log.Errorf(err, "trigger user event '%s' handler failed", event.EventType())
- return
- }
+ clusterName := util.BytesToStringWithNoCopy(event.Payload)
// Excludes notifications from self, as the gossip protocol inevitably has redundant notifications
- if s.conf.ClusterName == m.NodeName {
+ if s.conf.ClusterName == clusterName {
return
}
// Get member information and get synchronized data from it
- members := s.agent.GroupMembers(m.NodeName)
- if members == nil || len(members) == 0{
- log.Warnf("serf member = %s is not found", m.NodeName)
+ members := s.agent.GroupMembers(clusterName)
+ if members == nil || len(members) == 0 {
+ log.Warnf("serf member = %s is not found", clusterName)
return
}
// todo: grpc supports multi-address polling
// Get dta from remote member
- endpoint := fmt.Sprintf("%s:%d", members[0].Addr, m.RPCPort)
- log.Debugf("Going to pull data from %s %s", m.NodeName, endpoint)
+ endpoint := fmt.Sprintf("%s:%s", members[0].Addr, members[0].Tags[myserf.TagKeyRPCPort])
+ log.Debugf("Going to pull data from %s %s", members[0].Name, endpoint)
data, err := grpc.Pull(context.Background(), endpoint)
if err != nil {
- log.Errorf(err, "Pull other serf instances failed, node name is '%s'", m.NodeName)
+ log.Errorf(err, "Pull other serf instances failed, node name is '%s'", members[0].Name)
return
}
// Registry instances to servicecenter and update storage of it
- s.servicecenter.Registry(m.NodeName, data)
+ s.servicecenter.Registry(clusterName, data)
}
// queryEvent Handles "EventQuery" query events and respond if conditions are met
diff --git a/syncer/servicecenter/servicecenter.go b/syncer/servicecenter/servicecenter.go
index 5e7c8c3..b6f5446 100644
--- a/syncer/servicecenter/servicecenter.go
+++ b/syncer/servicecenter/servicecenter.go
@@ -28,7 +28,7 @@ import (
type Servicecenter interface {
SetStorage(storage Storage)
FlushData()
- Registry(nodeName string, data *pb.SyncData)
+ Registry(clusterName string, data *pb.SyncData)
Discovery() *pb.SyncData
}
@@ -42,8 +42,8 @@ type Storage interface {
UpdateData(data *pb.SyncData)
GetMaps() (maps pb.SyncMapping)
UpdateMaps(maps pb.SyncMapping)
- GetMapByNode(nodeName string) (mapping pb.SyncMapping)
- UpdateMapByNode(nodeName string, mapping pb.SyncMapping)
+ GetMapByNode(clusterName string) (mapping pb.SyncMapping)
+ UpdateMapByNode(clusterName string, mapping pb.SyncMapping)
}
// NewServicecenter new store with endpoints
@@ -78,8 +78,8 @@ func (s *servicecenter) FlushData() {
}
// Registry registry data to the servicecenter, update mapping data
-func (s *servicecenter) Registry(nodeName string, data *pb.SyncData) {
- mapping := s.storage.GetMapByNode(nodeName)
+func (s *servicecenter) Registry(clusterName string, data *pb.SyncData) {
+ mapping := s.storage.GetMapByNode(clusterName)
for _, svc := range data.Services {
log.Debugf("trying to do registration of service, serviceID = %s", svc.Service.ServiceId)
// If the svc is in the mapping, just do nothing, if not, created it in servicecenter and get the new serviceID
@@ -98,7 +98,7 @@ func (s *servicecenter) Registry(nodeName string, data *pb.SyncData) {
OrgServiceID: inst.ServiceId,
OrgInstanceID: inst.InstanceId,
CurServiceID: svcID,
- NodeName: nodeName,
+ ClusterName: clusterName,
}
item.CurInstanceID = s.registryInstances(svc.DomainProject, svcID, inst)
@@ -111,7 +111,7 @@ func (s *servicecenter) Registry(nodeName string, data *pb.SyncData) {
// UnRegistry instances that is not in the data which means the instance in the mapping is no longer actived
mapping = s.unRegistryInstances(data, mapping)
// Update mapping data of the node to the storage of the servicecenter
- s.storage.UpdateMapByNode(nodeName, mapping)
+ s.storage.UpdateMapByNode(clusterName, mapping)
}
// Discovery discovery data from storage
diff --git a/syncer/servicecenter/servicecenter_test.go b/syncer/servicecenter/servicecenter_test.go
index 3c38b03..7036b9d 100644
--- a/syncer/servicecenter/servicecenter_test.go
+++ b/syncer/servicecenter/servicecenter_test.go
@@ -79,8 +79,8 @@ func TestOnEvent(t *testing.T) {
return
}
- nodeName := "test_node"
- dc.Registry(nodeName, data)
+ clusterName := "test_node"
+ dc.Registry(clusterName, data)
mockplugin.SetGetAll(mockplugin.NewGetAll)
dc.FlushData()
@@ -90,25 +90,25 @@ func TestOnEvent(t *testing.T) {
return
}
- dc.Registry(nodeName, newData)
+ dc.Registry(clusterName, newData)
mockplugin.SetRegisterInstance(func(ctx context.Context, domainProject, serviceId string, instance *proto.MicroServiceInstance) (s string, e error) {
return "", errors.New("test error")
})
- dc.Registry(nodeName, data)
+ dc.Registry(clusterName, data)
mockplugin.SetRegisterInstance(nil)
- dc.Registry(nodeName, data)
+ dc.Registry(clusterName, data)
- dc.Registry(nodeName, data)
+ dc.Registry(clusterName, data)
mockplugin.SetHeartbeat(func(ctx context.Context, domainProject, serviceId, instanceId string) error {
return errors.New("test error")
})
- dc.Registry(nodeName, data)
+ dc.Registry(clusterName, data)
}
func initPlugin(conf *config.Config) {
[servicecomb-service-center] 02/02: Change nodeName to clusterName
in storage
Posted by ze...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zenlin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/servicecomb-service-center.git
commit 71b9cca812c8a5ef94e3b15924d1fa589d09afc7
Author: chinx <c5...@126.com>
AuthorDate: Thu Jun 20 18:40:11 2019 +0800
Change nodeName to clusterName in storage
---
syncer/etcd/storage.go | 10 +++++-----
syncer/pkg/mock/mocksotrage/storage.go | 8 ++++----
syncer/servicecenter/servicecenter.go | 10 +++++-----
syncer/servicecenter/servicecenter_test.go | 2 +-
4 files changed, 15 insertions(+), 15 deletions(-)
diff --git a/syncer/etcd/storage.go b/syncer/etcd/storage.go
index b1e4f7f..36995a3 100644
--- a/syncer/etcd/storage.go
+++ b/syncer/etcd/storage.go
@@ -94,8 +94,8 @@ next:
}
}
-// UpdateMapByNode update map to storage by clusterName of other node
-func (s *storage) UpdateMapByNode(clusterName string, mapping pb.SyncMapping) {
+// UpdateMapByCluster update map to storage by clusterName of other cluster
+func (s *storage) UpdateMapByCluster(clusterName string, mapping pb.SyncMapping) {
newMaps := make(pb.SyncMapping, 0, len(mapping))
for _, val := range mapping {
key := mappingsKey + "/" + clusterName + "/" + val.OrgInstanceID
@@ -110,11 +110,11 @@ func (s *storage) UpdateMapByNode(clusterName string, mapping pb.SyncMapping) {
}
newMaps = append(newMaps, val)
}
- s.cleanExpired(s.GetMapByNode(clusterName), newMaps)
+ s.cleanExpired(s.GetMapByCluster(clusterName), newMaps)
}
-// GetMapByNode get map by clusterName of other node
-func (s *storage) GetMapByNode(clusterName string) (mapping pb.SyncMapping) {
+// GetMapByCluster get map by clusterName of other cluster
+func (s *storage) GetMapByCluster(clusterName string) (mapping pb.SyncMapping) {
maps := make(pb.SyncMapping, 0, 10)
s.getPrefixKey(mappingsKey+"/"+clusterName, func(key, val []byte) (next bool) {
next = true
diff --git a/syncer/pkg/mock/mocksotrage/storage.go b/syncer/pkg/mock/mocksotrage/storage.go
index 36a1685..11b7284 100644
--- a/syncer/pkg/mock/mocksotrage/storage.go
+++ b/syncer/pkg/mock/mocksotrage/storage.go
@@ -57,15 +57,15 @@ func (s *Storage) GetData() (data *pb.SyncData) {
return
}
-// UpdateMapByNode update map to storage by clusterName of other node
-func (s *Storage) UpdateMapByNode(clusterName string, mapping pb.SyncMapping) {
+// UpdateMapByCluster update map to storage by clusterName of other cluster
+func (s *Storage) UpdateMapByCluster(clusterName string, mapping pb.SyncMapping) {
s.lock.Lock()
s.maps[clusterName] = mapping
s.lock.Unlock()
}
-// GetMapByNode get map by clusterName of other node
-func (s *Storage) GetMapByNode(clusterName string) (mapping pb.SyncMapping) {
+// GetMapByCluster get map by clusterName of other cluster
+func (s *Storage) GetMapByCluster(clusterName string) (mapping pb.SyncMapping) {
s.lock.RLock()
data, ok := s.maps[clusterName]
if !ok {
diff --git a/syncer/servicecenter/servicecenter.go b/syncer/servicecenter/servicecenter.go
index b6f5446..a2da4ad 100644
--- a/syncer/servicecenter/servicecenter.go
+++ b/syncer/servicecenter/servicecenter.go
@@ -42,8 +42,8 @@ type Storage interface {
UpdateData(data *pb.SyncData)
GetMaps() (maps pb.SyncMapping)
UpdateMaps(maps pb.SyncMapping)
- GetMapByNode(clusterName string) (mapping pb.SyncMapping)
- UpdateMapByNode(clusterName string, mapping pb.SyncMapping)
+ GetMapByCluster(clusterName string) (mapping pb.SyncMapping)
+ UpdateMapByCluster(clusterName string, mapping pb.SyncMapping)
}
// NewServicecenter new store with endpoints
@@ -79,7 +79,7 @@ func (s *servicecenter) FlushData() {
// Registry registry data to the servicecenter, update mapping data
func (s *servicecenter) Registry(clusterName string, data *pb.SyncData) {
- mapping := s.storage.GetMapByNode(clusterName)
+ mapping := s.storage.GetMapByCluster(clusterName)
for _, svc := range data.Services {
log.Debugf("trying to do registration of service, serviceID = %s", svc.Service.ServiceId)
// If the svc is in the mapping, just do nothing, if not, created it in servicecenter and get the new serviceID
@@ -110,8 +110,8 @@ func (s *servicecenter) Registry(clusterName string, data *pb.SyncData) {
}
// UnRegistry instances that is not in the data which means the instance in the mapping is no longer actived
mapping = s.unRegistryInstances(data, mapping)
- // Update mapping data of the node to the storage of the servicecenter
- s.storage.UpdateMapByNode(clusterName, mapping)
+ // Update mapping data of the cluster to the storage of the servicecenter
+ s.storage.UpdateMapByCluster(clusterName, mapping)
}
// Discovery discovery data from storage
diff --git a/syncer/servicecenter/servicecenter_test.go b/syncer/servicecenter/servicecenter_test.go
index 7036b9d..d12d874 100644
--- a/syncer/servicecenter/servicecenter_test.go
+++ b/syncer/servicecenter/servicecenter_test.go
@@ -79,7 +79,7 @@ func TestOnEvent(t *testing.T) {
return
}
- clusterName := "test_node"
+ clusterName := "test_cluster"
dc.Registry(clusterName, data)
mockplugin.SetGetAll(mockplugin.NewGetAll)