You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ha...@apache.org on 2024/07/28 01:00:08 UTC
(skywalking-banyandb) branch rr-selector created (now 3d915b4a)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a change to branch rr-selector
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
at 3d915b4a Imporve round robin selector
This branch includes the following new commits:
new 3d915b4a Imporve round robin selector
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.
(skywalking-banyandb) 01/01: Imporve round robin selector
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch rr-selector
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 3d915b4acc88d3d8180b541b249d86bd7f8c36e5
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Sun Jul 28 08:57:47 2024 +0800
Imporve round robin selector
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
pkg/cmdsetup/liaison.go | 4 +-
pkg/node/interface.go | 12 ++-
pkg/node/maglev.go | 9 ++-
pkg/node/round_robin.go | 169 ++++++++++++++++++++++---------------------
pkg/node/round_robin_test.go | 72 +++++++++++-------
5 files changed, 150 insertions(+), 116 deletions(-)
diff --git a/pkg/cmdsetup/liaison.go b/pkg/cmdsetup/liaison.go
index 460c8544..66f3da41 100644
--- a/pkg/cmdsetup/liaison.go
+++ b/pkg/cmdsetup/liaison.go
@@ -46,7 +46,7 @@ func newLiaisonCmd(runners ...run.Unit) *cobra.Command {
}
pipeline := pub.New(metaSvc)
localPipeline := queue.Local()
- nodeSel := node.NewRoundRobinSelector()
+ nodeSel := node.NewRoundRobinSelector(metaSvc)
nodeRegistry := grpc.NewClusterNodeRegistry(pipeline, nodeSel)
grpcServer := grpc.NewServer(ctx, pipeline, localPipeline, metaSvc, nodeRegistry)
profSvc := observability.NewProfService()
@@ -62,6 +62,7 @@ func newLiaisonCmd(runners ...run.Unit) *cobra.Command {
metaSvc,
localPipeline,
pipeline,
+ nodeSel,
dQuery,
grpcServer,
httpServer,
@@ -77,7 +78,6 @@ func newLiaisonCmd(runners ...run.Unit) *cobra.Command {
Version: version.Build(),
Short: "Run as the liaison server",
RunE: func(_ *cobra.Command, _ []string) (err error) {
- defer nodeSel.Close()
node, err := common.GenerateNode(grpcServer.GetPort(), httpServer.GetPort())
if err != nil {
return err
diff --git a/pkg/node/interface.go b/pkg/node/interface.go
index 1a4f5882..0f41aab5 100644
--- a/pkg/node/interface.go
+++ b/pkg/node/interface.go
@@ -19,12 +19,14 @@
package node
import (
+ "context"
"sync"
"github.com/pkg/errors"
"golang.org/x/exp/slices"
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ "github.com/apache/skywalking-banyandb/pkg/run"
)
var (
@@ -39,7 +41,7 @@ type Selector interface {
AddNode(node *databasev1.Node)
RemoveNode(node *databasev1.Node)
Pick(group, name string, shardID uint32) (string, error)
- Close()
+ run.PreRunner
}
// NewPickFirstSelector returns a simple selector that always returns the first node if exists.
@@ -56,7 +58,13 @@ type pickFirstSelector struct {
mu sync.RWMutex
}
-func (p *pickFirstSelector) Close() {}
+func (p *pickFirstSelector) PreRun(context.Context) error {
+ return nil
+}
+
+func (p *pickFirstSelector) Name() string {
+ return "pick-first"
+}
func (p *pickFirstSelector) AddNode(node *databasev1.Node) {
nodeID := node.GetMetadata().GetName()
diff --git a/pkg/node/maglev.go b/pkg/node/maglev.go
index 4ae21740..34a855f6 100644
--- a/pkg/node/maglev.go
+++ b/pkg/node/maglev.go
@@ -18,6 +18,7 @@
package node
import (
+ "context"
"sort"
"strconv"
"sync"
@@ -37,7 +38,13 @@ type maglevSelector struct {
mutex sync.RWMutex
}
-func (m *maglevSelector) Close() {}
+func (m *maglevSelector) Name() string {
+ return "maglev-selector"
+}
+
+func (m *maglevSelector) PreRun(context.Context) error {
+ return nil
+}
func (m *maglevSelector) AddNode(node *databasev1.Node) {
m.mutex.Lock()
diff --git a/pkg/node/round_robin.go b/pkg/node/round_robin.go
index da85c210..79c9f900 100644
--- a/pkg/node/round_robin.go
+++ b/pkg/node/round_robin.go
@@ -18,49 +18,101 @@
package node
import (
+ "context"
"fmt"
"slices"
"sort"
"strings"
"sync"
- "sync/atomic"
"time"
"github.com/pkg/errors"
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
- "github.com/apache/skywalking-banyandb/pkg/timestamp"
-)
-
-const (
- expiredKeyCleanupInterval = 1 * time.Hour
- keyTTL = 24 * time.Hour
+ "github.com/apache/skywalking-banyandb/banyand/metadata"
+ "github.com/apache/skywalking-banyandb/banyand/metadata/schema"
)
type roundRobinSelector struct {
- clock timestamp.Clock
- closeCh chan struct{}
- lookupTable sync.Map
- nodes []string
- mu sync.RWMutex
- once sync.Once
- tMu sync.Mutex
-}
-
-func (r *roundRobinSelector) Close() {
- close(r.closeCh)
+ schemaRegistry metadata.Repo
+ closeCh chan struct{}
+ lookupTable sync.Map
+ nodes []string
+ mu sync.RWMutex
}
// NewRoundRobinSelector creates a new round-robin selector.
-func NewRoundRobinSelector() Selector {
+func NewRoundRobinSelector(schemaRegistry metadata.Repo) Selector {
rrs := &roundRobinSelector{
- nodes: make([]string, 0),
- clock: timestamp.NewClock(),
- closeCh: make(chan struct{}),
+ nodes: make([]string, 0),
+ closeCh: make(chan struct{}),
+ schemaRegistry: schemaRegistry,
}
return rrs
}
+func (r *roundRobinSelector) Name() string {
+ return "round-robin-selector"
+}
+
+func (r *roundRobinSelector) PreRun(context.Context) error {
+ r.schemaRegistry.RegisterHandler("liaison", schema.KindGroup, r)
+ return nil
+}
+
+func (r *roundRobinSelector) OnAddOrUpdate(schemaMetadata schema.Metadata) {
+ if schemaMetadata.Kind != schema.KindGroup {
+ return
+ }
+ group := schemaMetadata.Spec.(*commonv1.Group)
+ for i := uint32(0); i < group.ResourceOpts.ShardNum; i++ {
+ k := key{group: group.Metadata.Name, shardID: i}
+ r.lookupTable.Store(k, 0)
+ }
+ r.sortEntries()
+}
+
+func (r *roundRobinSelector) OnDelete(schemaMetadata schema.Metadata) {
+ if schemaMetadata.Kind != schema.KindGroup {
+ return
+ }
+ group := schemaMetadata.Spec.(*commonv1.Group)
+ for i := uint32(0); i < group.ResourceOpts.ShardNum; i++ {
+ k := key{group: group.Metadata.Name, shardID: i}
+ r.lookupTable.Delete(k)
+ }
+ r.sortEntries()
+}
+
+func (r *roundRobinSelector) OnInit(kinds []schema.Kind) (bool, []int64) {
+ if len(kinds) != 1 {
+ return false, nil
+ }
+ if kinds[0] != schema.KindGroup {
+ return false, nil
+ }
+ ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
+ defer cancel()
+ gg, err := r.schemaRegistry.GroupRegistry().ListGroup(ctx)
+ if err != nil {
+ panic(fmt.Sprintf("failed to list group: %v", err))
+ }
+ var revision int64
+ r.lookupTable = sync.Map{}
+ for _, g := range gg {
+ if g.Metadata.ModRevision > revision {
+ revision = g.Metadata.ModRevision
+ }
+ for i := uint32(0); i < g.ResourceOpts.ShardNum; i++ {
+ k := key{group: g.Metadata.Name, shardID: i}
+ r.lookupTable.Store(k, 0)
+ }
+ }
+ r.sortEntries()
+ return true, []int64{revision}
+}
+
func (r *roundRobinSelector) AddNode(node *databasev1.Node) {
r.mu.Lock()
defer r.mu.Unlock()
@@ -90,13 +142,11 @@ func (r *roundRobinSelector) Pick(group, _ string, shardID uint32) (string, erro
if ok {
return r.selectNode(entry), nil
}
- r.tMu.Lock()
- defer r.tMu.Unlock()
- if entry, ok := r.lookupTable.Load(k); ok {
- return r.selectNode(entry), nil
- }
+ return "", fmt.Errorf("%s-%d is a unknown shard", group, shardID)
+}
- keys := []key{k}
+func (r *roundRobinSelector) sortEntries() {
+ var keys []key
r.lookupTable.Range(func(k, _ any) bool {
keys = append(keys, k.(key))
return true
@@ -109,69 +159,20 @@ func (r *roundRobinSelector) Pick(group, _ string, shardID uint32) (string, erro
return int(a.shardID) - int(b.shardID)
})
for i := range keys {
- if entry, ok := r.lookupTable.Load(keys[i]); ok {
- entry.(*tableEntry).index = i
- } else {
- r.lookupTable.Store(keys[i], r.newTableEntry(i))
- }
- }
- r.once.Do(r.startCleanupTicker)
- if entry, ok := r.lookupTable.Load(k); ok {
- return r.selectNode(entry), nil
+ r.lookupTable.Store(keys[i], i)
}
- panic(fmt.Sprintf("key %v not found", k))
+}
+
+func (r *roundRobinSelector) Close() {
+ close(r.closeCh)
}
func (r *roundRobinSelector) selectNode(entry any) string {
- e := entry.(*tableEntry)
- now := r.clock.Now()
- e.lastAccess.Store(&now)
- return r.nodes[e.index%len(r.nodes)]
+ index := entry.(int)
+ return r.nodes[index%len(r.nodes)]
}
type key struct {
group string
shardID uint32
}
-
-type tableEntry struct {
- lastAccess *atomic.Pointer[time.Time]
- index int
-}
-
-func (r *roundRobinSelector) newTableEntry(index int) *tableEntry {
- p := atomic.Pointer[time.Time]{}
- now := r.clock.Now()
- p.Store(&now)
- return &tableEntry{
- index: index,
- lastAccess: &p,
- }
-}
-
-func (r *roundRobinSelector) cleanupExpiredEntries() {
- now := r.clock.Now()
- r.tMu.Lock()
- defer r.tMu.Unlock()
-
- r.lookupTable.Range(func(k, value any) bool {
- e := value.(*tableEntry)
- if now.Sub(*e.lastAccess.Load()) > keyTTL {
- r.lookupTable.Delete(k)
- }
- return true
- })
-}
-
-func (r *roundRobinSelector) startCleanupTicker() {
- ticker := r.clock.Ticker(expiredKeyCleanupInterval)
- go func() {
- select {
- case <-r.closeCh:
- ticker.Stop()
- return
- case <-ticker.C:
- r.cleanupExpiredEntries()
- }
- }()
-}
diff --git a/pkg/node/round_robin_test.go b/pkg/node/round_robin_test.go
index 69c1822a..032cd525 100644
--- a/pkg/node/round_robin_test.go
+++ b/pkg/node/round_robin_test.go
@@ -19,23 +19,33 @@ package node
import (
"testing"
- "time"
"github.com/stretchr/testify/assert"
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
- "github.com/apache/skywalking-banyandb/pkg/timestamp"
+ "github.com/apache/skywalking-banyandb/banyand/metadata/schema"
)
func TestPickEmptySelector(t *testing.T) {
- selector := NewRoundRobinSelector()
+ selector := NewRoundRobinSelector(nil)
+ setupGroup(selector)
_, err := selector.Pick("group1", "", 0)
assert.Error(t, err)
}
+func TestPickUnknownGroup(t *testing.T) {
+ selector := NewRoundRobinSelector(nil)
+ _, err := selector.Pick("group1", "", 0)
+ assert.Error(t, err)
+ setupGroup(selector)
+ _, err = selector.Pick("group1", "", 100)
+ assert.Error(t, err)
+}
+
func TestPickSingleSelection(t *testing.T) {
- selector := NewRoundRobinSelector()
+ selector := NewRoundRobinSelector(nil)
+ setupGroup(selector)
selector.AddNode(&databasev1.Node{Metadata: &commonv1.Metadata{Name: "node1"}})
node, err := selector.Pick("group1", "", 0)
assert.NoError(t, err)
@@ -43,13 +53,12 @@ func TestPickSingleSelection(t *testing.T) {
}
func TestPickMultipleSelections(t *testing.T) {
- selector := NewRoundRobinSelector()
+ selector := NewRoundRobinSelector(nil)
+ setupGroup(selector)
selector.AddNode(&databasev1.Node{Metadata: &commonv1.Metadata{Name: "node1"}})
selector.AddNode(&databasev1.Node{Metadata: &commonv1.Metadata{Name: "node2"}})
- // load data
- _, err := selector.Pick("group1", "", 0)
- assert.NoError(t, err)
- _, err = selector.Pick("group1", "", 1)
+
+ _, err := selector.Pick("group1", "", 1)
assert.NoError(t, err)
node1, err := selector.Pick("group1", "", 0)
assert.NoError(t, err)
@@ -59,7 +68,8 @@ func TestPickMultipleSelections(t *testing.T) {
}
func TestPickNodeRemoval(t *testing.T) {
- selector := NewRoundRobinSelector()
+ selector := NewRoundRobinSelector(nil)
+ setupGroup(selector)
selector.AddNode(&databasev1.Node{Metadata: &commonv1.Metadata{Name: "node1"}})
selector.AddNode(&databasev1.Node{Metadata: &commonv1.Metadata{Name: "node2"}})
selector.RemoveNode(&databasev1.Node{Metadata: &commonv1.Metadata{Name: "node1"}})
@@ -69,7 +79,8 @@ func TestPickNodeRemoval(t *testing.T) {
}
func TestPickConsistentSelectionAfterRemoval(t *testing.T) {
- selector := NewRoundRobinSelector()
+ selector := NewRoundRobinSelector(nil)
+ setupGroup(selector)
selector.AddNode(&databasev1.Node{Metadata: &commonv1.Metadata{Name: "node1"}})
selector.AddNode(&databasev1.Node{Metadata: &commonv1.Metadata{Name: "node2"}})
selector.AddNode(&databasev1.Node{Metadata: &commonv1.Metadata{Name: "node3"}})
@@ -86,27 +97,34 @@ func TestPickConsistentSelectionAfterRemoval(t *testing.T) {
assert.Equal(t, "node3", node)
}
-func TestCleanupExpiredEntries(t *testing.T) {
- mc := timestamp.NewMockClock()
- mc.Set(time.Date(1970, 0o1, 0o1, 0, 0, 0, 0, time.Local))
+func TestCleanupGroup(t *testing.T) {
selector := &roundRobinSelector{
nodes: make([]string, 0),
- clock: mc,
}
+ setupGroup(selector)
selector.AddNode(&databasev1.Node{Metadata: &commonv1.Metadata{Name: "node1"}})
selector.AddNode(&databasev1.Node{Metadata: &commonv1.Metadata{Name: "node2"}})
_, err := selector.Pick("group1", "", 0)
assert.NoError(t, err)
- _, ok := selector.lookupTable.Load(key{group: "group1", shardID: 0})
- assert.True(t, ok)
- mc.Add(25 * time.Hour)
- _, err = selector.Pick("group1", "", 1)
- assert.NoError(t, err)
- _, ok = selector.lookupTable.Load(key{group: "group1", shardID: 1})
- assert.True(t, ok)
- selector.cleanupExpiredEntries()
- _, ok = selector.lookupTable.Load(key{group: "group1", shardID: 0})
- assert.False(t, ok)
- _, ok = selector.lookupTable.Load(key{group: "group1", shardID: 1})
- assert.True(t, ok)
+ selector.OnDelete(groupSchema)
+ _, err = selector.Pick("group1", "", 0)
+ assert.Error(t, err)
+}
+
+var groupSchema = schema.Metadata{
+ TypeMeta: schema.TypeMeta{
+ Kind: schema.KindGroup,
+ },
+ Spec: &commonv1.Group{
+ Metadata: &commonv1.Metadata{
+ Name: "group1",
+ },
+ ResourceOpts: &commonv1.ResourceOpts{
+ ShardNum: 2,
+ },
+ },
+}
+
+func setupGroup(selector Selector) {
+ selector.(*roundRobinSelector).OnAddOrUpdate(groupSchema)
}