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 2022/02/24 01:23:53 UTC
[skywalking-banyandb] 01/01: Move metadata event handler to th pkg package for reusing it in the measure
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch stream-metadata
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 8df1cab4f2510952c3a669ccf88f49a24b325356
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Thu Feb 24 01:21:48 2022 +0000
Move metadata event handler to th pkg package for
reusing it in the measure
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
banyand/stream/metadata.go | 501 +++++++--------------------------------
banyand/stream/metadata_test.go | 8 +-
banyand/stream/service.go | 12 +-
banyand/stream/stream.go | 20 ++
pkg/pb/v1/metadata.go | 10 +
pkg/schema/metadata.go | 507 ++++++++++++++++++++++++++++++++++++++++
6 files changed, 633 insertions(+), 425 deletions(-)
diff --git a/banyand/stream/metadata.go b/banyand/stream/metadata.go
index b58c066..eb150f9 100644
--- a/banyand/stream/metadata.go
+++ b/banyand/stream/metadata.go
@@ -19,14 +19,8 @@ package stream
import (
"context"
- "sync"
- "sync/atomic"
"time"
- "github.com/pkg/errors"
- "go.uber.org/multierr"
- "google.golang.org/protobuf/types/known/timestamppb"
-
"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"
@@ -34,63 +28,50 @@ import (
"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/encoding"
"github.com/apache/skywalking-banyandb/pkg/logger"
+ resourceSchema "github.com/apache/skywalking-banyandb/pkg/schema"
)
-type eventType uint8
-
-const (
- eventAddOrUpdate eventType = iota
- eventDelete
-)
-
-type metadataEvent struct {
- typ eventType
- kind schema.Kind
- metadata *commonv1.Metadata
-}
-
type schemaRepo struct {
- sync.RWMutex
- path string
- metadata metadata.Repo
- repo discovery.ServiceRepo
+ resourceSchema.Repository
l *logger.Logger
- data map[string]*group
-
- // stop channel for the inner worker
- workerStopCh chan struct{}
- eventCh chan metadataEvent
+ metadata metadata.Repo
}
func newSchemaRepo(path string, metadata metadata.Repo, repo discovery.ServiceRepo, l *logger.Logger) schemaRepo {
return schemaRepo{
- path: path,
- metadata: metadata,
- repo: repo,
- l: l,
- data: make(map[string]*group),
- eventCh: make(chan metadataEvent),
- workerStopCh: make(chan struct{}),
+ l: l,
+ metadata: metadata,
+ Repository: resourceSchema.NewRepository(
+ metadata,
+ repo,
+ l,
+ newSupplier(path, metadata, l),
+ event.StreamTopicShardEvent,
+ event.StreamTopicEntityEvent,
+ ),
}
}
func (sr *schemaRepo) OnAddOrUpdate(m schema.Metadata) {
switch m.Kind {
case schema.KindGroup:
- sr.eventCh <- metadataEvent{
- typ: eventAddOrUpdate,
- kind: schema.KindGroup,
- metadata: m.Spec.(*commonv1.Group).GetMetadata(),
+ g := m.Spec.(*commonv1.Group)
+ if g.Catalog != commonv1.Catalog_CATALOG_STREAM {
+ return
}
+ sr.SendMetadataEvent(resourceSchema.MetadataEvent{
+ Typ: resourceSchema.EventAddOrUpdate,
+ Kind: resourceSchema.EventKindGroup,
+ Metadata: g.GetMetadata(),
+ })
case schema.KindStream:
- sr.eventCh <- metadataEvent{
- typ: eventAddOrUpdate,
- kind: schema.KindStream,
- metadata: m.Spec.(*databasev1.Stream).GetMetadata(),
- }
+ sr.SendMetadataEvent(resourceSchema.MetadataEvent{
+ Typ: resourceSchema.EventAddOrUpdate,
+ Kind: resourceSchema.EventKindResource,
+ Metadata: m.Spec.(*databasev1.Stream).GetMetadata(),
+ })
case schema.KindIndexRuleBinding:
irb, ok := m.Spec.(*databasev1.IndexRuleBinding)
if !ok {
@@ -108,11 +89,11 @@ func (sr *schemaRepo) OnAddOrUpdate(m schema.Metadata) {
sr.l.Error().Err(err).Msg("fail to get subject")
return
}
- sr.eventCh <- metadataEvent{
- typ: eventAddOrUpdate,
- kind: schema.KindStream,
- metadata: stm.GetMetadata(),
- }
+ sr.SendMetadataEvent(resourceSchema.MetadataEvent{
+ Typ: resourceSchema.EventAddOrUpdate,
+ Kind: resourceSchema.EventKindResource,
+ Metadata: stm.GetMetadata(),
+ })
}
case schema.KindIndexRule:
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
@@ -123,11 +104,11 @@ func (sr *schemaRepo) OnAddOrUpdate(m schema.Metadata) {
return
}
for _, sub := range subjects {
- sr.eventCh <- metadataEvent{
- typ: eventAddOrUpdate,
- kind: schema.KindStream,
- metadata: sub.(*databasev1.Stream).GetMetadata(),
- }
+ sr.SendMetadataEvent(resourceSchema.MetadataEvent{
+ Typ: resourceSchema.EventAddOrUpdate,
+ Kind: resourceSchema.EventKindResource,
+ Metadata: sub.(*databasev1.Stream).GetMetadata(),
+ })
}
default:
}
@@ -136,17 +117,21 @@ func (sr *schemaRepo) OnAddOrUpdate(m schema.Metadata) {
func (sr *schemaRepo) OnDelete(m schema.Metadata) {
switch m.Kind {
case schema.KindGroup:
- sr.eventCh <- metadataEvent{
- typ: eventDelete,
- kind: schema.KindGroup,
- metadata: m.Spec.(*commonv1.Group).GetMetadata(),
+ g := m.Spec.(*commonv1.Group)
+ if g.Catalog != commonv1.Catalog_CATALOG_STREAM {
+ return
}
+ sr.SendMetadataEvent(resourceSchema.MetadataEvent{
+ Typ: resourceSchema.EventDelete,
+ Kind: resourceSchema.EventKindGroup,
+ Metadata: g.GetMetadata(),
+ })
case schema.KindStream:
- sr.eventCh <- metadataEvent{
- typ: eventDelete,
- kind: schema.KindStream,
- metadata: m.Spec.(*databasev1.Stream).GetMetadata(),
- }
+ sr.SendMetadataEvent(resourceSchema.MetadataEvent{
+ Typ: resourceSchema.EventDelete,
+ Kind: resourceSchema.EventKindResource,
+ Metadata: m.Spec.(*databasev1.Stream).GetMetadata(),
+ })
case schema.KindIndexRuleBinding:
if m.Spec.(*databasev1.IndexRuleBinding).GetSubject().Catalog == commonv1.Catalog_CATALOG_STREAM {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
@@ -159,378 +144,64 @@ func (sr *schemaRepo) OnDelete(m schema.Metadata) {
sr.l.Error().Err(err).Msg("fail to get subject")
return
}
- sr.eventCh <- metadataEvent{
- typ: eventDelete,
- kind: schema.KindStream,
- metadata: stm.GetMetadata(),
- }
+ sr.SendMetadataEvent(resourceSchema.MetadataEvent{
+ Typ: resourceSchema.EventDelete,
+ Kind: resourceSchema.EventKindResource,
+ Metadata: stm.GetMetadata(),
+ })
}
case schema.KindIndexRule:
default:
}
}
-func (sr *schemaRepo) watcher() {
- defer func() {
- if err := recover(); err != nil {
- sr.l.Warn().Interface("err", err).Msg("watching the events")
- }
- }()
- for {
- select {
- case evt := <-sr.eventCh:
- var err error
- switch evt.typ {
- case eventAddOrUpdate:
- switch evt.kind {
- case schema.KindGroup:
- _, err = sr.storeGroup(evt.metadata)
- case schema.KindStream:
- _, err = sr.storeStream(evt.metadata)
- }
- case eventDelete:
- switch evt.kind {
- case schema.KindGroup:
- err = sr.deleteGroup(evt.metadata)
- case schema.KindStream:
- err = sr.deleteStream(evt.metadata)
- }
- }
- if err != nil {
- sr.l.Err(err).Interface("event", evt).Msg("fail to handle the metadata event. retry...")
- sr.eventCh <- evt
- }
- case <-sr.workerStopCh:
- return
- }
- }
-}
-
-func (sr *schemaRepo) storeGroup(groupMeta *commonv1.Metadata) (*group, error) {
- ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
- groupSchema, err := sr.metadata.GroupRegistry().GetGroup(ctx, groupMeta.GetName())
- cancel()
- if err != nil {
- return nil, err
- }
- name := groupSchema.GetMetadata().GetName()
- sr.Lock()
- defer sr.Unlock()
- g, ok := sr.getGroup(name)
- if !ok {
- sr.l.Info().Str("group", name).Msg("creating a tsdb")
- var db tsdb.Database
- db, err = sr.openDB(groupSchema)
- if err != nil {
- return nil, err
- }
- g = newGroup(groupSchema, sr.repo, sr.metadata, db, sr.l)
- sr.data[name] = g
- return g, sr.notify(groupSchema, databasev1.Action_ACTION_PUT)
- }
- prevGroupSchema := g.groupSchema
- if groupSchema.GetMetadata().GetModRevision() <= prevGroupSchema.Metadata.ModRevision {
- return g, nil
- }
- 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.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 = groupSchema
- return g, nil
-}
-
-func (sr *schemaRepo) deleteGroup(groupMeta *commonv1.Metadata) error {
- name := groupMeta.GetName()
- sr.Lock()
- defer sr.Unlock()
- var ok bool
- g, ok := sr.getGroup(name)
- if !ok {
- return nil
- }
- err := g.close()
- if err != nil {
- return err
- }
- _ = sr.notify(g.groupSchema, databasev1.Action_ACTION_DELETE)
- delete(sr.data, name)
- return nil
-}
-
-func (sr *schemaRepo) openDB(groupSchema *commonv1.Group) (tsdb.Database, error) {
- return tsdb.OpenDatabase(
- context.TODO(),
- tsdb.DatabaseOpts{
- Location: sr.path,
- ShardNum: groupSchema.ResourceOpts.ShardNum,
- EncodingMethod: tsdb.EncodingMethod{
- EncoderPool: encoding.NewPlainEncoderPool(chunkSize),
- DecoderPool: encoding.NewPlainDecoderPool(chunkSize),
- },
- })
-}
-
-func (sr *schemaRepo) getGroup(name string) (*group, bool) {
- g := sr.data[name]
- if g == nil {
- return nil, false
- }
- return g, true
-}
-
-func (sr *schemaRepo) loadGroup(name string) (*group, bool) {
- sr.RLock()
- defer sr.RUnlock()
- return sr.getGroup(name)
-}
-
func (sr *schemaRepo) loadStream(metadata *commonv1.Metadata) (*stream, bool) {
- g, ok := sr.loadGroup(metadata.Group)
+ r, ok := sr.LoadResource(metadata)
if !ok {
return nil, false
}
- return g.loadStream(metadata.Name)
+ s, ok := r.(*stream)
+ return s, ok
}
-func (sr *schemaRepo) storeStream(metadata *commonv1.Metadata) (*stream, error) {
- group, ok := sr.loadGroup(metadata.Group)
- if !ok {
- return nil, errors.Errorf("unknown group")
- }
- ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
- stm, err := sr.metadata.StreamRegistry().GetStream(ctx, metadata)
- cancel()
- if err != nil {
- return nil, errors.Errorf("fails to get the stream")
- }
- return group.storeStream(stm)
-}
-
-func (sr *schemaRepo) deleteStream(metadata *commonv1.Metadata) error {
- group, ok := sr.loadGroup(metadata.Group)
- if !ok {
- return nil
- }
- return group.deleteStream(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(event.StreamTopicShardEvent, 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) {
- for _, g := range sr.getMap() {
- err = multierr.Append(err, sr.notify(g.groupSchema, databasev1.Action_ACTION_PUT))
- for _, s := range g.getMap() {
- err = multierr.Append(err, g.notify(s, databasev1.Action_ACTION_PUT))
- }
- }
- return err
-}
+var _ resourceSchema.ResourceSupplier = (*supplier)(nil)
-func (sr *schemaRepo) close() {
- defer func() {
- if err := recover(); err != nil {
- sr.l.Warn().Interface("err", err).Msg("closing stream")
- }
- }()
- if sr.eventCh != nil {
- close(sr.eventCh)
- }
- if sr.workerStopCh != nil {
- close(sr.workerStopCh)
- }
- for _, g := range sr.getMap() {
- err := g.close()
- if err != nil {
- sr.l.Err(err).Stringer("group", g.groupSchema.Metadata).Msg("closing")
- }
- }
-}
-
-func (sr *schemaRepo) getMap() map[string]*group {
- sr.RLock()
- defer sr.RUnlock()
- return sr.data
-}
-
-type group struct {
- groupSchema *commonv1.Group
- l *logger.Logger
- repo discovery.ServiceRepo
- metadata metadata.Repo
- db atomic.Value
- mapMutex sync.RWMutex
- schemaMap map[string]*stream
+type supplier struct {
+ path string
+ metadata metadata.Repo
+ l *logger.Logger
}
-func newGroup(groupSchema *commonv1.Group, repo discovery.ServiceRepo, metadata metadata.Repo, db tsdb.Database, l *logger.Logger) *group {
- g := &group{
- groupSchema: groupSchema,
- repo: repo,
- metadata: metadata,
- l: l,
- schemaMap: make(map[string]*stream),
+func newSupplier(path string, metadata metadata.Repo, l *logger.Logger) *supplier {
+ return &supplier{
+ path: path,
+ metadata: metadata,
+ l: l,
}
- g.db.Store(db)
- return g
-}
-
-func (g *group) SupplyTSDB() tsdb.Database {
- return g.db.Load().(tsdb.Database)
-}
-
-func (g *group) setDB(db tsdb.Database) {
- g.db.Store(db)
}
-func (g *group) storeStream(streamSchema *databasev1.Stream) (*stream, error) {
- g.mapMutex.Lock()
- defer g.mapMutex.Unlock()
- key := streamSchema.GetMetadata().GetName()
- preStream := g.schemaMap[key]
- if preStream != nil &&
- streamSchema.GetMetadata().GetModRevision() <= preStream.schema.GetMetadata().GetModRevision() {
- // we only need to check the max modifications revision observed for index rules
- ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
- idxRules, errIndexRules := g.metadata.IndexRules(ctx, streamSchema.GetMetadata())
- cancel()
- if errIndexRules != nil {
- return nil, errIndexRules
- }
- if len(idxRules) == len(preStream.indexRules) {
- maxModRevision := parseMaxModRevision(idxRules)
- if preStream.maxObservedModRevision >= maxModRevision {
- return preStream, nil
- }
- }
- }
- ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
- idxRules, errIndexRules := g.metadata.IndexRules(ctx, streamSchema.GetMetadata())
- cancel()
- if errIndexRules != nil {
- return nil, errIndexRules
- }
- sm, errTS := openStream(g.groupSchema.GetResourceOpts().ShardNum, g, streamSpec{
+func (s *supplier) OpenResource(shardNum uint32, db tsdb.Supplier, spec resourceSchema.ResourceSpec) (resourceSchema.Resource, error) {
+ streamSchema := spec.Schema.(*databasev1.Stream)
+ return openStream(shardNum, db, streamSpec{
schema: streamSchema,
- indexRules: idxRules,
- }, g.l)
- 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 preStream != nil {
- _ = preStream.Close()
- }
- return sm, nil
-}
-
-func (g *group) deleteStream(streamMetadata *commonv1.Metadata) error {
- g.mapMutex.Lock()
- defer g.mapMutex.Unlock()
- key := streamMetadata.GetName()
- preStream := g.schemaMap[key]
- if preStream == nil {
- return nil
- }
- if err := g.notify(preStream, databasev1.Action_ACTION_DELETE); err != nil {
- return err
- }
- delete(g.schemaMap, key)
- _ = preStream.Close()
- return nil
+ indexRules: spec.IndexRules,
+ }, s.l)
}
-
-func (g *group) loadStream(name string) (*stream, bool) {
- data := g.getMap()
- s := data[name]
- if s == nil {
- return nil, false
- }
- return s, true
+func (s *supplier) ResourceSchema(repo metadata.Repo, md *commonv1.Metadata) (resourceSchema.ResourceSchema, error) {
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ return s.metadata.StreamRegistry().GetStream(ctx, md)
}
-func (g *group) notify(stream *stream, action databasev1.Action) error {
- now := time.Now()
- nowPb := timestamppb.New(now)
- locator := make([]*databasev1.EntityEvent_TagLocator, 0, len(stream.entityLocator))
- for _, tagLocator := range stream.entityLocator {
- locator = append(locator, &databasev1.EntityEvent_TagLocator{
- FamilyOffset: uint32(tagLocator.FamilyOffset),
- TagOffset: uint32(tagLocator.TagOffset),
+func (s *supplier) OpenDB(groupSchema *commonv1.Group) (tsdb.Database, error) {
+ return tsdb.OpenDatabase(
+ context.TODO(),
+ tsdb.DatabaseOpts{
+ Location: s.path,
+ ShardNum: groupSchema.ResourceOpts.ShardNum,
+ EncodingMethod: tsdb.EncodingMethod{
+ EncoderPool: encoding.NewPlainEncoderPool(chunkSize),
+ DecoderPool: encoding.NewPlainDecoderPool(chunkSize),
+ },
})
- }
- _, err := g.repo.Publish(event.StreamTopicEntityEvent, bus.NewMessage(bus.MessageID(now.UnixNano()), &databasev1.EntityEvent{
- Subject: &commonv1.Metadata{
- Name: stream.name,
- Group: stream.group,
- },
- EntityLocator: locator,
- Time: nowPb,
- Action: action,
- }))
- if errors.Is(err, bus.ErrTopicNotExist) {
- return nil
- }
- return err
-}
-
-func (g *group) close() (err error) {
- for _, s := range g.getMap() {
- err = multierr.Append(err, s.Close())
- }
- return multierr.Append(err, g.SupplyTSDB().Close())
-}
-
-func (g *group) getMap() map[string]*stream {
- g.mapMutex.RLock()
- defer g.mapMutex.RUnlock()
- return g.schemaMap
}
diff --git a/banyand/stream/metadata_test.go b/banyand/stream/metadata_test.go
index 93db229..216bbd4 100644
--- a/banyand/stream/metadata_test.go
+++ b/banyand/stream/metadata_test.go
@@ -44,7 +44,7 @@ var _ = Describe("Metadata", func() {
Context("Manage group", func() {
It("should pass smoke test", func() {
Eventually(func() bool {
- _, ok := svcs.stream.schemaRepo.loadGroup("default")
+ _, ok := svcs.stream.schemaRepo.LoadGroup("default")
return ok
}).WithTimeout(10 * time.Second).Should(BeTrue())
})
@@ -54,7 +54,7 @@ var _ = Describe("Metadata", func() {
Expect(err).ShouldNot(HaveOccurred())
Expect(deleted).Should(BeTrue())
Eventually(func() bool {
- _, ok := svcs.stream.schemaRepo.loadGroup("default")
+ _, ok := svcs.stream.schemaRepo.LoadGroup("default")
return ok
}).WithTimeout(10 * time.Second).Should(BeFalse())
})
@@ -70,11 +70,11 @@ var _ = Describe("Metadata", func() {
Expect(svcs.metadataService.GroupRegistry().UpdateGroup(context.TODO(), groupSchema)).Should(Succeed())
Eventually(func() bool {
- group, ok := svcs.stream.schemaRepo.loadGroup("default")
+ group, ok := svcs.stream.schemaRepo.LoadGroup("default")
if !ok {
return false
}
- return group.groupSchema.GetResourceOpts().GetShardNum() == 4
+ return group.GetSchema().GetResourceOpts().GetShardNum() == 4
}).WithTimeout(10 * time.Second).Should(BeTrue())
})
})
diff --git a/banyand/stream/service.go b/banyand/stream/service.go
index ec95ba4..eab1f34 100644
--- a/banyand/stream/service.go
+++ b/banyand/stream/service.go
@@ -97,18 +97,18 @@ func (s *service) PreRun() error {
if g.Catalog != commonv1.Catalog_CATALOG_STREAM {
continue
}
- gp, err := s.schemaRepo.storeGroup(g.Metadata)
+ gp, err := s.schemaRepo.StoreGroup(g.Metadata)
if err != nil {
return err
}
ctx, cancel = context.WithTimeout(context.Background(), 5*time.Second)
- schemas, err := s.metadata.StreamRegistry().ListStream(ctx, schema.ListOpt{Group: gp.groupSchema.GetMetadata().Name})
+ schemas, err := s.metadata.StreamRegistry().ListStream(ctx, schema.ListOpt{Group: gp.GetSchema().GetMetadata().Name})
cancel()
if err != nil {
return err
}
for _, sa := range schemas {
- if _, innerErr := gp.storeStream(sa); innerErr != nil {
+ if _, innerErr := gp.StoreResource(sa); innerErr != nil {
return innerErr
}
}
@@ -124,9 +124,9 @@ func (s *service) PreRun() error {
}
func (s *service) Serve() run.StopNotify {
- _ = s.schemaRepo.notifyAll()
+ _ = s.schemaRepo.NotifyAll()
// run a serial watcher
- go s.schemaRepo.watcher()
+ go s.schemaRepo.Watcher()
s.metadata.StreamRegistry().RegisterHandler(schema.KindGroup|schema.KindStream|schema.KindIndexRuleBinding|schema.KindIndexRule,
&s.schemaRepo)
@@ -136,7 +136,7 @@ func (s *service) Serve() run.StopNotify {
}
func (s *service) GracefulStop() {
- s.schemaRepo.close()
+ s.schemaRepo.Close()
if s.stopCh != nil {
close(s.stopCh)
}
diff --git a/banyand/stream/stream.go b/banyand/stream/stream.go
index defe201..2856938 100644
--- a/banyand/stream/stream.go
+++ b/banyand/stream/stream.go
@@ -20,16 +20,20 @@ package stream
import (
"context"
+ 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/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"
)
// a chunk is 1MB
const chunkSize = 1 << 20
+var _ schema.Resource = (*stream)(nil)
+
type stream struct {
name string
group string
@@ -45,6 +49,22 @@ type stream struct {
indexWriter *index.Writer
}
+func (s *stream) GetMetadata() *commonv1.Metadata {
+ return s.schema.Metadata
+}
+
+func (s *stream) GetIndexRules() []*databasev1.IndexRule {
+ return s.indexRules
+}
+
+func (s *stream) MaxObservedModRevision() int64 {
+ return s.maxObservedModRevision
+}
+
+func (s *stream) EntityLocator() partition.EntityLocator {
+ return s.entityLocator
+}
+
func (s *stream) Close() error {
return s.indexWriter.Close()
}
diff --git a/pkg/pb/v1/metadata.go b/pkg/pb/v1/metadata.go
index 61f93c2..7d36bf3 100644
--- a/pkg/pb/v1/metadata.go
+++ b/pkg/pb/v1/metadata.go
@@ -64,3 +64,13 @@ func FieldValueTypeConv(tagValue *modelv1.FieldValue) (tagType databasev1.FieldT
}
return databasev1.FieldType_FIELD_TYPE_UNSPECIFIED, false
}
+
+func ParseMaxModRevision(indexRules []*databasev1.IndexRule) (maxRevisionForIdxRules int64) {
+ maxRevisionForIdxRules = int64(0)
+ for _, idxRule := range indexRules {
+ if idxRule.GetMetadata().GetModRevision() > maxRevisionForIdxRules {
+ maxRevisionForIdxRules = idxRule.GetMetadata().GetModRevision()
+ }
+ }
+ return
+}
diff --git a/pkg/schema/metadata.go b/pkg/schema/metadata.go
new file mode 100644
index 0000000..f9d0749
--- /dev/null
+++ b/pkg/schema/metadata.go
@@ -0,0 +1,507 @@
+// 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"
+ "io"
+ "sync"
+ "sync/atomic"
+ "time"
+
+ "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/tsdb"
+ "github.com/apache/skywalking-banyandb/pkg/bus"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/partition"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+)
+
+type EventType uint8
+
+const (
+ EventAddOrUpdate EventType = iota
+ EventDelete
+)
+
+type EventKind uint8
+
+const (
+ EventKindGroup EventKind = iota
+ EventKindResource
+)
+
+type Group interface {
+ GetSchema() *commonv1.Group
+ StoreResource(resourceSchema ResourceSchema) (Resource, error)
+ LoadResource(name string) (Resource, bool)
+}
+
+type MetadataEvent struct {
+ Typ EventType
+ Kind EventKind
+ Metadata *commonv1.Metadata
+}
+
+type ResourceSchema interface {
+ GetMetadata() *commonv1.Metadata
+}
+
+type ResourceSpec struct {
+ Schema ResourceSchema
+ IndexRules []*databasev1.IndexRule
+}
+
+type Resource interface {
+ GetIndexRules() []*databasev1.IndexRule
+ MaxObservedModRevision() int64
+ EntityLocator() partition.EntityLocator
+ ResourceSchema
+ io.Closer
+}
+
+type ResourceSupplier interface {
+ OpenResource(shardNum uint32, db tsdb.Supplier, spec ResourceSpec) (Resource, error)
+ ResourceSchema(repo metadata.Repo, metdata *commonv1.Metadata) (ResourceSchema, error)
+ OpenDB(groupSchema *commonv1.Group) (tsdb.Database, error)
+}
+
+type Repository interface {
+ Watcher()
+ SendMetadataEvent(MetadataEvent)
+ StoreGroup(groupMeta *commonv1.Metadata) (*group, error)
+ LoadGroup(name string) (Group, bool)
+ LoadResource(metadata *commonv1.Metadata) (Resource, bool)
+ NotifyAll() (err error)
+ Close()
+}
+
+var _ Repository = (*schemaRepo)(nil)
+
+type schemaRepo struct {
+ sync.RWMutex
+ metadata metadata.Repo
+ repo discovery.ServiceRepo
+ l *logger.Logger
+ resourceSupplier ResourceSupplier
+ shardTopic bus.Topic
+ entityTopic bus.Topic
+ data map[string]*group
+
+ // stop channel for the inner worker
+ workerStopCh chan struct{}
+ eventCh chan MetadataEvent
+}
+
+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),
+ workerStopCh: make(chan struct{}),
+ }
+}
+
+func (sr *schemaRepo) SendMetadataEvent(event MetadataEvent) {
+ sr.eventCh <- event
+}
+
+func (sr *schemaRepo) Watcher() {
+ defer func() {
+ if err := recover(); err != nil {
+ sr.l.Warn().Interface("err", err).Msg("watching the events")
+ }
+ }()
+ for {
+ select {
+ case evt := <-sr.eventCh:
+ var err error
+ switch evt.Typ {
+ case EventAddOrUpdate:
+ switch evt.Kind {
+ case EventKindGroup:
+ _, err = sr.StoreGroup(evt.Metadata)
+ case EventKindResource:
+ _, err = sr.storeResource(evt.Metadata)
+ }
+ case EventDelete:
+ switch evt.Kind {
+ case EventKindGroup:
+ err = sr.deleteGroup(evt.Metadata)
+ case EventKindResource:
+ err = sr.deleteResource(evt.Metadata)
+ }
+ }
+ if err != nil {
+ sr.l.Err(err).Interface("event", evt).Msg("fail to handle the metadata event. retry...")
+ sr.eventCh <- evt
+ }
+ case <-sr.workerStopCh:
+ return
+ }
+ }
+}
+
+func (sr *schemaRepo) StoreGroup(groupMeta *commonv1.Metadata) (*group, error) {
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ groupSchema, err := sr.metadata.GroupRegistry().GetGroup(ctx, groupMeta.GetName())
+ cancel()
+ if err != nil {
+ return nil, err
+ }
+ name := groupSchema.GetMetadata().GetName()
+ sr.Lock()
+ defer sr.Unlock()
+ g, ok := sr.getGroup(name)
+ if !ok {
+ sr.l.Info().Str("group", name).Msg("creating a tsdb")
+ var db tsdb.Database
+ db, err = sr.resourceSupplier.OpenDB(groupSchema)
+ if err != nil {
+ return nil, err
+ }
+ g = newGroup(groupSchema, sr.repo, sr.metadata, db, sr.l, sr.resourceSupplier, sr.entityTopic)
+ sr.data[name] = g
+ return g, sr.notify(groupSchema, databasev1.Action_ACTION_PUT)
+ }
+ prevGroupSchema := g.groupSchema
+ if groupSchema.GetMetadata().GetModRevision() <= prevGroupSchema.Metadata.ModRevision {
+ return g, nil
+ }
+ 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 = groupSchema
+ return g, nil
+}
+
+func (sr *schemaRepo) deleteGroup(groupMeta *commonv1.Metadata) error {
+ name := groupMeta.GetName()
+ sr.Lock()
+ defer sr.Unlock()
+ var ok bool
+ g, ok := sr.getGroup(name)
+ if !ok {
+ return nil
+ }
+ err := g.close()
+ if err != nil {
+ return err
+ }
+ _ = sr.notify(g.groupSchema, databasev1.Action_ACTION_DELETE)
+ delete(sr.data, name)
+ return nil
+}
+
+func (sr *schemaRepo) getGroup(name string) (*group, bool) {
+ g := sr.data[name]
+ if g == nil {
+ return nil, false
+ }
+ return g, true
+}
+
+func (sr *schemaRepo) LoadGroup(name string) (Group, bool) {
+ sr.RLock()
+ defer sr.RUnlock()
+ return sr.getGroup(name)
+}
+
+func (sr *schemaRepo) LoadResource(metadata *commonv1.Metadata) (Resource, bool) {
+ g, ok := sr.LoadGroup(metadata.Group)
+ if !ok {
+ return nil, false
+ }
+ return g.LoadResource(metadata.Name)
+}
+
+func (sr *schemaRepo) storeResource(metadata *commonv1.Metadata) (Resource, error) {
+ group, ok := sr.LoadGroup(metadata.Group)
+ if !ok {
+ return nil, errors.Errorf("unknown group")
+ }
+ stm, err := sr.resourceSupplier.ResourceSchema(sr.metadata, metadata)
+ if err != nil {
+ return nil, errors.Errorf("fails to get the resource")
+ }
+ return group.StoreResource(stm)
+}
+
+func (sr *schemaRepo) deleteResource(metadata *commonv1.Metadata) error {
+ g, ok := sr.LoadGroup(metadata.Group)
+ if !ok {
+ return nil
+ }
+ 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) {
+ for _, g := range sr.getMap() {
+ err = multierr.Append(err, sr.notify(g.groupSchema, databasev1.Action_ACTION_PUT))
+ for _, s := range g.getMap() {
+ err = multierr.Append(err, g.notify(s, databasev1.Action_ACTION_PUT))
+ }
+ }
+ return err
+}
+
+func (sr *schemaRepo) Close() {
+ defer func() {
+ if err := recover(); err != nil {
+ sr.l.Warn().Interface("err", err).Msg("closing resource")
+ }
+ }()
+ if sr.eventCh != nil {
+ close(sr.eventCh)
+ }
+ if sr.workerStopCh != nil {
+ close(sr.workerStopCh)
+ }
+ for _, g := range sr.getMap() {
+ err := g.close()
+ if err != nil {
+ sr.l.Err(err).Stringer("group", g.groupSchema.Metadata).Msg("closing")
+ }
+ }
+}
+
+func (sr *schemaRepo) getMap() map[string]*group {
+ sr.RLock()
+ defer sr.RUnlock()
+ return sr.data
+}
+
+var _ Group = (*group)(nil)
+
+type group struct {
+ groupSchema *commonv1.Group
+ l *logger.Logger
+ resourceSupplier ResourceSupplier
+ repo discovery.ServiceRepo
+ metadata metadata.Repo
+ entityTopic bus.Topic
+ db atomic.Value
+ mapMutex sync.RWMutex
+ schemaMap map[string]Resource
+}
+
+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: groupSchema,
+ repo: repo,
+ metadata: metadata,
+ l: l,
+ schemaMap: make(map[string]Resource),
+ resourceSupplier: resourceSupplier,
+ entityTopic: entityTopic,
+ }
+ g.db.Store(db)
+ return g
+}
+
+func (g *group) GetSchema() *commonv1.Group {
+ return g.groupSchema
+}
+
+func (g *group) SupplyTSDB() tsdb.Database {
+ return g.db.Load().(tsdb.Database)
+}
+
+func (g *group) setDB(db tsdb.Database) {
+ g.db.Store(db)
+}
+
+func (g *group) StoreResource(resourceSchema ResourceSchema) (Resource, error) {
+ g.mapMutex.Lock()
+ defer g.mapMutex.Unlock()
+ key := resourceSchema.GetMetadata().GetName()
+ preResource := g.schemaMap[key]
+ if preResource != nil &&
+ resourceSchema.GetMetadata().GetModRevision() <= preResource.GetMetadata().GetModRevision() {
+ // we only need to check the max modifications revision observed for index rules
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ idxRules, errIndexRules := g.metadata.IndexRules(ctx, resourceSchema.GetMetadata())
+ cancel()
+ if errIndexRules != nil {
+ return nil, errIndexRules
+ }
+ if len(idxRules) == len(preResource.GetIndexRules()) {
+ maxModRevision := pbv1.ParseMaxModRevision(idxRules)
+ if preResource.MaxObservedModRevision() >= maxModRevision {
+ return preResource, nil
+ }
+ }
+ }
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ idxRules, errIndexRules := g.metadata.IndexRules(ctx, resourceSchema.GetMetadata())
+ cancel()
+ if errIndexRules != nil {
+ return nil, errIndexRules
+ }
+ sm, errTS := g.resourceSupplier.OpenResource(g.groupSchema.GetResourceOpts().ShardNum, g, ResourceSpec{
+ Schema: resourceSchema,
+ IndexRules: idxRules,
+ })
+ 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()
+ }
+ return sm, nil
+}
+
+func (g *group) deleteResource(metadata *commonv1.Metadata) error {
+ g.mapMutex.Lock()
+ defer g.mapMutex.Unlock()
+ key := metadata.GetName()
+ preResource := g.schemaMap[key]
+ 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
+}
+
+func (g *group) LoadResource(name string) (Resource, bool) {
+ data := g.getMap()
+ s := data[name]
+ if s == nil {
+ return nil, false
+ }
+ return s, true
+}
+
+func (g *group) notify(resource Resource, action databasev1.Action) error {
+ 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) {
+ for _, s := range g.getMap() {
+ err = multierr.Append(err, s.Close())
+ }
+ return multierr.Append(err, g.SupplyTSDB().Close())
+}
+
+func (g *group) getMap() map[string]Resource {
+ g.mapMutex.RLock()
+ defer g.mapMutex.RUnlock()
+ return g.schemaMap
+}