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/23 00:07:45 UTC
[skywalking-banyandb] branch watch-etcd created (now 19ea90be)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a change to branch watch-etcd
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
at 19ea90be Add watcher to sync data in etcd to local cache
This branch includes the following new commits:
new 19ea90be Add watcher to sync data in etcd to local cache
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: Add watcher to sync data in etcd to local cache
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch watch-etcd
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 19ea90be0fa12837e591128080d279e93cf5ff2b
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Tue Aug 22 06:25:22 2023 +0000
Add watcher to sync data in etcd to local cache
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
CHANGES.md | 1 +
api/proto/banyandb/database/v1/database.proto | 2 +-
banyand/liaison/grpc/discovery.go | 4 +-
banyand/measure/metadata.go | 5 -
banyand/measure/service.go | 64 +---------
banyand/metadata/client.go | 10 +-
banyand/metadata/metadata.go | 2 +-
banyand/metadata/schema/etcd.go | 122 ++++++-------------
banyand/metadata/schema/etcd_test.go | 144 -----------------------
banyand/metadata/schema/group.go | 22 ++--
banyand/metadata/schema/index.go | 9 +-
banyand/metadata/schema/kind.go | 162 ++++++++++++++++++++++++++
banyand/metadata/schema/measure.go | 5 +-
banyand/metadata/schema/node.go | 8 +-
banyand/metadata/schema/property.go | 5 +-
banyand/metadata/schema/register_test.go | 7 +-
banyand/metadata/schema/schema.go | 55 +--------
banyand/metadata/schema/schema_suite_test.go | 35 +++++-
banyand/metadata/schema/shard.go | 5 +-
banyand/metadata/schema/stream.go | 5 +-
banyand/metadata/schema/topn.go | 5 +-
banyand/metadata/schema/watcher.go | 152 ++++++++++++++++++++++++
banyand/metadata/schema/watcher_test.go | 149 +++++++++++++++++++++++
banyand/stream/metadata.go | 3 -
banyand/stream/service.go | 31 +----
docs/api-reference.md | 2 +-
go.mod | 2 +-
pkg/logger/logger.go | 2 +-
pkg/run/closer.go | 8 ++
29 files changed, 587 insertions(+), 439 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 20d2bee1..92c6e2fd 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -20,6 +20,7 @@ Release Notes.
- Fix timer not released
- BanyanDB ui misses fields when creating a group
- Fix data duplicate writing
+- Syncing metadata change events from etcd instead of a local channel.
### Chores
diff --git a/api/proto/banyandb/database/v1/database.proto b/api/proto/banyandb/database/v1/database.proto
index 0e758d82..23309d0a 100644
--- a/api/proto/banyandb/database/v1/database.proto
+++ b/api/proto/banyandb/database/v1/database.proto
@@ -34,7 +34,7 @@ enum Role {
}
message Node {
- string name = 1;
+ common.v1.Metadata metadata = 1;
repeated Role roles = 2;
string grpc_address = 3;
string http_address = 4;
diff --git a/banyand/liaison/grpc/discovery.go b/banyand/liaison/grpc/discovery.go
index 606d93fb..ad0b3f58 100644
--- a/banyand/liaison/grpc/discovery.go
+++ b/banyand/liaison/grpc/discovery.go
@@ -130,8 +130,8 @@ func (ds *discoveryService) initialize(ctx context.Context) error {
return fmt.Errorf("unsupported kind: %d", ds.kind)
}
}
- ds.metadataRepo.RegisterHandler(schema.KindShard, ds.shardRepo)
- ds.metadataRepo.RegisterHandler(ds.kind, ds.entityRepo)
+ ds.metadataRepo.RegisterHandler("liaison", schema.KindShard, ds.shardRepo)
+ ds.metadataRepo.RegisterHandler("liaison", ds.kind, ds.entityRepo)
return nil
}
diff --git a/banyand/measure/metadata.go b/banyand/measure/metadata.go
index 21d44fc0..9df34dfa 100644
--- a/banyand/measure/metadata.go
+++ b/banyand/measure/metadata.go
@@ -93,7 +93,6 @@ func (sr *schemaRepo) OnAddOrUpdate(metadata schema.Metadata) {
})
cancel()
if err != nil {
- sr.l.Error().Err(err).Msg("fail to get subject")
return
}
sr.SendMetadataEvent(resourceSchema.MetadataEvent{
@@ -107,7 +106,6 @@ func (sr *schemaRepo) OnAddOrUpdate(metadata schema.Metadata) {
defer cancel()
subjects, err := sr.metadata.Subjects(ctx, metadata.Spec.(*databasev1.IndexRule), commonv1.Catalog_CATALOG_MEASURE)
if err != nil {
- sr.l.Error().Err(err).Msg("fail to get subjects(measure)")
return
}
for _, sub := range subjects {
@@ -121,7 +119,6 @@ func (sr *schemaRepo) OnAddOrUpdate(metadata schema.Metadata) {
// createOrUpdate TopN schemas in advance
_, err := createOrUpdateTopNMeasure(context.Background(), sr.metadata.MeasureRegistry(), metadata.Spec.(*databasev1.TopNAggregation))
if err != nil {
- sr.l.Error().Err(err).Msg("fail to create/update topN measure")
return
}
// reload source measure
@@ -231,7 +228,6 @@ func (sr *schemaRepo) OnDelete(metadata schema.Metadata) {
Group: metadata.Group,
})
if err != nil {
- sr.l.Error().Err(err).Msg("fail to get subject")
return
}
// we should update instead of delete
@@ -245,7 +241,6 @@ func (sr *schemaRepo) OnDelete(metadata schema.Metadata) {
case schema.KindTopNAggregation:
err := sr.removeTopNMeasure(metadata.Spec.(*databasev1.TopNAggregation).GetSourceMeasure())
if err != nil {
- sr.l.Error().Err(err).Msg("fail to remove topN measure")
return
}
// we should update instead of delete
diff --git a/banyand/measure/service.go b/banyand/measure/service.go
index 24689347..2ff76731 100644
--- a/banyand/measure/service.go
+++ b/banyand/measure/service.go
@@ -20,10 +20,8 @@ package measure
import (
"context"
"path"
- "time"
"github.com/pkg/errors"
- "go.uber.org/multierr"
"github.com/apache/skywalking-banyandb/api/data"
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
@@ -107,81 +105,25 @@ func (s *service) Role() databasev1.Role {
return databasev1.Role_ROLE_DATA
}
-func (s *service) PreRun(ctx context.Context) error {
+func (s *service) PreRun(_ context.Context) error {
s.l = logger.GetLogger(s.Name())
- ctxGroup, cancelGroup := context.WithTimeout(ctx, 5*time.Second)
- groups, err := s.metadata.GroupRegistry().ListGroup(ctxGroup)
- cancelGroup()
- if err != nil {
- return err
- }
path := path.Join(s.root, s.Name())
observability.UpdatePath(path)
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 {
- continue
- }
- gp, innerErr := s.schemaRepo.StoreGroup(g.Metadata)
- if innerErr != nil {
- return innerErr
- }
- ctxMeasure, cancelMeasure := context.WithTimeout(ctx, 5*time.Second)
- allMeasureSchemas, innerErr := s.metadata.MeasureRegistry().
- ListMeasure(ctxMeasure, schema.ListOpt{Group: gp.GetSchema().GetMetadata().GetName()})
- cancelMeasure()
- if innerErr != nil {
- return innerErr
- }
- for _, measureSchema := range allMeasureSchemas {
- // sanity check before calling StoreResource
- // since StoreResource may be called inside the event loop
- if checkErr := s.sanityCheck(ctx, gp, measureSchema); checkErr != nil {
- return checkErr
- }
- if _, innerErr := gp.StoreResource(ctx, measureSchema); innerErr != nil {
- return innerErr
- }
- }
- }
// run a serial watcher
go s.schemaRepo.Watcher()
s.metadata.
- RegisterHandler(schema.KindGroup|schema.KindMeasure|schema.KindIndexRuleBinding|schema.KindIndexRule|schema.KindTopNAggregation,
+ RegisterHandler("measure", 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)
+ err := s.pipeline.Subscribe(data.TopicMeasureWrite, s.writeListener)
if err != nil {
return err
}
return nil
}
-func (s *service) sanityCheck(ctx context.Context, group resourceSchema.Group, measureSchema *databasev1.Measure) error {
- var topNAggrs []*databasev1.TopNAggregation
- ctxLocal, cancel := context.WithTimeout(ctx, 10*time.Second)
- defer cancel()
- topNAggrs, err := s.metadata.MeasureRegistry().TopNAggregations(ctxLocal, measureSchema.GetMetadata())
- if err != nil || len(topNAggrs) == 0 {
- return err
- }
-
- for _, topNAggr := range topNAggrs {
- topNMeasure, innerErr := createOrUpdateTopNMeasure(ctx, s.metadata.MeasureRegistry(), topNAggr)
- err = multierr.Append(err, innerErr)
- if topNMeasure != nil {
- _, storeErr := group.StoreResource(ctx, topNMeasure)
- if storeErr != nil {
- err = multierr.Append(err, storeErr)
- }
- }
- }
-
- return err
-}
-
func (s *service) Serve() run.StopNotify {
return s.schemaRepo.StopCh()
}
diff --git a/banyand/metadata/client.go b/banyand/metadata/client.go
index 73768a1f..33c258be 100644
--- a/banyand/metadata/client.go
+++ b/banyand/metadata/client.go
@@ -83,7 +83,9 @@ func (s *clientService) PreRun(ctx context.Context) error {
ctxRegister, cancel := context.WithTimeout(ctx, time.Second*5)
defer cancel()
if err = s.schemaRegistry.RegisterNode(ctxRegister, &databasev1.Node{
- Name: node.NodeID,
+ Metadata: &commonv1.Metadata{
+ Name: node.NodeID,
+ },
GrpcAddress: node.GrpcAddress,
HttpAddress: node.HTTPAddress,
Roles: nodeRoles,
@@ -92,7 +94,7 @@ func (s *clientService) PreRun(ctx context.Context) error {
return err
}
s.alc = newAllocator(s.schemaRegistry, logger.GetLogger(s.Name()).Named("allocator"))
- s.schemaRegistry.RegisterHandler(schema.KindGroup|schema.KindNode, s.alc)
+ s.schemaRegistry.RegisterHandler("shard-allocator", schema.KindGroup|schema.KindNode, s.alc)
return nil
}
@@ -106,8 +108,8 @@ func (s *clientService) GracefulStop() {
_ = s.schemaRegistry.Close()
}
-func (s *clientService) RegisterHandler(kind schema.Kind, handler schema.EventHandler) {
- s.schemaRegistry.RegisterHandler(kind, handler)
+func (s *clientService) RegisterHandler(name string, kind schema.Kind, handler schema.EventHandler) {
+ s.schemaRegistry.RegisterHandler(name, kind, handler)
}
func (s *clientService) StreamRegistry() schema.Stream {
diff --git a/banyand/metadata/metadata.go b/banyand/metadata/metadata.go
index 9bad1e72..58d10318 100644
--- a/banyand/metadata/metadata.go
+++ b/banyand/metadata/metadata.go
@@ -47,7 +47,7 @@ type Repo interface {
TopNAggregationRegistry() schema.TopNAggregation
PropertyRegistry() schema.Property
ShardRegistry() schema.Shard
- RegisterHandler(schema.Kind, schema.EventHandler)
+ RegisterHandler(string, schema.Kind, schema.EventHandler)
}
// Service is the metadata repository.
diff --git a/banyand/metadata/schema/etcd.go b/banyand/metadata/schema/etcd.go
index 6cbb459c..aff8e9ad 100644
--- a/banyand/metadata/schema/etcd.go
+++ b/banyand/metadata/schema/etcd.go
@@ -19,6 +19,7 @@ package schema
import (
"context"
+ "fmt"
"sync"
"time"
@@ -29,8 +30,6 @@ import (
"google.golang.org/protobuf/proto"
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
- databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
- propertyv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/property/v1"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/run"
)
@@ -63,19 +62,11 @@ func ConfigureServerEndpoints(url []string) RegistryOption {
}
}
-type eventHandler struct {
- handler EventHandler
- interestKeys Kind
-}
-
-func (eh *eventHandler) interestOf(kind Kind) bool {
- return KindMask&kind&eh.interestKeys != 0
-}
-
type etcdSchemaRegistry struct {
client *clientv3.Client
closer *run.Closer
- handlers []*eventHandler
+ l *logger.Logger
+ watchers []*watcher
mux sync.RWMutex
}
@@ -83,33 +74,19 @@ type etcdSchemaRegistryConfig struct {
serverEndpoints []string
}
-func (e *etcdSchemaRegistry) RegisterHandler(kind Kind, handler EventHandler) {
+func (e *etcdSchemaRegistry) RegisterHandler(name string, kind Kind, handler EventHandler) {
+ // Validate kind
+ if kind&KindMask != kind {
+ panic(fmt.Sprintf("invalid kind %d", kind))
+ }
e.mux.Lock()
defer e.mux.Unlock()
- e.handlers = append(e.handlers, &eventHandler{
- interestKeys: kind,
- handler: handler,
- })
-}
-
-func (e *etcdSchemaRegistry) notifyUpdate(metadata Metadata) {
- e.mux.RLock()
- hh := e.handlers
- e.mux.RUnlock()
- for _, h := range hh {
- if h.interestOf(metadata.Kind) {
- h.handler.OnAddOrUpdate(metadata)
- }
- }
-}
-
-func (e *etcdSchemaRegistry) notifyDelete(metadata Metadata) {
- e.mux.RLock()
- hh := e.handlers
- e.mux.RUnlock()
- for _, h := range hh {
- if h.interestOf(metadata.Kind) {
- h.handler.OnDelete(metadata)
+ for i := 0; i < KindSize; i++ {
+ ki := Kind(1 << i)
+ if kind&ki > 0 {
+ e.l.Info().Str("name", name).Stringer("kind", kind).Msg("registering watcher")
+ w := e.newWatcher(name, ki, handler)
+ e.watchers = append(e.watchers, w)
}
}
}
@@ -117,6 +94,10 @@ func (e *etcdSchemaRegistry) notifyDelete(metadata Metadata) {
func (e *etcdSchemaRegistry) Close() error {
e.closer.Done()
e.closer.CloseThenWait()
+ for i, w := range e.watchers {
+ _ = w.Close()
+ e.watchers[i] = nil
+ }
return e.client.Close()
}
@@ -129,7 +110,8 @@ func NewEtcdSchemaRegistry(options ...RegistryOption) (Registry, error) {
if registryConfig.serverEndpoints == nil {
return nil, errors.New("server address is not set")
}
- zapCfg := logger.GetLogger("etcd").ToZapConfig()
+ log := logger.GetLogger("etcd")
+ zapCfg := log.ToZapConfig()
var l *zap.Logger
var err error
@@ -152,6 +134,7 @@ func NewEtcdSchemaRegistry(options ...RegistryOption) (Registry, error) {
reg := &etcdSchemaRegistry{
client: client,
closer: run.NewCloser(1),
+ l: log,
}
return reg, nil
}
@@ -207,7 +190,7 @@ func (e *etcdSchemaRegistry) update(ctx context.Context, metadata Metadata) erro
}
replace := getResp.Count > 0
if replace {
- existingVal, innerErr := metadata.Unmarshal(getResp.Kvs[0].Value)
+ existingVal, innerErr := metadata.Kind.Unmarshal(getResp.Kvs[0])
if innerErr != nil {
return innerErr
}
@@ -230,7 +213,6 @@ func (e *etcdSchemaRegistry) update(ctx context.Context, metadata Metadata) erro
} else {
return ErrGRPCResourceNotFound
}
- e.notifyUpdate(metadata)
return nil
}
@@ -265,40 +247,29 @@ func (e *etcdSchemaRegistry) create(ctx context.Context, metadata Metadata) erro
if err != nil {
return err
}
-
- e.notifyUpdate(metadata)
return nil
}
-func (e *etcdSchemaRegistry) listWithPrefix(ctx context.Context, prefix string, factory func() proto.Message) ([]proto.Message, error) {
+func (e *etcdSchemaRegistry) listWithPrefix(ctx context.Context, prefix string, kind Kind) ([]proto.Message, error) {
if !e.closer.AddRunning() {
return nil, ErrClosed
}
defer e.closer.Done()
- resp, err := e.client.Get(ctx, prefix, clientv3.WithFromKey(), clientv3.WithRange(incrementLastByte(prefix)))
+ resp, err := e.client.Get(ctx, prefix, clientv3.WithPrefix())
if err != nil {
return nil, err
}
entities := make([]proto.Message, resp.Count)
for i := int64(0); i < resp.Count; i++ {
- message := factory()
- if innerErr := proto.Unmarshal(resp.Kvs[i].Value, message); innerErr != nil {
- return nil, innerErr
- }
- entities[i] = message
- if messageWithMetadata, ok := message.(HasMetadata); ok {
- // Assign readonly fields
- messageWithMetadata.GetMetadata().CreateRevision = resp.Kvs[i].CreateRevision
- messageWithMetadata.GetMetadata().ModRevision = resp.Kvs[i].ModRevision
+ md, err := kind.Unmarshal(resp.Kvs[i])
+ if err != nil {
+ return nil, err
}
+ entities[i] = md.Spec.(proto.Message)
}
return entities, nil
}
-func listPrefixesForEntity(group, entityPrefix string) string {
- return groupsKeyPrefix + group + entityPrefix
-}
-
func (e *etcdSchemaRegistry) delete(ctx context.Context, metadata Metadata) (bool, error) {
if !e.closer.AddRunning() {
return false, ErrClosed
@@ -313,33 +284,6 @@ func (e *etcdSchemaRegistry) delete(ctx context.Context, metadata Metadata) (boo
return false, err
}
if resp.Deleted == 1 {
- var message proto.Message
- switch metadata.Kind {
- case KindMeasure:
- message = &databasev1.Measure{}
- case KindStream:
- message = &databasev1.Stream{}
- case KindIndexRuleBinding:
- message = &databasev1.IndexRuleBinding{}
- case KindIndexRule:
- message = &databasev1.IndexRule{}
- case KindProperty:
- message = &propertyv1.Property{}
- case KindTopNAggregation:
- message = &databasev1.TopNAggregation{}
- default:
- return false, nil
- }
- if unmarshalErr := proto.Unmarshal(resp.PrevKvs[0].Value, message); unmarshalErr == nil {
- e.notifyDelete(Metadata{
- TypeMeta: TypeMeta{
- Kind: metadata.Kind,
- Name: metadata.Name,
- Group: metadata.Group,
- },
- Spec: message,
- })
- }
return true, nil
}
return false, nil
@@ -404,8 +348,16 @@ func (e *etcdSchemaRegistry) register(ctx context.Context, metadata Metadata) er
return nil
}
+func (e *etcdSchemaRegistry) newWatcher(name string, kind Kind, handler EventHandler) *watcher {
+ return newWatcher(e.client, kind, handler, e.l.Named(fmt.Sprintf("watcher-%s[%s]", name, kind.String())))
+}
+
+func listPrefixesForEntity(group, entityPrefix string) string {
+ return entityPrefix + "/" + group
+}
+
func formatKey(entityPrefix string, metadata *commonv1.Metadata) string {
- return groupsKeyPrefix + metadata.GetGroup() + entityPrefix + metadata.GetName()
+ return listPrefixesForEntity(metadata.GetGroup(), entityPrefix) + "/" + metadata.GetName()
}
func incrementLastByte(key string) string {
diff --git a/banyand/metadata/schema/etcd_test.go b/banyand/metadata/schema/etcd_test.go
index 4d9a647c..2ed06549 100644
--- a/banyand/metadata/schema/etcd_test.go
+++ b/banyand/metadata/schema/etcd_test.go
@@ -20,7 +20,6 @@ package schema
import (
"context"
"embed"
- "errors"
"fmt"
"os"
"path"
@@ -378,149 +377,6 @@ func Test_Etcd_Delete(t *testing.T) {
}
}
-func Test_Notify(t *testing.T) {
- req := require.New(t)
- registry, closer := initServerAndRegister(t)
- defer closer()
-
- err := preloadSchema(registry)
- req.NoError(err)
-
- tests := []struct {
- testFunc func(context.Context, Registry) error
- validationFunc func(*mockedEventHandler) bool
- name string
- }{
- {
- name: "modify indexRule",
- testFunc: func(ctx context.Context, r Registry) error {
- ir, err := r.GetIndexRule(ctx, &commonv1.Metadata{
- Name: "db.instance",
- Group: "default",
- })
- if err != nil {
- return err
- }
-
- ir.Type = databasev1.IndexRule_TYPE_TREE
- return r.UpdateIndexRule(ctx, ir)
- },
- validationFunc: func(mocked *mockedEventHandler) bool {
- return mocked.AssertNumberOfCalls(t, "OnAddOrUpdate", 1) &&
- mocked.AssertNumberOfCalls(t, "OnDelete", 0)
- },
- },
- {
- name: "modify indexRule without modification",
- testFunc: func(ctx context.Context, r Registry) error {
- ir, err := r.GetIndexRule(ctx, &commonv1.Metadata{
- Name: "db.instance",
- Group: "default",
- })
- if err != nil {
- return err
- }
-
- return r.UpdateIndexRule(ctx, ir)
- },
- validationFunc: func(mocked *mockedEventHandler) bool {
- return mocked.AssertNumberOfCalls(t, "OnAddOrUpdate", 0) &&
- mocked.AssertNumberOfCalls(t, "OnDelete", 0)
- },
- },
- {
- name: "delete indexRule",
- testFunc: func(ctx context.Context, r Registry) error {
- deleted, err := r.DeleteIndexRule(ctx, &commonv1.Metadata{
- Name: "db.instance",
- Group: "default",
- })
-
- if !deleted {
- return errors.New("fail to delete object")
- }
-
- return err
- },
- validationFunc: func(mocked *mockedEventHandler) bool {
- return mocked.AssertNumberOfCalls(t, "OnAddOrUpdate", 0) &&
- mocked.AssertNumberOfCalls(t, "OnDelete", 1)
- },
- },
- {
- name: "update indexRuleBinding",
- testFunc: func(ctx context.Context, r Registry) error {
- irb, err := r.GetIndexRuleBinding(ctx, &commonv1.Metadata{
- Name: "sw-index-rule-binding",
- Group: "default",
- })
- if err != nil {
- return err
- }
-
- irb.Rules = []string{"trace_id", "duration"}
- return r.UpdateIndexRuleBinding(ctx, irb)
- },
- validationFunc: func(mocked *mockedEventHandler) bool {
- return mocked.AssertNumberOfCalls(t, "OnAddOrUpdate", 1) &&
- mocked.AssertNumberOfCalls(t, "OnDelete", 0)
- },
- },
- {
- name: "update indexRuleBinding without modification",
- testFunc: func(ctx context.Context, r Registry) error {
- irb, err := r.GetIndexRuleBinding(ctx, &commonv1.Metadata{
- Name: "sw-index-rule-binding",
- Group: "default",
- })
- if err != nil {
- return err
- }
-
- return r.UpdateIndexRuleBinding(ctx, irb)
- },
- validationFunc: func(mocked *mockedEventHandler) bool {
- return mocked.AssertNumberOfCalls(t, "OnAddOrUpdate", 0) &&
- mocked.AssertNumberOfCalls(t, "OnDelete", 0)
- },
- },
- {
- name: "delete indexRuleBinding",
- testFunc: func(ctx context.Context, r Registry) error {
- _, err := r.DeleteIndexRuleBinding(ctx, &commonv1.Metadata{
- Name: "sw-index-rule-binding",
- Group: "default",
- })
- if err != nil {
- return err
- }
-
- return nil
- },
- validationFunc: func(mocked *mockedEventHandler) bool {
- return mocked.AssertNumberOfCalls(t, "OnAddOrUpdate", 0) &&
- mocked.AssertNumberOfCalls(t, "OnDelete", 1)
- },
- },
- }
-
- for _, tt := range tests {
- t.Run(tt.name, func(t *testing.T) {
- req := require.New(t)
-
- mockedObj := new(mockedEventHandler)
- mockedObj.On("OnAddOrUpdate", mock.Anything).Return()
- mockedObj.On("OnDelete", mock.Anything).Return()
- registry.RegisterHandler(KindStream|KindIndexRuleBinding|KindIndexRule, mockedObj)
-
- err := tt.testFunc(context.TODO(), registry)
- req.NoError(err)
-
- req.True(tt.validationFunc(mockedObj))
- })
- }
-}
-
func Test_Etcd_Entity_Update(t *testing.T) {
tester := assert.New(t)
registry, closer := initServerAndRegister(t)
diff --git a/banyand/metadata/schema/group.go b/banyand/metadata/schema/group.go
index 584b75a0..f8d1545c 100644
--- a/banyand/metadata/schema/group.go
+++ b/banyand/metadata/schema/group.go
@@ -65,25 +65,23 @@ func (e *etcdSchemaRegistry) ListGroup(ctx context.Context) ([]*commonv1.Group,
}
func (e *etcdSchemaRegistry) DeleteGroup(ctx context.Context, group string) (bool, error) {
- g, err := e.GetGroup(ctx, group)
+ _, err := e.GetGroup(ctx, group)
if err != nil {
return false, errors.Wrap(err, group)
}
- keyPrefix := groupsKeyPrefix + g.GetMetadata().GetName() + "/"
- resp, err := e.client.Delete(ctx, keyPrefix, clientv3.WithRange(incrementLastByte(keyPrefix)))
+ keysToDelete := allKeys()
+ deleteOPs := make([]clientv3.Op, 0, len(keysToDelete)+1)
+ for _, key := range keysToDelete {
+ deleteOPs = append(deleteOPs, clientv3.OpDelete(listPrefixesForEntity(group, key), clientv3.WithPrefix()))
+ }
+ deleteOPs = append(deleteOPs, clientv3.OpDelete(formatGroupKey(group), clientv3.WithPrefix()))
+ txnResponse, err := e.client.Txn(ctx).Then(deleteOPs...).Commit()
if err != nil {
return false, err
}
- if resp.Deleted > 0 {
- e.notifyDelete(Metadata{
- TypeMeta: TypeMeta{
- Kind: KindGroup,
- Name: group,
- },
- Spec: g,
- })
+ if !txnResponse.Succeeded {
+ return false, errConcurrentModification
}
-
return true, nil
}
diff --git a/banyand/metadata/schema/index.go b/banyand/metadata/schema/index.go
index cfda6c6c..41ff541b 100644
--- a/banyand/metadata/schema/index.go
+++ b/banyand/metadata/schema/index.go
@@ -21,7 +21,6 @@ import (
"context"
"hash/crc32"
- "google.golang.org/protobuf/proto"
"google.golang.org/protobuf/types/known/timestamppb"
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
@@ -45,9 +44,7 @@ func (e *etcdSchemaRegistry) ListIndexRuleBinding(ctx context.Context, opt ListO
if opt.Group == "" {
return nil, BadRequest("group", "group should not be empty")
}
- messages, err := e.listWithPrefix(ctx, listPrefixesForEntity(opt.Group, indexRuleBindingKeyPrefix), func() proto.Message {
- return &databasev1.IndexRuleBinding{}
- })
+ messages, err := e.listWithPrefix(ctx, listPrefixesForEntity(opt.Group, indexRuleBindingKeyPrefix), KindIndexRuleBinding)
if err != nil {
return nil, err
}
@@ -108,9 +105,7 @@ func (e *etcdSchemaRegistry) ListIndexRule(ctx context.Context, opt ListOpt) ([]
if opt.Group == "" {
return nil, BadRequest("group", "group should not be empty")
}
- messages, err := e.listWithPrefix(ctx, listPrefixesForEntity(opt.Group, indexRuleKeyPrefix), func() proto.Message {
- return &databasev1.IndexRule{}
- })
+ messages, err := e.listWithPrefix(ctx, listPrefixesForEntity(opt.Group, indexRuleKeyPrefix), KindIndexRule)
if err != nil {
return nil, err
}
diff --git a/banyand/metadata/schema/kind.go b/banyand/metadata/schema/kind.go
new file mode 100644
index 00000000..4f406f86
--- /dev/null
+++ b/banyand/metadata/schema/kind.go
@@ -0,0 +1,162 @@
+// 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 (
+ "io"
+
+ "github.com/pkg/errors"
+ mvccpb "go.etcd.io/etcd/api/v3/mvccpb"
+ "google.golang.org/protobuf/proto"
+
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ propertyv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/property/v1"
+)
+
+// Kind is the type of a resource.
+type Kind int
+
+// KindMask tends to check whether an event is valid.
+const (
+ KindGroup Kind = 1 << iota
+ KindStream
+ KindMeasure
+ KindIndexRuleBinding
+ KindIndexRule
+ KindTopNAggregation
+ KindProperty
+ KindNode
+ KindShard
+ KindMask = KindGroup | KindStream | KindMeasure |
+ KindIndexRuleBinding | KindIndexRule |
+ KindTopNAggregation | KindProperty | KindNode | KindShard
+ KindSize = 9
+)
+
+func (k Kind) key() string {
+ switch k {
+ case KindGroup:
+ return groupsKeyPrefix
+ case KindStream:
+ return streamKeyPrefix
+ case KindMeasure:
+ return measureKeyPrefix
+ case KindIndexRuleBinding:
+ return indexRuleBindingKeyPrefix
+ case KindIndexRule:
+ return indexRuleKeyPrefix
+ case KindTopNAggregation:
+ return topNAggregationKeyPrefix
+ case KindProperty:
+ return propertyKeyPrefix
+ case KindNode:
+ return nodeKeyPrefix
+ case KindShard:
+ return shardKeyPrefix
+ default:
+ return "unknown"
+ }
+}
+
+// Unmarshal encode bytes to proto.Message.
+func (k Kind) Unmarshal(kv *mvccpb.KeyValue) (Metadata, error) {
+ if len(kv.Value) == 0 {
+ return Metadata{}, io.EOF
+ }
+ var m proto.Message
+ switch k {
+ case KindGroup:
+ m = &commonv1.Group{}
+ case KindStream:
+ m = &databasev1.Stream{}
+ case KindMeasure:
+ m = &databasev1.Measure{}
+ case KindIndexRuleBinding:
+ m = &databasev1.IndexRuleBinding{}
+ case KindIndexRule:
+ m = &databasev1.IndexRule{}
+ case KindProperty:
+ m = &propertyv1.Property{}
+ case KindTopNAggregation:
+ m = &databasev1.TopNAggregation{}
+ case KindNode:
+ m = &databasev1.Node{}
+ case KindShard:
+ m = &databasev1.Shard{}
+ default:
+ return Metadata{}, errUnsupportedEntityType
+ }
+ err := proto.Unmarshal(kv.Value, m)
+ if err != nil {
+ return Metadata{}, err
+ }
+ if messageWithMetadata, ok := m.(HasMetadata); ok {
+ if messageWithMetadata.GetMetadata() == nil {
+ return Metadata{}, errors.Errorf("message %s does not have metadata", k.key())
+ }
+ // Assign readonly fields
+ messageWithMetadata.GetMetadata().CreateRevision = kv.CreateRevision
+ messageWithMetadata.GetMetadata().ModRevision = kv.ModRevision
+ return Metadata{
+ TypeMeta: TypeMeta{
+ Kind: k,
+ Name: messageWithMetadata.GetMetadata().GetName(),
+ Group: messageWithMetadata.GetMetadata().GetGroup(),
+ },
+ Spec: messageWithMetadata,
+ }, nil
+ }
+ return Metadata{Spec: m}, nil
+}
+
+func (k Kind) String() string {
+ switch k {
+ case KindGroup:
+ return "group"
+ case KindStream:
+ return "stream"
+ case KindMeasure:
+ return "measure"
+ case KindIndexRuleBinding:
+ return "indexRuleBinding"
+ case KindIndexRule:
+ return "indexRule"
+ case KindTopNAggregation:
+ return "topNAggregation"
+ case KindProperty:
+ return "property"
+ case KindNode:
+ return "node"
+ case KindShard:
+ return "shard"
+ default:
+ return "unknown"
+ }
+}
+
+func allKeys() []string {
+ var keys []string
+ for i := 0; i < KindSize; i++ {
+ ki := Kind(1 << i)
+ if KindMask&ki > 0 {
+ keys = append(keys, ki.key())
+ }
+ }
+ return keys
+}
diff --git a/banyand/metadata/schema/measure.go b/banyand/metadata/schema/measure.go
index 0f4e2f9f..5b0b6f50 100644
--- a/banyand/metadata/schema/measure.go
+++ b/banyand/metadata/schema/measure.go
@@ -21,7 +21,6 @@ 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"
@@ -46,9 +45,7 @@ func (e *etcdSchemaRegistry) ListMeasure(ctx context.Context, opt ListOpt) ([]*d
if opt.Group == "" {
return nil, BadRequest("group", "group should not be empty")
}
- messages, err := e.listWithPrefix(ctx, listPrefixesForEntity(opt.Group, measureKeyPrefix), func() proto.Message {
- return &databasev1.Measure{}
- })
+ messages, err := e.listWithPrefix(ctx, listPrefixesForEntity(opt.Group, measureKeyPrefix), KindMeasure)
if err != nil {
return nil, err
}
diff --git a/banyand/metadata/schema/node.go b/banyand/metadata/schema/node.go
index d6d1fd4f..c383c134 100644
--- a/banyand/metadata/schema/node.go
+++ b/banyand/metadata/schema/node.go
@@ -20,8 +20,6 @@ package schema
import (
"context"
- "google.golang.org/protobuf/proto"
-
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
)
@@ -31,9 +29,7 @@ func (e *etcdSchemaRegistry) ListNode(ctx context.Context, role databasev1.Role)
if role == databasev1.Role_ROLE_UNSPECIFIED {
return nil, BadRequest("group", "group should not be empty")
}
- messages, err := e.listWithPrefix(ctx, nodeKeyPrefix, func() proto.Message {
- return &databasev1.Node{}
- })
+ messages, err := e.listWithPrefix(ctx, nodeKeyPrefix, KindNode)
if err != nil {
return nil, err
}
@@ -54,7 +50,7 @@ func (e *etcdSchemaRegistry) RegisterNode(ctx context.Context, node *databasev1.
return e.register(ctx, Metadata{
TypeMeta: TypeMeta{
Kind: KindNode,
- Name: node.Name,
+ Name: node.Metadata.Name,
},
Spec: node,
})
diff --git a/banyand/metadata/schema/property.go b/banyand/metadata/schema/property.go
index 0333af5e..7d9b1c5d 100644
--- a/banyand/metadata/schema/property.go
+++ b/banyand/metadata/schema/property.go
@@ -21,7 +21,6 @@ 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"
@@ -63,9 +62,7 @@ func (e *etcdSchemaRegistry) ListProperty(ctx context.Context, container *common
if container.Group == "" {
return nil, BadRequest("container.group", "group should not be empty")
}
- messages, err := e.listWithPrefix(ctx, listPrefixesForEntity(container.Group, propertyKeyPrefix+container.Name), func() proto.Message {
- return &propertyv1.Property{}
- })
+ messages, err := e.listWithPrefix(ctx, listPrefixesForEntity(container.Group+"/"+container.Name, propertyKeyPrefix), KindProperty)
if err != nil {
return nil, err
}
diff --git a/banyand/metadata/schema/register_test.go b/banyand/metadata/schema/register_test.go
index 2fd2ace9..28e7162a 100644
--- a/banyand/metadata/schema/register_test.go
+++ b/banyand/metadata/schema/register_test.go
@@ -25,6 +25,7 @@ import (
"github.com/onsi/gomega"
"github.com/onsi/gomega/gleak"
+ 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/embeddedetcd"
"github.com/apache/skywalking-banyandb/pkg/test"
@@ -41,7 +42,11 @@ var _ = ginkgo.Describe("etcd_register", func() {
Name: "test",
Kind: KindNode,
},
- Spec: &databasev1.Node{},
+ Spec: &databasev1.Node{
+ Metadata: &commonv1.Metadata{
+ Name: "test",
+ },
+ },
}
ginkgo.BeforeEach(func() {
goods = gleak.Goroutines()
diff --git a/banyand/metadata/schema/schema.go b/banyand/metadata/schema/schema.go
index fed58409..d5dc0185 100644
--- a/banyand/metadata/schema/schema.go
+++ b/banyand/metadata/schema/schema.go
@@ -32,31 +32,12 @@ import (
var errUnsupportedEntityType = errors.New("unsupported entity type")
-// Kind is the type of a resource.
-type Kind int
-
// EventHandler allows receiving and handling the resource change events.
type EventHandler interface {
OnAddOrUpdate(Metadata)
OnDelete(Metadata)
}
-// KindMask tends to check whether an event is valid.
-const (
- KindGroup Kind = 1 << iota
- KindStream
- KindMeasure
- KindIndexRuleBinding
- KindIndexRule
- KindTopNAggregation
- KindProperty
- KindNode
- KindShard
- KindMask = KindGroup | KindStream | KindMeasure |
- KindIndexRuleBinding | KindIndexRule |
- KindTopNAggregation | KindProperty | KindNode | KindShard
-)
-
// ListOpt contains options to list resources.
type ListOpt struct {
Group string
@@ -74,7 +55,7 @@ type Registry interface {
Property
Node
Shard
- RegisterHandler(Kind, EventHandler)
+ RegisterHandler(string, Kind, EventHandler)
}
// TypeMeta defines the identity and type of an Event.
@@ -93,34 +74,6 @@ type Metadata struct {
// Spec is a placeholder of a serialized resource.
type Spec interface{}
-// Unmarshal encode bytes to proto.Message.
-func (tm TypeMeta) Unmarshal(data []byte) (m proto.Message, err error) {
- switch tm.Kind {
- case KindGroup:
- m = &commonv1.Group{}
- case KindStream:
- m = &databasev1.Stream{}
- case KindMeasure:
- m = &databasev1.Measure{}
- case KindIndexRuleBinding:
- m = &databasev1.IndexRuleBinding{}
- case KindIndexRule:
- m = &databasev1.IndexRule{}
- case KindProperty:
- m = &propertyv1.Property{}
- case KindTopNAggregation:
- m = &databasev1.TopNAggregation{}
- case KindNode:
- m = &databasev1.Node{}
- case KindShard:
- m = &databasev1.Shard{}
- default:
- return nil, errUnsupportedEntityType
- }
- err = proto.Unmarshal(data, m)
- return
-}
-
func (m Metadata) key() (string, error) {
switch m.Kind {
case KindGroup:
@@ -167,13 +120,13 @@ func (m Metadata) key() (string, error) {
}
}
-func (m Metadata) equal(other proto.Message) bool {
- if other == nil {
+func (m Metadata) equal(other Metadata) bool {
+ if other.Spec == nil {
return false
}
if checker, ok := checkerMap[m.Kind]; ok {
- return checker(m.Spec.(proto.Message), other)
+ return checker(m.Spec.(proto.Message), other.Spec.(proto.Message))
}
return false
diff --git a/banyand/metadata/schema/schema_suite_test.go b/banyand/metadata/schema/schema_suite_test.go
index c5f64d3f..383a4f54 100644
--- a/banyand/metadata/schema/schema_suite_test.go
+++ b/banyand/metadata/schema/schema_suite_test.go
@@ -18,23 +18,50 @@
package schema
import (
+ "fmt"
"testing"
"github.com/onsi/ginkgo/v2"
- . "github.com/onsi/gomega"
+ "github.com/onsi/gomega"
+ "github.com/apache/skywalking-banyandb/banyand/metadata/embeddedetcd"
"github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/test"
"github.com/apache/skywalking-banyandb/pkg/test/flags"
)
func TestSchema(t *testing.T) {
- RegisterFailHandler(ginkgo.Fail)
+ gomega.RegisterFailHandler(ginkgo.Fail)
ginkgo.RunSpecs(t, "Schema Suite")
}
+var (
+ server embeddedetcd.Server
+ registry *etcdSchemaRegistry
+)
+
var _ = ginkgo.BeforeSuite(func() {
- Expect(logger.Init(logger.Logging{
+ gomega.Expect(logger.Init(logger.Logging{
Env: "dev",
Level: flags.LogLevel,
- })).To(Succeed())
+ })).To(gomega.Succeed())
+ ports, err := test.AllocateFreePorts(2)
+ if err != nil {
+ panic("fail to find free ports")
+ }
+ endpoints := []string{fmt.Sprintf("http://127.0.0.1:%d", ports[0])}
+ server, err = embeddedetcd.NewServer(
+ embeddedetcd.ConfigureListener(endpoints, []string{fmt.Sprintf("http://127.0.0.1:%d", ports[1])}),
+ embeddedetcd.RootDir(randomTempDir()))
+ gomega.Expect(err).ShouldNot(gomega.HaveOccurred())
+ <-server.ReadyNotify()
+ schemaRegistry, err := NewEtcdSchemaRegistry(ConfigureServerEndpoints(endpoints))
+ gomega.Expect(err).ShouldNot(gomega.HaveOccurred())
+ registry = schemaRegistry.(*etcdSchemaRegistry)
+})
+
+var _ = ginkgo.AfterSuite(func() {
+ registry.Close()
+ server.Close()
+ <-server.StopNotify()
})
diff --git a/banyand/metadata/schema/shard.go b/banyand/metadata/schema/shard.go
index caf8b9de..8f1c7a42 100644
--- a/banyand/metadata/schema/shard.go
+++ b/banyand/metadata/schema/shard.go
@@ -21,7 +21,6 @@ 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"
@@ -58,9 +57,7 @@ func (e *etcdSchemaRegistry) ListShard(ctx context.Context, opt ListOpt) ([]*dat
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{}
- })
+ messages, err := e.listWithPrefix(ctx, listPrefixesForEntity(opt.Group, shardKeyPrefix), KindShard)
if err != nil {
return nil, err
}
diff --git a/banyand/metadata/schema/stream.go b/banyand/metadata/schema/stream.go
index 4f8c8d70..d2e2491f 100644
--- a/banyand/metadata/schema/stream.go
+++ b/banyand/metadata/schema/stream.go
@@ -20,7 +20,6 @@ package schema
import (
"context"
- "google.golang.org/protobuf/proto"
"google.golang.org/protobuf/types/known/timestamppb"
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
@@ -41,9 +40,7 @@ func (e *etcdSchemaRegistry) ListStream(ctx context.Context, opt ListOpt) ([]*da
if opt.Group == "" {
return nil, BadRequest("group", "group should not be empty")
}
- messages, err := e.listWithPrefix(ctx, listPrefixesForEntity(opt.Group, streamKeyPrefix), func() proto.Message {
- return &databasev1.Stream{}
- })
+ messages, err := e.listWithPrefix(ctx, listPrefixesForEntity(opt.Group, streamKeyPrefix), KindStream)
if err != nil {
return nil, err
}
diff --git a/banyand/metadata/schema/topn.go b/banyand/metadata/schema/topn.go
index f1c341f7..858fdf04 100644
--- a/banyand/metadata/schema/topn.go
+++ b/banyand/metadata/schema/topn.go
@@ -20,7 +20,6 @@ package schema
import (
"context"
- "google.golang.org/protobuf/proto"
"google.golang.org/protobuf/types/known/timestamppb"
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
@@ -41,9 +40,7 @@ func (e *etcdSchemaRegistry) ListTopNAggregation(ctx context.Context, opt ListOp
if opt.Group == "" {
return nil, BadRequest("group", "group should not be empty")
}
- messages, err := e.listWithPrefix(ctx, listPrefixesForEntity(opt.Group, topNAggregationKeyPrefix), func() proto.Message {
- return &databasev1.TopNAggregation{}
- })
+ messages, err := e.listWithPrefix(ctx, listPrefixesForEntity(opt.Group, topNAggregationKeyPrefix), KindTopNAggregation)
if err != nil {
return nil, err
}
diff --git a/banyand/metadata/schema/watcher.go b/banyand/metadata/schema/watcher.go
new file mode 100644
index 00000000..3a93828c
--- /dev/null
+++ b/banyand/metadata/schema/watcher.go
@@ -0,0 +1,152 @@
+// 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 (
+ "time"
+
+ mvccpb "go.etcd.io/etcd/api/v3/mvccpb"
+ clientv3 "go.etcd.io/etcd/client/v3"
+
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/run"
+)
+
+type watcher struct {
+ handler EventHandler
+ cli *clientv3.Client
+ closer *run.Closer
+ l *logger.Logger
+ key string
+ kind Kind
+}
+
+func newWatcher(cli *clientv3.Client, kind Kind, handler EventHandler, l *logger.Logger) *watcher {
+ w := &watcher{
+ cli: cli,
+ key: kind.key(),
+ kind: kind,
+ handler: handler,
+ closer: run.NewCloser(1),
+ l: l,
+ }
+ revision := w.allEvents()
+ go w.watch(revision)
+ return w
+}
+
+func (w *watcher) Close() error {
+ w.closer.Done()
+ w.closer.CloseThenWait()
+ return nil
+}
+
+func (w *watcher) allEvents() int64 {
+ cli := w.cli
+ var resp *clientv3.GetResponse
+ for {
+ var err error
+ if resp, err = cli.Get(w.closer.Ctx(), w.key, clientv3.WithPrefix()); err == nil {
+ w.handleAllEvents(resp.Kvs)
+ break
+ }
+ select {
+ case <-w.closer.CloseNotify():
+ return -1
+ case <-time.After(1 * time.Second):
+ }
+ }
+ return resp.Header.Revision
+}
+
+func (w *watcher) watch(revision int64) {
+ if !w.closer.AddRunning() {
+ return
+ }
+ defer w.closer.Done()
+ cli := w.cli
+ for {
+ if revision > 0 {
+ revision = -1
+ } else {
+ revision = w.allEvents()
+ }
+ select {
+ case <-w.closer.CloseNotify():
+ return
+ default:
+ }
+
+ wch := cli.Watch(w.closer.Ctx(), w.key,
+ clientv3.WithPrefix(),
+ clientv3.WithRev(revision+1),
+ clientv3.WithPrevKV(),
+ )
+ if wch == nil {
+ continue
+ }
+ for watchResp := range wch {
+ if err := watchResp.Err(); err != nil {
+ select {
+ case <-w.closer.CloseNotify():
+ return
+ default:
+ continue
+ }
+ }
+ for _, event := range watchResp.Events {
+ select {
+ case <-w.closer.CloseNotify():
+ return
+ default:
+ w.handle(event)
+ }
+ }
+ }
+ }
+}
+
+func (w *watcher) handle(watchEvent *clientv3.Event) {
+ switch watchEvent.Type {
+ case mvccpb.PUT:
+ md, err := w.kind.Unmarshal(watchEvent.Kv)
+ if err != nil {
+ w.l.Error().AnErr("err", err).Msg("failed to unmarshal message")
+ return
+ }
+ w.handler.OnAddOrUpdate(md)
+ case mvccpb.DELETE:
+ md, err := w.kind.Unmarshal(watchEvent.PrevKv)
+ if err != nil {
+ w.l.Error().AnErr("err", err).Msg("failed to unmarshal message")
+ return
+ }
+ w.handler.OnDelete(md)
+ }
+}
+
+func (w *watcher) handleAllEvents(kvs []*mvccpb.KeyValue) {
+ for i := 0; i < len(kvs); i++ {
+ md, err := w.kind.Unmarshal(kvs[i])
+ if err != nil {
+ w.l.Error().AnErr("err", err).Msg("failed to unmarshal message")
+ continue
+ }
+ w.handler.OnAddOrUpdate(md)
+ }
+}
diff --git a/banyand/metadata/schema/watcher_test.go b/banyand/metadata/schema/watcher_test.go
new file mode 100644
index 00000000..04d22194
--- /dev/null
+++ b/banyand/metadata/schema/watcher_test.go
@@ -0,0 +1,149 @@
+// 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"
+ "fmt"
+ "sync"
+ "sync/atomic"
+
+ ginkgo "github.com/onsi/ginkgo/v2"
+ "github.com/onsi/gomega"
+
+ 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/flags"
+)
+
+type mockedHandler struct {
+ data map[string]Metadata
+ addOrUpdateCalledNum *atomic.Int32
+ deleteCalledNum *atomic.Int32
+ sync.RWMutex
+}
+
+func newMockedHandler() *mockedHandler {
+ return &mockedHandler{
+ data: make(map[string]Metadata),
+ addOrUpdateCalledNum: &atomic.Int32{},
+ deleteCalledNum: &atomic.Int32{},
+ }
+}
+
+func (m *mockedHandler) OnAddOrUpdate(obj Metadata) {
+ m.Lock()
+ defer m.Unlock()
+ m.data[obj.Name] = obj
+ m.addOrUpdateCalledNum.Add(1)
+}
+
+func (m *mockedHandler) OnDelete(obj Metadata) {
+ m.Lock()
+ defer m.Unlock()
+ delete(m.data, obj.Name)
+ m.deleteCalledNum.Add(1)
+}
+
+func (m *mockedHandler) Data() map[string]any {
+ m.RLock()
+ defer m.RUnlock()
+ ret := make(map[string]any)
+ for k, v := range m.data {
+ ret[k] = v
+ }
+ return ret
+}
+
+var _ = ginkgo.Describe("Watcher", func() {
+ var (
+ mockedObj *mockedHandler
+ watcher *watcher
+ )
+
+ ginkgo.BeforeEach(func() {
+ mockedObj = newMockedHandler()
+ })
+
+ ginkgo.It("should handle all existing key-value pairs on initial load", func() {
+ // Insert some key-value pairs
+ err := registry.CreateGroup(context.Background(), &commonv1.Group{
+ Metadata: &commonv1.Metadata{
+ Name: "testgroup-measure",
+ },
+ })
+ gomega.Expect(err).NotTo(gomega.HaveOccurred())
+ for i := 0; i < 2; i++ {
+ err = registry.CreateMeasure(context.Background(), &databasev1.Measure{
+ Metadata: &commonv1.Metadata{
+ Name: fmt.Sprintf("testkey%d", i+1),
+ Group: "testgroup-measure",
+ },
+ })
+ gomega.Expect(err).NotTo(gomega.HaveOccurred())
+ }
+
+ // Start the watcher
+ watcher = registry.newWatcher("test", KindMeasure, mockedObj)
+ ginkgo.DeferCleanup(func() {
+ gomega.Expect(watcher.Close()).ShouldNot(gomega.HaveOccurred())
+ })
+ gomega.Eventually(func() bool {
+ _, ok := mockedObj.Data()["testkey1"]
+ if !ok {
+ return false
+ }
+ _, ok = mockedObj.Data()["testkey2"]
+ if !ok {
+ return false
+ }
+ return mockedObj.addOrUpdateCalledNum.Load() == 2
+ }, flags.EventuallyTimeout).Should(gomega.BeTrue())
+ })
+ ginkgo.It("should handle watch events", func() {
+ watcher = registry.newWatcher("test", KindStream, mockedObj)
+ ginkgo.DeferCleanup(func() {
+ gomega.Expect(watcher.Close()).ShouldNot(gomega.HaveOccurred())
+ })
+ err := registry.CreateGroup(context.Background(), &commonv1.Group{
+ Metadata: &commonv1.Metadata{
+ Name: "testgroup-stream",
+ },
+ })
+ gomega.Expect(err).NotTo(gomega.HaveOccurred())
+ err = registry.CreateStream(context.Background(), &databasev1.Stream{
+ Metadata: &commonv1.Metadata{
+ Name: "testkey",
+ Group: "testgroup-stream",
+ },
+ })
+ gomega.Expect(err).NotTo(gomega.HaveOccurred())
+ ok, err := registry.DeleteStream(context.Background(), &commonv1.Metadata{
+ Name: "testkey",
+ Group: "testgroup-stream",
+ })
+ gomega.Expect(err).NotTo(gomega.HaveOccurred())
+ gomega.Expect(ok).To(gomega.BeTrue())
+
+ gomega.Eventually(func() bool {
+ return mockedObj.addOrUpdateCalledNum.Load() == 1 &&
+ mockedObj.deleteCalledNum.Load() == 1 &&
+ len(mockedObj.Data()) == 0
+ }, flags.EventuallyTimeout).Should(gomega.BeTrue())
+ })
+})
diff --git a/banyand/stream/metadata.go b/banyand/stream/metadata.go
index d69710b1..e349872e 100644
--- a/banyand/stream/metadata.go
+++ b/banyand/stream/metadata.go
@@ -85,7 +85,6 @@ func (sr *schemaRepo) OnAddOrUpdate(m schema.Metadata) {
})
cancel()
if err != nil {
- sr.l.Error().Err(err).Msg("fail to get subject")
return
}
sr.SendMetadataEvent(resourceSchema.MetadataEvent{
@@ -99,7 +98,6 @@ func (sr *schemaRepo) OnAddOrUpdate(m schema.Metadata) {
subjects, err := sr.metadata.Subjects(ctx, m.Spec.(*databasev1.IndexRule), commonv1.Catalog_CATALOG_STREAM)
cancel()
if err != nil {
- sr.l.Error().Err(err).Msg("fail to get subjects(stream)")
return
}
for _, sub := range subjects {
@@ -140,7 +138,6 @@ func (sr *schemaRepo) OnDelete(m schema.Metadata) {
})
cancel()
if err != nil {
- sr.l.Error().Err(err).Msg("fail to get subject")
return
}
sr.SendMetadataEvent(resourceSchema.MetadataEvent{
diff --git a/banyand/stream/service.go b/banyand/stream/service.go
index f614b9e9..b9fd39b3 100644
--- a/banyand/stream/service.go
+++ b/banyand/stream/service.go
@@ -20,7 +20,6 @@ package stream
import (
"context"
"path"
- "time"
"github.com/pkg/errors"
@@ -98,40 +97,14 @@ func (s *service) Role() databasev1.Role {
return databasev1.Role_ROLE_DATA
}
-func (s *service) PreRun(ctx context.Context) error {
+func (s *service) PreRun(_ context.Context) error {
s.l = logger.GetLogger(s.Name())
- ctxLocal, cancel := context.WithTimeout(ctx, 5*time.Second)
- groups, err := s.metadata.GroupRegistry().ListGroup(ctxLocal)
- cancel()
- if err != nil {
- return err
- }
path := path.Join(s.root, s.Name())
observability.UpdatePath(path)
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
- }
- gp, err := s.schemaRepo.StoreGroup(g.Metadata)
- if err != nil {
- return err
- }
- ctxLocal, cancel := context.WithTimeout(ctx, 5*time.Second)
- schemas, err := s.metadata.StreamRegistry().ListStream(ctxLocal, schema.ListOpt{Group: gp.GetSchema().GetMetadata().Name})
- cancel()
- if err != nil {
- return err
- }
- for _, sa := range schemas {
- if _, innerErr := gp.StoreResource(ctx, sa); innerErr != nil {
- return innerErr
- }
- }
- }
// run a serial watcher
s.schemaRepo.Watcher()
- s.metadata.RegisterHandler(schema.KindGroup|schema.KindStream|schema.KindIndexRuleBinding|schema.KindIndexRule,
+ s.metadata.RegisterHandler("stream", schema.KindGroup|schema.KindStream|schema.KindIndexRuleBinding|schema.KindIndexRule,
&s.schemaRepo)
s.writeListener = setUpWriteCallback(s.l, &s.schemaRepo)
diff --git a/docs/api-reference.md b/docs/api-reference.md
index 45fa7b25..956af736 100644
--- a/docs/api-reference.md
+++ b/docs/api-reference.md
@@ -335,7 +335,7 @@ Metadata is for multi-tenant, multi-model use
| Field | Type | Label | Description |
| ----- | ---- | ----- | ----------- |
-| name | [string](#string) | | |
+| metadata | [banyandb.common.v1.Metadata](#banyandb-common-v1-Metadata) | | |
| roles | [Role](#banyandb-database-v1-Role) | repeated | |
| grpc_address | [string](#string) | | |
| http_address | [string](#string) | | |
diff --git a/go.mod b/go.mod
index 7cf87065..2a013209 100644
--- a/go.mod
+++ b/go.mod
@@ -118,7 +118,7 @@ require (
github.com/xiang90/probing v0.0.0-20221125231312-a49e3df8f510 // indirect
github.com/yusufpapurcu/wmi v1.2.2 // indirect
go.etcd.io/bbolt v1.3.7 // indirect
- go.etcd.io/etcd/api/v3 v3.5.9 // indirect
+ go.etcd.io/etcd/api/v3 v3.5.9
go.etcd.io/etcd/client/pkg/v3 v3.5.9 // indirect
go.etcd.io/etcd/client/v2 v2.305.9 // indirect
go.etcd.io/etcd/pkg/v3 v3.5.9 // indirect
diff --git a/pkg/logger/logger.go b/pkg/logger/logger.go
index 64dbff37..ee8f8561 100644
--- a/pkg/logger/logger.go
+++ b/pkg/logger/logger.go
@@ -96,7 +96,7 @@ func (l *Logger) ToZapConfig() zap.Config {
}
if !l.development {
config := zap.NewProductionConfig()
- config.Level = level
+ config.Level = zap.NewAtomicLevelAt(zap.ErrorLevel)
return config
}
encoderConfig := zapcore.EncoderConfig{
diff --git a/pkg/run/closer.go b/pkg/run/closer.go
index 5f718ee4..5ada22af 100644
--- a/pkg/run/closer.go
+++ b/pkg/run/closer.go
@@ -55,6 +55,14 @@ func (c *Closer) AddRunning() bool {
return true
}
+// Ctx returns the context of the Closer.
+func (c *Closer) Ctx() context.Context {
+ if c == nil {
+ return context.Background()
+ }
+ return c.ctx
+}
+
// CloseNotify receives a signal from Close.
func (c *Closer) CloseNotify() <-chan struct{} {
if c == nil {