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 2023/08/02 00:55:53 UTC
[skywalking-banyandb] branch drop-discovery created (now c735498a)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a change to branch drop-discovery
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
at c735498a Drop discovery module
This branch includes the following new commits:
new c735498a Drop discovery module
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: Drop discovery module
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch drop-discovery
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit c735498acd9a7690ac76fac510c7fdb8a9194f80
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Tue Aug 1 16:12:07 2023 +0000
Drop discovery module
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
CHANGES.md | 1 +
README.md | 1 -
api/event/doc.go | 19 --
api/event/measure.go | 43 -----
api/event/stream.go | 43 -----
api/proto/banyandb/database/v1/database.proto | 7 +-
banyand/discovery/discovery.go | 78 --------
banyand/internal/cmd/liaison.go | 10 +-
banyand/internal/cmd/standalone.go | 14 +-
banyand/internal/cmd/storage.go | 12 +-
banyand/liaison/grpc/discovery.go | 252 ++++++++++++++++++++------
banyand/liaison/grpc/registry_test.go | 9 +-
banyand/liaison/grpc/server.go | 39 +---
banyand/liaison/liaison.go | 5 +-
banyand/measure/measure.go | 7 -
banyand/measure/measure_suite_test.go | 20 +-
banyand/measure/metadata.go | 7 +-
banyand/measure/metadata_test.go | 7 -
banyand/measure/service.go | 27 +--
banyand/metadata/allocator.go | 90 +++++++++
banyand/metadata/client.go | 17 +-
banyand/metadata/metadata.go | 2 +
banyand/metadata/schema/checker.go | 14 ++
banyand/metadata/schema/node.go | 53 ++++++
banyand/metadata/schema/schema.go | 60 +++++-
banyand/metadata/schema/shard.go | 76 ++++++++
banyand/queue/local.go | 2 -
banyand/queue/queue.go | 6 +-
banyand/stream/metadata.go | 7 +-
banyand/stream/metadata_test.go | 7 -
banyand/stream/service.go | 24 +--
banyand/stream/stream.go | 7 -
banyand/stream/stream_suite_test.go | 21 +--
dist/LICENSE | 6 +-
docs/api-reference.md | 1 +
go.mod | 6 +-
go.sum | 12 +-
pkg/schema/metadata.go | 133 ++------------
pkg/test/setup/setup.go | 19 +-
39 files changed, 592 insertions(+), 572 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 338d9c6b..bbdec150 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -22,6 +22,7 @@ Release Notes.
### Chores
- Bump several dependencies and tools.
+- Drop redundant "discovery" module from banyand. "metadata" module is enough to play the node and shard discovery role.
## 0.4.0
diff --git a/README.md b/README.md
index 83a32c53..73ae9c7c 100644
--- a/README.md
+++ b/README.md
@@ -29,7 +29,6 @@ The database research community usually uses [RUM conjecture](http://daslab.seas
- [ ] Sharding
- [ ] Load balance
- [ ] Distributed query optimizer
-- [ ] Node discovery
- [ ] Data queue
### Data processor
diff --git a/api/event/doc.go b/api/event/doc.go
deleted file mode 100644
index 8a9656c7..00000000
--- a/api/event/doc.go
+++ /dev/null
@@ -1,19 +0,0 @@
-// 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 event contains metadata syncing topics.
-package event
diff --git a/api/event/measure.go b/api/event/measure.go
deleted file mode 100644
index 68516d6c..00000000
--- a/api/event/measure.go
+++ /dev/null
@@ -1,43 +0,0 @@
-// 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 event
-
-import (
- "github.com/apache/skywalking-banyandb/api/common"
- "github.com/apache/skywalking-banyandb/pkg/bus"
-)
-
-var (
- // MeasureShardEventKindVersion is the version tag of measure shard event kind.
- MeasureShardEventKindVersion = common.KindVersion{
- Version: "v1",
- Kind: "measure-event-shard",
- }
-
- // MeasureTopicShardEvent is the measure shard event publishing topic.
- MeasureTopicShardEvent = bus.UniTopic(MeasureShardEventKindVersion.String())
-
- // MeasureEntityEventKindVersion is the version tag of measure entity kind.
- MeasureEntityEventKindVersion = common.KindVersion{
- Version: "v1",
- Kind: "measure-event-entity",
- }
-
- // MeasureTopicEntityEvent is the measure entity event publishing topic.
- MeasureTopicEntityEvent = bus.UniTopic(MeasureEntityEventKindVersion.String())
-)
diff --git a/api/event/stream.go b/api/event/stream.go
deleted file mode 100644
index e39f01fb..00000000
--- a/api/event/stream.go
+++ /dev/null
@@ -1,43 +0,0 @@
-// 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 event
-
-import (
- "github.com/apache/skywalking-banyandb/api/common"
- "github.com/apache/skywalking-banyandb/pkg/bus"
-)
-
-var (
- // StreamShardEventKindVersion is the version tag of stream shard entity kind.
- StreamShardEventKindVersion = common.KindVersion{
- Version: "v1",
- Kind: "stream-event-shard",
- }
-
- // StreamTopicShardEvent is the stream entity event publishing topic.
- StreamTopicShardEvent = bus.UniTopic(StreamShardEventKindVersion.String())
-
- // StreamEntityEventKindVersion is the version tag of stream entity kind.
- StreamEntityEventKindVersion = common.KindVersion{
- Version: "v1",
- Kind: "stream-event-entity",
- }
-
- // StreamTopicEntityEvent is the stream entity event publishing topic.
- StreamTopicEntityEvent = bus.UniTopic(StreamEntityEventKindVersion.String())
-)
diff --git a/api/proto/banyandb/database/v1/database.proto b/api/proto/banyandb/database/v1/database.proto
index 2e2ac78b..c530b60b 100644
--- a/api/proto/banyandb/database/v1/database.proto
+++ b/api/proto/banyandb/database/v1/database.proto
@@ -27,9 +27,10 @@ option java_package = "org.apache.skywalking.banyandb.database.v1";
message Node {
string id = 1;
- string addr = 2;
- google.protobuf.Timestamp updated_at = 3;
- google.protobuf.Timestamp created_at = 4;
+ common.v1.Metadata metadata = 2;
+ string addr = 3;
+ google.protobuf.Timestamp updated_at = 4;
+ google.protobuf.Timestamp created_at = 5;
}
message Shard {
diff --git a/banyand/discovery/discovery.go b/banyand/discovery/discovery.go
deleted file mode 100644
index 837ff4ec..00000000
--- a/banyand/discovery/discovery.go
+++ /dev/null
@@ -1,78 +0,0 @@
-// 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 discovery implements the service discovery.
-package discovery
-
-import (
- "context"
-
- "github.com/apache/skywalking-banyandb/pkg/bus"
- "github.com/apache/skywalking-banyandb/pkg/run"
-)
-
-// ServiceRepo provides service subscripting and publishing.
-//
-//go:generate mockgen -destination=./discovery_mock.go -package=discovery github.com/apache/skywalking-banyandb/banyand/discovery ServiceRepo
-type ServiceRepo interface {
- NodeID() string
- Name() string
- run.Unit
- bus.Subscriber
- bus.Publisher
- run.Service
-}
-
-type repo struct {
- local *bus.Bus
- stopCh chan struct{}
-}
-
-func (r *repo) NodeID() string {
- return "local"
-}
-
-func (r *repo) Name() string {
- return "service-discovery"
-}
-
-func (r *repo) Subscribe(topic bus.Topic, listener bus.MessageListener) error {
- return r.local.Subscribe(topic, listener)
-}
-
-func (r *repo) Publish(topic bus.Topic, message ...bus.Message) (bus.Future, error) {
- return r.local.Publish(topic, message...)
-}
-
-// NewServiceRepo return a new ServiceRepo.
-func NewServiceRepo(_ context.Context) (ServiceRepo, error) {
- return &repo{
- local: bus.NewBus(),
- stopCh: make(chan struct{}),
- }, nil
-}
-
-func (r *repo) Serve() run.StopNotify {
- return r.stopCh
-}
-
-func (r *repo) GracefulStop() {
- r.local.Close()
- if r.stopCh != nil {
- close(r.stopCh)
- }
-}
diff --git a/banyand/internal/cmd/liaison.go b/banyand/internal/cmd/liaison.go
index 6c883468..97dff1e7 100644
--- a/banyand/internal/cmd/liaison.go
+++ b/banyand/internal/cmd/liaison.go
@@ -24,7 +24,6 @@ import (
"github.com/spf13/cobra"
- "github.com/apache/skywalking-banyandb/banyand/discovery"
"github.com/apache/skywalking-banyandb/banyand/liaison"
"github.com/apache/skywalking-banyandb/banyand/liaison/http"
"github.com/apache/skywalking-banyandb/banyand/metadata"
@@ -42,12 +41,8 @@ var liaisonGroup = run.NewGroup("liaison")
func newLiaisonCmd() *cobra.Command {
l := logger.GetLogger("bootstrap")
ctx := context.Background()
- repo, err := discovery.NewServiceRepo(ctx)
- if err != nil {
- l.Fatal().Err(err).Msg("failed to initiate service repository")
- }
// nolint: staticcheck
- pipeline, err := queue.NewQueue(ctx, repo)
+ pipeline, err := queue.NewQueue(ctx)
if err != nil {
l.Fatal().Err(err).Msg("failed to initiate data pipeline")
}
@@ -55,7 +50,7 @@ func newLiaisonCmd() *cobra.Command {
if err != nil {
l.Fatal().Err(err).Msg("failed to initiate metadata service")
}
- tcp, err := liaison.NewEndpoint(ctx, pipeline, repo, metaSvc)
+ tcp, err := liaison.NewEndpoint(ctx, pipeline, metaSvc)
if err != nil {
l.Fatal().Err(err).Msg("failed to initiate Endpoint transport layer")
}
@@ -65,7 +60,6 @@ func newLiaisonCmd() *cobra.Command {
units := []run.Unit{
new(signal.Handler),
- repo,
pipeline,
tcp,
httpServer,
diff --git a/banyand/internal/cmd/standalone.go b/banyand/internal/cmd/standalone.go
index 027e0407..9d6eccaa 100644
--- a/banyand/internal/cmd/standalone.go
+++ b/banyand/internal/cmd/standalone.go
@@ -24,7 +24,6 @@ import (
"github.com/spf13/cobra"
- "github.com/apache/skywalking-banyandb/banyand/discovery"
"github.com/apache/skywalking-banyandb/banyand/liaison"
"github.com/apache/skywalking-banyandb/banyand/liaison/http"
"github.com/apache/skywalking-banyandb/banyand/measure"
@@ -45,11 +44,7 @@ var standaloneGroup = run.NewGroup("standalone")
func newStandaloneCmd() *cobra.Command {
l := logger.GetLogger("bootstrap")
ctx := context.Background()
- repo, err := discovery.NewServiceRepo(ctx)
- if err != nil {
- l.Fatal().Err(err).Msg("failed to initiate service repository")
- }
- pipeline, err := queue.NewQueue(ctx, repo)
+ pipeline, err := queue.NewQueue(ctx)
if err != nil {
l.Fatal().Err(err).Msg("failed to initiate data pipeline")
}
@@ -57,11 +52,11 @@ func newStandaloneCmd() *cobra.Command {
if err != nil {
l.Fatal().Err(err).Msg("failed to initiate metadata service")
}
- streamSvc, err := stream.NewService(ctx, metaSvc, repo, pipeline)
+ streamSvc, err := stream.NewService(ctx, metaSvc, pipeline)
if err != nil {
l.Fatal().Err(err).Msg("failed to initiate stream service")
}
- measureSvc, err := measure.NewService(ctx, metaSvc, repo, pipeline)
+ measureSvc, err := measure.NewService(ctx, metaSvc, pipeline)
if err != nil {
l.Fatal().Err(err).Msg("failed to initiate measure service")
}
@@ -69,7 +64,7 @@ func newStandaloneCmd() *cobra.Command {
if err != nil {
l.Fatal().Err(err).Msg("failed to initiate query processor")
}
- tcp, err := liaison.NewEndpoint(ctx, pipeline, repo, metaSvc)
+ tcp, err := liaison.NewEndpoint(ctx, pipeline, metaSvc)
if err != nil {
l.Fatal().Err(err).Msg("failed to initiate Endpoint transport layer")
}
@@ -79,7 +74,6 @@ func newStandaloneCmd() *cobra.Command {
units := []run.Unit{
new(signal.Handler),
- repo,
pipeline,
metaSvc,
measureSvc,
diff --git a/banyand/internal/cmd/storage.go b/banyand/internal/cmd/storage.go
index 80c3311b..7d297fc0 100644
--- a/banyand/internal/cmd/storage.go
+++ b/banyand/internal/cmd/storage.go
@@ -24,7 +24,6 @@ import (
"github.com/spf13/cobra"
- "github.com/apache/skywalking-banyandb/banyand/discovery"
"github.com/apache/skywalking-banyandb/banyand/measure"
"github.com/apache/skywalking-banyandb/banyand/metadata"
"github.com/apache/skywalking-banyandb/banyand/observability"
@@ -51,12 +50,8 @@ var flagStorageMode string
func newStorageCmd() *cobra.Command {
l := logger.GetLogger("bootstrap")
ctx := context.Background()
- repo, err := discovery.NewServiceRepo(ctx)
- if err != nil {
- l.Fatal().Err(err).Msg("failed to initiate service repository")
- }
// nolint: staticcheck
- pipeline, err := queue.NewQueue(ctx, repo)
+ pipeline, err := queue.NewQueue(ctx)
if err != nil {
l.Fatal().Err(err).Msg("failed to initiate data pipeline")
}
@@ -64,11 +59,11 @@ func newStorageCmd() *cobra.Command {
if err != nil {
l.Fatal().Err(err).Msg("failed to initiate metadata service")
}
- streamSvc, err := stream.NewService(ctx, metaSvc, repo, pipeline)
+ streamSvc, err := stream.NewService(ctx, metaSvc, pipeline)
if err != nil {
l.Fatal().Err(err).Msg("failed to initiate stream service")
}
- measureSvc, err := measure.NewService(ctx, metaSvc, repo, pipeline)
+ measureSvc, err := measure.NewService(ctx, metaSvc, pipeline)
if err != nil {
l.Fatal().Err(err).Msg("failed to initiate measure service")
}
@@ -82,7 +77,6 @@ func newStorageCmd() *cobra.Command {
units := []run.Unit{
new(signal.Handler),
- repo,
pipeline,
measureSvc,
streamSvc,
diff --git a/banyand/liaison/grpc/discovery.go b/banyand/liaison/grpc/discovery.go
index bb9f66e1..a2a38841 100644
--- a/banyand/liaison/grpc/discovery.go
+++ b/banyand/liaison/grpc/discovery.go
@@ -18,7 +18,10 @@
package grpc
import (
+ "context"
+ "fmt"
"sync"
+ "time"
"github.com/pkg/errors"
@@ -26,9 +29,10 @@ import (
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+ "github.com/apache/skywalking-banyandb/banyand/metadata"
+ "github.com/apache/skywalking-banyandb/banyand/metadata/schema"
"github.com/apache/skywalking-banyandb/banyand/queue"
"github.com/apache/skywalking-banyandb/banyand/tsdb"
- "github.com/apache/skywalking-banyandb/pkg/bus"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/partition"
)
@@ -36,20 +40,101 @@ import (
var errNotExist = errors.New("the object doesn't exist")
type discoveryService struct {
- shardRepo *shardRepo
- entityRepo *entityRepo
- pipeline queue.Queue
- log *logger.Logger
+ pipeline queue.Queue
+ metadataRepo metadata.Repo
+ shardRepo *shardRepo
+ entityRepo *entityRepo
+ log *logger.Logger
+ kind schema.Kind
}
-func newDiscoveryService(pipeline queue.Queue) *discoveryService {
+func newDiscoveryService(pipeline queue.Queue, kind schema.Kind, metadataRepo metadata.Repo) *discoveryService {
+ sr := &shardRepo{shardEventsMap: make(map[identity]uint32)}
+ er := &entityRepo{entitiesMap: make(map[identity]partition.EntityLocator)}
return &discoveryService{
- shardRepo: &shardRepo{shardEventsMap: make(map[identity]uint32)},
- entityRepo: &entityRepo{entitiesMap: make(map[identity]partition.EntityLocator)},
- pipeline: pipeline,
+ shardRepo: sr,
+ entityRepo: er,
+ pipeline: pipeline,
+ kind: kind,
+ metadataRepo: metadataRepo,
}
}
+func (ds *discoveryService) initialize() error {
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ groups, err := ds.metadataRepo.GroupRegistry().ListGroup(ctx)
+ cancel()
+ if err != nil {
+ return err
+ }
+ for _, g := range groups {
+ switch ds.kind {
+ case schema.KindMeasure:
+ case schema.KindStream:
+ default:
+ continue
+ }
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ shards, innerErr := ds.metadataRepo.ShardRegistry().ListShard(ctx, schema.ListOpt{Group: g.Metadata.Name})
+ cancel()
+ if innerErr != nil {
+ return innerErr
+ }
+ for _, s := range shards {
+ ds.shardRepo.OnAddOrUpdate(schema.Metadata{
+ TypeMeta: schema.TypeMeta{
+ Kind: schema.KindShard,
+ Name: s.Metadata.Name,
+ Group: s.Metadata.Group,
+ },
+ Spec: s,
+ })
+ }
+
+ switch ds.kind {
+ case schema.KindMeasure:
+ ctx, cancel = context.WithTimeout(context.Background(), 5*time.Second)
+ mm, innerErr := ds.metadataRepo.MeasureRegistry().ListMeasure(ctx, schema.ListOpt{Group: g.Metadata.Name})
+ cancel()
+ if innerErr != nil {
+ return innerErr
+ }
+ for _, m := range mm {
+ ds.entityRepo.OnAddOrUpdate(schema.Metadata{
+ TypeMeta: schema.TypeMeta{
+ Kind: schema.KindMeasure,
+ Name: m.Metadata.Name,
+ Group: m.Metadata.Group,
+ },
+ Spec: m,
+ })
+ }
+ case schema.KindStream:
+ ctx, cancel = context.WithTimeout(context.Background(), 5*time.Second)
+ ss, innerErr := ds.metadataRepo.StreamRegistry().ListStream(ctx, schema.ListOpt{Group: g.Metadata.Name})
+ cancel()
+ if innerErr != nil {
+ return innerErr
+ }
+ for _, s := range ss {
+ ds.entityRepo.OnAddOrUpdate(schema.Metadata{
+ TypeMeta: schema.TypeMeta{
+ Kind: schema.KindStream,
+ Name: s.Metadata.Name,
+ Group: s.Metadata.Group,
+ },
+ Spec: s,
+ })
+ }
+ default:
+ return fmt.Errorf("unsupported kind: %d", ds.kind)
+ }
+ }
+ ds.metadataRepo.RegisterHandler(schema.KindShard, ds.shardRepo)
+ ds.metadataRepo.RegisterHandler(ds.kind, ds.entityRepo)
+ return nil
+}
+
func (ds *discoveryService) SetLogger(log *logger.Logger) {
ds.log = log
ds.shardRepo.log = log
@@ -75,38 +160,46 @@ type identity struct {
group string
}
+func (i identity) String() string {
+ return fmt.Sprintf("%s/%s", i.group, i.name)
+}
+
+var _ schema.EventHandler = (*shardRepo)(nil)
+
type shardRepo struct {
log *logger.Logger
shardEventsMap map[identity]uint32
sync.RWMutex
}
-func (s *shardRepo) Rev(message bus.Message) (resp bus.Message) {
- e, ok := message.Data().(*databasev1.ShardEvent)
- if !ok {
- s.log.Warn().Msg("invalid e data type")
+// OnAddOrUpdate implements schema.EventHandler.
+func (s *shardRepo) OnAddOrUpdate(schemaMetadata schema.Metadata) {
+ if schemaMetadata.Kind != schema.KindShard {
return
}
- s.setShardNum(e)
-
+ shard := schemaMetadata.Spec.(*databasev1.Shard)
+ idx := getID(shard.GetMetadata())
if le := s.log.Debug(); le.Enabled() {
- le.
- Str("action", databasev1.Action_name[int32(e.Action)]).
- Uint64("shardID", e.Shard.Id).
- Msg("received a shard e")
+ le.Stringer("id", idx).Uint32("total", shard.Total).Msg("shard added or updated")
}
- return
+ s.RWMutex.Lock()
+ defer s.RWMutex.Unlock()
+ s.shardEventsMap[idx] = shard.Total
}
-func (s *shardRepo) setShardNum(eventVal *databasev1.ShardEvent) {
+// OnDelete implements schema.EventHandler.
+func (s *shardRepo) OnDelete(schemaMetadata schema.Metadata) {
+ if schemaMetadata.Kind != schema.KindShard {
+ return
+ }
+ shard := schemaMetadata.Spec.(*databasev1.Shard)
+ idx := getID(shard.GetMetadata())
+ if le := s.log.Debug(); le.Enabled() {
+ le.Stringer("id", idx).Msg("shard deleted")
+ }
s.RWMutex.Lock()
defer s.RWMutex.Unlock()
- idx := getID(eventVal.GetShard().GetMetadata())
- if eventVal.Action == databasev1.Action_ACTION_PUT {
- s.shardEventsMap[idx] = eventVal.Shard.Total
- } else if eventVal.Action == databasev1.Action_ACTION_DELETE {
- delete(s.shardEventsMap, idx)
- }
+ delete(s.shardEventsMap, idx)
}
func (s *shardRepo) shardNum(idx identity) (uint32, bool) {
@@ -126,49 +219,96 @@ func getID(metadata *commonv1.Metadata) identity {
}
}
+var _ schema.EventHandler = (*entityRepo)(nil)
+
type entityRepo struct {
log *logger.Logger
entitiesMap map[identity]partition.EntityLocator
sync.RWMutex
}
-func (s *entityRepo) Rev(message bus.Message) (resp bus.Message) {
- e, ok := message.Data().(*databasev1.EntityEvent)
- if !ok {
- s.log.Warn().Msg("invalid e data type")
+// OnAddOrUpdate implements schema.EventHandler.
+func (e *entityRepo) OnAddOrUpdate(schemaMetadata schema.Metadata) {
+ var el partition.EntityLocator
+ var id identity
+ switch schemaMetadata.Kind {
+ case schema.KindMeasure:
+ measure := schemaMetadata.Spec.(*databasev1.Measure)
+ el = partition.NewEntityLocator(measure.TagFamilies, measure.Entity)
+ id = getID(measure.GetMetadata())
+ case schema.KindStream:
+ stream := schemaMetadata.Spec.(*databasev1.Stream)
+ el = partition.NewEntityLocator(stream.TagFamilies, stream.Entity)
+ id = getID(stream.GetMetadata())
+ default:
return
}
- id := getID(e.GetSubject())
- if le := s.log.Debug(); le.Enabled() {
+ if le := e.log.Debug(); le.Enabled() {
+ var kind string
+ switch schemaMetadata.Kind {
+ case schema.KindMeasure:
+ kind = "measure"
+ case schema.KindStream:
+ kind = "stream"
+ default:
+ kind = "unknown"
+ }
le.
- Str("action", databasev1.Action_name[int32(e.Action)]).
- Interface("subject", id).
- Msg("received an entity event")
+ Str("action", "add_or_update").
+ Stringer("subject", id).
+ Str("kind", kind).
+ Msg("entity added or updated")
}
- s.RWMutex.Lock()
- defer s.RWMutex.Unlock()
- switch e.Action {
- case databasev1.Action_ACTION_PUT:
- en := make(partition.EntityLocator, 0, len(e.GetEntityLocator()))
- for _, l := range e.GetEntityLocator() {
- en = append(en, partition.TagLocator{
- FamilyOffset: int(l.FamilyOffset),
- TagOffset: int(l.TagOffset),
- })
+ en := make(partition.EntityLocator, 0, len(el))
+ for _, l := range el {
+ en = append(en, partition.TagLocator{
+ FamilyOffset: l.FamilyOffset,
+ TagOffset: l.TagOffset,
+ })
+ }
+ e.RWMutex.Lock()
+ defer e.RWMutex.Unlock()
+ e.entitiesMap[id] = en
+}
+
+// OnDelete implements schema.EventHandler.
+func (e *entityRepo) OnDelete(schemaMetadata schema.Metadata) {
+ var id identity
+ switch schemaMetadata.Kind {
+ case schema.KindMeasure:
+ measure := schemaMetadata.Spec.(*databasev1.Measure)
+ id = getID(measure.GetMetadata())
+ case schema.KindStream:
+ stream := schemaMetadata.Spec.(*databasev1.Stream)
+ id = getID(stream.GetMetadata())
+ default:
+ return
+ }
+ if le := e.log.Debug(); le.Enabled() {
+ var kind string
+ switch schemaMetadata.Kind {
+ case schema.KindMeasure:
+ kind = "measure"
+ case schema.KindStream:
+ kind = "stream"
+ default:
+ kind = "unknown"
}
- s.entitiesMap[id] = en
- case databasev1.Action_ACTION_DELETE:
- delete(s.entitiesMap, id)
- case databasev1.Action_ACTION_UNSPECIFIED:
- s.log.Warn().RawJSON("event", logger.Proto(e)).Msg("ignored unspecified event")
+ le.
+ Str("action", "delete").
+ Stringer("subject", id).
+ Str("kind", kind).
+ Msg("entity deleted")
}
- return
+ e.RWMutex.Lock()
+ defer e.RWMutex.Unlock()
+ delete(e.entitiesMap, id)
}
-func (s *entityRepo) getLocator(id identity) (partition.EntityLocator, bool) {
- s.RWMutex.RLock()
- defer s.RWMutex.RUnlock()
- el, ok := s.entitiesMap[id]
+func (e *entityRepo) getLocator(id identity) (partition.EntityLocator, bool) {
+ e.RWMutex.RLock()
+ defer e.RWMutex.RUnlock()
+ el, ok := e.entitiesMap[id]
if !ok {
return nil, false
}
diff --git a/banyand/liaison/grpc/registry_test.go b/banyand/liaison/grpc/registry_test.go
index cc48ce86..4b12a1ac 100644
--- a/banyand/liaison/grpc/registry_test.go
+++ b/banyand/liaison/grpc/registry_test.go
@@ -31,7 +31,6 @@ import (
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/banyand/discovery"
"github.com/apache/skywalking-banyandb/banyand/liaison/grpc"
"github.com/apache/skywalking-banyandb/banyand/metadata"
"github.com/apache/skywalking-banyandb/banyand/queue"
@@ -173,17 +172,14 @@ var _ = Describe("Registry", func() {
})
func setupForRegistry() func() {
- // Init `Discovery` module
- repo, err := discovery.NewServiceRepo(context.Background())
- Expect(err).NotTo(HaveOccurred())
// Init `Queue` module
- pipeline, err := queue.NewQueue(context.TODO(), repo)
+ pipeline, err := queue.NewQueue(context.TODO())
Expect(err).NotTo(HaveOccurred())
// Init `Metadata` module
metaSvc, err := metadata.NewService(context.TODO())
Expect(err).NotTo(HaveOccurred())
- tcp := grpc.NewServer(context.TODO(), pipeline, repo, metaSvc)
+ tcp := grpc.NewServer(context.TODO(), pipeline, metaSvc)
preloadStreamSvc := &preloadStreamService{metaSvc: metaSvc}
var flags []string
metaPath, metaDeferFunc, err := test.NewSpace()
@@ -194,7 +190,6 @@ func setupForRegistry() func() {
"--etcd-listen-peer-url="+listenPeerURL)
deferFunc := test.SetupModules(
flags,
- repo,
pipeline,
metaSvc,
preloadStreamSvc,
diff --git a/banyand/liaison/grpc/server.go b/banyand/liaison/grpc/server.go
index 9ead16c3..2bb90664 100644
--- a/banyand/liaison/grpc/server.go
+++ b/banyand/liaison/grpc/server.go
@@ -34,16 +34,14 @@ import (
"google.golang.org/grpc/health/grpc_health_v1"
"google.golang.org/grpc/status"
- "github.com/apache/skywalking-banyandb/api/event"
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
measurev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1"
propertyv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/property/v1"
streamv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/stream/v1"
- "github.com/apache/skywalking-banyandb/banyand/discovery"
"github.com/apache/skywalking-banyandb/banyand/metadata"
+ "github.com/apache/skywalking-banyandb/banyand/metadata/schema"
"github.com/apache/skywalking-banyandb/banyand/observability"
"github.com/apache/skywalking-banyandb/banyand/queue"
- "github.com/apache/skywalking-banyandb/pkg/bus"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/run"
)
@@ -61,7 +59,6 @@ var (
type server struct {
pipeline queue.Queue
creds credentials.TransportCredentials
- repo discovery.ServiceRepo
*indexRuleRegistryServer
measureSVC *measureService
log *logger.Logger
@@ -85,16 +82,15 @@ type server struct {
}
// NewServer returns a new gRPC server.
-func NewServer(_ context.Context, pipeline queue.Queue, repo discovery.ServiceRepo, schemaRegistry metadata.Repo) run.Unit {
+func NewServer(_ context.Context, pipeline queue.Queue, schemaRegistry metadata.Repo) run.Unit {
streamSVC := &streamService{
- discoveryService: newDiscoveryService(pipeline),
+ discoveryService: newDiscoveryService(pipeline, schema.KindStream, schemaRegistry),
}
measureSVC := &measureService{
- discoveryService: newDiscoveryService(pipeline),
+ discoveryService: newDiscoveryService(pipeline, schema.KindMeasure, schemaRegistry),
}
s := &server{
pipeline: pipeline,
- repo: repo,
streamSVC: streamSVC,
measureSVC: measureSVC,
streamRegistryServer: &streamRegistryServer{
@@ -127,30 +123,13 @@ func (s *server) PreRun() error {
s.log = logger.GetLogger("liaison-grpc")
s.streamSVC.setLogger(s.log)
s.measureSVC.setLogger(s.log)
- components := []struct {
- discoverySVC *discoveryService
- shardEvent bus.Topic
- entityEvent bus.Topic
- }{
- {
- shardEvent: event.StreamTopicShardEvent,
- entityEvent: event.StreamTopicEntityEvent,
- discoverySVC: s.streamSVC.discoveryService,
- },
- {
- shardEvent: event.MeasureTopicShardEvent,
- entityEvent: event.MeasureTopicEntityEvent,
- discoverySVC: s.measureSVC.discoveryService,
- },
+ components := []*discoveryService{
+ s.streamSVC.discoveryService,
+ s.measureSVC.discoveryService,
}
for _, c := range components {
- c.discoverySVC.SetLogger(s.log)
- err := s.repo.Subscribe(c.shardEvent, c.discoverySVC.shardRepo)
- if err != nil {
- return err
- }
- err = s.repo.Subscribe(c.entityEvent, c.discoverySVC.entityRepo)
- if err != nil {
+ c.SetLogger(s.log)
+ if err := c.initialize(); err != nil {
return err
}
}
diff --git a/banyand/liaison/liaison.go b/banyand/liaison/liaison.go
index 539d2f45..a1b657cf 100644
--- a/banyand/liaison/liaison.go
+++ b/banyand/liaison/liaison.go
@@ -21,7 +21,6 @@ package liaison
import (
"context"
- "github.com/apache/skywalking-banyandb/banyand/discovery"
"github.com/apache/skywalking-banyandb/banyand/liaison/grpc"
"github.com/apache/skywalking-banyandb/banyand/metadata"
"github.com/apache/skywalking-banyandb/banyand/queue"
@@ -29,6 +28,6 @@ import (
)
// NewEndpoint return a new endpoint which is the entry point for the database server.
-func NewEndpoint(ctx context.Context, pipeline queue.Queue, repo discovery.ServiceRepo, schemaRegistry metadata.Repo) (run.Unit, error) {
- return grpc.NewServer(ctx, pipeline, repo, schemaRegistry), nil
+func NewEndpoint(ctx context.Context, pipeline queue.Queue, schemaRegistry metadata.Repo) (run.Unit, error) {
+ return grpc.NewServer(ctx, pipeline, schemaRegistry), nil
}
diff --git a/banyand/measure/measure.go b/banyand/measure/measure.go
index a5ee5857..9fbfe653 100644
--- a/banyand/measure/measure.go
+++ b/banyand/measure/measure.go
@@ -31,7 +31,6 @@ import (
"github.com/apache/skywalking-banyandb/banyand/tsdb"
"github.com/apache/skywalking-banyandb/banyand/tsdb/index"
"github.com/apache/skywalking-banyandb/pkg/logger"
- "github.com/apache/skywalking-banyandb/pkg/partition"
"github.com/apache/skywalking-banyandb/pkg/query/logical"
resourceSchema "github.com/apache/skywalking-banyandb/pkg/schema"
"github.com/apache/skywalking-banyandb/pkg/timestamp"
@@ -51,7 +50,6 @@ type measure struct {
processorManager *topNProcessorManager
name string
group string
- entityLocator partition.EntityLocator
indexRules []*databasev1.IndexRule
topNAggregations []*databasev1.TopNAggregation
maxObservedModRevision int64
@@ -98,10 +96,6 @@ func (s *measure) MaxObservedModRevision() int64 {
return s.maxObservedModRevision
}
-func (s *measure) EntityLocator() partition.EntityLocator {
- return s.entityLocator
-}
-
func (s *measure) Close() error {
if s.processorManager == nil {
return nil
@@ -111,7 +105,6 @@ func (s *measure) Close() error {
func (s *measure) parseSpec() (err error) {
s.name, s.group = s.schema.GetMetadata().GetName(), s.schema.GetMetadata().GetGroup()
- s.entityLocator = partition.NewEntityLocator(s.schema.GetTagFamilies(), s.schema.GetEntity())
s.maxObservedModRevision = int64(math.Max(float64(resourceSchema.ParseMaxModRevision(s.indexRules)), float64(resourceSchema.ParseMaxModRevision(s.topNAggregations))))
if s.schema.Interval != "" {
s.interval, err = timestamp.ParseDuration(s.schema.Interval)
diff --git a/banyand/measure/measure_suite_test.go b/banyand/measure/measure_suite_test.go
index 704720d2..a1c829bb 100644
--- a/banyand/measure/measure_suite_test.go
+++ b/banyand/measure/measure_suite_test.go
@@ -25,9 +25,6 @@ import (
"github.com/onsi/gomega"
"go.uber.org/mock/gomock"
- "github.com/apache/skywalking-banyandb/api/event"
- databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
- "github.com/apache/skywalking-banyandb/banyand/discovery"
"github.com/apache/skywalking-banyandb/banyand/measure"
"github.com/apache/skywalking-banyandb/banyand/metadata"
"github.com/apache/skywalking-banyandb/banyand/queue"
@@ -64,26 +61,15 @@ func (p *preloadMeasureService) PreRun() error {
type services struct {
measure measure.Service
metadataService metadata.Service
- repo *discovery.MockServiceRepo
pipeline queue.Queue
}
func setUp() (*services, func()) {
ctrl := gomock.NewController(ginkgo.GinkgoT())
gomega.Expect(ctrl).ShouldNot(gomega.BeNil())
- // Init Discovery
- repo := discovery.NewMockServiceRepo(ctrl)
- repo.EXPECT().NodeID().AnyTimes()
- repo.EXPECT().Name().AnyTimes()
- stopCh := make(chan struct{})
- repo.EXPECT().Serve().Return(stopCh).Times(1)
- repo.EXPECT().GracefulStop().Do(func() { close(stopCh) }).Times(1)
- // Both PreRun and Serve phases send events
- repo.EXPECT().Publish(event.MeasureTopicEntityEvent, test.NewEntityEventMatcher(databasev1.Action_ACTION_PUT)).AnyTimes()
- repo.EXPECT().Publish(event.MeasureTopicShardEvent, test.NewShardEventMatcher(databasev1.Action_ACTION_PUT)).AnyTimes()
// Init Pipeline
- pipeline, err := queue.NewQueue(context.TODO(), repo)
+ pipeline, err := queue.NewQueue(context.TODO())
gomega.Expect(err).NotTo(gomega.HaveOccurred())
// Init Metadata Service
@@ -91,7 +77,7 @@ func setUp() (*services, func()) {
gomega.Expect(err).NotTo(gomega.HaveOccurred())
// Init Measure Service
- measureService, err := measure.NewService(context.TODO(), metadataService, repo, pipeline)
+ measureService, err := measure.NewService(context.TODO(), metadataService, pipeline)
gomega.Expect(err).NotTo(gomega.HaveOccurred())
preloadMeasureSvc := &preloadMeasureService{metaSvc: metadataService}
var flags []string
@@ -106,7 +92,6 @@ func setUp() (*services, func()) {
flags = append(flags, "--etcd-listen-client-url="+listenClientURL, "--etcd-listen-peer-url="+listenPeerURL)
moduleDeferFunc := test.SetupModules(
flags,
- repo,
pipeline,
metadataService,
preloadMeasureSvc,
@@ -115,7 +100,6 @@ func setUp() (*services, func()) {
return &services{
measure: measureService,
metadataService: metadataService,
- repo: repo,
pipeline: pipeline,
}, func() {
moduleDeferFunc()
diff --git a/banyand/measure/metadata.go b/banyand/measure/metadata.go
index 2a577b26..a86bfcc4 100644
--- a/banyand/measure/metadata.go
+++ b/banyand/measure/metadata.go
@@ -28,10 +28,8 @@ import (
"google.golang.org/protobuf/testing/protocmp"
"github.com/apache/skywalking-banyandb/api/common"
- "github.com/apache/skywalking-banyandb/api/event"
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/banyand/discovery"
"github.com/apache/skywalking-banyandb/banyand/metadata"
"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
"github.com/apache/skywalking-banyandb/banyand/queue"
@@ -49,7 +47,7 @@ type schemaRepo struct {
metadata metadata.Repo
}
-func newSchemaRepo(path string, metadata metadata.Repo, repo discovery.ServiceRepo,
+func newSchemaRepo(path string, metadata metadata.Repo,
dbOpts tsdb.DatabaseOpts, l *logger.Logger, pipeline queue.Queue, encoderBufferSize, bufferSize int64,
) schemaRepo {
return schemaRepo{
@@ -57,11 +55,8 @@ func newSchemaRepo(path string, metadata metadata.Repo, repo discovery.ServiceRe
metadata: metadata,
Repository: resourceSchema.NewRepository(
metadata,
- repo,
l,
newSupplier(path, metadata, dbOpts, l, pipeline, encoderBufferSize, bufferSize),
- event.MeasureTopicShardEvent,
- event.MeasureTopicEntityEvent,
),
}
}
diff --git a/banyand/measure/metadata_test.go b/banyand/measure/metadata_test.go
index fdbada2f..9504c1f9 100644
--- a/banyand/measure/metadata_test.go
+++ b/banyand/measure/metadata_test.go
@@ -24,11 +24,9 @@ import (
. "github.com/onsi/gomega"
"github.com/onsi/gomega/gleak"
- "github.com/apache/skywalking-banyandb/api/event"
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/banyand/measure"
- "github.com/apache/skywalking-banyandb/pkg/test"
"github.com/apache/skywalking-banyandb/pkg/test/flags"
)
@@ -54,7 +52,6 @@ var _ = Describe("Metadata", func() {
}).WithTimeout(flags.EventuallyTimeout).Should(BeTrue())
})
It("should close the group", func() {
- svcs.repo.EXPECT().Publish(event.MeasureTopicShardEvent, test.NewShardEventMatcher(databasev1.Action_ACTION_DELETE)).Times(2)
deleted, err := svcs.metadataService.GroupRegistry().DeleteGroup(context.TODO(), "sw_metric")
Expect(err).ShouldNot(HaveOccurred())
Expect(deleted).Should(BeTrue())
@@ -65,8 +62,6 @@ var _ = Describe("Metadata", func() {
})
It("should add shards", func() {
- svcs.repo.EXPECT().Publish(event.MeasureTopicShardEvent, test.NewShardEventMatcher(databasev1.Action_ACTION_DELETE)).AnyTimes()
- svcs.repo.EXPECT().Publish(event.MeasureTopicShardEvent, test.NewShardEventMatcher(databasev1.Action_ACTION_PUT)).AnyTimes()
groupSchema, err := svcs.metadataService.GroupRegistry().GetGroup(context.TODO(), "sw_metric")
Expect(err).ShouldNot(HaveOccurred())
Expect(groupSchema).ShouldNot(BeNil())
@@ -95,7 +90,6 @@ var _ = Describe("Metadata", func() {
}).WithTimeout(flags.EventuallyTimeout).Should(BeTrue())
})
It("should close the measure", func() {
- svcs.repo.EXPECT().Publish(event.MeasureTopicEntityEvent, test.NewEntityEventMatcher(databasev1.Action_ACTION_DELETE)).Times(1)
deleted, err := svcs.metadataService.MeasureRegistry().DeleteMeasure(context.TODO(), &commonv1.Metadata{
Name: "service_cpm_minute",
Group: "sw_metric",
@@ -126,7 +120,6 @@ var _ = Describe("Metadata", func() {
})
It("should update a new measure", func() {
- svcs.repo.EXPECT().Publish(event.MeasureTopicEntityEvent, test.NewEntityEventMatcher(databasev1.Action_ACTION_PUT)).AnyTimes()
// Remove the first tag from the entity
measureSchema.Entity.TagNames = measureSchema.Entity.TagNames[1:]
entitySize := len(measureSchema.Entity.TagNames)
diff --git a/banyand/measure/service.go b/banyand/measure/service.go
index 04c1b9ce..6e46f841 100644
--- a/banyand/measure/service.go
+++ b/banyand/measure/service.go
@@ -28,7 +28,6 @@ import (
"github.com/apache/skywalking-banyandb/api/data"
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/banyand/discovery"
"github.com/apache/skywalking-banyandb/banyand/metadata"
"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
"github.com/apache/skywalking-banyandb/banyand/observability"
@@ -61,9 +60,7 @@ type service struct {
writeListener bus.MessageListener
metadata metadata.Repo
pipeline queue.Queue
- repo discovery.ServiceRepo
l *logger.Logger
- stopCh chan struct{}
root string
dbOpts tsdb.DatabaseOpts
BlockEncoderBufferSize run.Bytes
@@ -116,7 +113,7 @@ func (s *service) PreRun() error {
}
path := path.Join(s.root, s.Name())
observability.UpdatePath(path)
- s.schemaRepo = newSchemaRepo(path, s.metadata, s.repo, s.dbOpts,
+ s.schemaRepo = newSchemaRepo(path, s.metadata, s.dbOpts,
s.l, s.pipeline, int64(s.BlockEncoderBufferSize), int64(s.BlockBufferSize))
for _, g := range groups {
if g.Catalog != commonv1.Catalog_CATALOG_MEASURE {
@@ -144,6 +141,11 @@ func (s *service) PreRun() error {
}
}
}
+ // run a serial watcher
+ go s.schemaRepo.Watcher()
+ s.metadata.
+ RegisterHandler(schema.KindGroup|schema.KindMeasure|schema.KindIndexRuleBinding|schema.KindIndexRule|schema.KindTopNAggregation,
+ &s.schemaRepo)
s.writeListener = setUpWriteCallback(s.l, &s.schemaRepo)
err = s.pipeline.Subscribe(data.TopicMeasureWrite, s.writeListener)
@@ -177,31 +179,18 @@ func (s *service) sanityCheck(group resourceSchema.Group, measureSchema *databas
}
func (s *service) Serve() run.StopNotify {
- _ = s.schemaRepo.NotifyAll()
- // run a serial watcher
- go s.schemaRepo.Watcher()
-
- s.metadata.MeasureRegistry().
- RegisterHandler(schema.KindGroup|schema.KindMeasure|schema.KindIndexRuleBinding|schema.KindIndexRule|schema.KindTopNAggregation,
- &s.schemaRepo)
-
- return s.stopCh
+ return s.schemaRepo.StopCh()
}
func (s *service) GracefulStop() {
s.schemaRepo.Close()
- if s.stopCh != nil {
- close(s.stopCh)
- }
}
// NewService returns a new service.
-func NewService(_ context.Context, metadata metadata.Repo, repo discovery.ServiceRepo, pipeline queue.Queue) (Service, error) {
+func NewService(_ context.Context, metadata metadata.Repo, pipeline queue.Queue) (Service, error) {
return &service{
metadata: metadata,
- repo: repo,
pipeline: pipeline,
- stopCh: make(chan struct{}),
dbOpts: tsdb.DatabaseOpts{
IndexGranularity: tsdb.IndexGranularitySeries,
},
diff --git a/banyand/metadata/allocator.go b/banyand/metadata/allocator.go
new file mode 100644
index 00000000..acf52dbf
--- /dev/null
+++ b/banyand/metadata/allocator.go
@@ -0,0 +1,90 @@
+// 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 metadata
+
+import (
+ "context"
+ "time"
+
+ "google.golang.org/protobuf/types/known/timestamppb"
+
+ 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/banyand/metadata/schema"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+)
+
+var _ schema.EventHandler = (*allocator)(nil)
+
+type allocator struct {
+ schemaRegistry schema.Registry
+ l *logger.Logger
+}
+
+func newAllocator(schemaRegistry schema.Registry, logger *logger.Logger) *allocator {
+ return &allocator{
+ schemaRegistry: schemaRegistry,
+ l: logger,
+ }
+}
+
+// OnAddOrUpdate implements EventHandler.
+func (a *allocator) OnAddOrUpdate(metadata schema.Metadata) {
+ switch metadata.Kind {
+ case schema.KindGroup:
+ groupSchema := metadata.Spec.(*commonv1.Group)
+ if groupSchema.Catalog == commonv1.Catalog_CATALOG_UNSPECIFIED {
+ return
+ }
+ now := time.Now()
+ nowPb := timestamppb.New(now)
+ shardNum := groupSchema.GetResourceOpts().GetShardNum()
+ syncShard := func(id uint64) error {
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ return a.schemaRegistry.CreateOrUpdateShard(ctx, &databasev1.Shard{
+ Id: id,
+ Total: shardNum,
+ Metadata: &commonv1.Metadata{
+ Name: groupSchema.GetMetadata().GetName(),
+ },
+ Node: &databasev1.Node{
+ Id: "TODO",
+ CreatedAt: nowPb,
+ UpdatedAt: nowPb,
+ Addr: "TODO",
+ },
+ })
+ }
+ for i := 0; i < int(shardNum); i++ {
+ if err := syncShard(uint64(i)); err != nil {
+ // TODO: handle error. retry? or do a full sync?
+ a.l.Error().Err(err).Msg("failed to sync shard")
+ }
+ }
+ case schema.KindNode:
+ // TODO: handle node
+ default:
+ return
+ }
+}
+
+// OnDelete implements EventHandler.
+func (*allocator) OnDelete(schema.Metadata) {
+ // TODO: handle delete
+}
diff --git a/banyand/metadata/client.go b/banyand/metadata/client.go
index cfcfec84..0fc967c7 100644
--- a/banyand/metadata/client.go
+++ b/banyand/metadata/client.go
@@ -27,6 +27,7 @@ import (
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/banyand/metadata/schema"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/run"
)
@@ -39,10 +40,15 @@ func NewClient(_ context.Context) (Service, error) {
type clientService struct {
schemaRegistry schema.Registry
+ alc *allocator
closer *run.Closer
endpoints []string
}
+func (s *clientService) SchemaRegistry() schema.Registry {
+ return s.schemaRegistry
+}
+
func (s *clientService) FlagSet() *run.FlagSet {
fs := run.NewFlagSet("metadata")
fs.StringArrayVar(&s.endpoints, flagEtcdEndpointsName, []string{"http://localhost:2379"}, "A comma-delimited list of etcd endpoints")
@@ -62,6 +68,9 @@ func (s *clientService) PreRun() error {
if err != nil {
return err
}
+
+ s.alc = newAllocator(s.schemaRegistry, logger.GetLogger(s.Name()).Named("allocator"))
+ s.schemaRegistry.RegisterHandler(schema.KindGroup|schema.KindNode, s.alc)
return nil
}
@@ -75,8 +84,8 @@ func (s *clientService) GracefulStop() {
_ = s.schemaRegistry.Close()
}
-func (s *clientService) SchemaRegistry() schema.Registry {
- return s.schemaRegistry
+func (s *clientService) RegisterHandler(kind schema.Kind, handler schema.EventHandler) {
+ s.schemaRegistry.RegisterHandler(kind, handler)
}
func (s *clientService) StreamRegistry() schema.Stream {
@@ -107,6 +116,10 @@ func (s *clientService) PropertyRegistry() schema.Property {
return s.schemaRegistry
}
+func (s *clientService) ShardRegistry() schema.Shard {
+ return s.schemaRegistry
+}
+
func (s *clientService) Name() string {
return "metadata"
}
diff --git a/banyand/metadata/metadata.go b/banyand/metadata/metadata.go
index f9ab33d7..9bad1e72 100644
--- a/banyand/metadata/metadata.go
+++ b/banyand/metadata/metadata.go
@@ -46,6 +46,8 @@ type Repo interface {
GroupRegistry() schema.Group
TopNAggregationRegistry() schema.TopNAggregation
PropertyRegistry() schema.Property
+ ShardRegistry() schema.Shard
+ RegisterHandler(schema.Kind, schema.EventHandler)
}
// Service is the metadata repository.
diff --git a/banyand/metadata/schema/checker.go b/banyand/metadata/schema/checker.go
index 4a1a8751..3d669cbc 100644
--- a/banyand/metadata/schema/checker.go
+++ b/banyand/metadata/schema/checker.go
@@ -82,6 +82,20 @@ var checkerMap = map[Kind]equalityChecker{
protocmp.IgnoreFields(&commonv1.Metadata{}, "id", "create_revision", "mod_revision"),
protocmp.Transform())
},
+ KindNode: func(a, b proto.Message) bool {
+ return cmp.Equal(a, b,
+ protocmp.IgnoreUnknown(),
+ protocmp.IgnoreFields(&databasev1.Node{}, "updated_at"),
+ protocmp.IgnoreFields(&commonv1.Metadata{}, "id", "create_revision", "mod_revision"),
+ protocmp.Transform())
+ },
+ KindShard: func(a, b proto.Message) bool {
+ return cmp.Equal(a, b,
+ protocmp.IgnoreUnknown(),
+ protocmp.IgnoreFields(&databasev1.Shard{}, "updated_at"),
+ protocmp.IgnoreFields(&commonv1.Metadata{}, "id", "create_revision", "mod_revision"),
+ protocmp.Transform())
+ },
KindMask: func(a, b proto.Message) bool {
return false
},
diff --git a/banyand/metadata/schema/node.go b/banyand/metadata/schema/node.go
new file mode 100644
index 00000000..0c6dcdb1
--- /dev/null
+++ b/banyand/metadata/schema/node.go
@@ -0,0 +1,53 @@
+// 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 schema
+
+import (
+ "context"
+
+ "google.golang.org/protobuf/proto"
+
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+)
+
+var nodeKeyPrefix = "/nodes/"
+
+func (e *etcdSchemaRegistry) ListNode(ctx context.Context, role Role) ([]*databasev1.Node, error) {
+ if role == "" {
+ return nil, BadRequest("group", "group should not be empty")
+ }
+ messages, err := e.listWithPrefix(ctx, listPrefixesForEntity(string(role), nodeKeyPrefix), func() proto.Message {
+ return &databasev1.Node{}
+ })
+ if err != nil {
+ return nil, err
+ }
+ entities := make([]*databasev1.Node, 0, len(messages))
+ for _, message := range messages {
+ entities = append(entities, message.(*databasev1.Node))
+ }
+ return entities, nil
+}
+
+func formatNodePrefix(role Role) string {
+ return nodeKeyPrefix + string(role)
+}
+
+func formatNodeKey(role Role, id string) string {
+ return formatNodePrefix(role) + "/" + id
+}
diff --git a/banyand/metadata/schema/schema.go b/banyand/metadata/schema/schema.go
index f9614938..fd97a1d6 100644
--- a/banyand/metadata/schema/schema.go
+++ b/banyand/metadata/schema/schema.go
@@ -50,7 +50,11 @@ const (
KindIndexRule
KindTopNAggregation
KindProperty
- KindMask = KindGroup | KindStream | KindMeasure | KindIndexRuleBinding | KindIndexRule | KindTopNAggregation
+ KindNode
+ KindShard
+ KindMask = KindGroup | KindStream | KindMeasure |
+ KindIndexRuleBinding | KindIndexRule |
+ KindTopNAggregation | KindProperty | KindNode | KindShard
)
// ListOpt contains options to list resources.
@@ -68,6 +72,9 @@ type Registry interface {
Group
TopNAggregation
Property
+ Node
+ Shard
+ RegisterHandler(Kind, EventHandler)
}
// TypeMeta defines the identity and type of an Event.
@@ -103,6 +110,10 @@ func (tm TypeMeta) Unmarshal(data []byte) (m proto.Message, err error) {
m = &propertyv1.Property{}
case KindTopNAggregation:
m = &databasev1.TopNAggregation{}
+ case KindNode:
+ m = &databasev1.Node{}
+ case KindShard:
+ m = &databasev1.Shard{}
default:
return nil, errUnsupportedEntityType
}
@@ -144,6 +155,17 @@ func (m Metadata) key() (string, error) {
Group: m.Group,
Name: m.Name,
}), nil
+ case KindNode:
+ r, err := strToRole(m.Group)
+ if err != nil {
+ return "", err
+ }
+ return formatNodeKey(r, m.Name), nil
+ case KindShard:
+ return formatShardKey(&commonv1.Metadata{
+ Group: m.Group,
+ Name: m.Name,
+ }), nil
default:
return "", errUnsupportedEntityType
}
@@ -168,7 +190,6 @@ type Stream interface {
CreateStream(ctx context.Context, stream *databasev1.Stream) error
UpdateStream(ctx context.Context, stream *databasev1.Stream) error
DeleteStream(ctx context.Context, metadata *commonv1.Metadata) (bool, error)
- RegisterHandler(Kind, EventHandler)
}
// IndexRule allows CRUD index rule schemas in a group.
@@ -196,7 +217,6 @@ type Measure interface {
CreateMeasure(ctx context.Context, measure *databasev1.Measure) error
UpdateMeasure(ctx context.Context, measure *databasev1.Measure) error
DeleteMeasure(ctx context.Context, metadata *commonv1.Metadata) (bool, error)
- RegisterHandler(Kind, EventHandler)
TopNAggregations(ctx context.Context, metadata *commonv1.Metadata) ([]*databasev1.TopNAggregation, error)
}
@@ -226,3 +246,37 @@ type Property interface {
ApplyProperty(ctx context.Context, property *propertyv1.Property, strategy propertyv1.ApplyRequest_Strategy) (bool, uint32, error)
DeleteProperty(ctx context.Context, metadata *propertyv1.Metadata, tags []string) (bool, uint32, error)
}
+
+// Role is the role of node.
+type Role string
+
+const (
+ // RoleMeta is the role of meta node.
+ RoleMeta = "meta"
+ // RoleData is the role of data node.
+ RoleData = "data"
+ // RoleQuery is the role of query node.
+ RoleQuery = "query"
+ // RoleLiaison is the role of liaison node.
+ RoleLiaison = "liaison"
+)
+
+func strToRole(role string) (Role, error) {
+ switch role {
+ case RoleMeta, RoleData, RoleQuery, RoleLiaison:
+ return Role(role), nil
+ default:
+ return "", errors.New("invalid role")
+ }
+}
+
+// Node allows CRUD node schemas in a group.
+type Node interface {
+ ListNode(ctx context.Context, role Role) ([]*databasev1.Node, error)
+}
+
+// Shard allows CRUD shard schemas in a group.
+type Shard interface {
+ CreateOrUpdateShard(ctx context.Context, shard *databasev1.Shard) error
+ ListShard(ctx context.Context, opt ListOpt) ([]*databasev1.Shard, error)
+}
diff --git a/banyand/metadata/schema/shard.go b/banyand/metadata/schema/shard.go
new file mode 100644
index 00000000..caf8b9de
--- /dev/null
+++ b/banyand/metadata/schema/shard.go
@@ -0,0 +1,76 @@
+// 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 schema
+
+import (
+ "context"
+
+ "github.com/pkg/errors"
+ "google.golang.org/protobuf/proto"
+ "google.golang.org/protobuf/types/known/timestamppb"
+
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+)
+
+var shardKeyPrefix = "/shards/"
+
+func (e *etcdSchemaRegistry) CreateOrUpdateShard(ctx context.Context, shard *databasev1.Shard) error {
+ if shard.UpdatedAt != nil {
+ shard.UpdatedAt = timestamppb.Now()
+ }
+ md := Metadata{
+ TypeMeta: TypeMeta{
+ Kind: KindShard,
+ Group: shard.GetMetadata().GetGroup(),
+ Name: shard.GetMetadata().GetName(),
+ },
+ Spec: shard,
+ }
+ err := e.update(ctx, md)
+ if err == nil {
+ return nil
+ }
+ if errors.Is(err, ErrGRPCResourceNotFound) {
+ shard.CreatedAt = shard.UpdatedAt
+ md.Spec = shard
+ return e.create(ctx, md)
+ }
+ return err
+}
+
+func (e *etcdSchemaRegistry) ListShard(ctx context.Context, opt ListOpt) ([]*databasev1.Shard, error) {
+ if opt.Group == "" {
+ return nil, BadRequest("group", "group should not be empty")
+ }
+ messages, err := e.listWithPrefix(ctx, listPrefixesForEntity(opt.Group, shardKeyPrefix), func() proto.Message {
+ return &databasev1.Shard{}
+ })
+ if err != nil {
+ return nil, err
+ }
+ entities := make([]*databasev1.Shard, 0, len(messages))
+ for _, message := range messages {
+ entities = append(entities, message.(*databasev1.Shard))
+ }
+ return entities, nil
+}
+
+func formatShardKey(metadata *commonv1.Metadata) string {
+ return formatKey(shardKeyPrefix, metadata)
+}
diff --git a/banyand/queue/local.go b/banyand/queue/local.go
index ec4a95e1..14a8632b 100644
--- a/banyand/queue/local.go
+++ b/banyand/queue/local.go
@@ -19,7 +19,6 @@
package queue
import (
- "github.com/apache/skywalking-banyandb/banyand/discovery"
"github.com/apache/skywalking-banyandb/pkg/bus"
"github.com/apache/skywalking-banyandb/pkg/run"
)
@@ -31,7 +30,6 @@ var (
type local struct {
local *bus.Bus
- repo discovery.ServiceRepo
stopCh chan struct{}
}
diff --git a/banyand/queue/queue.go b/banyand/queue/queue.go
index 0946ecb1..e4bc0b33 100644
--- a/banyand/queue/queue.go
+++ b/banyand/queue/queue.go
@@ -20,7 +20,6 @@ package queue
import (
"context"
- "github.com/apache/skywalking-banyandb/banyand/discovery"
"github.com/apache/skywalking-banyandb/pkg/bus"
"github.com/apache/skywalking-banyandb/pkg/run"
)
@@ -35,10 +34,9 @@ type Queue interface {
run.Service
}
-// NewQueue return a new Queue which relies on the discovery service.
-func NewQueue(_ context.Context, repo discovery.ServiceRepo) (Queue, error) {
+// NewQueue return a new Queue.
+func NewQueue(_ context.Context) (Queue, error) {
return &local{
- repo: repo,
local: bus.NewBus(),
stopCh: make(chan struct{}),
}, nil
diff --git a/banyand/stream/metadata.go b/banyand/stream/metadata.go
index 28e78da8..d69710b1 100644
--- a/banyand/stream/metadata.go
+++ b/banyand/stream/metadata.go
@@ -23,10 +23,8 @@ import (
"time"
"github.com/apache/skywalking-banyandb/api/common"
- "github.com/apache/skywalking-banyandb/api/event"
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/banyand/discovery"
"github.com/apache/skywalking-banyandb/banyand/metadata"
"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
"github.com/apache/skywalking-banyandb/banyand/tsdb"
@@ -41,7 +39,7 @@ type schemaRepo struct {
metadata metadata.Repo
}
-func newSchemaRepo(path string, metadata metadata.Repo, repo discovery.ServiceRepo,
+func newSchemaRepo(path string, metadata metadata.Repo,
bufferSize int64, dbOpts tsdb.DatabaseOpts, l *logger.Logger,
) schemaRepo {
return schemaRepo{
@@ -49,11 +47,8 @@ func newSchemaRepo(path string, metadata metadata.Repo, repo discovery.ServiceRe
metadata: metadata,
Repository: resourceSchema.NewRepository(
metadata,
- repo,
l,
newSupplier(path, metadata, bufferSize, dbOpts, l),
- event.StreamTopicShardEvent,
- event.StreamTopicEntityEvent,
),
}
}
diff --git a/banyand/stream/metadata_test.go b/banyand/stream/metadata_test.go
index 3377d556..dc184d40 100644
--- a/banyand/stream/metadata_test.go
+++ b/banyand/stream/metadata_test.go
@@ -24,10 +24,8 @@ import (
. "github.com/onsi/gomega"
"github.com/onsi/gomega/gleak"
- "github.com/apache/skywalking-banyandb/api/event"
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/test"
"github.com/apache/skywalking-banyandb/pkg/test/flags"
)
@@ -54,7 +52,6 @@ var _ = Describe("Metadata", func() {
}).WithTimeout(flags.EventuallyTimeout).Should(BeTrue())
})
It("should close the group", func() {
- svcs.repo.EXPECT().Publish(event.StreamTopicShardEvent, test.NewShardEventMatcher(databasev1.Action_ACTION_DELETE)).Times(2)
deleted, err := svcs.metadataService.GroupRegistry().DeleteGroup(context.TODO(), "default")
Expect(err).ShouldNot(HaveOccurred())
Expect(deleted).Should(BeTrue())
@@ -65,8 +62,6 @@ var _ = Describe("Metadata", func() {
})
It("should add shards", func() {
- svcs.repo.EXPECT().Publish(event.StreamTopicShardEvent, test.NewShardEventMatcher(databasev1.Action_ACTION_DELETE)).Times(2)
- svcs.repo.EXPECT().Publish(event.StreamTopicShardEvent, test.NewShardEventMatcher(databasev1.Action_ACTION_PUT)).Times(4)
groupSchema, err := svcs.metadataService.GroupRegistry().GetGroup(context.TODO(), "default")
Expect(err).ShouldNot(HaveOccurred())
Expect(groupSchema).ShouldNot(BeNil())
@@ -95,7 +90,6 @@ var _ = Describe("Metadata", func() {
}).WithTimeout(flags.EventuallyTimeout).Should(BeTrue())
})
It("should close the stream", func() {
- svcs.repo.EXPECT().Publish(event.StreamTopicEntityEvent, test.NewEntityEventMatcher(databasev1.Action_ACTION_DELETE)).Times(1)
deleted, err := svcs.metadataService.StreamRegistry().DeleteStream(context.TODO(), &commonv1.Metadata{
Name: "sw",
Group: "default",
@@ -126,7 +120,6 @@ var _ = Describe("Metadata", func() {
})
It("should update a new stream", func() {
- svcs.repo.EXPECT().Publish(event.StreamTopicEntityEvent, test.NewEntityEventMatcher(databasev1.Action_ACTION_PUT)).Times(1)
// Remove the first tag from the entity
streamSchema.Entity.TagNames = streamSchema.Entity.TagNames[1:]
entitySize := len(streamSchema.Entity.TagNames)
diff --git a/banyand/stream/service.go b/banyand/stream/service.go
index d395e9ca..07749f5b 100644
--- a/banyand/stream/service.go
+++ b/banyand/stream/service.go
@@ -26,7 +26,6 @@ import (
"github.com/apache/skywalking-banyandb/api/data"
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
- "github.com/apache/skywalking-banyandb/banyand/discovery"
"github.com/apache/skywalking-banyandb/banyand/metadata"
"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
"github.com/apache/skywalking-banyandb/banyand/observability"
@@ -55,10 +54,8 @@ type service struct {
schemaRepo schemaRepo
metadata metadata.Repo
pipeline queue.Queue
- repo discovery.ServiceRepo
writeListener *writeCallback
l *logger.Logger
- stopCh chan struct{}
root string
dbOpts tsdb.DatabaseOpts
blockBufferSize run.Bytes
@@ -106,7 +103,7 @@ func (s *service) PreRun() error {
}
path := path.Join(s.root, s.Name())
observability.UpdatePath(path)
- s.schemaRepo = newSchemaRepo(path, s.metadata, s.repo, int64(s.blockBufferSize), s.dbOpts, s.l)
+ s.schemaRepo = newSchemaRepo(path, s.metadata, int64(s.blockBufferSize), s.dbOpts, s.l)
for _, g := range groups {
if g.Catalog != commonv1.Catalog_CATALOG_STREAM {
continue
@@ -127,6 +124,10 @@ func (s *service) PreRun() error {
}
}
}
+ // run a serial watcher
+ s.schemaRepo.Watcher()
+ s.metadata.RegisterHandler(schema.KindGroup|schema.KindStream|schema.KindIndexRuleBinding|schema.KindIndexRule,
+ &s.schemaRepo)
s.writeListener = setUpWriteCallback(s.l, &s.schemaRepo)
@@ -138,32 +139,21 @@ func (s *service) PreRun() error {
}
func (s *service) Serve() run.StopNotify {
- _ = s.schemaRepo.NotifyAll()
- // run a serial watcher
- s.schemaRepo.Watcher()
-
- s.metadata.StreamRegistry().RegisterHandler(schema.KindGroup|schema.KindStream|schema.KindIndexRuleBinding|schema.KindIndexRule,
- &s.schemaRepo)
- return s.stopCh
+ return s.schemaRepo.StopCh()
}
func (s *service) GracefulStop() {
s.schemaRepo.Close()
- if s.stopCh != nil {
- close(s.stopCh)
- }
}
// NewService returns a new service.
-func NewService(_ context.Context, metadata metadata.Repo, repo discovery.ServiceRepo, pipeline queue.Queue) (Service, error) {
+func NewService(_ context.Context, metadata metadata.Repo, pipeline queue.Queue) (Service, error) {
return &service{
metadata: metadata,
- repo: repo,
pipeline: pipeline,
dbOpts: tsdb.DatabaseOpts{
EnableGlobalIndex: true,
IndexGranularity: tsdb.IndexGranularityBlock,
},
- stopCh: make(chan struct{}),
}, nil
}
diff --git a/banyand/stream/stream.go b/banyand/stream/stream.go
index ded686fe..fb60bd64 100644
--- a/banyand/stream/stream.go
+++ b/banyand/stream/stream.go
@@ -27,7 +27,6 @@ import (
"github.com/apache/skywalking-banyandb/banyand/tsdb"
"github.com/apache/skywalking-banyandb/banyand/tsdb/index"
"github.com/apache/skywalking-banyandb/pkg/logger"
- "github.com/apache/skywalking-banyandb/pkg/partition"
"github.com/apache/skywalking-banyandb/pkg/schema"
)
@@ -43,7 +42,6 @@ type stream struct {
indexWriter *index.Writer
name string
group string
- entityLocator partition.EntityLocator
indexRules []*databasev1.IndexRule
maxObservedModRevision int64
shardNum uint32
@@ -69,17 +67,12 @@ func (s *stream) GetTopN() []*databasev1.TopNAggregation {
return nil
}
-func (s *stream) EntityLocator() partition.EntityLocator {
- return s.entityLocator
-}
-
func (s *stream) Close() error {
return nil
}
func (s *stream) parseSpec() {
s.name, s.group = s.schema.GetMetadata().GetName(), s.schema.GetMetadata().GetGroup()
- s.entityLocator = partition.NewEntityLocator(s.schema.GetTagFamilies(), s.schema.GetEntity())
s.maxObservedModRevision = schema.ParseMaxModRevision(s.indexRules)
}
diff --git a/banyand/stream/stream_suite_test.go b/banyand/stream/stream_suite_test.go
index e89987f2..fa5fa2c3 100644
--- a/banyand/stream/stream_suite_test.go
+++ b/banyand/stream/stream_suite_test.go
@@ -25,9 +25,6 @@ import (
. "github.com/onsi/gomega"
"go.uber.org/mock/gomock"
- "github.com/apache/skywalking-banyandb/api/event"
- databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
- "github.com/apache/skywalking-banyandb/banyand/discovery"
"github.com/apache/skywalking-banyandb/banyand/metadata"
"github.com/apache/skywalking-banyandb/banyand/queue"
"github.com/apache/skywalking-banyandb/pkg/logger"
@@ -63,25 +60,13 @@ func (p *preloadStreamService) PreRun() error {
type services struct {
stream *service
metadataService metadata.Service
- repo *discovery.MockServiceRepo
}
func setUp() (*services, func()) {
ctrl := gomock.NewController(GinkgoT())
Expect(ctrl).ShouldNot(BeNil())
- // Init Discovery
- repo := discovery.NewMockServiceRepo(ctrl)
- repo.EXPECT().NodeID().AnyTimes()
- repo.EXPECT().Name().AnyTimes()
- stopCh := make(chan struct{})
- repo.EXPECT().Serve().Return(stopCh).Times(1)
- repo.EXPECT().GracefulStop().Do(func() { close(stopCh) }).Times(1)
- // Both PreRun and Serve phases send events
- repo.EXPECT().Publish(event.StreamTopicEntityEvent, test.NewEntityEventMatcher(databasev1.Action_ACTION_PUT)).Times(2 * 1)
- repo.EXPECT().Publish(event.StreamTopicShardEvent, test.NewShardEventMatcher(databasev1.Action_ACTION_PUT)).Times(2 * 2)
-
// Init Pipeline
- pipeline, err := queue.NewQueue(context.TODO(), repo)
+ pipeline, err := queue.NewQueue(context.TODO())
Expect(err).NotTo(HaveOccurred())
// Init Metadata Service
@@ -90,7 +75,7 @@ func setUp() (*services, func()) {
Expect(err).NotTo(HaveOccurred())
// Init Stream Service
- streamService, err := NewService(context.TODO(), metadataService, repo, pipeline)
+ streamService, err := NewService(context.TODO(), metadataService, pipeline)
Expect(err).NotTo(HaveOccurred())
preloadStreamSvc := &preloadStreamService{metaSvc: metadataService}
var flags []string
@@ -105,7 +90,6 @@ func setUp() (*services, func()) {
flags = append(flags, "--etcd-listen-client-url="+listenClientURL, "--etcd-listen-peer-url="+listenPeerURL)
moduleDeferFunc := test.SetupModules(
flags,
- repo,
pipeline,
metadataService,
preloadStreamSvc,
@@ -114,7 +98,6 @@ func setUp() (*services, func()) {
return &services{
stream: streamService.(*service),
metadataService: metadataService,
- repo: repo,
}, func() {
moduleDeferFunc()
metaDeferFunc()
diff --git a/dist/LICENSE b/dist/LICENSE
index 55639375..87b47d39 100644
--- a/dist/LICENSE
+++ b/dist/LICENSE
@@ -191,7 +191,7 @@ Apache-2.0 licenses
github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da Apache-2.0
github.com/google/btree v1.1.2 Apache-2.0
github.com/google/flatbuffers v1.12.1 Apache-2.0
- github.com/google/pprof v0.0.0-20230705174524-200ffdc848b8 Apache-2.0
+ github.com/google/pprof v0.0.0-20230728192033-2ba5b33183c6 Apache-2.0
github.com/grpc-ecosystem/go-grpc-middleware v1.4.0 Apache-2.0
github.com/grpc-ecosystem/go-grpc-middleware/providers/prometheus v1.0.0-rc.0 Apache-2.0
github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.0.0-rc.5 Apache-2.0
@@ -287,7 +287,7 @@ BSD-3-Clause licenses
golang.org/x/sys v0.10.0 BSD-3-Clause
golang.org/x/text v0.11.0 BSD-3-Clause
golang.org/x/time v0.3.0 BSD-3-Clause
- golang.org/x/tools v0.11.0 BSD-3-Clause
+ golang.org/x/tools v0.11.1 BSD-3-Clause
google.golang.org/protobuf v1.31.0 BSD-3-Clause
========================================================================
@@ -326,7 +326,7 @@ MIT licenses
github.com/mattn/go-isatty v0.0.19 MIT
github.com/mitchellh/mapstructure v1.5.0 MIT
github.com/onsi/ginkgo/v2 v2.11.0 MIT
- github.com/onsi/gomega v1.27.8 MIT
+ github.com/onsi/gomega v1.27.10 MIT
github.com/pelletier/go-toml/v2 v2.0.8 MIT
github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c MIT
github.com/robfig/cron/v3 v3.0.1 MIT
diff --git a/docs/api-reference.md b/docs/api-reference.md
index e587be94..96931a18 100644
--- a/docs/api-reference.md
+++ b/docs/api-reference.md
@@ -341,6 +341,7 @@ Metadata is for multi-tenant, multi-model use
| Field | Type | Label | Description |
| ----- | ---- | ----- | ----------- |
| id | [string](#string) | | |
+| metadata | [banyandb.common.v1.Metadata](#banyandb-common-v1-Metadata) | | |
| addr | [string](#string) | | |
| updated_at | [google.protobuf.Timestamp](#google-protobuf-Timestamp) | | |
| created_at | [google.protobuf.Timestamp](#google-protobuf-Timestamp) | | |
diff --git a/go.mod b/go.mod
index 5eea559e..7cf87065 100644
--- a/go.mod
+++ b/go.mod
@@ -21,7 +21,7 @@ require (
github.com/hashicorp/golang-lru/v2 v2.0.4
github.com/oklog/run v1.1.0
github.com/onsi/ginkgo/v2 v2.11.0
- github.com/onsi/gomega v1.27.8
+ github.com/onsi/gomega v1.27.10
github.com/pkg/errors v0.9.1
github.com/prometheus/client_golang v1.16.0
github.com/rs/zerolog v1.29.1
@@ -77,7 +77,7 @@ require (
github.com/golang/snappy v0.0.3
github.com/google/btree v1.1.2 // indirect
github.com/google/flatbuffers v1.12.1 // indirect
- github.com/google/pprof v0.0.0-20230705174524-200ffdc848b8 // indirect
+ github.com/google/pprof v0.0.0-20230728192033-2ba5b33183c6 // indirect
github.com/gorilla/websocket v1.5.0 // indirect
github.com/grpc-ecosystem/go-grpc-middleware v1.4.0 // indirect
github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 // indirect
@@ -140,7 +140,7 @@ require (
golang.org/x/sys v0.10.0 // indirect
golang.org/x/text v0.11.0 // indirect
golang.org/x/time v0.3.0 // indirect
- golang.org/x/tools v0.11.0 // indirect
+ golang.org/x/tools v0.11.1 // indirect
google.golang.org/genproto v0.0.0-20230706204954-ccb25ca9f130 // indirect
gopkg.in/ini.v1 v1.67.0 // indirect
gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect
diff --git a/go.sum b/go.sum
index cd1e625b..eb7ac7e6 100644
--- a/go.sum
+++ b/go.sum
@@ -222,8 +222,8 @@ github.com/google/pprof v0.0.0-20200708004538-1a94d8640e99/go.mod h1:ZgVRPoUq/hf
github.com/google/pprof v0.0.0-20201023163331-3e6fc7fc9c4c/go.mod h1:kpwsk12EmLew5upagYY7GY0pfYCcupk39gWOCRROcvE=
github.com/google/pprof v0.0.0-20201203190320-1bf35d6f28c2/go.mod h1:kpwsk12EmLew5upagYY7GY0pfYCcupk39gWOCRROcvE=
github.com/google/pprof v0.0.0-20201218002935-b9804c9f04c2/go.mod h1:kpwsk12EmLew5upagYY7GY0pfYCcupk39gWOCRROcvE=
-github.com/google/pprof v0.0.0-20230705174524-200ffdc848b8 h1:n6vlPhxsA+BW/XsS5+uqi7GyzaLa5MH7qlSLBZtRdiA=
-github.com/google/pprof v0.0.0-20230705174524-200ffdc848b8/go.mod h1:Jh3hGz2jkYak8qXPD19ryItVnUgpgeqzdkY/D0EaeuA=
+github.com/google/pprof v0.0.0-20230728192033-2ba5b33183c6 h1:ZgoomqkdjGbQ3+qQXCkvYMCDvGDNg2k5JJDjjdTB6jY=
+github.com/google/pprof v0.0.0-20230728192033-2ba5b33183c6/go.mod h1:Jh3hGz2jkYak8qXPD19ryItVnUgpgeqzdkY/D0EaeuA=
github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
@@ -311,8 +311,8 @@ github.com/oklog/run v1.1.0 h1:GEenZ1cK0+q0+wsJew9qUg/DyD8k3JzYsZAi5gYi2mA=
github.com/oklog/run v1.1.0/go.mod h1:sVPdnTZT1zYwAJeCMu2Th4T21pA3FPOQRfWjQlk7DVU=
github.com/onsi/ginkgo/v2 v2.11.0 h1:WgqUCUt/lT6yXoQ8Wef0fsNn5cAuMK7+KT9UFRz2tcU=
github.com/onsi/ginkgo/v2 v2.11.0/go.mod h1:ZhrRA5XmEE3x3rhlzamx/JJvujdZoJ2uvgI7kR0iZvM=
-github.com/onsi/gomega v1.27.8 h1:gegWiwZjBsf2DgiSbf5hpokZ98JVDMcWkUiigk6/KXc=
-github.com/onsi/gomega v1.27.8/go.mod h1:2J8vzI/s+2shY9XHRApDkdgPo1TKT7P2u6fXeJKFnNQ=
+github.com/onsi/gomega v1.27.10 h1:naR28SdDFlqrG6kScpT8VWpu1xWY5nJRCF3XaYyBjhI=
+github.com/onsi/gomega v1.27.10/go.mod h1:RsS8tutOdbdgzbPtzzATp12yT7kM5I5aElG3evPbQ0M=
github.com/opentracing/opentracing-go v1.1.0/go.mod h1:UkNAQd3GIcIGf0SeVgPpRdFStlNbqXla1AfSYxPUl2o=
github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=
github.com/pelletier/go-toml/v2 v2.0.8 h1:0ctb6s9mE31h0/lhu+J6OPmVeDxJn+kYnJc2jZR9tGQ=
@@ -707,8 +707,8 @@ golang.org/x/tools v0.0.0-20210105154028-b0ab187a4818/go.mod h1:emZCQorbCU4vsT4f
golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
golang.org/x/tools v0.0.0-20210108195828-e2f9c7f1fc8e/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
golang.org/x/tools v0.1.0/go.mod h1:xkSsbof2nBLbhDlRMhhhyNLN/zl3eTqcnHD5viDpcZ0=
-golang.org/x/tools v0.11.0 h1:EMCa6U9S2LtZXLAMoWiR/R8dAQFRqbAitmbJ2UKhoi8=
-golang.org/x/tools v0.11.0/go.mod h1:anzJrxPjNtfgiYQYirP2CPGzGLxrH2u2QBhn6Bf3qY8=
+golang.org/x/tools v0.11.1 h1:ojD5zOW8+7dOGzdnNgersm8aPfcDjhMp12UfG93NIMc=
+golang.org/x/tools v0.11.1/go.mod h1:anzJrxPjNtfgiYQYirP2CPGzGLxrH2u2QBhn6Bf3qY8=
golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
diff --git a/pkg/schema/metadata.go b/pkg/schema/metadata.go
index 31f23f57..6d20cd30 100644
--- a/pkg/schema/metadata.go
+++ b/pkg/schema/metadata.go
@@ -26,20 +26,15 @@ import (
"sync/atomic"
"time"
- "github.com/onsi/ginkgo/v2"
"github.com/pkg/errors"
"go.uber.org/multierr"
- "google.golang.org/protobuf/types/known/timestamppb"
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/banyand/discovery"
"github.com/apache/skywalking-banyandb/banyand/metadata"
"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
"github.com/apache/skywalking-banyandb/banyand/tsdb"
- "github.com/apache/skywalking-banyandb/pkg/bus"
"github.com/apache/skywalking-banyandb/pkg/logger"
- "github.com/apache/skywalking-banyandb/pkg/partition"
"github.com/apache/skywalking-banyandb/pkg/run"
)
@@ -95,7 +90,6 @@ type Resource interface {
GetIndexRules() []*databasev1.IndexRule
GetTopN() []*databasev1.TopNAggregation
MaxObservedModRevision() int64
- EntityLocator() partition.EntityLocator
ResourceSchema
io.Closer
}
@@ -114,8 +108,8 @@ type Repository interface {
StoreGroup(groupMeta *commonv1.Metadata) (*group, error)
LoadGroup(name string) (Group, bool)
LoadResource(metadata *commonv1.Metadata) (Resource, bool)
- NotifyAll() (err error)
Close()
+ StopCh() <-chan struct{}
}
const defaultWorkerNum = 10
@@ -124,38 +118,36 @@ var _ Repository = (*schemaRepo)(nil)
type schemaRepo struct {
metadata metadata.Repo
- repo discovery.ServiceRepo
resourceSupplier ResourceSupplier
l *logger.Logger
data map[string]*group
+ workerCloser *run.Closer
closer *run.Closer
eventCh chan MetadataEvent
- shardTopic bus.Topic
- entityTopic bus.Topic
workerNum int
sync.RWMutex
}
+// StopCh implements Repository.
+func (sr *schemaRepo) StopCh() <-chan struct{} {
+ return sr.closer.CloseNotify()
+}
+
// NewRepository return a new Repository.
func NewRepository(
metadata metadata.Repo,
- repo discovery.ServiceRepo,
l *logger.Logger,
resourceSupplier ResourceSupplier,
- shardTopic bus.Topic,
- entityTopic bus.Topic,
) Repository {
return &schemaRepo{
metadata: metadata,
- repo: repo,
l: l,
resourceSupplier: resourceSupplier,
- shardTopic: shardTopic,
- entityTopic: entityTopic,
data: make(map[string]*group),
eventCh: make(chan MetadataEvent, defaultWorkerNum),
workerNum: defaultWorkerNum,
- closer: run.NewCloser(defaultWorkerNum),
+ workerCloser: run.NewCloser(defaultWorkerNum),
+ closer: run.NewCloser(1),
}
}
@@ -167,7 +159,7 @@ func (sr *schemaRepo) Watcher() {
for i := 0; i < sr.workerNum; i++ {
go func() {
defer func() {
- sr.closer.Done()
+ sr.workerCloser.Done()
if err := recover(); err != nil {
sr.l.Warn().Interface("err", err).Msg("watching the events")
}
@@ -202,11 +194,11 @@ func (sr *schemaRepo) Watcher() {
sr.l.Err(err).Interface("event", evt).Msg("fail to handle the metadata event. retry...")
select {
case sr.eventCh <- evt:
- case <-sr.closer.CloseNotify():
+ case <-sr.workerCloser.CloseNotify():
return
}
}
- case <-sr.closer.CloseNotify():
+ case <-sr.workerCloser.CloseNotify():
return
}
}
@@ -232,9 +224,9 @@ func (sr *schemaRepo) StoreGroup(groupMeta *commonv1.Metadata) (*group, error) {
if err != nil {
return nil, err
}
- g = newGroup(groupSchema, sr.repo, sr.metadata, db, sr.l, sr.resourceSupplier, sr.entityTopic)
+ g = newGroup(groupSchema, sr.metadata, db, sr.l, sr.resourceSupplier)
sr.data[name] = g
- return g, sr.notify(groupSchema, databasev1.Action_ACTION_PUT)
+ return g, nil
}
prevGroupSchema := g.GetSchema()
if groupSchema.GetMetadata().GetModRevision() <= prevGroupSchema.Metadata.ModRevision {
@@ -243,20 +235,12 @@ func (sr *schemaRepo) StoreGroup(groupMeta *commonv1.Metadata) (*group, error) {
sr.l.Info().Str("group", name).Msg("closing the previous tsdb")
db := g.SupplyTSDB()
db.Close()
- err = sr.notify(prevGroupSchema, databasev1.Action_ACTION_DELETE)
- if err != nil {
- return nil, err
- }
sr.l.Info().Str("group", name).Msg("creating a new tsdb")
newDB, err := sr.resourceSupplier.OpenDB(groupSchema)
if err != nil {
return nil, err
}
g.setDB(newDB)
- err = sr.notify(groupSchema, databasev1.Action_ACTION_PUT)
- if err != nil {
- return nil, err
- }
g.groupSchema.Store(groupSchema)
return g, nil
}
@@ -274,7 +258,6 @@ func (sr *schemaRepo) deleteGroup(groupMeta *commonv1.Metadata) error {
if err != nil {
return err
}
- _ = sr.notify(g.GetSchema(), databasev1.Action_ACTION_DELETE)
delete(sr.data, name)
return nil
}
@@ -324,61 +307,13 @@ func (sr *schemaRepo) deleteResource(metadata *commonv1.Metadata) error {
return g.(*group).deleteResource(metadata)
}
-func (sr *schemaRepo) notify(groupSchema *commonv1.Group, action databasev1.Action) (err error) {
- now := time.Now()
- nowPb := timestamppb.New(now)
- shardNum := groupSchema.GetResourceOpts().GetShardNum()
- for i := 0; i < int(shardNum); i++ {
- _, errInternal := sr.repo.Publish(sr.shardTopic, bus.NewMessage(bus.MessageID(now.UnixNano()), &databasev1.ShardEvent{
- Shard: &databasev1.Shard{
- Id: uint64(i),
- Total: shardNum,
- Metadata: &commonv1.Metadata{
- Name: groupSchema.GetMetadata().GetName(),
- },
- Node: &databasev1.Node{
- Id: sr.repo.NodeID(),
- CreatedAt: nowPb,
- UpdatedAt: nowPb,
- Addr: "localhost",
- },
- UpdatedAt: nowPb,
- CreatedAt: nowPb,
- },
- Time: nowPb,
- Action: action,
- }))
- if errors.Is(errInternal, bus.ErrTopicNotExist) {
- return nil
- }
- if errInternal != nil {
- err = multierr.Append(err, errInternal)
- }
- }
- return err
-}
-
-func (sr *schemaRepo) NotifyAll() (err error) {
- sr.RLock()
- defer sr.RUnlock()
- for _, g := range sr.data {
- err = multierr.Append(err, sr.notify(g.GetSchema(), databasev1.Action_ACTION_PUT))
- g.mapMutex.RLock()
- for _, s := range g.schemaMap {
- err = multierr.Append(err, g.notify(s, databasev1.Action_ACTION_PUT))
- }
- g.mapMutex.RUnlock()
- }
- return err
-}
-
func (sr *schemaRepo) Close() {
defer func() {
if err := recover(); err != nil {
sr.l.Warn().Interface("err", err).Msg("closing resource")
}
}()
- sr.closer.CloseThenWait()
+ sr.workerCloser.CloseThenWait()
sr.RLock()
defer sr.RUnlock()
@@ -388,39 +323,35 @@ func (sr *schemaRepo) Close() {
sr.l.Err(err).RawJSON("group", logger.Proto(g.GetSchema().Metadata)).Msg("closing")
}
}
+ sr.closer.Done()
+ sr.closer.CloseThenWait()
}
var _ Group = (*group)(nil)
type group struct {
resourceSupplier ResourceSupplier
- repo discovery.ServiceRepo
metadata metadata.Repo
db atomic.Value
groupSchema atomic.Pointer[commonv1.Group]
l *logger.Logger
schemaMap map[string]Resource
- entityTopic bus.Topic
mapMutex sync.RWMutex
}
func newGroup(
groupSchema *commonv1.Group,
- repo discovery.ServiceRepo,
metadata metadata.Repo,
db tsdb.Database,
l *logger.Logger,
resourceSupplier ResourceSupplier,
- entityTopic bus.Topic,
) *group {
g := &group{
groupSchema: atomic.Pointer[commonv1.Group]{},
- repo: repo,
metadata: metadata,
l: l,
schemaMap: make(map[string]Resource),
resourceSupplier: resourceSupplier,
- entityTopic: entityTopic,
}
g.groupSchema.Store(groupSchema)
g.db.Store(db)
@@ -492,9 +423,6 @@ func (g *group) StoreResource(resourceSchema ResourceSchema) (Resource, error) {
if errTS != nil {
return nil, errTS
}
- if err := g.notify(sm, databasev1.Action_ACTION_PUT); err != nil {
- return nil, err
- }
g.schemaMap[key] = sm
if preResource != nil {
_ = preResource.Close()
@@ -510,9 +438,6 @@ func (g *group) deleteResource(metadata *commonv1.Metadata) error {
if preResource == nil {
return nil
}
- if err := g.notify(preResource, databasev1.Action_ACTION_DELETE); err != nil {
- return err
- }
delete(g.schemaMap, key)
_ = preResource.Close()
return nil
@@ -528,30 +453,6 @@ func (g *group) LoadResource(name string) (Resource, bool) {
return s, true
}
-func (g *group) notify(resource Resource, action databasev1.Action) error {
- defer ginkgo.GinkgoRecover()
- now := time.Now()
- nowPb := timestamppb.New(now)
- entityLocator := resource.EntityLocator()
- locator := make([]*databasev1.EntityEvent_TagLocator, 0, len(entityLocator))
- for _, tagLocator := range entityLocator {
- locator = append(locator, &databasev1.EntityEvent_TagLocator{
- FamilyOffset: uint32(tagLocator.FamilyOffset),
- TagOffset: uint32(tagLocator.TagOffset),
- })
- }
- _, err := g.repo.Publish(g.entityTopic, bus.NewMessage(bus.MessageID(now.UnixNano()), &databasev1.EntityEvent{
- Subject: resource.GetMetadata(),
- EntityLocator: locator,
- Time: nowPb,
- Action: action,
- }))
- if errors.Is(err, bus.ErrTopicNotExist) {
- return nil
- }
- return err
-}
-
func (g *group) close() (err error) {
g.mapMutex.RLock()
for _, s := range g.schemaMap {
diff --git a/pkg/test/setup/setup.go b/pkg/test/setup/setup.go
index 7b654471..18070973 100644
--- a/pkg/test/setup/setup.go
+++ b/pkg/test/setup/setup.go
@@ -24,7 +24,6 @@ import (
"github.com/onsi/gomega"
- "github.com/apache/skywalking-banyandb/banyand/discovery"
"github.com/apache/skywalking-banyandb/banyand/liaison/grpc"
"github.com/apache/skywalking-banyandb/banyand/liaison/http"
"github.com/apache/skywalking-banyandb/banyand/measure"
@@ -77,37 +76,37 @@ func CommonWithSchemaLoaders(schemaLoaders []SchemaLoader, flags ...string) (str
}
func modules(schemaLoaders []SchemaLoader, flags []string) func() {
- // Init `Discovery` module
- repo, err := discovery.NewServiceRepo(context.Background())
- gomega.Expect(err).NotTo(gomega.HaveOccurred())
// Init `Queue` module
- pipeline, err := queue.NewQueue(context.TODO(), repo)
+ pipeline, err := queue.NewQueue(context.TODO())
gomega.Expect(err).NotTo(gomega.HaveOccurred())
// Init `Metadata` module
metaSvc, err := metadata.NewService(context.TODO())
gomega.Expect(err).NotTo(gomega.HaveOccurred())
// Init `Stream` module
- streamSvc, err := stream.NewService(context.TODO(), metaSvc, repo, pipeline)
+ streamSvc, err := stream.NewService(context.TODO(), metaSvc, pipeline)
gomega.Expect(err).NotTo(gomega.HaveOccurred())
// Init `Measure` module
- measureSvc, err := measure.NewService(context.TODO(), metaSvc, repo, pipeline)
+ measureSvc, err := measure.NewService(context.TODO(), metaSvc, pipeline)
gomega.Expect(err).NotTo(gomega.HaveOccurred())
// 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, repo, metaSvc)
+ tcp := grpc.NewServer(context.TODO(), pipeline, metaSvc)
httpServer := http.NewService()
units := []run.Unit{
- repo,
pipeline,
metaSvc,
+ streamSvc,
+ measureSvc,
+ q,
+ tcp,
+ httpServer,
}
for _, sl := range schemaLoaders {
sl.SetMeta(metaSvc)
units = append(units, sl)
}
- units = append(units, streamSvc, measureSvc, q, tcp, httpServer)
return test.SetupModules(
flags,