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:11 UTC
(skywalking-banyandb) 03/03: Remove SetSchema from measure cache
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