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/05/23 12:15:08 UTC
(skywalking-banyandb) branch patch-metadata updated (fdf7b1fd -> b74efc55)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a change to branch patch-metadata
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
discard fdf7b1fd Add debug messages
add 7dfd3b98 Add benchmarks for stream filtering and sorting (#449)
add 030e548b [GsoC][BanyanDB] Self-Observability: Create Internal Group (#446)
new 8aab993f Add debug messages
new 37e6037d Add debug info
new b74efc55 Remove SetSchema from measure cache
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (fdf7b1fd)
\
N -- N -- N refs/heads/patch-metadata (b74efc55)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 3 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.
Summary of changes:
CHANGES.md | 3 +
banyand/liaison/grpc/registry_test.go | 3 +-
banyand/measure/measure_suite_test.go | 3 +-
banyand/measure/query.go | 5 -
banyand/measure/write.go | 5 +-
banyand/metadata/client.go | 7 +-
banyand/metadata/{ => embeddedserver}/server.go | 12 +-
banyand/metadata/metadata_test.go | 8 +-
banyand/metadata/schema/checker.go | 6 +-
banyand/metadata/schema/checker_test.go | 15 +-
banyand/metadata/schema/etcd.go | 6 +-
banyand/metadata/schema/etcd_test.go | 75 ++--
banyand/metadata/schema/node.go | 13 +-
banyand/metadata/schema/register_test.go | 42 +--
banyand/metadata/schema/schema.go | 5 +-
banyand/metadata/schema/schema_suite_test.go | 2 +-
banyand/metadata/schema/watcher.go | 8 +-
banyand/metadata/schema/watcher_test.go | 25 +-
.../{meter_noop.go => self_observability.go} | 35 +-
banyand/observability/service.go | 13 +-
banyand/query/processor_topn.go | 13 +-
banyand/stream/benchmark_test.go | 408 +++++++++++++++++++++
banyand/stream/index.go | 19 +-
banyand/stream/iter.go | 3 -
banyand/stream/query.go | 2 +-
banyand/stream/stream_suite_test.go | 3 +-
bydbctl/internal/cmd/group_test.go | 2 +-
pkg/cmdsetup/data.go | 2 +-
pkg/cmdsetup/liaison.go | 2 +-
pkg/cmdsetup/standalone.go | 6 +-
pkg/schema/cache.go | 18 +-
pkg/timestamp/range.go | 2 +-
32 files changed, 605 insertions(+), 166 deletions(-)
rename banyand/metadata/{ => embeddedserver}/server.go (89%)
copy banyand/observability/{meter_noop.go => self_observability.go} (56%)
create mode 100644 banyand/stream/benchmark_test.go
(skywalking-banyandb) 01/03: Add debug messages
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch patch-metadata
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 8aab993fbd7eb3856657b0b278ef6c9ad8de08a9
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Tue May 21 22:59:01 2024 +0800
Add debug messages
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
banyand/measure/write.go | 5 ++++-
pkg/schema/cache.go | 25 ++++++++++++++++++++-----
2 files changed, 24 insertions(+), 6 deletions(-)
diff --git a/banyand/measure/write.go b/banyand/measure/write.go
index 62642f68..95fc6ff0 100644
--- a/banyand/measure/write.go
+++ b/banyand/measure/write.go
@@ -93,10 +93,13 @@ func (w *writeCallback) handle(dst map[string]*dataPointsInGroup, writeEvent *me
dpg.tables = append(dpg.tables, dpt)
}
dpt.dataPoints.timestamps = append(dpt.dataPoints.timestamps, ts)
- stm, ok := w.schemaRepo.loadMeasure(writeEvent.GetRequest().GetMetadata())
+ stm, ok := w.schemaRepo.loadMeasure(req.GetMetadata())
if !ok {
return nil, fmt.Errorf("cannot find measure definition: %s", writeEvent.GetRequest().GetMetadata())
}
+ if stm.name != req.Metadata.Name {
+ return nil, fmt.Errorf("measure name mismatch: %s != %s", stm.name, req.Metadata.Name)
+ }
fLen := len(req.DataPoint.GetTagFamilies())
if fLen < 1 {
return nil, fmt.Errorf("%s has no tag family", req.Metadata)
diff --git a/pkg/schema/cache.go b/pkg/schema/cache.go
index 2e8a9c85..64db1eea 100644
--- a/pkg/schema/cache.go
+++ b/pkg/schema/cache.go
@@ -20,6 +20,8 @@ package schema
import (
"context"
"io"
+ "path"
+ "strings"
"sync"
"sync/atomic"
"time"
@@ -295,13 +297,15 @@ func (sr *schemaRepo) LoadGroup(name string) (Group, bool) {
}
func (sr *schemaRepo) LoadResource(metadata *commonv1.Metadata) (Resource, bool) {
- _, ok := sr.LoadGroup(metadata.Group)
+ k := getKey(metadata)
+ s, ok := sr.resourceMap.Load(k)
if !ok {
return nil, false
}
- s, ok := sr.resourceMap.Load(metadata.Name)
- if !ok {
- return nil, false
+ r := s.(Resource)
+ rk := getKey(r.Schema().GetMetadata())
+ if k != rk {
+ logger.Panicf("'%s' is not the expected metadata by '%s', gotten [%s]", rk, k, r.Schema())
}
return s.(Resource), true
}
@@ -316,7 +320,7 @@ func (sr *schemaRepo) storeResource(g Group, stm ResourceSchema,
indexRules: idxRules,
aggregations: topNAggrs,
}
- key := stm.GetMetadata().GetName()
+ key := getKey(stm.GetMetadata())
pre, loadedPre := sr.resourceMap.Load(key)
var preResource *resourceSpec
if loadedPre {
@@ -334,6 +338,13 @@ func (sr *schemaRepo) storeResource(g Group, stm ResourceSchema,
return err
}
resource.delegated = sm
+ rk := getKey(resource.schema.GetMetadata())
+ if rk != key {
+ return errors.Errorf("'%s' is not the expected metadata by '%s'", rk, key)
+ }
+ if strings.HasPrefix(key, "measure-minute/endpoint_sla_minute") {
+ logger.Infof("storeResource: %s: %s", key, resource)
+ }
sr.resourceMap.Store(key, resource)
if loadedPre {
return preResource.Close()
@@ -341,6 +352,10 @@ func (sr *schemaRepo) storeResource(g Group, stm ResourceSchema,
return nil
}
+func getKey(metadata *commonv1.Metadata) string {
+ return path.Join(metadata.GetGroup(), metadata.GetName())
+}
+
func (sr *schemaRepo) initResource(metadata *commonv1.Metadata) error {
g, ok := sr.LoadGroup(metadata.Group)
if !ok {
(skywalking-banyandb) 02/03: Add debug info
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch patch-metadata
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 37e6037d6139e77bdd39c115224aeb0447f63238
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Tue May 21 22:08:37 2024 +0000
Add debug info
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
banyand/measure/write.go | 4 ++--
banyand/metadata/schema/watcher.go | 8 ++++++--
pkg/schema/cache.go | 17 ++++++++++++++---
3 files changed, 22 insertions(+), 7 deletions(-)
diff --git a/banyand/measure/write.go b/banyand/measure/write.go
index 95fc6ff0..849e0a9c 100644
--- a/banyand/measure/write.go
+++ b/banyand/measure/write.go
@@ -95,10 +95,10 @@ func (w *writeCallback) handle(dst map[string]*dataPointsInGroup, writeEvent *me
dpt.dataPoints.timestamps = append(dpt.dataPoints.timestamps, ts)
stm, ok := w.schemaRepo.loadMeasure(req.GetMetadata())
if !ok {
- return nil, fmt.Errorf("cannot find measure definition: %s", writeEvent.GetRequest().GetMetadata())
+ return nil, fmt.Errorf("cannot find measure definition: %s", req.GetMetadata())
}
if stm.name != req.Metadata.Name {
- return nil, fmt.Errorf("measure name mismatch: %s != %s", stm.name, req.Metadata.Name)
+ return nil, fmt.Errorf("measure name mismatch: %s != %s", stm.schema, req.Metadata)
}
fLen := len(req.DataPoint.GetTagFamilies())
if fLen < 1 {
diff --git a/banyand/metadata/schema/watcher.go b/banyand/metadata/schema/watcher.go
index b95000d2..148dab64 100644
--- a/banyand/metadata/schema/watcher.go
+++ b/banyand/metadata/schema/watcher.go
@@ -19,6 +19,7 @@ package schema
import (
"errors"
+ "strings"
"time"
mvccpb "go.etcd.io/etcd/api/v3/mvccpb"
@@ -155,7 +156,7 @@ OUTER:
case <-w.closer.CloseNotify():
return
default:
- w.handle(event)
+ w.handle(event, &watchResp)
}
}
}
@@ -163,7 +164,7 @@ OUTER:
}
}
-func (w *watcher) handle(watchEvent *clientv3.Event) {
+func (w *watcher) handle(watchEvent *clientv3.Event, watchResp *clientv3.WatchResponse) {
switch watchEvent.Type {
case mvccpb.PUT:
md, err := w.kind.Unmarshal(watchEvent.Kv)
@@ -171,6 +172,9 @@ func (w *watcher) handle(watchEvent *clientv3.Event) {
w.l.Error().AnErr("err", err).Msg("failed to unmarshal message")
return
}
+ if strings.HasPrefix(md.Name, "endpoint_sla_minute") {
+ logger.Infof("watching '%s' handle: header[%s] all events[%d] body[%v]", w.key, &watchResp.Header, len(watchResp.Events), md)
+ }
w.handler.OnAddOrUpdate(md)
case mvccpb.DELETE:
md, err := w.kind.Unmarshal(watchEvent.PrevKv)
diff --git a/pkg/schema/cache.go b/pkg/schema/cache.go
index 64db1eea..2bfb4b1d 100644
--- a/pkg/schema/cache.go
+++ b/pkg/schema/cache.go
@@ -89,7 +89,7 @@ func (rs *resourceSpec) isNewThan(other *resourceSpec) bool {
return true
}
-const defaultWorkerNum = 10
+const defaultWorkerNum = 1
var _ Repository = (*schemaRepo)(nil)
@@ -177,6 +177,10 @@ func (sr *schemaRepo) Watcher() {
if e := sr.l.Debug(); e.Enabled() {
e.Interface("event", evt).Msg("received an event")
}
+ k := getKey(evt.Metadata.GetMetadata())
+ if strings.HasPrefix(k, "measure-minute/endpoint_sla_minute") {
+ logger.Infof("EventKindResource: %v", evt)
+ }
var err error
switch evt.Typ {
case EventAddOrUpdate:
@@ -191,7 +195,11 @@ func (sr *schemaRepo) Watcher() {
if err != nil {
break
}
- err = sr.initResource(topNSchema.SourceMeasure)
+ k := getKey(topNSchema.SourceMeasure)
+ if strings.HasPrefix(k, "measure-minute/endpoint_sla_minute") {
+ logger.Infof("EventKindTopNAgg: %s", topNSchema.SourceMeasure)
+ }
+ // err = sr.initResource(topNSchema.SourceMeasure)
}
case EventDelete:
switch evt.Kind {
@@ -307,6 +315,9 @@ func (sr *schemaRepo) LoadResource(metadata *commonv1.Metadata) (Resource, bool)
if k != rk {
logger.Panicf("'%s' is not the expected metadata by '%s', gotten [%s]", rk, k, r.Schema())
}
+ if strings.HasPrefix(k, "measure-minute/endpoint_sla_minute") {
+ logger.Infof("LoadResource: %s: %s", k, r.Schema())
+ }
return s.(Resource), true
}
@@ -343,7 +354,7 @@ func (sr *schemaRepo) storeResource(g Group, stm ResourceSchema,
return errors.Errorf("'%s' is not the expected metadata by '%s'", rk, key)
}
if strings.HasPrefix(key, "measure-minute/endpoint_sla_minute") {
- logger.Infof("storeResource: %s: %s", key, resource)
+ logger.Infof("storeResource: %s: %s, agg: %s", key, resource.Schema(), resource.aggregations)
}
sr.resourceMap.Store(key, resource)
if loadedPre {
(skywalking-banyandb) 03/03: Remove SetSchema from measure cache
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch patch-metadata
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit b74efc55af14d8e0b5c1455634bff6eb538156d3
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Thu May 23 08:05:09 2024 +0000
Remove SetSchema from measure cache
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
CHANGES.md | 1 +
banyand/measure/query.go | 5 -----
banyand/measure/write.go | 3 ---
banyand/metadata/schema/watcher.go | 8 ++------
banyand/query/processor_topn.go | 13 +++++--------
pkg/schema/cache.go | 31 ++++---------------------------
6 files changed, 12 insertions(+), 49 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index d4a81f21..8d029b2d 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -12,6 +12,7 @@ Release Notes.
- Fix the bug that topN query doesn't return when an error occurs.
- Data race in the hot series index selection.
+- Remove SetSchema from measure cache which could change the schema in the cache.
## 0.6.0
diff --git a/banyand/measure/query.go b/banyand/measure/query.go
index 88a91880..57bd381d 100644
--- a/banyand/measure/query.go
+++ b/banyand/measure/query.go
@@ -53,15 +53,10 @@ type Measure interface {
Query(ctx context.Context, opts pbv1.MeasureQueryOptions) (pbv1.MeasureQueryResult, error)
GetSchema() *databasev1.Measure
GetIndexRules() []*databasev1.IndexRule
- SetSchema(schema *databasev1.Measure)
}
var _ Measure = (*measure)(nil)
-func (s *measure) SetSchema(schema *databasev1.Measure) {
- s.schema = schema
-}
-
type queryOptions struct {
pbv1.MeasureQueryOptions
minTimestamp int64
diff --git a/banyand/measure/write.go b/banyand/measure/write.go
index 849e0a9c..32dd4a71 100644
--- a/banyand/measure/write.go
+++ b/banyand/measure/write.go
@@ -97,9 +97,6 @@ func (w *writeCallback) handle(dst map[string]*dataPointsInGroup, writeEvent *me
if !ok {
return nil, fmt.Errorf("cannot find measure definition: %s", req.GetMetadata())
}
- if stm.name != req.Metadata.Name {
- return nil, fmt.Errorf("measure name mismatch: %s != %s", stm.schema, req.Metadata)
- }
fLen := len(req.DataPoint.GetTagFamilies())
if fLen < 1 {
return nil, fmt.Errorf("%s has no tag family", req.Metadata)
diff --git a/banyand/metadata/schema/watcher.go b/banyand/metadata/schema/watcher.go
index 148dab64..e4945866 100644
--- a/banyand/metadata/schema/watcher.go
+++ b/banyand/metadata/schema/watcher.go
@@ -19,7 +19,6 @@ package schema
import (
"errors"
- "strings"
"time"
mvccpb "go.etcd.io/etcd/api/v3/mvccpb"
@@ -169,17 +168,14 @@ func (w *watcher) handle(watchEvent *clientv3.Event, watchResp *clientv3.WatchRe
case mvccpb.PUT:
md, err := w.kind.Unmarshal(watchEvent.Kv)
if err != nil {
- w.l.Error().AnErr("err", err).Msg("failed to unmarshal message")
+ w.l.Error().Stringer("event_header", &watchResp.Header).AnErr("err", err).Msg("failed to unmarshal message")
return
}
- if strings.HasPrefix(md.Name, "endpoint_sla_minute") {
- logger.Infof("watching '%s' handle: header[%s] all events[%d] body[%v]", w.key, &watchResp.Header, len(watchResp.Events), md)
- }
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")
+ w.l.Error().Stringer("event_header", &watchResp.Header).AnErr("err", err).Msg("failed to unmarshal message")
return
}
w.handler.OnDelete(md)
diff --git a/banyand/query/processor_topn.go b/banyand/query/processor_topn.go
index 99dc8b42..de54a1da 100644
--- a/banyand/query/processor_topn.go
+++ b/banyand/query/processor_topn.go
@@ -91,23 +91,21 @@ func (t *topNQueryProcessor) Rev(message bus.Message) (resp bus.Message) {
return
}
- schema, err := t.metaService.MeasureRegistry().GetMeasure(context.TODO(), topNMetadata)
+ topNResultMeasure, err := t.measureService.Measure(topNMetadata)
if err != nil {
t.log.Error().Err(err).
Str("topN", topNMetadata.GetName()).
- Msg("fail to find topN measure")
+ Msg("fail to find topN result measure")
return
}
- sourceMeasureSchema := sourceMeasure.GetSchema()
- sourceMeasure.SetSchema(schema)
- s, err := logical_measure.BuildTopNSchema(schema)
+ s, err := logical_measure.BuildTopNSchema(topNResultMeasure.GetSchema())
if err != nil {
t.log.Error().Err(err).
Str("topN", topNMetadata.GetName()).
Msg("fail to build schema")
}
- plan, err := logical_measure.TopNAnalyze(context.TODO(), request, schema, sourceMeasureSchema, topNSchema, s)
+ plan, err := logical_measure.TopNAnalyze(context.TODO(), request, topNResultMeasure.GetSchema(), sourceMeasure.GetSchema(), topNSchema, s)
if err != nil {
resp = bus.NewMessage(bus.MessageID(now), common.NewError("fail to analyze the query request for topn %s: %v", topNMetadata.GetName(), err))
return
@@ -117,7 +115,7 @@ func (t *topNQueryProcessor) Rev(message bus.Message) (resp bus.Message) {
e.Str("plan", plan.String()).Msg("topn plan")
}
- mIterator, err := plan.(executor.MeasureExecutable).Execute(executor.WithMeasureExecutionContext(context.Background(), sourceMeasure))
+ mIterator, err := plan.(executor.MeasureExecutable).Execute(executor.WithMeasureExecutionContext(context.Background(), topNResultMeasure))
if err != nil {
ml.Error().Err(err).RawJSON("req", logger.Proto(request)).Msg("fail to close the topn plan")
resp = bus.NewMessage(bus.MessageID(now), common.NewError("fail to execute the topn plan for measure %s: %v", topNMetadata.GetName(), err))
@@ -129,7 +127,6 @@ func (t *topNQueryProcessor) Rev(message bus.Message) (resp bus.Message) {
}
}()
- sourceMeasure.SetSchema(sourceMeasureSchema)
result := make([]*measurev1.DataPoint, 0)
for mIterator.Next() {
current := mIterator.Current()
diff --git a/pkg/schema/cache.go b/pkg/schema/cache.go
index 2bfb4b1d..70045f3e 100644
--- a/pkg/schema/cache.go
+++ b/pkg/schema/cache.go
@@ -21,7 +21,7 @@ import (
"context"
"io"
"path"
- "strings"
+ "runtime"
"sync"
"sync/atomic"
"time"
@@ -89,7 +89,7 @@ func (rs *resourceSpec) isNewThan(other *resourceSpec) bool {
return true
}
-const defaultWorkerNum = 1
+var defaultWorkerNum = runtime.GOMAXPROCS(-1)
var _ Repository = (*schemaRepo)(nil)
@@ -177,10 +177,6 @@ func (sr *schemaRepo) Watcher() {
if e := sr.l.Debug(); e.Enabled() {
e.Interface("event", evt).Msg("received an event")
}
- k := getKey(evt.Metadata.GetMetadata())
- if strings.HasPrefix(k, "measure-minute/endpoint_sla_minute") {
- logger.Infof("EventKindResource: %v", evt)
- }
var err error
switch evt.Typ {
case EventAddOrUpdate:
@@ -195,11 +191,7 @@ func (sr *schemaRepo) Watcher() {
if err != nil {
break
}
- k := getKey(topNSchema.SourceMeasure)
- if strings.HasPrefix(k, "measure-minute/endpoint_sla_minute") {
- logger.Infof("EventKindTopNAgg: %s", topNSchema.SourceMeasure)
- }
- // err = sr.initResource(topNSchema.SourceMeasure)
+ err = sr.initResource(topNSchema.SourceMeasure)
}
case EventDelete:
switch evt.Kind {
@@ -310,14 +302,6 @@ func (sr *schemaRepo) LoadResource(metadata *commonv1.Metadata) (Resource, bool)
if !ok {
return nil, false
}
- r := s.(Resource)
- rk := getKey(r.Schema().GetMetadata())
- if k != rk {
- logger.Panicf("'%s' is not the expected metadata by '%s', gotten [%s]", rk, k, r.Schema())
- }
- if strings.HasPrefix(k, "measure-minute/endpoint_sla_minute") {
- logger.Infof("LoadResource: %s: %s", k, r.Schema())
- }
return s.(Resource), true
}
@@ -349,13 +333,6 @@ func (sr *schemaRepo) storeResource(g Group, stm ResourceSchema,
return err
}
resource.delegated = sm
- rk := getKey(resource.schema.GetMetadata())
- if rk != key {
- return errors.Errorf("'%s' is not the expected metadata by '%s'", rk, key)
- }
- if strings.HasPrefix(key, "measure-minute/endpoint_sla_minute") {
- logger.Infof("storeResource: %s: %s, agg: %s", key, resource.Schema(), resource.aggregations)
- }
sr.resourceMap.Store(key, resource)
if loadedPre {
return preResource.Close()
@@ -406,7 +383,7 @@ func (sr *schemaRepo) initResource(metadata *commonv1.Metadata) error {
}
func (sr *schemaRepo) deleteResource(metadata *commonv1.Metadata) error {
- key := metadata.GetName()
+ key := getKey(metadata)
pre, loaded := sr.resourceMap.LoadAndDelete(key)
if !loaded {
return nil