You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ha...@apache.org on 2022/02/21 23:24:50 UTC
[skywalking-banyandb] branch tsdb-blocks created (now db5ff6b)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a change to branch tsdb-blocks
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git.
at db5ff6b Reopen closed blocks on touching
This branch includes the following new commits:
new db5ff6b Reopen closed blocks on touching
The 1 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] 01/01: Reopen closed blocks on touching
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch tsdb-blocks
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit db5ff6bfeafdec9eb100179b906c625aabbb3786
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Mon Feb 21 14:53:28 2022 +0000
Reopen closed blocks on touching
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
banyand/stream/stream_query.go | 4 ++
banyand/tsdb/block.go | 82 +++++++++++++++----------
banyand/tsdb/segment.go | 135 ++++++++++++++++++++++-------------------
banyand/tsdb/seriesdb.go | 5 +-
banyand/tsdb/shard.go | 63 ++++++++++++-------
banyand/tsdb/shard_test.go | 53 +++++++++++++++-
banyand/tsdb/tsdb.go | 16 +++++
go.mod | 2 +-
8 files changed, 241 insertions(+), 119 deletions(-)
diff --git a/banyand/stream/stream_query.go b/banyand/stream/stream_query.go
index 678335c..2896f0d 100644
--- a/banyand/stream/stream_query.go
+++ b/banyand/stream/stream_query.go
@@ -163,6 +163,10 @@ func (sd *shardDelegate) Index() tsdb.IndexDatabase {
return sd.delegated.Index()
}
+func (sd *shardDelegate) State() tsdb.ShardState {
+ return sd.delegated.State()
+}
+
var _ tsdb.SeriesDatabase = (*seriesDatabaseDelegate)(nil)
type seriesDatabaseDelegate struct {
diff --git a/banyand/tsdb/block.go b/banyand/tsdb/block.go
index 98c0b73..77a7ad3 100644
--- a/banyand/tsdb/block.go
+++ b/banyand/tsdb/block.go
@@ -21,10 +21,10 @@ import (
"context"
"io"
"strconv"
+ "sync"
"time"
"github.com/dgraph-io/ristretto/z"
- "go.uber.org/atomic"
"github.com/apache/skywalking-banyandb/api/common"
"github.com/apache/skywalking-banyandb/banyand/kv"
@@ -38,11 +38,12 @@ import (
)
type block struct {
- path string
- l *logger.Logger
- suffix string
- ref *z.Closer
- closing *atomic.Bool
+ path string
+ l *logger.Logger
+ suffix string
+ ref *z.Closer
+ lock sync.RWMutex
+ closed bool
store kv.TimeSeriesStore
primaryIndex index.Store
@@ -51,8 +52,9 @@ type block struct {
closableLst []io.Closer
timestamp.TimeRange
bucket.Reporter
- segID uint16
- blockID uint16
+ segID uint16
+ blockID uint16
+ encodingMethod EncodingMethod
}
type blockOpts struct {
@@ -70,16 +72,6 @@ func newBlock(ctx context.Context, opts blockOpts) (b *block, err error) {
}
id := uint16(opts.blockSize.Unit)<<12 | ((uint16(suffixInteger) << 4) >> 4)
timeRange := timestamp.NewTimeRange(opts.startTime, opts.blockSize.NextTime(opts.startTime), true, false)
- b = &block{
- segID: opts.segID,
- blockID: id,
- path: opts.path,
- ref: z.NewCloser(1),
- l: logger.Fetch(ctx, "block"),
- TimeRange: timeRange,
- Reporter: bucket.NewTimeBasedReporter(timeRange),
- closing: atomic.NewBool(false),
- }
encodingMethodObject := ctx.Value(encodingMethodKey)
if encodingMethodObject == nil {
encodingMethodObject = EncodingMethod{
@@ -87,44 +79,60 @@ func newBlock(ctx context.Context, opts blockOpts) (b *block, err error) {
DecoderPool: encoding.NewPlainDecoderPool(0),
}
}
- encodingMethod := encodingMethodObject.(EncodingMethod)
+ b = &block{
+ segID: opts.segID,
+ blockID: id,
+ path: opts.path,
+ l: logger.Fetch(ctx, "block"),
+ TimeRange: timeRange,
+ Reporter: bucket.NewTimeBasedReporter(timeRange),
+ closed: true,
+ encodingMethod: encodingMethodObject.(EncodingMethod),
+ }
+ return b, err
+}
+
+func (b *block) open() (err error) {
+ b.lock.Lock()
+ defer b.lock.Unlock()
+ if !b.closed {
+ return nil
+ }
+ b.ref = z.NewCloser(1)
if b.store, err = kv.OpenTimeSeriesStore(
0,
b.path+"/store",
- kv.TSSWithEncoding(encodingMethod.EncoderPool, encodingMethod.DecoderPool),
+ kv.TSSWithEncoding(b.encodingMethod.EncoderPool, b.encodingMethod.DecoderPool),
kv.TSSWithLogger(b.l),
); err != nil {
- return nil, err
+ return err
}
if b.primaryIndex, err = lsm.NewStore(lsm.StoreOpts{
Path: b.path + "/primary",
Logger: b.l,
}); err != nil {
- return nil, err
+ return err
}
b.closableLst = append(b.closableLst, b.store, b.primaryIndex)
if b.invertedIndex, err = inverted.NewStore(inverted.StoreOpts{
Path: b.path + "/inverted",
Logger: b.l,
}); err != nil {
- return nil, err
+ return err
}
if b.lsmIndex, err = lsm.NewStore(lsm.StoreOpts{
Path: b.path + "/lsm",
Logger: b.l,
}); err != nil {
- return nil, err
+ return err
}
b.closableLst = append(b.closableLst, b.invertedIndex, b.lsmIndex)
- go func() {
- <-b.ref.HasBeenClosed()
- b.closing.Store(true)
- }()
- return b, err
+ b.closed = false
+ return nil
}
func (b *block) delegate() blockDelegate {
- if b.closing.Load() {
+ if b.isClosed() {
return nil
}
b.incRef()
@@ -142,14 +150,26 @@ func (b *block) incRef() {
}
func (b *block) close() {
+ b.lock.Lock()
+ defer b.lock.Unlock()
+ if b.closed {
+ return
+ }
b.dscRef()
b.ref.SignalAndWait()
for _, closer := range b.closableLst {
_ = closer.Close()
}
+ b.closed = true
+}
+
+func (b *block) isClosed() bool {
+ b.lock.RLock()
+ defer b.lock.RUnlock()
+ return b.closed
}
-func (b block) String() string {
+func (b *block) String() string {
return b.Reporter.String()
}
diff --git a/banyand/tsdb/segment.go b/banyand/tsdb/segment.go
index 2063318..0053a35 100644
--- a/banyand/tsdb/segment.go
+++ b/banyand/tsdb/segment.go
@@ -37,9 +37,8 @@ type segment struct {
path string
suffix string
- blockCloseCh chan uint16
- globalIndex kv.Store
- l *logger.Logger
+ globalIndex kv.Store
+ l *logger.Logger
timestamp.TimeRange
bucket.Reporter
blockController *blockController
@@ -63,7 +62,6 @@ func openSegment(ctx context.Context, startTime time.Time, path, suffix string,
blockController: newBlockController(id, path, timeRange, blockSize, l, blockQueue),
TimeRange: timeRange,
Reporter: bucket.NewTimeBasedReporter(timeRange),
- blockCloseCh: make(chan uint16, 24),
}
err = s.blockController.open(context.WithValue(ctx, logger.ContextKey, s.l))
if err != nil {
@@ -82,7 +80,6 @@ func openSegment(ctx context.Context, startTime time.Time, path, suffix string,
return nil, err
}
s.blockManageStrategy.Run()
- s.blockController.listenToBlockClosing(s.blockCloseCh)
return s, nil
}
@@ -93,8 +90,8 @@ func (s *segment) close() {
s.Stop()
}
-func (s *segment) notifyCloseBlock(id uint16) {
- s.blockCloseCh <- id
+func (s *segment) closeBlock(id uint16) {
+ s.blockController.closeBlock(id)
}
func (s segment) String() string {
@@ -109,7 +106,6 @@ type blockController struct {
blockSize IntervalRule
lst []*block
blockQueue bucket.Queue
- closing chan struct{}
l *logger.Logger
}
@@ -121,7 +117,6 @@ func newBlockController(segID uint16, location string, segTimeRange timestamp.Ti
location: location,
blockSize: blockSize,
segTimeRange: segTimeRange,
- closing: make(chan struct{}),
blockQueue: blockQueue,
l: l,
}
@@ -150,49 +145,21 @@ func (bc *blockController) Next() (bucket.Reporter, error) {
if errors.Is(err, ErrEndOfSegment) {
return nil, bucket.ErrNoMoreBucket
}
+ err = reporter.open()
+ if err != nil {
+ return nil, err
+ }
return reporter, err
}
func (bc *blockController) OnMove(prev bucket.Reporter, next bucket.Reporter) {
- bc.blockQueue.Push(blockIDAndSegID{
- segID: bc.segID,
- blockID: prev.(*block).blockID,
+ bc.l.Info().Stringer("prev", prev).Stringer("next", next).Msg("move to the next block")
+ bc.blockQueue.Push(BlockID{
+ SegID: bc.segID,
+ BlockID: prev.(*block).blockID,
})
}
-func (bc *blockController) listenToBlockClosing(closeCh <-chan uint16) {
- go func() {
- for {
- select {
- case id := <-closeCh:
- b := bc.removeBlock(id)
- if b != nil {
- b.close()
- continue
- }
- bc.l.Warn().Uint16("blockID", id).Uint16("segID", id).
- Msg("block is absent on removing it from the segment")
- case <-bc.closing:
- return
- }
- }
- }()
-}
-
-func (bc *blockController) removeBlock(blockID uint16) *block {
- bc.Lock()
- defer bc.Unlock()
- for i := range bc.lst {
- b := bc.lst[i]
- if b.blockID == blockID {
- //remove the block from the internal lst
- bc.lst = append(bc.lst[:i], bc.lst[i+1:]...)
- return b
- }
- }
- return nil
-}
-
func (bc *blockController) Format(tm time.Time) string {
switch bc.blockSize.Unit {
case HOUR:
@@ -218,29 +185,75 @@ func (bc *blockController) Parse(value string) (time.Time, error) {
}
func (bc *blockController) span(timeRange timestamp.TimeRange) (bb []*block) {
+ return bc.ensureBlockOpen(bc.search(func(b *block) bool {
+ return b.Overlapping(timeRange)
+ }))
+}
+
+func (bc *blockController) get(blockID uint16) *block {
+ b := bc.getBlock(blockID)
+ if b != nil {
+ return bc.ensureBlockOpen([]*block{b})[0]
+ }
+ return nil
+}
+
+func (bc *blockController) getBlock(blockID uint16) *block {
+ bb := bc.search(func(b *block) bool {
+ return b.blockID == blockID
+ })
+ if len(bb) > 0 {
+ return bb[0]
+ }
+ return nil
+}
+
+func (bc *blockController) blocks() []*block {
+ bc.RLock()
+ defer bc.RUnlock()
+ r := make([]*block, len(bc.lst))
+ copy(r, bc.lst)
+ return r
+}
+
+func (bc *blockController) search(matcher func(*block) bool) (bb []*block) {
bc.RLock()
defer bc.RUnlock()
last := len(bc.lst) - 1
for i := range bc.lst {
b := bc.lst[last-i]
- if b.Overlapping(timeRange) {
+ if matcher(b) {
bb = append(bb, b)
}
}
return bb
}
-func (bc *blockController) get(blockID uint16) *block {
+func (bc *blockController) ensureBlockOpen(blocks []*block) (openedBlocks []*block) {
+ if blocks == nil {
+ return nil
+ }
+ for _, b := range blocks {
+ if b.isClosed() {
+ if err := b.open(); err != nil {
+ bc.l.Error().Err(err).Stringer("block", b).Msg("fail to open block")
+ continue
+ }
+ }
+ openedBlocks = append(openedBlocks, b)
+ bc.blockQueue.Push(b.blockID)
+ }
+ return openedBlocks
+}
+
+func (bc *blockController) closeBlock(blockID uint16) {
bc.RLock()
defer bc.RUnlock()
- last := len(bc.lst) - 1
- for i := range bc.lst {
- b := bc.lst[last-i]
- if b.blockID == blockID {
- return b
- }
+ b := bc.getBlock(blockID)
+ if b == nil {
+ return
}
- return nil
+ b.close()
}
func (bc *blockController) startTime(suffix string) (time.Time, error) {
@@ -274,10 +287,13 @@ func (bc *blockController) open(ctx context.Context) error {
return err
}
if bc.Current() == nil {
- _, err = bc.create(ctx, time.Now())
+ b, err := bc.create(ctx, time.Now())
if err != nil {
return err
}
+ if err = b.open(); err != nil {
+ return err
+ }
}
return nil
}
@@ -311,16 +327,13 @@ func (bc *blockController) load(ctx context.Context, suffix, path string) (b *bl
}); err != nil {
return nil, err
}
- {
- bc.Lock()
- defer bc.Unlock()
- bc.lst = append(bc.lst, b)
- }
+ bc.Lock()
+ defer bc.Unlock()
+ bc.lst = append(bc.lst, b)
return b, nil
}
func (bc *blockController) close() {
- bc.closing <- struct{}{}
for _, s := range bc.lst {
s.close()
}
diff --git a/banyand/tsdb/seriesdb.go b/banyand/tsdb/seriesdb.go
index 1273de9..4c36389 100644
--- a/banyand/tsdb/seriesdb.go
+++ b/banyand/tsdb/seriesdb.go
@@ -230,7 +230,10 @@ func (s *seriesDB) span(timeRange timestamp.TimeRange) []blockDelegate {
result := make([]blockDelegate, 0)
for _, s := range s.segCtrl.span(timeRange) {
for _, b := range s.blockController.span(timeRange) {
- result = append(result, b.delegate())
+ bd := b.delegate()
+ if bd != nil {
+ result = append(result, bd)
+ }
}
}
return result
diff --git a/banyand/tsdb/shard.go b/banyand/tsdb/shard.go
index 135dd58..df6c5f4 100644
--- a/banyand/tsdb/shard.go
+++ b/banyand/tsdb/shard.go
@@ -45,18 +45,6 @@ type shard struct {
segmentManageStrategy *bucket.Strategy
}
-func (s *shard) ID() common.ShardID {
- return s.id
-}
-
-func (s *shard) Series() SeriesDatabase {
- return s.seriesDatabase
-}
-
-func (s *shard) Index() IndexDatabase {
- return s.indexDatabase
-}
-
func OpenShard(ctx context.Context, id common.ShardID,
root string, segmentSize, blockSize IntervalRule, openedBlockSize int) (Shard, error) {
path, err := mkdir(shardTemplate, root, int(id))
@@ -104,6 +92,34 @@ func OpenShard(ctx context.Context, id common.ShardID,
return s, nil
}
+func (s *shard) ID() common.ShardID {
+ return s.id
+}
+
+func (s *shard) Series() SeriesDatabase {
+ return s.seriesDatabase
+}
+
+func (s *shard) Index() IndexDatabase {
+ return s.indexDatabase
+}
+
+func (s *shard) State() (shardState ShardState) {
+ for _, seg := range s.segmentController.segments() {
+ for _, b := range seg.blockController.blocks() {
+ shardState.OpenedBlocks = append(shardState.OpenedBlocks, BlockState{
+ ID: BlockID{
+ SegID: b.segID,
+ BlockID: b.blockID,
+ },
+ TimeRange: b.TimeRange,
+ Closed: b.isClosed(),
+ })
+ }
+ }
+ return shardState
+}
+
func (s *shard) Close() error {
s.segmentManageStrategy.Close()
s.segmentController.close()
@@ -171,11 +187,6 @@ type segmentController struct {
l *logger.Logger
}
-type blockIDAndSegID struct {
- segID uint16
- blockID uint16
-}
-
func newSegmentController(location string, segmentSize, blockSize IntervalRule, openedBlockSize int, l *logger.Logger) (*segmentController, error) {
sc := &segmentController{
location: location,
@@ -185,13 +196,14 @@ func newSegmentController(location string, segmentSize, blockSize IntervalRule,
}
var err error
sc.blockQueue, err = bucket.NewQueue(openedBlockSize, func(id interface{}) {
- bsID := id.(blockIDAndSegID)
- seg := sc.get(bsID.segID)
+ bsID := id.(BlockID)
+ seg := sc.get(bsID.SegID)
if seg == nil {
- l.Warn().Uint16("segID", bsID.segID).Msg("segment is absent")
+ l.Warn().Uint16("segID", bsID.SegID).Msg("segment is absent")
return
}
- seg.notifyCloseBlock(bsID.blockID)
+ l.Info().Uint16("blockID", bsID.BlockID).Uint16("segID", bsID.SegID).Msg("closing the block")
+ seg.closeBlock(bsID.BlockID)
})
return sc, err
}
@@ -253,7 +265,14 @@ func (sc *segmentController) Next() (bucket.Reporter, error) {
}
func (sc *segmentController) OnMove(prev bucket.Reporter, next bucket.Reporter) {
- sc.l.Info().Stringer("prev", prev).Stringer("next", next).Msg("move to the next segment")
+ event := sc.l.Info()
+ if prev != nil {
+ event.Stringer("prev", prev)
+ }
+ if next != nil {
+ event.Stringer("next", next)
+ }
+ event.Msg("move to the next segment")
}
func (sc *segmentController) Format(tm time.Time) string {
diff --git a/banyand/tsdb/shard_test.go b/banyand/tsdb/shard_test.go
index a7b45ed..87756b6 100644
--- a/banyand/tsdb/shard_test.go
+++ b/banyand/tsdb/shard_test.go
@@ -33,7 +33,7 @@ import (
)
var _ = Describe("Shard", func() {
- Describe("Generate segments", func() {
+ Describe("Generate segments and blocks", func() {
var tmp string
var deferFn func()
var shard tsdb.Shard
@@ -47,7 +47,7 @@ var _ = Describe("Shard", func() {
shard.Close()
deferFn()
})
- It("generates several segments", func() {
+ It("generates several segments and blocks", func() {
var err error
shard, err = tsdb.OpenShard(context.TODO(), common.ShardID(0), tmp,
tsdb.IntervalRule{
@@ -125,6 +125,53 @@ var _ = Describe("Shard", func() {
return num
}).WithTimeout(30 * time.Second).Should(BeNumerically(">=", 1))
})
-
+ It("reopens closed blocks", func() {
+ var err error
+ shard, err = tsdb.OpenShard(context.TODO(), common.ShardID(0), tmp,
+ tsdb.IntervalRule{
+ Unit: tsdb.MILLISECOND,
+ Num: 3000,
+ },
+ tsdb.IntervalRule{
+ Unit: tsdb.MILLISECOND,
+ Num: 1000,
+ },
+ 2,
+ )
+ Expect(err).NotTo(HaveOccurred())
+ Eventually(func() int {
+ num := 0
+ for _, bs := range shard.State().OpenedBlocks {
+ if !bs.Closed {
+ num++
+ }
+ }
+ return num
+ }).WithTimeout(10 * time.Second).Should(BeNumerically(">=", 2))
+ var closedBlocks []tsdb.BlockState
+ Eventually(func() int {
+ closedBlocks = nil
+ for _, ob := range shard.State().OpenedBlocks {
+ if ob.Closed {
+ closedBlocks = append(closedBlocks, ob)
+ }
+ }
+ return len(closedBlocks)
+ }).WithTimeout(10 * time.Second).Should(BeNumerically(">=", 1))
+ series, err := shard.Series().GetByID(common.SeriesID(11))
+ Expect(err).NotTo(HaveOccurred())
+ writeFn := func(bs tsdb.BlockState) {
+ span, err := series.Span(bs.TimeRange)
+ Expect(err).NotTo(HaveOccurred())
+ defer span.Close()
+ writer, err := span.WriterBuilder().Family([]byte("test"), []byte("test")).Time(bs.TimeRange.Start).Build()
+ Expect(err).NotTo(HaveOccurred())
+ _, err = writer.Write()
+ Expect(err).NotTo(HaveOccurred())
+ }
+ for _, bs := range closedBlocks {
+ writeFn(bs)
+ }
+ })
})
})
diff --git a/banyand/tsdb/tsdb.go b/banyand/tsdb/tsdb.go
index b863a98..cf40414 100644
--- a/banyand/tsdb/tsdb.go
+++ b/banyand/tsdb/tsdb.go
@@ -34,6 +34,7 @@ import (
"github.com/apache/skywalking-banyandb/api/common"
"github.com/apache/skywalking-banyandb/pkg/encoding"
"github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/timestamp"
)
const (
@@ -80,6 +81,7 @@ type Shard interface {
ID() common.ShardID
Series() SeriesDatabase
Index() IndexDatabase
+ State() ShardState
}
var _ Database = (*database)(nil)
@@ -97,6 +99,20 @@ type EncodingMethod struct {
DecoderPool encoding.SeriesDecoderPool
}
+type BlockID struct {
+ SegID uint16
+ BlockID uint16
+}
+
+type BlockState struct {
+ ID BlockID
+ TimeRange timestamp.TimeRange
+ Closed bool
+}
+type ShardState struct {
+ OpenedBlocks []BlockState
+}
+
type database struct {
logger *logger.Logger
location string
diff --git a/go.mod b/go.mod
index a445dc6..6bd45bd 100644
--- a/go.mod
+++ b/go.mod
@@ -94,7 +94,7 @@ require (
go.opentelemetry.io/otel/sdk/metric v0.20.0 // indirect
go.opentelemetry.io/otel/trace v0.20.0 // indirect
go.opentelemetry.io/proto/otlp v0.7.0 // indirect
- go.uber.org/atomic v1.9.0
+ go.uber.org/atomic v1.9.0 // indirect
go.uber.org/zap v1.17.0 // indirect
golang.org/x/crypto v0.0.0-20201002170205-7f63de1d35b0 // indirect
golang.org/x/text v0.3.6 // indirect