You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ha...@apache.org on 2024/03/14 01:14:43 UTC
(skywalking-banyandb) branch metadata created (now 9444e509)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a change to branch metadata
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
at 9444e509 Add a initilization phase on starting
This branch includes the following new commits:
new d78d3b00 Refactor stress test
new 9444e509 Add a initilization phase on starting
The 2 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) 02/02: Add a initilization phase on starting
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch metadata
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 9444e509188efbdf3c59f3304d116df912300f21
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Thu Mar 14 01:10:27 2024 +0000
Add a initilization phase on starting
In this phase, etcd client would load all necessary data
using a prefix instead of invoking server several times.
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
banyand/liaison/grpc/discovery.go | 2 +
banyand/liaison/grpc/node.go | 1 +
banyand/measure/metadata.go | 132 +------
banyand/measure/topn.go | 8 -
banyand/measure/write.go | 1 +
banyand/metadata/schema/etcd.go | 39 +-
banyand/metadata/schema/etcd_test.go | 15 -
banyand/metadata/schema/schema.go | 9 +
banyand/metadata/schema/watcher.go | 52 ++-
banyand/queue/pub/pub.go | 1 +
banyand/stream/metadata.go | 22 +-
banyand/stream/write.go | 1 +
dist/LICENSE | 1 -
dist/licenses/license-github.com-stretchr-objx.txt | 22 --
go.mod | 1 -
go.sum | 3 +-
pkg/schema/init.go | 392 +++++++++++++++++++++
pkg/schema/metadata.go | 41 ++-
pkg/schema/schema.go | 6 +-
19 files changed, 543 insertions(+), 206 deletions(-)
diff --git a/banyand/liaison/grpc/discovery.go b/banyand/liaison/grpc/discovery.go
index 57914fec..07f308d5 100644
--- a/banyand/liaison/grpc/discovery.go
+++ b/banyand/liaison/grpc/discovery.go
@@ -95,6 +95,7 @@ func (i identity) String() string {
var _ schema.EventHandler = (*shardRepo)(nil)
type shardRepo struct {
+ schema.UnimplementedOnInitHandler
log *logger.Logger
shardEventsMap map[identity]uint32
sync.RWMutex
@@ -154,6 +155,7 @@ func getID(metadata *commonv1.Metadata) identity {
var _ schema.EventHandler = (*entityRepo)(nil)
type entityRepo struct {
+ schema.UnimplementedOnInitHandler
log *logger.Logger
entitiesMap map[identity]partition.EntityLocator
sync.RWMutex
diff --git a/banyand/liaison/grpc/node.go b/banyand/liaison/grpc/node.go
index edae58aa..b43a3bdb 100644
--- a/banyand/liaison/grpc/node.go
+++ b/banyand/liaison/grpc/node.go
@@ -40,6 +40,7 @@ type NodeRegistry interface {
}
type clusterNodeService struct {
+ schema.UnimplementedOnInitHandler
pipeline queue.Client
sel node.Selector
sync.Once
diff --git a/banyand/measure/metadata.go b/banyand/measure/metadata.go
index 681e36c0..c6a89320 100644
--- a/banyand/measure/metadata.go
+++ b/banyand/measure/metadata.go
@@ -24,9 +24,7 @@ import (
"path"
"time"
- "github.com/google/go-cmp/cmp"
"github.com/pkg/errors"
- "google.golang.org/protobuf/testing/protocmp"
"github.com/apache/skywalking-banyandb/api/common"
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
@@ -94,6 +92,14 @@ func (sr *schemaRepo) Measure(metadata *commonv1.Metadata) (Measure, error) {
return sm, nil
}
+func (sr *schemaRepo) OnInit(kinds []schema.Kind) (bool, []int64) {
+ if len(kinds) != 5 {
+ logger.Panicf("unexpected kinds: %v", kinds)
+ return false, nil
+ }
+ return true, sr.Repository.Init(schema.KindMeasure)
+}
+
func (sr *schemaRepo) OnAddOrUpdate(metadata schema.Metadata) {
switch metadata.Kind {
case schema.KindGroup:
@@ -104,13 +110,13 @@ func (sr *schemaRepo) OnAddOrUpdate(metadata schema.Metadata) {
sr.SendMetadataEvent(resourceSchema.MetadataEvent{
Typ: resourceSchema.EventAddOrUpdate,
Kind: resourceSchema.EventKindGroup,
- Metadata: g.GetMetadata(),
+ Metadata: g,
})
case schema.KindMeasure:
sr.SendMetadataEvent(resourceSchema.MetadataEvent{
Typ: resourceSchema.EventAddOrUpdate,
Kind: resourceSchema.EventKindResource,
- Metadata: metadata.Spec.(*databasev1.Measure).GetMetadata(),
+ Metadata: metadata.Spec.(*databasev1.Measure),
})
case schema.KindIndexRuleBinding:
irb, ok := metadata.Spec.(*databasev1.IndexRuleBinding)
@@ -131,7 +137,7 @@ func (sr *schemaRepo) OnAddOrUpdate(metadata schema.Metadata) {
sr.SendMetadataEvent(resourceSchema.MetadataEvent{
Typ: resourceSchema.EventAddOrUpdate,
Kind: resourceSchema.EventKindResource,
- Metadata: stm.GetMetadata(),
+ Metadata: stm,
})
}
case schema.KindIndexRule:
@@ -145,109 +151,19 @@ func (sr *schemaRepo) OnAddOrUpdate(metadata schema.Metadata) {
sr.SendMetadataEvent(resourceSchema.MetadataEvent{
Typ: resourceSchema.EventAddOrUpdate,
Kind: resourceSchema.EventKindResource,
- Metadata: sub.(*databasev1.Measure).GetMetadata(),
+ Metadata: sub.(*databasev1.Measure),
})
}
case schema.KindTopNAggregation:
- // createOrUpdate TopN schemas in advance
- _, err := createOrUpdateTopNMeasure(context.Background(), sr.metadata.MeasureRegistry(), metadata.Spec.(*databasev1.TopNAggregation))
- if err != nil {
- return
- }
- // reload source measure
sr.SendMetadataEvent(resourceSchema.MetadataEvent{
Typ: resourceSchema.EventAddOrUpdate,
- Kind: resourceSchema.EventKindResource,
- Metadata: metadata.Spec.(*databasev1.TopNAggregation).GetSourceMeasure(),
+ Kind: resourceSchema.EventKindTopNAgg,
+ Metadata: metadata.Spec.(*databasev1.TopNAggregation),
})
default:
}
}
-func createOrUpdateTopNMeasure(ctx context.Context, measureSchemaRegistry schema.Measure, topNSchema *databasev1.TopNAggregation) (*databasev1.Measure, error) {
- oldTopNSchema, err := measureSchemaRegistry.GetMeasure(ctx, topNSchema.GetMetadata())
- if err != nil && !errors.Is(err, schema.ErrGRPCResourceNotFound) {
- return nil, err
- }
-
- sourceMeasureSchema, err := measureSchemaRegistry.GetMeasure(ctx, topNSchema.GetSourceMeasure())
- if err != nil {
- return nil, err
- }
-
- tagNames := sourceMeasureSchema.GetEntity().GetTagNames()
- seriesSpecs := make([]*databasev1.TagSpec, 0, len(tagNames))
-
- for _, tagName := range tagNames {
- var found bool
- for _, fSpec := range sourceMeasureSchema.GetTagFamilies() {
- for _, tSpec := range fSpec.GetTags() {
- if tSpec.GetName() == tagName {
- seriesSpecs = append(seriesSpecs, tSpec)
- found = true
- goto CHECK
- }
- }
- }
-
- CHECK:
- if !found {
- return nil, fmt.Errorf("fail to find tag spec %s", tagName)
- }
- }
-
- // create a new "derived" measure for TopN result
- newTopNMeasure := &databasev1.Measure{
- Metadata: topNSchema.GetMetadata(),
- Interval: sourceMeasureSchema.GetInterval(),
- TagFamilies: []*databasev1.TagFamilySpec{
- {
- Name: topNTagFamily,
- Tags: append([]*databasev1.TagSpec{
- {
- Name: "measure_id",
- Type: databasev1.TagType_TAG_TYPE_STRING,
- },
- {
- Name: "sortDirection",
- Type: databasev1.TagType_TAG_TYPE_INT,
- },
- {
- Name: "rankNumber",
- Type: databasev1.TagType_TAG_TYPE_INT,
- },
- }, seriesSpecs...),
- },
- },
- Fields: []*databasev1.FieldSpec{topNValueFieldSpec},
- Entity: &databasev1.Entity{
- TagNames: append([]string{
- "sortDirection",
- "rankNumber",
- }, topNSchema.GetGroupByTagNames()...),
- },
- }
- if oldTopNSchema == nil {
- if _, innerErr := measureSchemaRegistry.CreateMeasure(ctx, newTopNMeasure); innerErr != nil {
- return nil, innerErr
- }
- return newTopNMeasure, nil
- }
- // compare with the old one
- if cmp.Diff(newTopNMeasure, oldTopNSchema,
- protocmp.IgnoreUnknown(),
- protocmp.IgnoreFields(&databasev1.Measure{}, "updated_at"),
- protocmp.IgnoreFields(&commonv1.Metadata{}, "id", "create_revision", "mod_revision"),
- protocmp.Transform()) == "" {
- return oldTopNSchema, nil
- }
- // update
- if _, err = measureSchemaRegistry.UpdateMeasure(ctx, newTopNMeasure); err != nil {
- return nil, err
- }
- return newTopNMeasure, nil
-}
-
func (sr *schemaRepo) OnDelete(metadata schema.Metadata) {
switch metadata.Kind {
case schema.KindGroup:
@@ -258,13 +174,13 @@ func (sr *schemaRepo) OnDelete(metadata schema.Metadata) {
sr.SendMetadataEvent(resourceSchema.MetadataEvent{
Typ: resourceSchema.EventDelete,
Kind: resourceSchema.EventKindGroup,
- Metadata: g.GetMetadata(),
+ Metadata: g,
})
case schema.KindMeasure:
sr.SendMetadataEvent(resourceSchema.MetadataEvent{
Typ: resourceSchema.EventDelete,
Kind: resourceSchema.EventKindResource,
- Metadata: metadata.Spec.(*databasev1.Measure).GetMetadata(),
+ Metadata: metadata.Spec.(*databasev1.Measure),
})
case schema.KindIndexRuleBinding:
if metadata.Spec.(*databasev1.IndexRuleBinding).GetSubject().Catalog == commonv1.Catalog_CATALOG_MEASURE {
@@ -281,30 +197,20 @@ func (sr *schemaRepo) OnDelete(metadata schema.Metadata) {
sr.SendMetadataEvent(resourceSchema.MetadataEvent{
Typ: resourceSchema.EventAddOrUpdate,
Kind: resourceSchema.EventKindResource,
- Metadata: m.GetMetadata(),
+ Metadata: m,
})
}
case schema.KindIndexRule:
case schema.KindTopNAggregation:
- err := sr.removeTopNMeasure(metadata.Spec.(*databasev1.TopNAggregation).GetSourceMeasure())
- if err != nil {
- return
- }
- // we should update instead of delete
sr.SendMetadataEvent(resourceSchema.MetadataEvent{
Typ: resourceSchema.EventAddOrUpdate,
- Kind: resourceSchema.EventKindResource,
- Metadata: metadata.Spec.(*databasev1.TopNAggregation).GetSourceMeasure(),
+ Kind: resourceSchema.EventKindTopNAgg,
+ Metadata: metadata.Spec.(*databasev1.TopNAggregation),
})
default:
}
}
-func (sr *schemaRepo) removeTopNMeasure(metadata *commonv1.Metadata) error {
- _, err := sr.metadata.MeasureRegistry().DeleteMeasure(context.Background(), metadata)
- return err
-}
-
func (sr *schemaRepo) loadMeasure(metadata *commonv1.Metadata) (*measure, bool) {
r, ok := sr.LoadResource(metadata)
if !ok {
diff --git a/banyand/measure/topn.go b/banyand/measure/topn.go
index 10774efc..90b2d473 100644
--- a/banyand/measure/topn.go
+++ b/banyand/measure/topn.go
@@ -51,20 +51,12 @@ import (
const (
timeBucketFormat = "200601021504"
- topNTagFamily = "__topN__"
)
var (
_ io.Closer = (*topNStreamingProcessor)(nil)
_ io.Closer = (*topNProcessorManager)(nil)
_ flow.Sink = (*topNStreamingProcessor)(nil)
-
- topNValueFieldSpec = &databasev1.FieldSpec{
- Name: "value",
- FieldType: databasev1.FieldType_FIELD_TYPE_INT,
- EncodingMethod: databasev1.EncodingMethod_ENCODING_METHOD_GORILLA,
- CompressionMethod: databasev1.CompressionMethod_COMPRESSION_METHOD_ZSTD,
- }
)
type dataPointWithEntityValues struct {
diff --git a/banyand/measure/write.go b/banyand/measure/write.go
index a4075dab..d6292bb3 100644
--- a/banyand/measure/write.go
+++ b/banyand/measure/write.go
@@ -240,6 +240,7 @@ func (w *writeCallback) Rev(message bus.Message) (resp bus.Message) {
var err error
if groups, err = w.handle(groups, writeEvent); err != nil {
w.l.Error().Err(err).Msg("cannot handle write event")
+ groups = make(map[string]*dataPointsInGroup)
continue
}
}
diff --git a/banyand/metadata/schema/etcd.go b/banyand/metadata/schema/etcd.go
index 213f02a5..7de1f39c 100644
--- a/banyand/metadata/schema/etcd.go
+++ b/banyand/metadata/schema/etcd.go
@@ -124,11 +124,33 @@ func (e *etcdSchemaRegistry) RegisterHandler(name string, kind Kind, handler Eve
}
e.mux.Lock()
defer e.mux.Unlock()
+ var kinds []Kind
for i := 0; i < KindSize; i++ {
ki := Kind(1 << i)
if kind&ki > 0 {
- e.l.Info().Str("name", name).Stringer("kind", ki).Msg("registering watcher")
- w := e.newWatcher(name, ki, handler)
+ kinds = append(kinds, ki)
+ }
+ }
+ e.l.Info().Str("name", name).Interface("kinds", kinds).Msg("initializing schema cache")
+ ok, revisions := handler.OnInit(kinds)
+ if ok {
+ if len(revisions) != len(kinds) {
+ logger.Panicf("invalid number of revisions for %s", name)
+ return
+ }
+ for i := range kinds {
+ e.l.Info().Str("name", name).Stringer("kind", kinds[i]).Msg("registering watcher")
+ w := e.newWatcherWithRevision(name, kinds[i], revisions[i], handler)
+ if w != nil {
+ e.watchers = append(e.watchers, w)
+ }
+ }
+ return
+ }
+ for i := range kinds {
+ e.l.Info().Str("name", name).Stringer("kind", kinds[i]).Msg("registering watcher")
+ w := e.newWatcher(name, kinds[i], handler)
+ if w != nil {
e.watchers = append(e.watchers, w)
}
}
@@ -419,11 +441,16 @@ func (e *etcdSchemaRegistry) register(ctx context.Context, metadata Metadata, fo
return nil
}
-func (e *etcdSchemaRegistry) newWatcher(name string, kind Kind, handler EventHandler) *watcher {
+func (e *etcdSchemaRegistry) newWatcher(name string, kind Kind, handler watchEventHandler) *watcher {
+ return e.newWatcherWithRevision(name, kind, 0, handler)
+}
+
+func (e *etcdSchemaRegistry) newWatcherWithRevision(name string, kind Kind, revision int64, handler watchEventHandler) *watcher {
return newWatcher(e.client, watcherConfig{
- key: e.prependNamespace(kind.key()),
- kind: kind,
- handler: handler,
+ key: e.prependNamespace(kind.key()),
+ kind: kind,
+ handler: handler,
+ revision: revision,
}, e.l.Named(fmt.Sprintf("watcher-%s[%s]", name, kind.String())))
}
diff --git a/banyand/metadata/schema/etcd_test.go b/banyand/metadata/schema/etcd_test.go
index e4fc3913..45aa7fd8 100644
--- a/banyand/metadata/schema/etcd_test.go
+++ b/banyand/metadata/schema/etcd_test.go
@@ -27,7 +27,6 @@ import (
"github.com/google/uuid"
"github.com/stretchr/testify/assert"
- "github.com/stretchr/testify/mock"
"github.com/stretchr/testify/require"
"google.golang.org/protobuf/encoding/protojson"
@@ -48,22 +47,8 @@ var (
streamJSON string
//go:embed testdata/group.json
groupJSON string
-
- _ EventHandler = (*mockedEventHandler)(nil)
)
-type mockedEventHandler struct {
- mock.Mock
-}
-
-func (m *mockedEventHandler) OnAddOrUpdate(metadata Metadata) {
- m.Called(metadata)
-}
-
-func (m *mockedEventHandler) OnDelete(metadata Metadata) {
- m.Called(metadata)
-}
-
func preloadSchema(e Registry) error {
g := &commonv1.Group{}
if err := protojson.Unmarshal([]byte(groupJSON), g); err != nil {
diff --git a/banyand/metadata/schema/schema.go b/banyand/metadata/schema/schema.go
index 3a35c13f..aaedd502 100644
--- a/banyand/metadata/schema/schema.go
+++ b/banyand/metadata/schema/schema.go
@@ -34,10 +34,19 @@ var errUnsupportedEntityType = errors.New("unsupported entity type")
// EventHandler allows receiving and handling the resource change events.
type EventHandler interface {
+ OnInit([]Kind) (bool, []int64)
OnAddOrUpdate(Metadata)
OnDelete(Metadata)
}
+// UnimplementedOnInitHandler is a placeholder for unimplemented OnInitHandler.
+type UnimplementedOnInitHandler struct{}
+
+// OnInit is a placeholder for unimplemented OnInitHandler.
+func (u UnimplementedOnInitHandler) OnInit([]Kind) (bool, []int64) {
+ return false, nil
+}
+
// ListOpt contains options to list resources.
type ListOpt struct {
Group string
diff --git a/banyand/metadata/schema/watcher.go b/banyand/metadata/schema/watcher.go
index c21e3dec..aa203485 100644
--- a/banyand/metadata/schema/watcher.go
+++ b/banyand/metadata/schema/watcher.go
@@ -18,40 +18,52 @@
package schema
import (
+ "errors"
"time"
mvccpb "go.etcd.io/etcd/api/v3/mvccpb"
+ v3rpc "go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
clientv3 "go.etcd.io/etcd/client/v3"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/run"
)
+type watchEventHandler interface {
+ OnAddOrUpdate(Metadata)
+ OnDelete(Metadata)
+}
type watcherConfig struct {
- handler EventHandler
- key string
- kind Kind
+ handler watchEventHandler
+ key string
+ revision int64
+ kind Kind
}
type watcher struct {
- handler EventHandler
- cli *clientv3.Client
- closer *run.Closer
- l *logger.Logger
- key string
- kind Kind
+ handler watchEventHandler
+ cli *clientv3.Client
+ closer *run.Closer
+ l *logger.Logger
+ key string
+ revision int64
+ kind Kind
}
func newWatcher(cli *clientv3.Client, wc watcherConfig, l *logger.Logger) *watcher {
w := &watcher{
- cli: cli,
- key: wc.key,
- kind: wc.kind,
- handler: wc.handler,
- closer: run.NewCloser(1),
- l: l,
+ cli: cli,
+ key: wc.key,
+ kind: wc.kind,
+ handler: wc.handler,
+ revision: wc.revision,
+ closer: run.NewCloser(1),
+ l: l,
+ }
+ revision := w.revision
+ if revision < 1 {
+ revision = w.allEvents()
}
- revision := w.allEvents()
go w.watch(revision)
return w
}
@@ -95,9 +107,7 @@ func (w *watcher) watch(revision int64) {
defer w.closer.Done()
cli := w.cli
for {
- if revision > 0 {
- revision = -1
- } else {
+ if revision < 0 {
revision = w.allEvents()
}
select {
@@ -120,6 +130,10 @@ func (w *watcher) watch(revision int64) {
case <-w.closer.CloseNotify():
return
default:
+ if errors.Is(err, v3rpc.ErrCompacted) {
+ revision = -1
+ break
+ }
continue
}
}
diff --git a/banyand/queue/pub/pub.go b/banyand/queue/pub/pub.go
index 8e6de582..18a9b214 100644
--- a/banyand/queue/pub/pub.go
+++ b/banyand/queue/pub/pub.go
@@ -46,6 +46,7 @@ var (
)
type pub struct {
+ schema.UnimplementedOnInitHandler
metadata metadata.Repo
handler schema.EventHandler
log *logger.Logger
diff --git a/banyand/stream/metadata.go b/banyand/stream/metadata.go
index d494ed8b..9ef65105 100644
--- a/banyand/stream/metadata.go
+++ b/banyand/stream/metadata.go
@@ -92,6 +92,14 @@ func (sr *schemaRepo) Stream(metadata *commonv1.Metadata) (Stream, error) {
return sm, nil
}
+func (sr *schemaRepo) OnInit(kinds []schema.Kind) (bool, []int64) {
+ if len(kinds) != 4 {
+ logger.Panicf("invalid kinds: %v", kinds)
+ return false, nil
+ }
+ return true, sr.Repository.Init(schema.KindStream)
+}
+
func (sr *schemaRepo) OnAddOrUpdate(metadata schema.Metadata) {
switch metadata.Kind {
case schema.KindGroup:
@@ -102,13 +110,13 @@ func (sr *schemaRepo) OnAddOrUpdate(metadata schema.Metadata) {
sr.SendMetadataEvent(resourceSchema.MetadataEvent{
Typ: resourceSchema.EventAddOrUpdate,
Kind: resourceSchema.EventKindGroup,
- Metadata: g.GetMetadata(),
+ Metadata: g,
})
case schema.KindStream:
sr.SendMetadataEvent(resourceSchema.MetadataEvent{
Typ: resourceSchema.EventAddOrUpdate,
Kind: resourceSchema.EventKindResource,
- Metadata: metadata.Spec.(*databasev1.Stream).GetMetadata(),
+ Metadata: metadata.Spec.(*databasev1.Stream),
})
case schema.KindIndexRuleBinding:
irb, ok := metadata.Spec.(*databasev1.IndexRuleBinding)
@@ -129,7 +137,7 @@ func (sr *schemaRepo) OnAddOrUpdate(metadata schema.Metadata) {
sr.SendMetadataEvent(resourceSchema.MetadataEvent{
Typ: resourceSchema.EventAddOrUpdate,
Kind: resourceSchema.EventKindResource,
- Metadata: stm.GetMetadata(),
+ Metadata: stm,
})
}
case schema.KindIndexRule:
@@ -143,7 +151,7 @@ func (sr *schemaRepo) OnAddOrUpdate(metadata schema.Metadata) {
sr.SendMetadataEvent(resourceSchema.MetadataEvent{
Typ: resourceSchema.EventAddOrUpdate,
Kind: resourceSchema.EventKindResource,
- Metadata: sub.(*databasev1.Stream).GetMetadata(),
+ Metadata: sub.(*databasev1.Stream),
})
}
default:
@@ -160,13 +168,13 @@ func (sr *schemaRepo) OnDelete(metadata schema.Metadata) {
sr.SendMetadataEvent(resourceSchema.MetadataEvent{
Typ: resourceSchema.EventDelete,
Kind: resourceSchema.EventKindGroup,
- Metadata: g.GetMetadata(),
+ Metadata: g,
})
case schema.KindStream:
sr.SendMetadataEvent(resourceSchema.MetadataEvent{
Typ: resourceSchema.EventDelete,
Kind: resourceSchema.EventKindResource,
- Metadata: metadata.Spec.(*databasev1.Stream).GetMetadata(),
+ Metadata: metadata.Spec.(*databasev1.Stream),
})
case schema.KindIndexRuleBinding:
if metadata.Spec.(*databasev1.IndexRuleBinding).GetSubject().Catalog == commonv1.Catalog_CATALOG_STREAM {
@@ -183,7 +191,7 @@ func (sr *schemaRepo) OnDelete(metadata schema.Metadata) {
sr.SendMetadataEvent(resourceSchema.MetadataEvent{
Typ: resourceSchema.EventAddOrUpdate,
Kind: resourceSchema.EventKindResource,
- Metadata: m.GetMetadata(),
+ Metadata: m,
})
}
case schema.KindIndexRule:
diff --git a/banyand/stream/write.go b/banyand/stream/write.go
index 4cc06aa4..216a340f 100644
--- a/banyand/stream/write.go
+++ b/banyand/stream/write.go
@@ -217,6 +217,7 @@ func (w *writeCallback) Rev(message bus.Message) (resp bus.Message) {
var err error
if groups, err = w.handle(groups, writeEvent); err != nil {
w.l.Error().Err(err).Msg("cannot handle write event")
+ groups = make(map[string]*elementsInGroup)
continue
}
}
diff --git a/dist/LICENSE b/dist/LICENSE
index 43c34b3a..06b1cfc3 100644
--- a/dist/LICENSE
+++ b/dist/LICENSE
@@ -344,7 +344,6 @@ MIT licenses
github.com/sourcegraph/conc v0.3.0 MIT
github.com/spf13/cast v1.6.0 MIT
github.com/spf13/viper v1.18.2 MIT
- github.com/stretchr/objx v0.5.0 MIT
github.com/stretchr/testify v1.8.4 MIT
github.com/subosito/gotenv v1.6.0 MIT
github.com/tmc/grpc-websocket-proxy v0.0.0-20220101234140-673ab2c3ae75 MIT
diff --git a/dist/licenses/license-github.com-stretchr-objx.txt b/dist/licenses/license-github.com-stretchr-objx.txt
deleted file mode 100644
index 44d4d9d5..00000000
--- a/dist/licenses/license-github.com-stretchr-objx.txt
+++ /dev/null
@@ -1,22 +0,0 @@
-The MIT License
-
-Copyright (c) 2014 Stretchr, Inc.
-Copyright (c) 2017-2018 objx contributors
-
-Permission is hereby granted, free of charge, to any person obtaining a copy
-of this software and associated documentation files (the "Software"), to deal
-in the Software without restriction, including without limitation the rights
-to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-copies of the Software, and to permit persons to whom the Software is
-furnished to do so, subject to the following conditions:
-
-The above copyright notice and this permission notice shall be included in all
-copies or substantial portions of the Software.
-
-THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
-SOFTWARE.
diff --git a/go.mod b/go.mod
index dabad54e..bfaf6890 100644
--- a/go.mod
+++ b/go.mod
@@ -115,7 +115,6 @@ require (
github.com/sourcegraph/conc v0.3.0 // indirect
github.com/spf13/afero v1.11.0 // indirect
github.com/spf13/cast v1.6.0 // indirect
- github.com/stretchr/objx v0.5.0 // indirect
github.com/subosito/gotenv v1.6.0 // indirect
github.com/tklauser/go-sysconf v0.3.13 // indirect
github.com/tklauser/numcpus v0.7.0 // indirect
diff --git a/go.sum b/go.sum
index 2bc1fba7..3cd60e04 100644
--- a/go.sum
+++ b/go.sum
@@ -321,7 +321,6 @@ github.com/spf13/viper v1.18.2/go.mod h1:EKmWIqdnk5lOcmR72yw6hS+8OPYcwD0jteitLMV
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
-github.com/stretchr/objx v0.5.0 h1:1zr/of2m5FGMsad5YfcqgdqdWrIhu+EBEJRhR1U7z/c=
github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
@@ -600,4 +599,4 @@ honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWh
honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
rsc.io/pdf v0.1.1/go.mod h1:n8OzWcQ6Sp37PL01nO98y4iUCRdTGarVfzxY20ICaU4=
sigs.k8s.io/yaml v1.4.0 h1:Mk1wCc2gy/F0THH0TAp1QYyJNzRm2KCLy3o5ASXVI5E=
-sigs.k8s.io/yaml v1.4.0/go.mod h1:Ejl7/uTz7PSA4eKMyQCUTnhZYNmLIl+5c2lQPGR2BPY=
\ No newline at end of file
+sigs.k8s.io/yaml v1.4.0/go.mod h1:Ejl7/uTz7PSA4eKMyQCUTnhZYNmLIl+5c2lQPGR2BPY=
diff --git a/pkg/schema/init.go b/pkg/schema/init.go
new file mode 100644
index 00000000..4813be53
--- /dev/null
+++ b/pkg/schema/init.go
@@ -0,0 +1,392 @@
+// 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"
+ "time"
+
+ "github.com/google/go-cmp/cmp"
+ "github.com/pkg/errors"
+ "google.golang.org/protobuf/testing/protocmp"
+
+ 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"
+)
+
+const topNTagFamily = "__topN__"
+
+var (
+ initTimeout = 10 * time.Second
+ topNValueFieldSpec = &databasev1.FieldSpec{
+ Name: "value",
+ FieldType: databasev1.FieldType_FIELD_TYPE_INT,
+ EncodingMethod: databasev1.EncodingMethod_ENCODING_METHOD_GORILLA,
+ CompressionMethod: databasev1.CompressionMethod_COMPRESSION_METHOD_ZSTD,
+ }
+)
+
+type revisionContext struct {
+ group int64
+ measure int64
+ stream int64
+ indexRule int64
+ indexRuleBinding int64
+ topNAgg int64
+}
+
+func (r revisionContext) String() string {
+ return fmt.Sprintf("Group: %d, Measure: %d, Stream: %d, IndexRule: %d, IndexRuleBinding: %d, TopNAgg: %d",
+ r.group, r.measure, r.stream, r.indexRule, r.indexRuleBinding, r.topNAgg)
+}
+
+type revisionContextKey struct{}
+
+var revCtxKey = revisionContextKey{}
+
+func (sr *schemaRepo) Init(kind schema.Kind) []int64 {
+ if kind != schema.KindMeasure && kind != schema.KindStream {
+ return nil
+ }
+ catalog := sr.getCatalog(kind)
+ ctx := context.Background()
+ groups, err := sr.metadata.GroupRegistry().ListGroup(ctx)
+ if err != nil {
+ logger.Panicf("fails to get the groups: %v", err)
+ return nil
+ }
+ var revCtx revisionContext
+ for _, g := range groups {
+ if g.Catalog != catalog {
+ continue
+ }
+ if g.Metadata.ModRevision > revCtx.group {
+ revCtx.group = g.Metadata.ModRevision
+ }
+ sr.processGroup(context.WithValue(ctx, revCtxKey, &revCtx), g, catalog)
+ }
+ if kind == schema.KindMeasure {
+ sr.l.Info().Stringer("revision", revCtx).Msg("init measures")
+ return []int64{revCtx.group, revCtx.measure, revCtx.indexRuleBinding, revCtx.indexRule, revCtx.topNAgg}
+ }
+ sr.l.Info().Stringer("revision", revCtx).Msg("init stream")
+ return []int64{revCtx.group, revCtx.stream, revCtx.indexRuleBinding, revCtx.indexRule}
+}
+
+func (sr *schemaRepo) getCatalog(kind schema.Kind) commonv1.Catalog {
+ if kind == schema.KindMeasure {
+ return commonv1.Catalog_CATALOG_MEASURE
+ }
+ return commonv1.Catalog_CATALOG_STREAM
+}
+
+func (sr *schemaRepo) processGroup(ctx context.Context, g *commonv1.Group, catalog commonv1.Catalog) {
+ group, err := sr.initGroup(g)
+ if err != nil {
+ logger.Panicf("fails to init the group: %v", err)
+ }
+ bindings := sr.getBindings(ctx, g.Metadata.GetName())
+ rules := sr.getRules(ctx, g.Metadata.GetName())
+ if catalog == commonv1.Catalog_CATALOG_MEASURE {
+ sr.processMeasure(ctx, g.Metadata.Name, group, bindings, rules)
+ return
+ }
+ sr.processStream(ctx, g.Metadata.Name, group, bindings, rules)
+}
+
+func (sr *schemaRepo) getBindings(ctx context.Context, gName string) map[string][]string {
+ ctx, cancel := context.WithTimeout(ctx, initTimeout)
+ defer cancel()
+ start := time.Now()
+ ibb, err := sr.metadata.IndexRuleBindingRegistry().ListIndexRuleBinding(ctx, schema.ListOpt{Group: gName})
+ if err != nil {
+ logger.Panicf("fails to get the index rule bindings: %v", err)
+ }
+ revCtx := ctx.Value(revCtxKey).(*revisionContext)
+ bindings := make(map[string][]string, len(ibb))
+ for _, ib := range ibb {
+ bindings[ib.GetSubject().GetName()] = ib.GetRules()
+ if ib.Metadata.ModRevision > revCtx.indexRuleBinding {
+ revCtx.indexRuleBinding = ib.Metadata.ModRevision
+ }
+ }
+ sr.l.Info().Str("group", gName).Dur("duration", time.Since(start)).Int("size", len(bindings)).Msg("get index rule bindings")
+ return bindings
+}
+
+func (sr *schemaRepo) getRules(ctx context.Context, gName string) map[string]*databasev1.IndexRule {
+ ctx, cancel := context.WithTimeout(ctx, initTimeout)
+ defer cancel()
+ start := time.Now()
+ rr, err := sr.metadata.IndexRuleRegistry().ListIndexRule(ctx, schema.ListOpt{Group: gName})
+ if err != nil {
+ logger.Panicf("fails to get the index rules: %v", err)
+ }
+ revCtx := ctx.Value(revCtxKey).(*revisionContext)
+ rules := make(map[string]*databasev1.IndexRule, len(rr))
+ for _, r := range rr {
+ rules[r.Metadata.Name] = r
+ if r.Metadata.ModRevision > revCtx.indexRule {
+ revCtx.indexRule = r.Metadata.ModRevision
+ }
+ }
+ sr.l.Info().Str("group", gName).Dur("duration", time.Since(start)).Int("size", len(rules)).Msg("get index rules")
+ return rules
+}
+
+func (sr *schemaRepo) processMeasure(ctx context.Context, gName string, group *group, bindings map[string][]string, rules map[string]*databasev1.IndexRule) {
+ aggMap := sr.getAggMap(ctx, gName)
+
+ ctx, cancel := context.WithTimeout(ctx, initTimeout)
+ defer cancel()
+ start := time.Now()
+ mm, err := sr.metadata.MeasureRegistry().ListMeasure(ctx, schema.ListOpt{Group: gName})
+ if err != nil {
+ logger.Panicf("fails to get the measures: %v", err)
+ return
+ }
+ revCtx := ctx.Value(revCtxKey).(*revisionContext)
+ measureMap := make(map[string]*databasev1.Measure, len(mm))
+ for _, m := range mm {
+ measureMap[m.Metadata.Name] = m
+ }
+ for _, aa := range aggMap {
+ for _, a := range aa {
+ sourceMeasre, ok := measureMap[a.SourceMeasure.Name]
+ if !ok {
+ sr.l.Warn().Str("group", gName).Str("measure", a.SourceMeasure.Name).Str("agg", a.Metadata.Name).Msg("source measure not found")
+ continue
+ }
+ if _, ok := measureMap[a.Metadata.Name]; !ok {
+ sr.l.Info().Str("group", gName).Str("measure", a.SourceMeasure.Name).Str("agg", a.Metadata.Name).Msg("remove topN aggregation")
+ m, err := createTopNMeasure(ctx, sr.metadata.MeasureRegistry(), a, sourceMeasre)
+ if err != nil {
+ logger.Panicf("fails to create or update the topN measure: %v", err)
+ return
+ }
+ mm = append(mm, m)
+ }
+ }
+ }
+
+ for _, m := range mm {
+ if m.Metadata.ModRevision > revCtx.measure {
+ revCtx.measure = m.Metadata.ModRevision
+ }
+ sr.storeMeasure(m, group, bindings, rules, aggMap)
+ }
+ sr.l.Info().Str("group", gName).Dur("duration", time.Since(start)).Int("size", len(mm)).Msg("store measures")
+}
+
+func (sr *schemaRepo) getAggMap(ctx context.Context, gName string) map[string][]*databasev1.TopNAggregation {
+ ctx, cancel := context.WithTimeout(ctx, initTimeout)
+ defer cancel()
+ start := time.Now()
+ agg, err := sr.metadata.TopNAggregationRegistry().ListTopNAggregation(ctx, schema.ListOpt{Group: gName})
+ if err != nil {
+ logger.Panicf("fails to get the topN aggregations: %v", err)
+ }
+ revCtx := ctx.Value(revCtxKey).(*revisionContext)
+ aggMap := make(map[string][]*databasev1.TopNAggregation, len(agg))
+ for _, a := range agg {
+ aggs, ok := aggMap[a.SourceMeasure.Name]
+ if ok {
+ aggs = append(aggs, a)
+ } else {
+ aggs = []*databasev1.TopNAggregation{a}
+ }
+ aggMap[a.SourceMeasure.Name] = aggs
+ if a.Metadata.ModRevision > revCtx.topNAgg {
+ revCtx.topNAgg = a.Metadata.ModRevision
+ }
+ }
+ sr.l.Info().Str("group", gName).Dur("duration", time.Since(start)).Int("size", len(agg)).Msg("get topN aggregations")
+ return aggMap
+}
+
+func (sr *schemaRepo) storeMeasure(m *databasev1.Measure, group *group, bindings map[string][]string,
+ rules map[string]*databasev1.IndexRule, aggMap map[string][]*databasev1.TopNAggregation,
+) {
+ var indexRules []*databasev1.IndexRule
+ if rr, ok := bindings[m.Metadata.GetName()]; ok {
+ for _, r := range rr {
+ indexRules = append(indexRules, rules[r])
+ }
+ }
+ var topNAggr []*databasev1.TopNAggregation
+ if aa, ok := aggMap[m.Metadata.GetName()]; ok {
+ topNAggr = append(topNAggr, aa...)
+ }
+ _, err := group.storeResource(m, indexRules, topNAggr)
+ if err != nil {
+ logger.Panicf("fails to store the measure: %v", err)
+ }
+}
+
+func (sr *schemaRepo) processStream(ctx context.Context, gName string, group *group, bindings map[string][]string, rules map[string]*databasev1.IndexRule) {
+ ctx, cancel := context.WithTimeout(ctx, initTimeout)
+ defer cancel()
+ start := time.Now()
+ ss, err := sr.metadata.StreamRegistry().ListStream(ctx, schema.ListOpt{Group: gName})
+ if err != nil {
+ logger.Panicf("fails to get the streams: %v", err)
+ return
+ }
+ revCtx := ctx.Value(revCtxKey).(*revisionContext)
+ for _, s := range ss {
+ sr.storeStream(s, group, bindings, rules)
+ if s.Metadata.ModRevision > revCtx.stream {
+ revCtx.stream = s.Metadata.ModRevision
+ }
+ }
+ sr.l.Info().Str("group", gName).Dur("duration", time.Since(start)).Int("size", len(ss)).Msg("store streams")
+}
+
+func (sr *schemaRepo) storeStream(s *databasev1.Stream, group *group, bindings map[string][]string, rules map[string]*databasev1.IndexRule) {
+ var indexRules []*databasev1.IndexRule
+ if rr, ok := bindings[s.Metadata.GetName()]; ok {
+ for _, r := range rr {
+ indexRules = append(indexRules, rules[r])
+ }
+ }
+ _, err := group.storeResource(s, indexRules, nil)
+ if err != nil {
+ logger.Panicf("fails to store the stream: %v", err)
+ }
+}
+
+func (sr *schemaRepo) initGroup(groupSchema *commonv1.Group) (*group, error) {
+ g, ok := sr.getGroup(groupSchema.Metadata.Name)
+ if ok {
+ return g, nil
+ }
+ sr.l.Info().Str("group", groupSchema.Metadata.Name).Msg("creating a tsdb")
+ g = sr.createGroup(groupSchema.Metadata.Name)
+ if err := g.initBySchema(groupSchema); err != nil {
+ return nil, err
+ }
+ return g, nil
+}
+
+func createOrUpdateTopNMeasure(ctx context.Context, measureSchemaRegistry schema.Measure, topNSchema *databasev1.TopNAggregation) (*databasev1.Measure, error) {
+ oldTopNSchema, err := measureSchemaRegistry.GetMeasure(ctx, topNSchema.GetMetadata())
+ if err != nil && !errors.Is(err, schema.ErrGRPCResourceNotFound) {
+ return nil, err
+ }
+
+ sourceMeasureSchema, err := measureSchemaRegistry.GetMeasure(ctx, topNSchema.GetSourceMeasure())
+ if err != nil {
+ return nil, err
+ }
+
+ // create a new "derived" measure for TopN result
+ newTopNMeasure, err := buildTopNSourceMeasure(topNSchema, sourceMeasureSchema)
+ if err != nil {
+ return nil, err
+ }
+ if oldTopNSchema == nil {
+ if _, innerErr := measureSchemaRegistry.CreateMeasure(ctx, newTopNMeasure); innerErr != nil {
+ return nil, innerErr
+ }
+ return newTopNMeasure, nil
+ }
+ // compare with the old one
+ if cmp.Diff(newTopNMeasure, oldTopNSchema,
+ protocmp.IgnoreUnknown(),
+ protocmp.IgnoreFields(&databasev1.Measure{}, "updated_at"),
+ protocmp.IgnoreFields(&commonv1.Metadata{}, "id", "create_revision", "mod_revision"),
+ protocmp.Transform()) == "" {
+ return oldTopNSchema, nil
+ }
+ // update
+ if _, err = measureSchemaRegistry.UpdateMeasure(ctx, newTopNMeasure); err != nil {
+ return nil, err
+ }
+ return newTopNMeasure, nil
+}
+
+func createTopNMeasure(ctx context.Context, measureSchemaRegistry schema.Measure, topNSchema *databasev1.TopNAggregation,
+ sourceMeasureSchema *databasev1.Measure,
+) (*databasev1.Measure, error) {
+ newTopNMeasure, err := buildTopNSourceMeasure(topNSchema, sourceMeasureSchema)
+ if err != nil {
+ return nil, err
+ }
+ if _, err := measureSchemaRegistry.CreateMeasure(ctx, newTopNMeasure); err != nil {
+ return nil, err
+ }
+ return newTopNMeasure, nil
+}
+
+func buildTopNSourceMeasure(topNSchema *databasev1.TopNAggregation, sourceMeasureSchema *databasev1.Measure) (*databasev1.Measure, error) {
+ tagNames := sourceMeasureSchema.GetEntity().GetTagNames()
+ seriesSpecs := make([]*databasev1.TagSpec, 0, len(tagNames))
+
+ for _, tagName := range tagNames {
+ var found bool
+ for _, fSpec := range sourceMeasureSchema.GetTagFamilies() {
+ for _, tSpec := range fSpec.GetTags() {
+ if tSpec.GetName() == tagName {
+ seriesSpecs = append(seriesSpecs, tSpec)
+ found = true
+ goto CHECK
+ }
+ }
+ }
+
+ CHECK:
+ if !found {
+ return nil, fmt.Errorf("fail to find tag spec %s", tagName)
+ }
+ }
+
+ // create a new "derived" measure for TopN result
+ return &databasev1.Measure{
+ Metadata: topNSchema.Metadata,
+ Interval: sourceMeasureSchema.GetInterval(),
+ TagFamilies: []*databasev1.TagFamilySpec{
+ {
+ Name: topNTagFamily,
+ Tags: append([]*databasev1.TagSpec{
+ {
+ Name: "measure_id",
+ Type: databasev1.TagType_TAG_TYPE_STRING,
+ },
+ {
+ Name: "sortDirection",
+ Type: databasev1.TagType_TAG_TYPE_INT,
+ },
+ {
+ Name: "rankNumber",
+ Type: databasev1.TagType_TAG_TYPE_INT,
+ },
+ }, seriesSpecs...),
+ },
+ },
+ Fields: []*databasev1.FieldSpec{topNValueFieldSpec},
+ Entity: &databasev1.Entity{
+ TagNames: append([]string{
+ "sortDirection",
+ "rankNumber",
+ }, topNSchema.GetGroupByTagNames()...),
+ },
+ }, nil
+}
diff --git a/pkg/schema/metadata.go b/pkg/schema/metadata.go
index 7692bdf2..75979ebd 100644
--- a/pkg/schema/metadata.go
+++ b/pkg/schema/metadata.go
@@ -180,16 +180,29 @@ func (sr *schemaRepo) Watcher() {
case EventAddOrUpdate:
switch evt.Kind {
case EventKindGroup:
- _, err = sr.StoreGroup(evt.Metadata)
+ _, err = sr.storeGroup(evt.Metadata.GetMetadata())
case EventKindResource:
- err = sr.storeResource(evt.Metadata)
+ err = sr.storeResource(evt.Metadata.GetMetadata())
+ case EventKindTopNAgg:
+ topNSchema := evt.Metadata.(*databasev1.TopNAggregation)
+ _, err = createOrUpdateTopNMeasure(context.Background(), sr.metadata.MeasureRegistry(), topNSchema)
+ if err != nil {
+ break
+ }
+ err = sr.storeResource(topNSchema.SourceMeasure)
}
case EventDelete:
switch evt.Kind {
case EventKindGroup:
- err = sr.deleteGroup(evt.Metadata)
+ err = sr.deleteGroup(evt.Metadata.GetMetadata())
case EventKindResource:
- err = sr.deleteResource(evt.Metadata)
+ err = sr.deleteResource(evt.Metadata.GetMetadata())
+ case EventKindTopNAgg:
+ topNSchema := evt.Metadata.(*databasev1.TopNAggregation)
+ err = multierr.Combine(
+ sr.deleteResource(topNSchema.SourceMeasure),
+ sr.storeResource(topNSchema.SourceMeasure),
+ )
}
}
if err != nil && !errors.Is(err, schema.ErrClosed) {
@@ -209,7 +222,7 @@ func (sr *schemaRepo) Watcher() {
}
}
-func (sr *schemaRepo) StoreGroup(groupMeta *commonv1.Metadata) (*group, error) {
+func (sr *schemaRepo) storeGroup(groupMeta *commonv1.Metadata) (*group, error) {
name := groupMeta.GetName()
sr.Lock()
defer sr.Unlock()
@@ -305,7 +318,7 @@ func (sr *schemaRepo) storeResource(metadata *commonv1.Metadata) error {
g, ok := sr.LoadGroup(metadata.Group)
if !ok {
var err error
- if g, err = sr.StoreGroup(&commonv1.Metadata{Name: metadata.Group}); err != nil {
+ if g, err = sr.storeGroup(&commonv1.Metadata{Name: metadata.Group}); err != nil {
return errors.WithMessagef(err, "create unknown group:%s", metadata.Group)
}
}
@@ -319,7 +332,7 @@ func (sr *schemaRepo) storeResource(metadata *commonv1.Metadata) error {
}
return errors.WithMessage(err, "fails to get the resource")
}
- _, err = g.(*group).storeResource(context.Background(), stm)
+ _, err = g.(*group).initResource(context.Background(), stm)
return err
}
@@ -404,6 +417,10 @@ func (g *group) init(name string) error {
if err != nil {
return err
}
+ return g.initBySchema(groupSchema)
+}
+
+func (g *group) initBySchema(groupSchema *commonv1.Group) error {
g.groupSchema.Store(groupSchema)
if g.isPortable() {
return nil
@@ -428,9 +445,7 @@ func (g *group) SupplyTSDB() io.Closer {
return g.db.Load().(io.Closer)
}
-func (g *group) storeResource(ctx context.Context, resourceSchema ResourceSchema) (Resource, error) {
- g.mapMutex.Lock()
- defer g.mapMutex.Unlock()
+func (g *group) initResource(ctx context.Context, resourceSchema ResourceSchema) (Resource, error) {
localCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()
idxRules, err := g.metadata.IndexRules(localCtx, resourceSchema.GetMetadata())
@@ -447,6 +462,12 @@ func (g *group) storeResource(ctx context.Context, resourceSchema ResourceSchema
return nil, innerErr
}
}
+ return g.storeResource(resourceSchema, idxRules, topNAggrs)
+}
+
+func (g *group) storeResource(resourceSchema ResourceSchema, idxRules []*databasev1.IndexRule, topNAggrs []*databasev1.TopNAggregation) (Resource, error) {
+ g.mapMutex.Lock()
+ defer g.mapMutex.Unlock()
resource := &resourceSpec{
schema: resourceSchema,
indexRules: idxRules,
diff --git a/pkg/schema/schema.go b/pkg/schema/schema.go
index 372f3035..025c29ec 100644
--- a/pkg/schema/schema.go
+++ b/pkg/schema/schema.go
@@ -23,6 +23,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"
)
// EventType defines actions of events.
@@ -42,6 +43,7 @@ type EventKind uint8
const (
EventKindGroup EventKind = iota
EventKindResource
+ EventKindTopNAgg
)
// Group is the root node, allowing get resources from its sub nodes.
@@ -53,7 +55,7 @@ type Group interface {
// MetadataEvent is the syncing message between metadata repo and this framework.
type MetadataEvent struct {
- Metadata *commonv1.Metadata
+ Metadata ResourceSchema
Typ EventType
Kind EventKind
}
@@ -92,8 +94,8 @@ type ResourceSupplier interface {
// Repository is the collection of several hierarchies groups by a "Group".
type Repository interface {
Watcher()
+ Init(schema.Kind) []int64
SendMetadataEvent(MetadataEvent)
- StoreGroup(groupMeta *commonv1.Metadata) (*group, error)
LoadGroup(name string) (Group, bool)
LoadResource(metadata *commonv1.Metadata) (Resource, bool)
Close()
(skywalking-banyandb) 01/02: Refactor stress test
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch metadata
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit d78d3b003ee6bac892f1a944c125394375d90e7e
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Mon Mar 11 22:11:45 2024 +0000
Refactor stress test
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
banyand/metadata/schema/watcher.go | 8 +++
banyand/queue/pub/pub.go | 2 +-
test/stress/{ => classic}/Makefile | 0
test/stress/{ => classic}/docker-compose.yaml | 14 ++--
test/stress/{ => classic}/env | 0
test/stress/{ => classic}/env.dev | 0
test/stress/{ => classic}/log4j2.xml | 0
test/stress/{ => classic}/prom/prometheus.yml | 0
test/stress/{ => classic}/scripts/consumer.js | 0
test/stress/{cases => }/istio/Makefile | 0
test/stress/{cases => }/istio/istio_suite_test.go | 0
test/stress/{cases => }/istio/repo.go | 0
test/stress/{cases => }/istio/report.md | 0
.../{cases => }/istio/testdata/access.tar.bz2 | Bin
.../{cases => }/istio/testdata/groups/group.yaml | 0
.../measure-default-index-rule-binding.yaml | 0
.../measure-minute-index-rule-binding.yaml | 0
.../index-rules/measure-default-index-rule.yaml | 0
.../index-rules/measure-minute-index-rule.yaml | 0
.../istio/testdata/measures/measure-default.yaml | 0
.../istio/testdata/measures/measure-minute.yaml | 0
.../{cases => }/istio/testdata/metrics/data.csv | 0
.../istio/testdata/metrics/intermediate.csv | 0
.../{cases => }/istio/testdata/metrics/result.csv | 0
test/stress/{ => trace}/Makefile | 6 +-
test/stress/{ => trace}/docker-compose.yaml | 78 +++++----------------
test/stress/{env.dev => trace/env} | 3 +-
test/stress/{ => trace}/log4j2.xml | 0
test/stress/{prom => trace}/prometheus.yml | 0
29 files changed, 35 insertions(+), 76 deletions(-)
diff --git a/banyand/metadata/schema/watcher.go b/banyand/metadata/schema/watcher.go
index bd9b2b8f..c21e3dec 100644
--- a/banyand/metadata/schema/watcher.go
+++ b/banyand/metadata/schema/watcher.go
@@ -65,10 +65,16 @@ func (w *watcher) Close() error {
func (w *watcher) allEvents() int64 {
cli := w.cli
var resp *clientv3.GetResponse
+ start := time.Now()
+ var eventHandleTime time.Duration
+ var eventSize int
for {
var err error
if resp, err = cli.Get(w.closer.Ctx(), w.key, clientv3.WithPrefix()); err == nil {
+ startHandle := time.Now()
+ eventSize = len(resp.Kvs)
w.handleAllEvents(resp.Kvs)
+ eventHandleTime = time.Since(startHandle)
break
}
select {
@@ -77,6 +83,8 @@ func (w *watcher) allEvents() int64 {
case <-time.After(1 * time.Second):
}
}
+ w.l.Info().Dur("event_handle_time", eventHandleTime).Dur("total_time", time.Since(start)).
+ Int("event_size", eventSize).Str("key", w.key).Msg("watcher all events")
return resp.Header.Revision
}
diff --git a/banyand/queue/pub/pub.go b/banyand/queue/pub/pub.go
index 73cf9927..8e6de582 100644
--- a/banyand/queue/pub/pub.go
+++ b/banyand/queue/pub/pub.go
@@ -107,7 +107,7 @@ func (p *pub) Publish(topic bus.Topic, messages ...bus.Message) (bus.Future, err
return multierr.Append(err, fmt.Errorf("failed to get client for node %s", node))
}
stream, errCreateStream := client.client.Send(context.Background())
- if err != nil {
+ if errCreateStream != nil {
return multierr.Append(err, fmt.Errorf("failed to get stream for node %s: %w", node, errCreateStream))
}
errSend = stream.Send(r)
diff --git a/test/stress/Makefile b/test/stress/classic/Makefile
similarity index 100%
copy from test/stress/Makefile
copy to test/stress/classic/Makefile
diff --git a/test/stress/docker-compose.yaml b/test/stress/classic/docker-compose.yaml
similarity index 92%
copy from test/stress/docker-compose.yaml
copy to test/stress/classic/docker-compose.yaml
index 9d28ff79..af8a24f8 100644
--- a/test/stress/docker-compose.yaml
+++ b/test/stress/classic/docker-compose.yaml
@@ -25,11 +25,11 @@ services:
banyandb:
user: "${USER_ID}:${GROUP_ID}"
extends:
- file: ../docker/base-compose.yml
+ file: ../../docker/base-compose.yml
service: banyandb
build:
dockerfile: ./docker/Dockerfile
- context: ../..
+ context: ../../..
volumes:
- /tmp/banyandb-stress-agent:/tmp:rw,delgated
ports:
@@ -45,7 +45,7 @@ services:
oap:
extends:
- file: ../docker/base-compose.yml
+ file: ../../docker/base-compose.yml
service: oap
environment:
SW_STORAGE: banyandb
@@ -61,14 +61,14 @@ services:
agent:
extends:
- file: ../docker/base-compose.yml
+ file: ../../docker/base-compose.yml
service: agent
networks:
- test
provider:
extends:
- file: ../docker/base-compose.yml
+ file: ../../docker/base-compose.yml
service: provider
ports:
- 9090
@@ -82,7 +82,7 @@ services:
consumer:
extends:
- file: ../docker/base-compose.yml
+ file: ../../docker/base-compose.yml
service: consumer
ports:
- 9092
@@ -98,7 +98,7 @@ services:
traffic_loader:
extends:
- file: ../docker/base-compose.yml
+ file: ../../docker/base-compose.yml
service: traffic_loader
command: run --vus ${VUS} --duration 1h /scripts/consumer.js
volumes:
diff --git a/test/stress/env b/test/stress/classic/env
similarity index 100%
rename from test/stress/env
rename to test/stress/classic/env
diff --git a/test/stress/env.dev b/test/stress/classic/env.dev
similarity index 100%
copy from test/stress/env.dev
copy to test/stress/classic/env.dev
diff --git a/test/stress/log4j2.xml b/test/stress/classic/log4j2.xml
similarity index 100%
copy from test/stress/log4j2.xml
copy to test/stress/classic/log4j2.xml
diff --git a/test/stress/prom/prometheus.yml b/test/stress/classic/prom/prometheus.yml
similarity index 100%
copy from test/stress/prom/prometheus.yml
copy to test/stress/classic/prom/prometheus.yml
diff --git a/test/stress/scripts/consumer.js b/test/stress/classic/scripts/consumer.js
similarity index 100%
rename from test/stress/scripts/consumer.js
rename to test/stress/classic/scripts/consumer.js
diff --git a/test/stress/cases/istio/Makefile b/test/stress/istio/Makefile
similarity index 100%
rename from test/stress/cases/istio/Makefile
rename to test/stress/istio/Makefile
diff --git a/test/stress/cases/istio/istio_suite_test.go b/test/stress/istio/istio_suite_test.go
similarity index 100%
rename from test/stress/cases/istio/istio_suite_test.go
rename to test/stress/istio/istio_suite_test.go
diff --git a/test/stress/cases/istio/repo.go b/test/stress/istio/repo.go
similarity index 100%
rename from test/stress/cases/istio/repo.go
rename to test/stress/istio/repo.go
diff --git a/test/stress/cases/istio/report.md b/test/stress/istio/report.md
similarity index 100%
rename from test/stress/cases/istio/report.md
rename to test/stress/istio/report.md
diff --git a/test/stress/cases/istio/testdata/access.tar.bz2 b/test/stress/istio/testdata/access.tar.bz2
similarity index 100%
rename from test/stress/cases/istio/testdata/access.tar.bz2
rename to test/stress/istio/testdata/access.tar.bz2
diff --git a/test/stress/cases/istio/testdata/groups/group.yaml b/test/stress/istio/testdata/groups/group.yaml
similarity index 100%
rename from test/stress/cases/istio/testdata/groups/group.yaml
rename to test/stress/istio/testdata/groups/group.yaml
diff --git a/test/stress/cases/istio/testdata/index-rule-bindings/measure-default-index-rule-binding.yaml b/test/stress/istio/testdata/index-rule-bindings/measure-default-index-rule-binding.yaml
similarity index 100%
rename from test/stress/cases/istio/testdata/index-rule-bindings/measure-default-index-rule-binding.yaml
rename to test/stress/istio/testdata/index-rule-bindings/measure-default-index-rule-binding.yaml
diff --git a/test/stress/cases/istio/testdata/index-rule-bindings/measure-minute-index-rule-binding.yaml b/test/stress/istio/testdata/index-rule-bindings/measure-minute-index-rule-binding.yaml
similarity index 100%
rename from test/stress/cases/istio/testdata/index-rule-bindings/measure-minute-index-rule-binding.yaml
rename to test/stress/istio/testdata/index-rule-bindings/measure-minute-index-rule-binding.yaml
diff --git a/test/stress/cases/istio/testdata/index-rules/measure-default-index-rule.yaml b/test/stress/istio/testdata/index-rules/measure-default-index-rule.yaml
similarity index 100%
rename from test/stress/cases/istio/testdata/index-rules/measure-default-index-rule.yaml
rename to test/stress/istio/testdata/index-rules/measure-default-index-rule.yaml
diff --git a/test/stress/cases/istio/testdata/index-rules/measure-minute-index-rule.yaml b/test/stress/istio/testdata/index-rules/measure-minute-index-rule.yaml
similarity index 100%
rename from test/stress/cases/istio/testdata/index-rules/measure-minute-index-rule.yaml
rename to test/stress/istio/testdata/index-rules/measure-minute-index-rule.yaml
diff --git a/test/stress/cases/istio/testdata/measures/measure-default.yaml b/test/stress/istio/testdata/measures/measure-default.yaml
similarity index 100%
rename from test/stress/cases/istio/testdata/measures/measure-default.yaml
rename to test/stress/istio/testdata/measures/measure-default.yaml
diff --git a/test/stress/cases/istio/testdata/measures/measure-minute.yaml b/test/stress/istio/testdata/measures/measure-minute.yaml
similarity index 100%
rename from test/stress/cases/istio/testdata/measures/measure-minute.yaml
rename to test/stress/istio/testdata/measures/measure-minute.yaml
diff --git a/test/stress/cases/istio/testdata/metrics/data.csv b/test/stress/istio/testdata/metrics/data.csv
similarity index 100%
rename from test/stress/cases/istio/testdata/metrics/data.csv
rename to test/stress/istio/testdata/metrics/data.csv
diff --git a/test/stress/cases/istio/testdata/metrics/intermediate.csv b/test/stress/istio/testdata/metrics/intermediate.csv
similarity index 100%
rename from test/stress/cases/istio/testdata/metrics/intermediate.csv
rename to test/stress/istio/testdata/metrics/intermediate.csv
diff --git a/test/stress/cases/istio/testdata/metrics/result.csv b/test/stress/istio/testdata/metrics/result.csv
similarity index 100%
rename from test/stress/cases/istio/testdata/metrics/result.csv
rename to test/stress/istio/testdata/metrics/result.csv
diff --git a/test/stress/Makefile b/test/stress/trace/Makefile
similarity index 89%
rename from test/stress/Makefile
rename to test/stress/trace/Makefile
index 789c4934..ca57b2db 100644
--- a/test/stress/Makefile
+++ b/test/stress/trace/Makefile
@@ -28,11 +28,7 @@ cli_env := DOCKER_BUILDKIT=1 USER_ID=$(shell id -u) GROUP_ID=$(shell id -g)
.PHONY: clean
clean:
- rm -rf /tmp/banyandb-stress-agent
-
-.PHONY: dev-up
-dev-up: clean
- $(cli_env) docker compose $(CLI_ARGS) --env-file ./env.dev up --build
+ rm -rf /tmp/banyandb-stress-trace
.PHONY: up
up: clean
diff --git a/test/stress/docker-compose.yaml b/test/stress/trace/docker-compose.yaml
similarity index 61%
rename from test/stress/docker-compose.yaml
rename to test/stress/trace/docker-compose.yaml
index 9d28ff79..9554ddf3 100644
--- a/test/stress/docker-compose.yaml
+++ b/test/stress/trace/docker-compose.yaml
@@ -20,22 +20,28 @@ services:
group_add:
- '${GROUP_ID}'
volumes:
- - /tmp/banyandb-stress-agent:/tmp/change-ownership
+ - /tmp/banyandb-stress-trace:/tmp/change-ownership
command: chown -R ${USER_ID}:${GROUP_ID} /tmp/change-ownership
+
banyandb:
user: "${USER_ID}:${GROUP_ID}"
extends:
- file: ../docker/base-compose.yml
+ file: ../../docker/base-compose.yml
service: banyandb
build:
dockerfile: ./docker/Dockerfile
- context: ../..
+ context: ../../..
volumes:
- - /tmp/banyandb-stress-agent:/tmp:rw,delgated
+ - /tmp/banyandb-stress-trace:/tmp:rw,delgated
ports:
- 17913:17913
- 6060:6060
- 2121:2121
+ deploy:
+ resources:
+ limits:
+ cpus: "4"
+ memory: 4G
networks:
- test
- monitoring
@@ -45,12 +51,17 @@ services:
oap:
extends:
- file: ../docker/base-compose.yml
+ file: ../../docker/base-compose.yml
service: oap
environment:
SW_STORAGE: banyandb
ports:
- 12800
+ deploy:
+ resources:
+ limits:
+ cpus: "4"
+ memory: 4G
volumes:
- ./log4j2.xml:/skywalking/config/log4j2.xml
networks:
@@ -59,60 +70,6 @@ services:
banyandb:
condition: service_healthy
- agent:
- extends:
- file: ../docker/base-compose.yml
- service: agent
- networks:
- - test
-
- provider:
- extends:
- file: ../docker/base-compose.yml
- service: provider
- ports:
- - 9090
- networks:
- - test
- depends_on:
- oap:
- condition: service_healthy
- agent:
- condition: service_completed_successfully
-
- consumer:
- extends:
- file: ../docker/base-compose.yml
- service: consumer
- ports:
- - 9092
- networks:
- - test
- depends_on:
- oap:
- condition: service_healthy
- provider:
- condition: service_healthy
- agent:
- condition: service_completed_successfully
-
- traffic_loader:
- extends:
- file: ../docker/base-compose.yml
- service: traffic_loader
- command: run --vus ${VUS} --duration 1h /scripts/consumer.js
- volumes:
- - ./scripts:/scripts
- networks:
- - test
- depends_on:
- oap:
- condition: service_healthy
- provider:
- condition: service_healthy
- consumer:
- condition: service_healthy
-
prometheus:
image: prom/prometheus:latest
container_name: prometheus
@@ -120,7 +77,7 @@ services:
profiles:
- "monitoring"
volumes:
- - ./prom/prometheus.yml:/etc/prometheus/prometheus.yml
+ - ./prometheus.yml:/etc/prometheus/prometheus.yml
- prometheus_data:/prometheus
command:
- '--config.file=/etc/prometheus/prometheus.yml'
@@ -137,5 +94,4 @@ networks:
monitoring:
volumes:
- sw_agent: {}
prometheus_data: {}
diff --git a/test/stress/env.dev b/test/stress/trace/env
similarity index 99%
rename from test/stress/env.dev
rename to test/stress/trace/env
index 6e0d3224..94bce219 100644
--- a/test/stress/env.dev
+++ b/test/stress/trace/env
@@ -13,6 +13,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
+
SW_AGENT_JAVA_COMMIT=3f88d735ba2bfd1196aff946502447d4b14450c8
SW_AGENT_SATELLITE_COMMIT=ea27a3f4e126a24775fe12e2aa2695bcb23d99c3
SW_AGENT_NGINX_LUA_COMMIT=c3cee4841798a147d83b96a10914d4ac0e11d0aa
@@ -27,5 +28,3 @@ SW_CTL_COMMIT=6b2eb0011e38b630db6af7203db215806bd141ed
SW_OAP_COMMIT=e5c308d4358d5b02a658b8ad7f153aec99b8e63a
SW_AGENT_E2E_SERVICE_PROVIDER_COMMIT=cc7a2c9e97fd2c421adbe3e9c471688459a446d9
-
-VUS=1
diff --git a/test/stress/log4j2.xml b/test/stress/trace/log4j2.xml
similarity index 100%
rename from test/stress/log4j2.xml
rename to test/stress/trace/log4j2.xml
diff --git a/test/stress/prom/prometheus.yml b/test/stress/trace/prometheus.yml
similarity index 100%
rename from test/stress/prom/prometheus.yml
rename to test/stress/trace/prometheus.yml