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 2025/01/31 03:23:06 UTC
(skywalking-banyandb) branch metadata updated: Fix issues found in test
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
The following commit(s) were added to refs/heads/metadata by this push:
new 7afa72c6 Fix issues found in test
7afa72c6 is described below
commit 7afa72c67274897537a1c0e849fe0bd9030f77b1
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Fri Jan 31 03:05:17 2025 +0000
Fix issues found in test
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
CHANGES.md | 1 +
banyand/measure/block.go | 14 ++---
banyand/metadata/client.go | 28 ++++++----
banyand/metadata/embeddedserver/server.go | 1 +
banyand/metadata/schema/etcd.go | 45 ++++++++++------
banyand/metadata/schema/schema.go | 3 +-
banyand/metadata/schema/watcher.go | 61 +++++++++++++++++-----
banyand/metadata/schema/watcher_test.go | 20 +++++--
banyand/stream/block.go | 11 +++-
banyand/stream/query.go | 2 +-
banyand/stream/query_by_ts.go | 2 +-
go.mod | 2 +-
pkg/query/model/model.go | 27 ++++++++++
scripts/ci/check/version_test.go | 2 +-
.../standalone/cold_query/query_suite_test.go | 4 +-
15 files changed, 162 insertions(+), 61 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 34656715..1db15cdf 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -22,6 +22,7 @@ Release Notes.
- Metadata: Wait for the existing registration to be removed before registering the node.
- Stream: Introduce the batch scan to improve the performance of the query and limit the memory usage.
- Add memory protector to protect the memory usage of the system. It will limit the memory usage of the querying.
+- Metadata: Introduce the periodic sync to sync the metadata from the etcd to the local cache in case of the loss of the events.
### Bug Fixes
diff --git a/banyand/measure/block.go b/banyand/measure/block.go
index 37ce7a3b..80f6ca00 100644
--- a/banyand/measure/block.go
+++ b/banyand/measure/block.go
@@ -692,24 +692,24 @@ func (bc *blockCursor) loadData(tmpBlock *block) bool {
tmpBlock.reset()
cfm := make([]columnMetadata, 0, len(bc.fieldProjection))
NEXT_FIELD:
- for j := range bc.fieldProjection {
- for i := range bc.bm.field.columnMetadata {
- if bc.bm.field.columnMetadata[i].name == bc.fieldProjection[j] {
- cfm = append(cfm, bc.bm.field.columnMetadata[i])
+ for _, fp := range bc.fieldProjection {
+ for _, cm := range bc.bm.field.columnMetadata {
+ if cm.name == fp {
+ cfm = append(cfm, cm)
continue NEXT_FIELD
}
}
cfm = append(cfm, columnMetadata{
- name: bc.fieldProjection[j],
+ name: fp,
valueType: pbv1.ValueTypeUnknown,
})
}
bc.bm.field.columnMetadata = cfm
bc.bm.tagProjection = bc.tagProjection
var tf map[string]*dataBlock
- for i := range bc.tagProjection {
+ for _, tp := range bc.tagProjection {
for tfName, block := range bc.bm.tagFamilies {
- if bc.tagProjection[i].Family == tfName {
+ if tp.Family == tfName {
if tf == nil {
tf = make(map[string]*dataBlock, len(bc.tagProjection))
}
diff --git a/banyand/metadata/client.go b/banyand/metadata/client.go
index ba0f57a0..b70336f8 100644
--- a/banyand/metadata/client.go
+++ b/banyand/metadata/client.go
@@ -62,17 +62,18 @@ func NewClient(forceRegisterNode bool) (Service, error) {
}
type clientService struct {
- schemaRegistry schema.Registry
- closer *run.Closer
- namespace string
- etcdUsername string
- etcdPassword string
- etcdTLSCAFile string
- etcdTLSCertFile string
- etcdTLSKeyFile string
- endpoints []string
- registryTimeout time.Duration
- forceRegisterNode bool
+ schemaRegistry schema.Registry
+ closer *run.Closer
+ namespace string
+ etcdUsername string
+ etcdPassword string
+ etcdTLSCAFile string
+ etcdTLSCertFile string
+ etcdTLSKeyFile string
+ endpoints []string
+ registryTimeout time.Duration
+ etcdFullSyncInterval time.Duration
+ forceRegisterNode bool
}
func (s *clientService) SchemaRegistry() schema.Registry {
@@ -89,6 +90,7 @@ func (s *clientService) FlagSet() *run.FlagSet {
fs.StringVar(&s.etcdTLSCertFile, flagEtcdTLSCertFile, "", "Etcd client certificate")
fs.StringVar(&s.etcdTLSKeyFile, flagEtcdTLSKeyFile, "", "Private key for the etcd client certificate.")
fs.DurationVar(&s.registryTimeout, "node-registry-timeout", 2*time.Minute, "The timeout for the node registry")
+ fs.DurationVar(&s.etcdFullSyncInterval, "etcd-full-sync-interval", 30*time.Minute, "The interval for full sync etcd")
return fs
}
@@ -123,6 +125,7 @@ func (s *clientService) PreRun(ctx context.Context) error {
schema.ConfigureEtcdUser(s.etcdUsername, s.etcdPassword),
schema.ConfigureEtcdTLSCAFile(s.etcdTLSCAFile),
schema.ConfigureEtcdTLSCertAndKey(s.etcdTLSCertFile, s.etcdTLSKeyFile),
+ schema.ConfigureWatchCheckInterval(s.etcdFullSyncInterval),
)
if errors.Is(err, context.DeadlineExceeded) {
select {
@@ -193,6 +196,9 @@ func (s *clientService) PreRun(ctx context.Context) error {
}
func (s *clientService) Serve() run.StopNotify {
+ if s.schemaRegistry != nil {
+ s.schemaRegistry.StartWatcher()
+ }
return s.closer.CloseNotify()
}
diff --git a/banyand/metadata/embeddedserver/server.go b/banyand/metadata/embeddedserver/server.go
index 44bd4864..908cbe7b 100644
--- a/banyand/metadata/embeddedserver/server.go
+++ b/banyand/metadata/embeddedserver/server.go
@@ -81,6 +81,7 @@ func (s *server) PreRun(ctx context.Context) error {
}
func (s *server) Serve() run.StopNotify {
+ _ = s.Service.Serve()
return s.metaServer.StoppingNotify()
}
diff --git a/banyand/metadata/schema/etcd.go b/banyand/metadata/schema/etcd.go
index 33490399..4e3d51e9 100644
--- a/banyand/metadata/schema/etcd.go
+++ b/banyand/metadata/schema/etcd.go
@@ -125,13 +125,13 @@ func CheckInterval(d time.Duration) WatcherOption {
}
type etcdSchemaRegistry struct {
- namespace string
client *clientv3.Client
closer *run.Closer
l *logger.Logger
- watchers []*watcher
- mux sync.RWMutex
+ watchers map[Kind]*watcher
+ namespace string
checkInterval time.Duration
+ mux sync.RWMutex
}
type etcdSchemaRegistryConfig struct {
@@ -167,21 +167,28 @@ func (e *etcdSchemaRegistry) RegisterHandler(name string, kind Kind, handler Eve
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, CheckInterval(e.checkInterval))
- if w != nil {
- e.watchers = append(e.watchers, w)
- }
+ e.registerToWatcher(name, kinds[i], revisions[i], handler)
}
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, CheckInterval(e.checkInterval))
- if w != nil {
- e.watchers = append(e.watchers, w)
+ e.registerToWatcher(name, kinds[i], 0, handler)
+ }
+}
+
+func (e *etcdSchemaRegistry) registerToWatcher(name string, kind Kind, revision int64, handler EventHandler) {
+ if w, ok := e.watchers[kind]; ok {
+ e.l.Info().Str("name", name).Stringer("kind", kind).Msg("registering to an existing watcher")
+ w.AddHandler(handler)
+ if w.revision > revision {
+ w.revision = revision
}
+ return
}
+ e.l.Info().Str("name", name).Stringer("kind", kind).Msg("registering to a new watcher")
+ w := e.newWatcherWithRevision(name, kind, revision, CheckInterval(e.checkInterval))
+ w.AddHandler(handler)
+ e.watchers[kind] = w
}
func (e *etcdSchemaRegistry) Compact(ctx context.Context, revision int64) error {
@@ -193,6 +200,12 @@ func (e *etcdSchemaRegistry) Compact(ctx context.Context, revision int64) error
return err
}
+func (e *etcdSchemaRegistry) StartWatcher() {
+ for _, w := range e.watchers {
+ w.Start()
+ }
+}
+
func (e *etcdSchemaRegistry) Close() error {
e.closer.Done()
e.closer.CloseThenWait()
@@ -241,6 +254,7 @@ func NewEtcdSchemaRegistry(options ...RegistryOption) (Registry, error) {
closer: run.NewCloser(1),
l: logger.GetLogger("schema-registry"),
checkInterval: registryConfig.checkInterval,
+ watchers: make(map[Kind]*watcher),
}
return reg, nil
}
@@ -560,15 +574,14 @@ func (e *etcdSchemaRegistry) revokeLease(lease *clientv3.LeaseGrantResponse) {
}
}
-func (e *etcdSchemaRegistry) NewWatcher(name string, kind Kind, handler watchEventHandler, opts ...WatcherOption) *watcher {
- return e.newWatcherWithRevision(name, kind, 0, handler, opts...)
+func (e *etcdSchemaRegistry) NewWatcher(name string, kind Kind, opts ...WatcherOption) *watcher {
+ return e.newWatcherWithRevision(name, kind, 0, opts...)
}
-func (e *etcdSchemaRegistry) newWatcherWithRevision(name string, kind Kind, revision int64, handler watchEventHandler, opts ...WatcherOption) *watcher {
+func (e *etcdSchemaRegistry) newWatcherWithRevision(name string, kind Kind, revision int64, opts ...WatcherOption) *watcher {
wc := watcherConfig{
key: e.prependNamespace(kind.key()),
kind: kind,
- handler: handler,
revision: revision,
checkInterval: 5 * time.Minute, // Default value
}
diff --git a/banyand/metadata/schema/schema.go b/banyand/metadata/schema/schema.go
index a1ca94d2..8f4a8e35 100644
--- a/banyand/metadata/schema/schema.go
+++ b/banyand/metadata/schema/schema.go
@@ -65,9 +65,10 @@ type Registry interface {
TopNAggregation
Node
RegisterHandler(string, Kind, EventHandler)
- NewWatcher(string, Kind, watchEventHandler, ...WatcherOption) *watcher
+ NewWatcher(string, Kind, ...WatcherOption) *watcher
Register(context.Context, Metadata, bool) error
Compact(context.Context, int64) error
+ StartWatcher()
}
// TypeMeta defines the identity and type of an Event.
diff --git a/banyand/metadata/schema/watcher.go b/banyand/metadata/schema/watcher.go
index 6eccc143..3c635237 100644
--- a/banyand/metadata/schema/watcher.go
+++ b/banyand/metadata/schema/watcher.go
@@ -18,6 +18,7 @@
package schema
import (
+ "context"
"sync"
"time"
@@ -36,7 +37,6 @@ type watchEventHandler interface {
OnDelete(Metadata)
}
type watcherConfig struct {
- handler watchEventHandler
key string
revision int64
kind Kind
@@ -49,17 +49,18 @@ type cacheEntry struct {
}
type watcher struct {
- handler watchEventHandler
cli *clientv3.Client
closer *run.Closer
l *logger.Logger
ticker *time.Ticker
cache map[string]cacheEntry
key string
+ handlers []watchEventHandler
revision int64
kind Kind
checkInterval time.Duration
mu sync.RWMutex
+ startOnce sync.Once
}
func newWatcher(cli *clientv3.Client, wc watcherConfig, l *logger.Logger) *watcher {
@@ -70,20 +71,29 @@ func newWatcher(cli *clientv3.Client, wc watcherConfig, l *logger.Logger) *watch
cli: cli,
key: wc.key,
kind: wc.kind,
- handler: wc.handler,
revision: wc.revision,
closer: run.NewCloser(1),
l: l,
checkInterval: wc.checkInterval,
cache: make(map[string]cacheEntry),
}
+ return w
+}
- if w.revision < 1 {
- w.periodicSync()
- }
+func (w *watcher) Start() {
+ w.startOnce.Do(func() {
+ if w.revision < 1 {
+ w.periodicSync()
+ }
+ go w.watch(w.revision)
+ })
+}
- go w.watch(w.revision)
- return w
+func (w *watcher) AddHandler(handler watchEventHandler) {
+ if w.handlers == nil {
+ w.handlers = make([]watchEventHandler, 0)
+ }
+ w.handlers = append(w.handlers, handler)
}
func (w *watcher) Close() {
@@ -165,6 +175,12 @@ func (w *watcher) handle(watchEvent *clientv3.Event, watchResp *clientv3.WatchRe
modRevision: watchEvent.Kv.ModRevision,
}
+ handlers := w.handlers
+ if len(handlers) == 0 {
+ w.l.Panic().Msg("no handlers registered")
+ return
+ }
+
w.mu.Lock()
defer w.mu.Unlock()
@@ -178,7 +194,9 @@ func (w *watcher) handle(watchEvent *clientv3.Event, watchResp *clientv3.WatchRe
w.l.Error().Stringer("event_header", &watchResp.Header).AnErr("err", err).Msg("failed to unmarshal message")
return
}
- w.handler.OnAddOrUpdate(md)
+ for i := range handlers {
+ handlers[i].OnAddOrUpdate(md)
+ }
}
case mvccpb.DELETE:
delete(w.cache, keyStr)
@@ -187,14 +205,18 @@ func (w *watcher) handle(watchEvent *clientv3.Event, watchResp *clientv3.WatchRe
w.l.Error().Stringer("event_header", &watchResp.Header).AnErr("err", err).Msg("failed to unmarshal message")
return
}
- w.handler.OnDelete(md)
+ for i := range handlers {
+ handlers[i].OnDelete(md)
+ }
}
}
func (w *watcher) periodicSync() {
resp, err := w.cli.Get(w.closer.Ctx(), w.key, clientv3.WithPrefix())
if err != nil {
- w.l.Error().Err(err).Msg("periodic sync failed to fetch keys")
+ if !errors.Is(err, context.Canceled) {
+ w.l.Error().Err(err).Msg("periodic sync failed to fetch keys")
+ }
return
}
@@ -206,6 +228,11 @@ func (w *watcher) periodicSync() {
}
}
+ handlers := w.handlers
+ if len(handlers) == 0 {
+ w.l.Panic().Msg("no handlers registered")
+ return
+ }
w.mu.Lock()
defer w.mu.Unlock()
@@ -216,7 +243,9 @@ func (w *watcher) periodicSync() {
// Handle deletion
delete(w.cache, cachedKey)
if md, err := w.getFromStore(cachedKey); err == nil {
- w.handler.OnDelete(*md)
+ for i := range handlers {
+ handlers[i].OnDelete(*md)
+ }
}
continue
}
@@ -224,7 +253,9 @@ func (w *watcher) periodicSync() {
if currentEntry.valueHash != cachedEntry.valueHash {
// Handle update
if md, err := w.getFromStore(cachedKey); err == nil {
- w.handler.OnAddOrUpdate(*md)
+ for i := range handlers {
+ handlers[i].OnAddOrUpdate(*md)
+ }
w.cache[cachedKey] = currentEntry
}
}
@@ -234,7 +265,9 @@ func (w *watcher) periodicSync() {
for key, entry := range currentState {
if _, exists := w.cache[key]; !exists {
if md, err := w.getFromStore(key); err == nil {
- w.handler.OnAddOrUpdate(*md)
+ for i := range handlers {
+ handlers[i].OnAddOrUpdate(*md)
+ }
w.cache[key] = entry
}
}
diff --git a/banyand/metadata/schema/watcher_test.go b/banyand/metadata/schema/watcher_test.go
index 38bdc163..c2fa90dd 100644
--- a/banyand/metadata/schema/watcher_test.go
+++ b/banyand/metadata/schema/watcher_test.go
@@ -172,7 +172,9 @@ var _ = ginkgo.Describe("Watcher", func() {
}
// Start the watcher
- watcher := registry.NewWatcher("test", schema.KindMeasure, mockedObj)
+ watcher := registry.NewWatcher("test", schema.KindMeasure)
+ watcher.AddHandler(mockedObj)
+ watcher.Start()
ginkgo.DeferCleanup(func() {
watcher.Close()
})
@@ -189,7 +191,9 @@ var _ = ginkgo.Describe("Watcher", func() {
}, flags.EventuallyTimeout).Should(gomega.BeTrue())
})
ginkgo.It("should handle watch events", func() {
- watcher := registry.NewWatcher("test", schema.KindStream, mockedObj)
+ watcher := registry.NewWatcher("test", schema.KindStream)
+ watcher.AddHandler(mockedObj)
+ watcher.Start()
ginkgo.DeferCleanup(func() {
watcher.Close()
})
@@ -312,7 +316,9 @@ var _ = ginkgo.Describe("Watcher", func() {
})
gomega.Expect(err).NotTo(gomega.HaveOccurred())
- watcher := registry.NewWatcher("test", schema.KindMeasure, mockedObj, schema.CheckInterval(1*time.Second))
+ watcher := registry.NewWatcher("test", schema.KindMeasure, schema.CheckInterval(1*time.Second))
+ watcher.AddHandler(mockedObj)
+ watcher.Start()
ginkgo.DeferCleanup(func() {
watcher.Close()
})
@@ -325,7 +331,9 @@ var _ = ginkgo.Describe("Watcher", func() {
})
ginkgo.It("should detect deletions", func() {
- watcher := registry.NewWatcher("test", schema.KindMeasure, mockedObj, schema.CheckInterval(1*time.Second))
+ watcher := registry.NewWatcher("test", schema.KindMeasure, schema.CheckInterval(1*time.Second))
+ watcher.AddHandler(mockedObj)
+ watcher.Start()
ginkgo.DeferCleanup(func() {
watcher.Close()
})
@@ -392,7 +400,9 @@ var _ = ginkgo.Describe("Watcher", func() {
})
ginkgo.It("should recover state after compaction", func() {
- watcher := registry.NewWatcher("test", schema.KindMeasure, mockedObj, schema.CheckInterval(1*time.Hour))
+ watcher := registry.NewWatcher("test", schema.KindMeasure, schema.CheckInterval(1*time.Hour))
+ watcher.AddHandler(mockedObj)
+ watcher.Start()
ginkgo.DeferCleanup(func() {
watcher.Close()
})
diff --git a/banyand/stream/block.go b/banyand/stream/block.go
index 00c4b337..e4a27945 100644
--- a/banyand/stream/block.go
+++ b/banyand/stream/block.go
@@ -510,9 +510,9 @@ func (bc *blockCursor) loadData(tmpBlock *block) bool {
tmpBlock.reset()
bc.bm.tagProjection = bc.tagProjection
var tf map[string]*dataBlock
- for i := range bc.tagProjection {
+ for _, tp := range bc.tagProjection {
for tfName, block := range bc.bm.tagFamilies {
- if bc.tagProjection[i].Family == tfName {
+ if tp.Family == tfName {
if tf == nil {
tf = make(map[string]*dataBlock, len(bc.tagProjection))
}
@@ -520,8 +520,15 @@ func (bc *blockCursor) loadData(tmpBlock *block) bool {
}
}
}
+ if len(tf) == 0 {
+ return false
+ }
+
bc.bm.tagFamilies = tf
tmpBlock.mustReadFrom(&bc.tagValuesDecoder, bc.p, bc.bm)
+ if len(tmpBlock.timestamps) == 0 {
+ return false
+ }
idxList := make([]int, 0)
var start, end int
diff --git a/banyand/stream/query.go b/banyand/stream/query.go
index d312c7fa..488fd1c3 100644
--- a/banyand/stream/query.go
+++ b/banyand/stream/query.go
@@ -149,7 +149,7 @@ type queryOptions struct {
}
func (qo *queryOptions) copyFrom(other *queryOptions) {
- qo.StreamQueryOptions = other.StreamQueryOptions
+ qo.StreamQueryOptions.CopyFrom(&other.StreamQueryOptions)
qo.elementFilter = other.elementFilter
qo.seriesToEntity = other.seriesToEntity
qo.sortedSids = other.sortedSids
diff --git a/banyand/stream/query_by_ts.go b/banyand/stream/query_by_ts.go
index b7fa2a99..d45a46ef 100644
--- a/banyand/stream/query_by_ts.go
+++ b/banyand/stream/query_by_ts.go
@@ -123,11 +123,11 @@ func (t *tsResult) scanSegment(ctx context.Context) error {
blockHeap.Push(bc)
}
}
- releaseBlockScanResultBatch(batch)
heap.Init(blockHeap)
result := blockHeap.merge(t.qo.MaxElementSize)
t.shards[workerID].CopyFrom(tmpResult, result)
blockHeap.reset()
+ releaseBlockScanResultBatch(batch)
}
workerWg.Done()
}(i)
diff --git a/go.mod b/go.mod
index 7f18d6ae..8ab8f6f8 100644
--- a/go.mod
+++ b/go.mod
@@ -2,7 +2,7 @@ module github.com/apache/skywalking-banyandb
go 1.23
-toolchain go1.23.1
+toolchain go1.23.5
require (
github.com/RoaringBitmap/roaring v1.9.4
diff --git a/pkg/query/model/model.go b/pkg/query/model/model.go
index d910ddb9..1565e40b 100644
--- a/pkg/query/model/model.go
+++ b/pkg/query/model/model.go
@@ -91,6 +91,33 @@ type StreamQueryOptions struct {
MaxElementSize int
}
+// CopyFrom copies the StreamQueryOptions from other to s.
+func (s *StreamQueryOptions) CopyFrom(other *StreamQueryOptions) {
+ s.Name = other.Name
+ s.TimeRange = other.TimeRange
+
+ // Deep copy for Entities if it's a slice
+ if other.Entities != nil {
+ s.Entities = make([][]*modelv1.TagValue, len(other.Entities))
+ copy(s.Entities, other.Entities)
+ } else {
+ s.Entities = nil
+ }
+
+ s.Filter = other.Filter
+ s.Order = other.Order
+
+ // Deep copy if TagProjection is a slice
+ if other.TagProjection != nil {
+ s.TagProjection = make([]TagProjection, len(other.TagProjection))
+ copy(s.TagProjection, other.TagProjection)
+ } else {
+ s.TagProjection = nil
+ }
+
+ s.MaxElementSize = other.MaxElementSize
+}
+
// StreamResult is the result of a query.
type StreamResult struct {
Error error
diff --git a/scripts/ci/check/version_test.go b/scripts/ci/check/version_test.go
index 26555a59..a5855cb5 100644
--- a/scripts/ci/check/version_test.go
+++ b/scripts/ci/check/version_test.go
@@ -31,7 +31,7 @@ import (
)
const (
- GoVersion = "1.23.0"
+ GoVersion = "1.23.5"
CPUType = 8
)
diff --git a/test/integration/standalone/cold_query/query_suite_test.go b/test/integration/standalone/cold_query/query_suite_test.go
index a28c145e..36f03901 100644
--- a/test/integration/standalone/cold_query/query_suite_test.go
+++ b/test/integration/standalone/cold_query/query_suite_test.go
@@ -89,7 +89,9 @@ var _ = SynchronizedAfterSuite(func() {
Expect(connection.Close()).To(Succeed())
}
}, func() {
- deferFunc()
+ if deferFunc != nil {
+ deferFunc()
+ }
Eventually(gleak.Goroutines, flags.EventuallyTimeout).ShouldNot(gleak.HaveLeaked(goods))
Eventually(pool.AllRefsCount, flags.EventuallyTimeout).Should(gmatcher.HaveZeroRef())
})