You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by lu...@apache.org on 2023/09/06 06:35:20 UTC
[skywalking-banyandb] branch liaison-topology updated (841ca0c8 -> 7982f866)
This is an automated email from the ASF dual-hosted git repository.
lujiajing pushed a change to branch liaison-topology
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
discard 841ca0c8 add discovery module
new 7982f866 setup selector
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (841ca0c8)
\
N -- N -- N refs/heads/liaison-topology (7982f866)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 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:
banyand/discovery/service.go | 77 -------------------------------
banyand/internal/cmd/liaison.go | 2 +-
banyand/internal/cmd/standalone.go | 2 +-
banyand/liaison/grpc/discovery.go | 4 +-
banyand/liaison/grpc/measure.go | 9 +++-
banyand/liaison/grpc/node.go | 84 ++++++++++++++++++++++++++++++++++
banyand/liaison/grpc/registry_test.go | 2 +-
banyand/liaison/grpc/server.go | 6 +--
banyand/liaison/grpc/stream.go | 8 +++-
pkg/discovery/register.go | 67 ---------------------------
pkg/node/interface.go | 86 +++++++++++++++++++++++++++++++++++
pkg/test/setup/setup.go | 2 +-
12 files changed, 194 insertions(+), 155 deletions(-)
delete mode 100644 banyand/discovery/service.go
create mode 100644 banyand/liaison/grpc/node.go
delete mode 100644 pkg/discovery/register.go
create mode 100644 pkg/node/interface.go
[skywalking-banyandb] 01/01: setup selector
Posted by lu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lujiajing pushed a commit to branch liaison-topology
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 7982f866310251e8f279225906057b8dcea7dc11
Author: Megrez Lu <lu...@gmail.com>
AuthorDate: Wed Sep 6 14:35:10 2023 +0800
setup selector
---
banyand/internal/cmd/liaison.go | 2 +-
banyand/internal/cmd/standalone.go | 2 +-
banyand/liaison/grpc/discovery.go | 4 +-
banyand/liaison/grpc/measure.go | 9 +++-
banyand/liaison/grpc/node.go | 84 ++++++++++++++++++++++++++++++++++
banyand/liaison/grpc/registry_test.go | 2 +-
banyand/liaison/grpc/server.go | 6 +--
banyand/liaison/grpc/stream.go | 8 +++-
pkg/node/interface.go | 86 +++++++++++++++++++++++++++++++++++
pkg/test/setup/setup.go | 2 +-
10 files changed, 194 insertions(+), 11 deletions(-)
diff --git a/banyand/internal/cmd/liaison.go b/banyand/internal/cmd/liaison.go
index 0b842201..0a8ad2cc 100644
--- a/banyand/internal/cmd/liaison.go
+++ b/banyand/internal/cmd/liaison.go
@@ -45,7 +45,7 @@ func newLiaisonCmd() *cobra.Command {
l.Fatal().Err(err).Msg("failed to initiate metadata service")
}
pipeline := pub.New(metaSvc)
- grpcServer := grpc.NewServer(ctx, pipeline, metaSvc)
+ grpcServer := grpc.NewServer(ctx, pipeline, metaSvc, grpc.NewClusterNodeRegistry(metaSvc))
profSvc := observability.NewProfService()
metricSvc := observability.NewMetricService()
httpServer := http.NewServer()
diff --git a/banyand/internal/cmd/standalone.go b/banyand/internal/cmd/standalone.go
index 8686db07..16a91192 100644
--- a/banyand/internal/cmd/standalone.go
+++ b/banyand/internal/cmd/standalone.go
@@ -60,7 +60,7 @@ func newStandaloneCmd() *cobra.Command {
if err != nil {
l.Fatal().Err(err).Msg("failed to initiate query processor")
}
- grpcServer := grpc.NewServer(ctx, pipeline, metaSvc)
+ grpcServer := grpc.NewServer(ctx, pipeline, metaSvc, grpc.NewLocalNodeRegistry())
profSvc := observability.NewProfService()
metricSvc := observability.NewMetricService()
httpServer := http.NewServer()
diff --git a/banyand/liaison/grpc/discovery.go b/banyand/liaison/grpc/discovery.go
index e2fb24c9..b10e16f7 100644
--- a/banyand/liaison/grpc/discovery.go
+++ b/banyand/liaison/grpc/discovery.go
@@ -41,6 +41,7 @@ var errNotExist = errors.New("the object doesn't exist")
type discoveryService struct {
pipeline queue.Client
+ nodeRegistry NodeRegistry
metadataRepo metadata.Repo
shardRepo *shardRepo
entityRepo *entityRepo
@@ -48,10 +49,11 @@ type discoveryService struct {
kind schema.Kind
}
-func newDiscoveryService(pipeline queue.Client, kind schema.Kind, metadataRepo metadata.Repo) *discoveryService {
+func newDiscoveryService(pipeline queue.Client, kind schema.Kind, metadataRepo metadata.Repo, nodeRegistry NodeRegistry) *discoveryService {
sr := &shardRepo{shardEventsMap: make(map[identity]uint32)}
er := &entityRepo{entitiesMap: make(map[identity]partition.EntityLocator)}
return &discoveryService{
+ nodeRegistry: nodeRegistry,
shardRepo: sr,
entityRepo: er,
pipeline: pipeline,
diff --git a/banyand/liaison/grpc/measure.go b/banyand/liaison/grpc/measure.go
index 8a541147..38deab8c 100644
--- a/banyand/liaison/grpc/measure.go
+++ b/banyand/liaison/grpc/measure.go
@@ -101,8 +101,13 @@ func (ms *measureService) Write(measure measurev1.MeasureService_WriteServer) er
SeriesHash: tsdb.HashEntity(entity),
EntityValues: tagValues.Encode(),
}
- // TODO: set node id
- message := bus.NewMessageWithNode(bus.MessageID(time.Now().UnixNano()), "todo", iwr)
+ nodeId, errPickNode := ms.nodeRegistry.Locate(writeRequest.GetMetadata().GetGroup(), writeRequest.GetMetadata().GetName(), uint32(shardID))
+ if errPickNode != nil {
+ ms.sampled.Error().Err(errPickNode).RawJSON("written", logger.Proto(writeRequest)).Msg("failed to pick an available node")
+ reply(measure, ms.sampled)
+ continue
+ }
+ message := bus.NewMessageWithNode(bus.MessageID(time.Now().UnixNano()), nodeId, iwr)
_, errWritePub := publisher.Publish(data.TopicMeasureWrite, message)
if errWritePub != nil {
ms.sampled.Error().Err(errWritePub).RawJSON("written", logger.Proto(writeRequest)).Msg("failed to send a message")
diff --git a/banyand/liaison/grpc/node.go b/banyand/liaison/grpc/node.go
new file mode 100644
index 00000000..dff0ca22
--- /dev/null
+++ b/banyand/liaison/grpc/node.go
@@ -0,0 +1,84 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package grpc
+
+import (
+ "github.com/pkg/errors"
+
+ "github.com/apache/skywalking-banyandb/banyand/metadata"
+ "github.com/apache/skywalking-banyandb/banyand/metadata/schema"
+ "github.com/apache/skywalking-banyandb/pkg/node"
+)
+
+var (
+ _ schema.EventHandler = (*clusterNodeService)(nil)
+ _ NodeRegistry = (*clusterNodeService)(nil)
+)
+
+type NodeRegistry interface {
+ Locate(group, name string, shardId uint32) (string, error)
+}
+
+type clusterNodeService struct {
+ metaRepo metadata.Repo
+ sel node.Selector
+}
+
+func NewClusterNodeRegistry(metaRepo metadata.Repo) NodeRegistry {
+ cns := &clusterNodeService{
+ metaRepo: metaRepo,
+ sel: node.NewPickFirstSelector(),
+ }
+ cns.metaRepo.RegisterHandler("cluster-node-service", schema.KindNode, cns)
+ return cns
+}
+
+func (n *clusterNodeService) Locate(group, name string, shardId uint32) (string, error) {
+ nodeId, err := n.sel.Pick(group, name, shardId)
+ if err != nil {
+ return "", errors.Wrapf(err, "fail to locate %s/%s(%d)", group, name, shardId)
+ }
+ return nodeId, nil
+}
+
+func (n *clusterNodeService) OnAddOrUpdate(metadata schema.Metadata) {
+ switch metadata.Kind {
+ case schema.KindNode:
+ n.sel.AddNode(metadata.Name)
+ default:
+ }
+}
+
+func (n *clusterNodeService) OnDelete(metadata schema.Metadata) {
+ switch metadata.Kind {
+ case schema.KindNode:
+ n.sel.RemoveNode(metadata.Name)
+ default:
+ }
+}
+
+type localNodeService struct{}
+
+func NewLocalNodeRegistry() NodeRegistry {
+ return localNodeService{}
+}
+
+// Locate of localNodeService always returns
+func (localNodeService) Locate(_group, _name string, _shardId uint32) (string, error) {
+ return "local", nil
+}
diff --git a/banyand/liaison/grpc/registry_test.go b/banyand/liaison/grpc/registry_test.go
index 6c86d780..6dff21fa 100644
--- a/banyand/liaison/grpc/registry_test.go
+++ b/banyand/liaison/grpc/registry_test.go
@@ -178,7 +178,7 @@ func setupForRegistry() func() {
metaSvc, err := metadata.NewService(context.TODO())
Expect(err).NotTo(HaveOccurred())
- tcp := grpc.NewServer(context.TODO(), pipeline, metaSvc)
+ tcp := grpc.NewServer(context.TODO(), pipeline, metaSvc, grpc.NewLocalNodeRegistry())
preloadStreamSvc := &preloadStreamService{metaSvc: metaSvc}
var flags []string
metaPath, metaDeferFunc, err := test.NewSpace()
diff --git a/banyand/liaison/grpc/server.go b/banyand/liaison/grpc/server.go
index cc7e0817..ba98016e 100644
--- a/banyand/liaison/grpc/server.go
+++ b/banyand/liaison/grpc/server.go
@@ -91,12 +91,12 @@ type server struct {
}
// NewServer returns a new gRPC server.
-func NewServer(_ context.Context, pipeline queue.Client, schemaRegistry metadata.Repo) Server {
+func NewServer(_ context.Context, pipeline queue.Client, schemaRegistry metadata.Repo, nodeRegistry NodeRegistry) Server {
streamSVC := &streamService{
- discoveryService: newDiscoveryService(pipeline, schema.KindStream, schemaRegistry),
+ discoveryService: newDiscoveryService(pipeline, schema.KindStream, schemaRegistry, nodeRegistry),
}
measureSVC := &measureService{
- discoveryService: newDiscoveryService(pipeline, schema.KindMeasure, schemaRegistry),
+ discoveryService: newDiscoveryService(pipeline, schema.KindMeasure, schemaRegistry, nodeRegistry),
}
s := &server{
pipeline: pipeline,
diff --git a/banyand/liaison/grpc/stream.go b/banyand/liaison/grpc/stream.go
index e4cc1ed1..06b78112 100644
--- a/banyand/liaison/grpc/stream.go
+++ b/banyand/liaison/grpc/stream.go
@@ -103,7 +103,13 @@ func (s *streamService) Write(stream streamv1.StreamService_WriteServer) error {
if s.log.Debug().Enabled() {
iwr.EntityValues = tagValues.Encode()
}
- message := bus.NewMessageWithNode(bus.MessageID(time.Now().UnixNano()), "TODO", iwr)
+ nodeId, errPickNode := s.nodeRegistry.Locate(writeEntity.GetMetadata().GetGroup(), writeEntity.GetMetadata().GetName(), uint32(shardID))
+ if errPickNode != nil {
+ s.sampled.Error().Err(errPickNode).RawJSON("written", logger.Proto(writeEntity)).Msg("failed to pick an available node")
+ reply(stream, s.sampled)
+ continue
+ }
+ message := bus.NewMessageWithNode(bus.MessageID(time.Now().UnixNano()), nodeId, iwr)
_, errWritePub := publisher.Publish(data.TopicStreamWrite, message)
if errWritePub != nil {
s.sampled.Error().Err(errWritePub).RawJSON("written", logger.Proto(writeEntity)).Msg("failed to send a message")
diff --git a/pkg/node/interface.go b/pkg/node/interface.go
new file mode 100644
index 00000000..a12c36e0
--- /dev/null
+++ b/pkg/node/interface.go
@@ -0,0 +1,86 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Package node provides node selector for liaison
+package node
+
+import (
+ "sync"
+
+ "github.com/pkg/errors"
+ "golang.org/x/exp/slices"
+)
+
+var (
+ _ Selector = (*pickFirstSelector)(nil)
+
+ ErrNoAvailableNode = errors.New("selector: no available node")
+)
+
+type Selector interface {
+ AddNode(nodeId string)
+ RemoveNode(nodeId string)
+ Pick(group, name string, shardId uint32) (string, error)
+}
+
+func NewPickFirstSelector() Selector {
+ return &pickFirstSelector{}
+}
+
+// pickFirstSelector always pick the first node in the sorted node ids list
+type pickFirstSelector struct {
+ nodeIds []string
+ nodeIdMap map[string]struct{}
+ mu sync.RWMutex
+}
+
+func (p *pickFirstSelector) AddNode(nodeId string) {
+ p.mu.RLock()
+ if _, ok := p.nodeIdMap[nodeId]; !ok {
+ p.mu.RUnlock()
+ return
+ }
+ p.mu.RUnlock()
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ p.nodeIdMap[nodeId] = struct{}{}
+ p.nodeIds = append(p.nodeIds, nodeId)
+ slices.Sort(p.nodeIds)
+}
+
+func (p *pickFirstSelector) RemoveNode(nodeId string) {
+ p.mu.RLock()
+ if _, ok := p.nodeIdMap[nodeId]; !ok {
+ p.mu.RUnlock()
+ return
+ }
+ p.mu.RUnlock()
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ p.nodeIdMap[nodeId] = struct{}{}
+ p.nodeIds = append(p.nodeIds, nodeId)
+ slices.Sort(p.nodeIds)
+}
+
+func (p *pickFirstSelector) Pick(group, name string, shardId uint32) (string, error) {
+ p.mu.RLock()
+ defer p.mu.RUnlock()
+ if len(p.nodeIds) == 0 {
+ return "", ErrNoAvailableNode
+ }
+ return p.nodeIds[0], nil
+}
diff --git a/pkg/test/setup/setup.go b/pkg/test/setup/setup.go
index e7fb94c7..281ffe8a 100644
--- a/pkg/test/setup/setup.go
+++ b/pkg/test/setup/setup.go
@@ -92,7 +92,7 @@ func modules(schemaLoaders []SchemaLoader, flags []string) func() {
// Init `Query` module
q, err := query.NewService(context.TODO(), streamSvc, measureSvc, metaSvc, pipeline)
gomega.Expect(err).NotTo(gomega.HaveOccurred())
- tcp := grpc.NewServer(context.TODO(), pipeline, metaSvc)
+ tcp := grpc.NewServer(context.TODO(), pipeline, metaSvc, grpc.NewLocalNodeRegistry())
httpServer := http.NewServer()
units := []run.Unit{