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 2021/07/12 03:26:25 UTC
[skywalking-banyandb] branch series-trace created (now ccb0d43)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a change to branch series-trace
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git.
at ccb0d43 Add local discovery and shard event
This branch includes the following new commits:
new b34bada Add trace series
new ccb0d43 Add local discovery and shard event
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] 01/02: Add trace series
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch series-trace
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit b34bada2d65f4c53ba4aade6f163f04ec613ccc9
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Fri Jul 9 20:07:22 2021 +0800
Add trace series
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
api/common/id.go | 7 +-
api/common/metadata.go | 2 +-
api/data/trace.go | 8 +-
api/fbs/v1/schema.fbs | 7 +
api/fbs/v1/schema_generated.go | 100 +++++-
api/fbs/v1/write.fbs | 7 +-
api/fbs/v1/write_generated.go | 52 +--
banyand/internal/cmd/standalone.go | 12 +-
banyand/kv/badger.go | 52 ++-
banyand/kv/kv.go | 73 ++--
banyand/series/idgen.go | 60 ++++
banyand/series/schema/sw/index_rule.bin | Bin 520 -> 560 bytes
banyand/series/schema/sw/index_rule.json | 6 +
banyand/series/schema/sw/trace_series.bin | Bin 712 -> 856 bytes
banyand/series/schema/sw/trace_series.json | 18 +-
banyand/series/series.go | 19 +-
banyand/series/trace/common_test.go | 270 +++++++++++++++
banyand/series/trace/query.go | 253 ++++++++++++++
banyand/series/trace/query_test.go | 301 +++++++++++++++++
banyand/series/{service.go => trace/schema.go} | 46 +--
.../{service_test.go => trace/schema_test.go} | 7 +-
banyand/series/trace/trace.go | 368 +++++++++++++++++++++
banyand/series/trace/write.go | 90 +++++
banyand/series/trace/write_test.go | 207 ++++++++++++
banyand/storage/block.go | 76 +----
banyand/storage/database.go | 170 ++++++++--
banyand/storage/database_test.go | 118 +++----
banyand/storage/storage.go | 53 ++-
go.mod | 3 +-
go.sum | 12 +-
api/common/id.go => pkg/convert/hash.go | 8 +-
pkg/convert/number.go | 6 +-
pkg/fb/api.go | 199 +++++++----
pkg/fb/util.go | 228 +++++++++++++
pkg/logical/analyzer_test.go | 20 +-
api/common/id.go => pkg/partition/route.go | 9 +-
36 files changed, 2446 insertions(+), 421 deletions(-)
diff --git a/api/common/id.go b/api/common/id.go
index 64d4fbf..3a549d7 100644
--- a/api/common/id.go
+++ b/api/common/id.go
@@ -17,4 +17,9 @@
package common
-type ChunkID uint64
+type ChunkID []byte
+type SeriesID uint64
+
+const (
+ DataBinaryFieldName = "data_binary"
+)
diff --git a/api/common/metadata.go b/api/common/metadata.go
index 9b493e0..f292e32 100644
--- a/api/common/metadata.go
+++ b/api/common/metadata.go
@@ -27,7 +27,7 @@ var MetadataKindVersion = KindVersion{Version: "v1", Kind: "metadata"}
type Metadata struct {
KindVersion
- Spec v1.Metadata
+ Spec *v1.Metadata
}
func (md Metadata) Equal(other Metadata) bool {
diff --git a/api/data/trace.go b/api/data/trace.go
index 8339577..49a621e 100644
--- a/api/data/trace.go
+++ b/api/data/trace.go
@@ -26,11 +26,15 @@ var TraceKindVersion = common.KindVersion{Version: "v1", Kind: "data-trace"}
type Trace struct {
common.KindVersion
- entities []Entity
+ Entities []Entity
}
type Entity struct {
- v1.Entity
+ *v1.Entity
+}
+
+type EntityValue struct {
+ *v1.EntityValue
}
func NewTrace() *Trace {
diff --git a/api/fbs/v1/schema.fbs b/api/fbs/v1/schema.fbs
index e764f1c..d2de568 100644
--- a/api/fbs/v1/schema.fbs
+++ b/api/fbs/v1/schema.fbs
@@ -19,6 +19,11 @@ include "database.fbs";
namespace banyandb.v1;
+table ShardInfo {
+ number:uint;
+ routing_fields:[string];
+}
+
enum DurationUint:byte { Hour = 0, Day = 1, Week = 2, Month = 3 }
// Duration represents the elapsed time between two instants
@@ -59,6 +64,8 @@ table TraceSeries {
fields:[FieldSpec];
// reserved_fields_map indicates how to index reserved fields to ingested fields
reserved_fields_map:TraceFieldMap;
+ // shard indicates how trace data distributes
+ shard:ShardInfo;
// duration determines how long a TraceSeries keeps its data
duration:Duration;
// updated_at_nanoseconds indicates when the TraceSeries is updated
diff --git a/api/fbs/v1/schema_generated.go b/api/fbs/v1/schema_generated.go
index c132693..c221b33 100644
--- a/api/fbs/v1/schema_generated.go
+++ b/api/fbs/v1/schema_generated.go
@@ -131,6 +131,78 @@ func (v IndexType) String() string {
return "IndexType(" + strconv.FormatInt(int64(v), 10) + ")"
}
+type ShardInfo struct {
+ _tab flatbuffers.Table
+}
+
+func GetRootAsShardInfo(buf []byte, offset flatbuffers.UOffsetT) *ShardInfo {
+ n := flatbuffers.GetUOffsetT(buf[offset:])
+ x := &ShardInfo{}
+ x.Init(buf, n+offset)
+ return x
+}
+
+func GetSizePrefixedRootAsShardInfo(buf []byte, offset flatbuffers.UOffsetT) *ShardInfo {
+ n := flatbuffers.GetUOffsetT(buf[offset+flatbuffers.SizeUint32:])
+ x := &ShardInfo{}
+ x.Init(buf, n+offset+flatbuffers.SizeUint32)
+ return x
+}
+
+func (rcv *ShardInfo) Init(buf []byte, i flatbuffers.UOffsetT) {
+ rcv._tab.Bytes = buf
+ rcv._tab.Pos = i
+}
+
+func (rcv *ShardInfo) Table() flatbuffers.Table {
+ return rcv._tab
+}
+
+func (rcv *ShardInfo) Number() uint32 {
+ o := flatbuffers.UOffsetT(rcv._tab.Offset(4))
+ if o != 0 {
+ return rcv._tab.GetUint32(o + rcv._tab.Pos)
+ }
+ return 0
+}
+
+func (rcv *ShardInfo) MutateNumber(n uint32) bool {
+ return rcv._tab.MutateUint32Slot(4, n)
+}
+
+func (rcv *ShardInfo) RoutingFields(j int) []byte {
+ o := flatbuffers.UOffsetT(rcv._tab.Offset(6))
+ if o != 0 {
+ a := rcv._tab.Vector(o)
+ return rcv._tab.ByteVector(a + flatbuffers.UOffsetT(j*4))
+ }
+ return nil
+}
+
+func (rcv *ShardInfo) RoutingFieldsLength() int {
+ o := flatbuffers.UOffsetT(rcv._tab.Offset(6))
+ if o != 0 {
+ return rcv._tab.VectorLen(o)
+ }
+ return 0
+}
+
+func ShardInfoStart(builder *flatbuffers.Builder) {
+ builder.StartObject(2)
+}
+func ShardInfoAddNumber(builder *flatbuffers.Builder, number uint32) {
+ builder.PrependUint32Slot(0, number, 0)
+}
+func ShardInfoAddRoutingFields(builder *flatbuffers.Builder, routingFields flatbuffers.UOffsetT) {
+ builder.PrependUOffsetTSlot(1, flatbuffers.UOffsetT(routingFields), 0)
+}
+func ShardInfoStartRoutingFieldsVector(builder *flatbuffers.Builder, numElems int) flatbuffers.UOffsetT {
+ return builder.StartVector(4, numElems, 4)
+}
+func ShardInfoEnd(builder *flatbuffers.Builder) flatbuffers.UOffsetT {
+ return builder.EndObject()
+}
+
type Duration struct {
_tab flatbuffers.Struct
}
@@ -450,9 +522,22 @@ func (rcv *TraceSeries) ReservedFieldsMap(obj *TraceFieldMap) *TraceFieldMap {
return nil
}
-func (rcv *TraceSeries) Duration(obj *Duration) *Duration {
+func (rcv *TraceSeries) Shard(obj *ShardInfo) *ShardInfo {
o := flatbuffers.UOffsetT(rcv._tab.Offset(10))
if o != 0 {
+ x := rcv._tab.Indirect(o + rcv._tab.Pos)
+ if obj == nil {
+ obj = new(ShardInfo)
+ }
+ obj.Init(rcv._tab.Bytes, x)
+ return obj
+ }
+ return nil
+}
+
+func (rcv *TraceSeries) Duration(obj *Duration) *Duration {
+ o := flatbuffers.UOffsetT(rcv._tab.Offset(12))
+ if o != 0 {
x := o + rcv._tab.Pos
if obj == nil {
obj = new(Duration)
@@ -464,7 +549,7 @@ func (rcv *TraceSeries) Duration(obj *Duration) *Duration {
}
func (rcv *TraceSeries) UpdatedAtNanoseconds() uint64 {
- o := flatbuffers.UOffsetT(rcv._tab.Offset(12))
+ o := flatbuffers.UOffsetT(rcv._tab.Offset(14))
if o != 0 {
return rcv._tab.GetUint64(o + rcv._tab.Pos)
}
@@ -472,11 +557,11 @@ func (rcv *TraceSeries) UpdatedAtNanoseconds() uint64 {
}
func (rcv *TraceSeries) MutateUpdatedAtNanoseconds(n uint64) bool {
- return rcv._tab.MutateUint64Slot(12, n)
+ return rcv._tab.MutateUint64Slot(14, n)
}
func TraceSeriesStart(builder *flatbuffers.Builder) {
- builder.StartObject(5)
+ builder.StartObject(6)
}
func TraceSeriesAddMetadata(builder *flatbuffers.Builder, metadata flatbuffers.UOffsetT) {
builder.PrependUOffsetTSlot(0, flatbuffers.UOffsetT(metadata), 0)
@@ -490,11 +575,14 @@ func TraceSeriesStartFieldsVector(builder *flatbuffers.Builder, numElems int) fl
func TraceSeriesAddReservedFieldsMap(builder *flatbuffers.Builder, reservedFieldsMap flatbuffers.UOffsetT) {
builder.PrependUOffsetTSlot(2, flatbuffers.UOffsetT(reservedFieldsMap), 0)
}
+func TraceSeriesAddShard(builder *flatbuffers.Builder, shard flatbuffers.UOffsetT) {
+ builder.PrependUOffsetTSlot(3, flatbuffers.UOffsetT(shard), 0)
+}
func TraceSeriesAddDuration(builder *flatbuffers.Builder, duration flatbuffers.UOffsetT) {
- builder.PrependStructSlot(3, flatbuffers.UOffsetT(duration), 0)
+ builder.PrependStructSlot(4, flatbuffers.UOffsetT(duration), 0)
}
func TraceSeriesAddUpdatedAtNanoseconds(builder *flatbuffers.Builder, updatedAtNanoseconds uint64) {
- builder.PrependUint64Slot(4, updatedAtNanoseconds, 0)
+ builder.PrependUint64Slot(5, updatedAtNanoseconds, 0)
}
func TraceSeriesEnd(builder *flatbuffers.Builder) flatbuffers.UOffsetT {
return builder.EndObject()
diff --git a/api/fbs/v1/write.fbs b/api/fbs/v1/write.fbs
index ff000f2..1d00f49 100644
--- a/api/fbs/v1/write.fbs
+++ b/api/fbs/v1/write.fbs
@@ -46,7 +46,7 @@ table Field {
value: ValueType;
}
-table entityValue {
+table EntityValue {
// entity_id could be span_id of a Span or segment_id of a Segment in the context of Trace
entity_id: string;
// timestamp_nanoseconds is in the timeunit of nanoseconds. It represents
@@ -66,7 +66,10 @@ table WriteEntity {
// the mate_data is only required in the first write.
meta_data: Metadata;
// the entity is required.
- entity: entityValue;
+ entity: EntityValue;
}
table WriteResponse {}
+
+root_type WriteEntity;
+root_type EntityValue;
diff --git a/api/fbs/v1/write_generated.go b/api/fbs/v1/write_generated.go
index 3d71f94..d3b561c 100644
--- a/api/fbs/v1/write_generated.go
+++ b/api/fbs/v1/write_generated.go
@@ -319,34 +319,34 @@ func FieldEnd(builder *flatbuffers.Builder) flatbuffers.UOffsetT {
return builder.EndObject()
}
-type entityValue struct {
+type EntityValue struct {
_tab flatbuffers.Table
}
-func GetRootAsentityValue(buf []byte, offset flatbuffers.UOffsetT) *entityValue {
+func GetRootAsEntityValue(buf []byte, offset flatbuffers.UOffsetT) *EntityValue {
n := flatbuffers.GetUOffsetT(buf[offset:])
- x := &entityValue{}
+ x := &EntityValue{}
x.Init(buf, n+offset)
return x
}
-func GetSizePrefixedRootAsentityValue(buf []byte, offset flatbuffers.UOffsetT) *entityValue {
+func GetSizePrefixedRootAsEntityValue(buf []byte, offset flatbuffers.UOffsetT) *EntityValue {
n := flatbuffers.GetUOffsetT(buf[offset+flatbuffers.SizeUint32:])
- x := &entityValue{}
+ x := &EntityValue{}
x.Init(buf, n+offset+flatbuffers.SizeUint32)
return x
}
-func (rcv *entityValue) Init(buf []byte, i flatbuffers.UOffsetT) {
+func (rcv *EntityValue) Init(buf []byte, i flatbuffers.UOffsetT) {
rcv._tab.Bytes = buf
rcv._tab.Pos = i
}
-func (rcv *entityValue) Table() flatbuffers.Table {
+func (rcv *EntityValue) Table() flatbuffers.Table {
return rcv._tab
}
-func (rcv *entityValue) EntityId() []byte {
+func (rcv *EntityValue) EntityId() []byte {
o := flatbuffers.UOffsetT(rcv._tab.Offset(4))
if o != 0 {
return rcv._tab.ByteVector(o + rcv._tab.Pos)
@@ -354,7 +354,7 @@ func (rcv *entityValue) EntityId() []byte {
return nil
}
-func (rcv *entityValue) TimestampNanoseconds() uint64 {
+func (rcv *EntityValue) TimestampNanoseconds() uint64 {
o := flatbuffers.UOffsetT(rcv._tab.Offset(6))
if o != 0 {
return rcv._tab.GetUint64(o + rcv._tab.Pos)
@@ -362,11 +362,11 @@ func (rcv *entityValue) TimestampNanoseconds() uint64 {
return 0
}
-func (rcv *entityValue) MutateTimestampNanoseconds(n uint64) bool {
+func (rcv *EntityValue) MutateTimestampNanoseconds(n uint64) bool {
return rcv._tab.MutateUint64Slot(6, n)
}
-func (rcv *entityValue) DataBinary(j int) byte {
+func (rcv *EntityValue) DataBinary(j int) byte {
o := flatbuffers.UOffsetT(rcv._tab.Offset(8))
if o != 0 {
a := rcv._tab.Vector(o)
@@ -375,7 +375,7 @@ func (rcv *entityValue) DataBinary(j int) byte {
return 0
}
-func (rcv *entityValue) DataBinaryLength() int {
+func (rcv *EntityValue) DataBinaryLength() int {
o := flatbuffers.UOffsetT(rcv._tab.Offset(8))
if o != 0 {
return rcv._tab.VectorLen(o)
@@ -383,7 +383,7 @@ func (rcv *entityValue) DataBinaryLength() int {
return 0
}
-func (rcv *entityValue) DataBinaryBytes() []byte {
+func (rcv *EntityValue) DataBinaryBytes() []byte {
o := flatbuffers.UOffsetT(rcv._tab.Offset(8))
if o != 0 {
return rcv._tab.ByteVector(o + rcv._tab.Pos)
@@ -391,7 +391,7 @@ func (rcv *entityValue) DataBinaryBytes() []byte {
return nil
}
-func (rcv *entityValue) MutateDataBinary(j int, n byte) bool {
+func (rcv *EntityValue) MutateDataBinary(j int, n byte) bool {
o := flatbuffers.UOffsetT(rcv._tab.Offset(8))
if o != 0 {
a := rcv._tab.Vector(o)
@@ -400,7 +400,7 @@ func (rcv *entityValue) MutateDataBinary(j int, n byte) bool {
return false
}
-func (rcv *entityValue) Fields(obj *Field, j int) bool {
+func (rcv *EntityValue) Fields(obj *Field, j int) bool {
o := flatbuffers.UOffsetT(rcv._tab.Offset(10))
if o != 0 {
x := rcv._tab.Vector(o)
@@ -412,7 +412,7 @@ func (rcv *entityValue) Fields(obj *Field, j int) bool {
return false
}
-func (rcv *entityValue) FieldsLength() int {
+func (rcv *EntityValue) FieldsLength() int {
o := flatbuffers.UOffsetT(rcv._tab.Offset(10))
if o != 0 {
return rcv._tab.VectorLen(o)
@@ -420,28 +420,28 @@ func (rcv *entityValue) FieldsLength() int {
return 0
}
-func entityValueStart(builder *flatbuffers.Builder) {
+func EntityValueStart(builder *flatbuffers.Builder) {
builder.StartObject(4)
}
-func entityValueAddEntityId(builder *flatbuffers.Builder, entityId flatbuffers.UOffsetT) {
+func EntityValueAddEntityId(builder *flatbuffers.Builder, entityId flatbuffers.UOffsetT) {
builder.PrependUOffsetTSlot(0, flatbuffers.UOffsetT(entityId), 0)
}
-func entityValueAddTimestampNanoseconds(builder *flatbuffers.Builder, timestampNanoseconds uint64) {
+func EntityValueAddTimestampNanoseconds(builder *flatbuffers.Builder, timestampNanoseconds uint64) {
builder.PrependUint64Slot(1, timestampNanoseconds, 0)
}
-func entityValueAddDataBinary(builder *flatbuffers.Builder, dataBinary flatbuffers.UOffsetT) {
+func EntityValueAddDataBinary(builder *flatbuffers.Builder, dataBinary flatbuffers.UOffsetT) {
builder.PrependUOffsetTSlot(2, flatbuffers.UOffsetT(dataBinary), 0)
}
-func entityValueStartDataBinaryVector(builder *flatbuffers.Builder, numElems int) flatbuffers.UOffsetT {
+func EntityValueStartDataBinaryVector(builder *flatbuffers.Builder, numElems int) flatbuffers.UOffsetT {
return builder.StartVector(1, numElems, 1)
}
-func entityValueAddFields(builder *flatbuffers.Builder, fields flatbuffers.UOffsetT) {
+func EntityValueAddFields(builder *flatbuffers.Builder, fields flatbuffers.UOffsetT) {
builder.PrependUOffsetTSlot(3, flatbuffers.UOffsetT(fields), 0)
}
-func entityValueStartFieldsVector(builder *flatbuffers.Builder, numElems int) flatbuffers.UOffsetT {
+func EntityValueStartFieldsVector(builder *flatbuffers.Builder, numElems int) flatbuffers.UOffsetT {
return builder.StartVector(4, numElems, 4)
}
-func entityValueEnd(builder *flatbuffers.Builder) flatbuffers.UOffsetT {
+func EntityValueEnd(builder *flatbuffers.Builder) flatbuffers.UOffsetT {
return builder.EndObject()
}
@@ -485,12 +485,12 @@ func (rcv *WriteEntity) MetaData(obj *Metadata) *Metadata {
return nil
}
-func (rcv *WriteEntity) Entity(obj *entityValue) *entityValue {
+func (rcv *WriteEntity) Entity(obj *EntityValue) *EntityValue {
o := flatbuffers.UOffsetT(rcv._tab.Offset(6))
if o != 0 {
x := rcv._tab.Indirect(o + rcv._tab.Pos)
if obj == nil {
- obj = new(entityValue)
+ obj = new(EntityValue)
}
obj.Init(rcv._tab.Bytes, x)
return obj
diff --git a/banyand/internal/cmd/standalone.go b/banyand/internal/cmd/standalone.go
index 6f1ba26..69e069e 100644
--- a/banyand/internal/cmd/standalone.go
+++ b/banyand/internal/cmd/standalone.go
@@ -28,7 +28,7 @@ import (
"github.com/apache/skywalking-banyandb/banyand/liaison"
"github.com/apache/skywalking-banyandb/banyand/query"
"github.com/apache/skywalking-banyandb/banyand/queue"
- "github.com/apache/skywalking-banyandb/banyand/series"
+ "github.com/apache/skywalking-banyandb/banyand/series/trace"
"github.com/apache/skywalking-banyandb/banyand/storage"
"github.com/apache/skywalking-banyandb/pkg/config"
"github.com/apache/skywalking-banyandb/pkg/logger"
@@ -61,11 +61,11 @@ func newStandaloneCmd() *cobra.Command {
if err != nil {
l.Fatal().Err(err).Msg("failed to initiate index builder")
}
- s, err := series.NewService(ctx, db)
+ traceSeries, err := trace.NewService(ctx, db)
if err != nil {
- l.Fatal().Err(err).Msg("failed to initiate series")
+ l.Fatal().Err(err).Msg("failed to initiate trace series")
}
- q, err := query.NewExecutor(ctx, idx, s)
+ q, err := query.NewExecutor(ctx, idx, traceSeries)
if err != nil {
l.Fatal().Err(err).Msg("failed to initiate query executor")
}
@@ -74,12 +74,12 @@ func newStandaloneCmd() *cobra.Command {
l.Fatal().Err(err).Msg("failed to initiate Endpoint transport layer")
}
- // Register the run Group units.
+ // Meta the run Group units.
g.Register(
new(signal.Handler),
repo,
+ traceSeries,
db,
- s,
idx,
q,
tcp,
diff --git a/banyand/kv/badger.go b/banyand/kv/badger.go
index 8cf6187..d118a58 100644
--- a/banyand/kv/badger.go
+++ b/banyand/kv/badger.go
@@ -22,15 +22,15 @@ import (
"github.com/dgraph-io/badger/v3"
- "github.com/apache/skywalking-banyandb/pkg/convert"
"github.com/apache/skywalking-banyandb/pkg/logger"
)
var _ TimeSeriesStore = (*badgerTSS)(nil)
type badgerTSS struct {
- dbOpts badger.Options
- db *badger.DB
+ shardID int
+ dbOpts badger.Options
+ db *badger.DB
badger.TSet
}
@@ -44,18 +44,44 @@ func (b *badgerTSS) Close() error {
var _ Store = (*badgerDB)(nil)
type badgerDB struct {
- dbOpts badger.Options
- db *badger.DB
- seqKey string
- seq *badger.Sequence
+ shardID int
+ dbOpts badger.Options
+ db *badger.DB
+ seqKey string
+ seq *badger.Sequence
}
-func (b *badgerDB) Add(val []byte) (uint64, error) {
- id, err := b.seq.Next()
- if err != nil {
- return 0, err
- }
- return id, b.Put(convert.Uint64ToBytes(id), val)
+func (b *badgerDB) Scan(key []byte, opt ScanOpts, f ScanFunc) error {
+ err := b.db.View(func(txn *badger.Txn) error {
+ opts := badger.DefaultIteratorOptions
+ opts.PrefetchSize = opt.PrefetchSize
+ opts.PrefetchValues = opt.PrefetchValues
+ it := txn.NewIterator(opts)
+ defer it.Close()
+ for it.Seek(key); it.Valid(); it.Next() {
+ item := it.Item()
+ k := item.Key()
+ err := f(b.shardID, k, func() ([]byte, error) {
+ var val []byte
+ err := item.Value(func(v []byte) error {
+ val = v
+ return nil
+ })
+ if err != nil {
+ return nil, err
+ }
+ return val, nil
+ })
+ if err == ErrStopScan {
+ break
+ }
+ if err != nil {
+ return err
+ }
+ }
+ return nil
+ })
+ return err
}
func (b *badgerDB) Close() error {
diff --git a/banyand/kv/kv.go b/banyand/kv/kv.go
index 22dbc76..34a5188 100644
--- a/banyand/kv/kv.go
+++ b/banyand/kv/kv.go
@@ -15,55 +15,71 @@
// specific language governing permissions and limitations
// under the License.
-//go:generate mockgen -destination=./kv_mock.go -package=kv . Hook
package kv
import (
- "bytes"
"fmt"
"io"
"github.com/dgraph-io/badger/v3"
- "github.com/dgraph-io/badger/v3/y"
+ "github.com/pkg/errors"
"github.com/apache/skywalking-banyandb/pkg/logger"
)
-// Hook contains several functions which are invoked on MemTables flushing or LevelTables retrieving
-type Hook interface {
- // Reduce the values with an identical key on MemTables flushing
- Reduce(left bytes.Buffer, right y.ValueStruct) bytes.Buffer
- // Extract an chunk from a reduced value on LevelTables retrieving
- Extract(raw []byte, ts uint64) ([]byte, error)
- // Split a reduced value into chunks on LevelTables retrieving
- Split(raw []byte) ([][]byte, error)
+var ErrStopScan = errors.New("stop scanning")
+var DefaultScanOpts = ScanOpts{
+ PrefetchSize: 100,
+ PrefetchValues: true,
}
-// Store is a common kv storage with auto-generated key
-type Store interface {
- io.Closer
- // Add a value whose key can be generated automatically
- Add(val []byte) (uint64, error)
+type Writer interface {
// Put a value
Put(key, val []byte) error
+}
+
+type ScanFunc func(shardID int, key []byte, getVal func() ([]byte, error)) error
+
+type ScanOpts struct {
+ PrefetchSize int
+ PrefetchValues bool
+}
+
+type Reader interface {
// Get a value by its key
Get(key []byte) ([]byte, error)
+ Scan(prefix []byte, opt ScanOpts, f ScanFunc) error
}
-// TimeSeriesStore is time series storage
-type TimeSeriesStore interface {
+// Store is a common kv storage with auto-generated key
+type Store interface {
io.Closer
+ Writer
+ Reader
+}
+
+type TimeSeriesWriter interface {
// Put a value with a timestamp/version
Put(key, val []byte, ts uint64) error
// PutAsync a value with a timestamp/version asynchronously.
// Injected "f" func will notice the result of value write.
PutAsync(key, val []byte, ts uint64, f func(error)) error
+}
+
+type TimeSeriesReader interface {
// Get a value by its key and timestamp/version
Get(key []byte, ts uint64) ([]byte, error)
// GetAll values with an identical key
GetAll(key []byte) ([][]byte, error)
}
+// TimeSeriesStore is time series storage
+type TimeSeriesStore interface {
+ io.Closer
+ TimeSeriesWriter
+ TimeSeriesReader
+}
+
type TimeSeriesOptions func(TimeSeriesStore)
// TSSWithLogger sets a external logger into underlying TimeSeriesStore
@@ -71,15 +87,16 @@ func TSSWithLogger(l *logger.Logger) TimeSeriesOptions {
return func(store TimeSeriesStore) {
if btss, ok := store.(*badgerTSS); ok {
btss.dbOpts = btss.dbOpts.WithLogger(&badgerLog{
- delegated: l,
+ delegated: l.Named("ts-kv"),
})
}
}
}
// OpenTimeSeriesStore creates a new TimeSeriesStore
-func OpenTimeSeriesStore(path string, hook Hook, options ...TimeSeriesOptions) (TimeSeriesStore, error) {
+func OpenTimeSeriesStore(shardID int, path string, compressLevel int, valueSize int, options ...TimeSeriesOptions) (TimeSeriesStore, error) {
btss := new(badgerTSS)
+ btss.shardID = shardID
btss.dbOpts = badger.DefaultOptions(path)
for _, opt := range options {
opt(btss)
@@ -90,7 +107,7 @@ func OpenTimeSeriesStore(path string, hook Hook, options ...TimeSeriesOptions) (
if err != nil {
return nil, fmt.Errorf("failed to open time series store: %v", err)
}
- btss.TSet = *badger.NewTSet(btss.db, hook.Reduce, hook.Extract, hook.Split)
+ btss.TSet = *badger.NewTSet(btss.db, compressLevel, valueSize)
return btss, nil
}
@@ -101,24 +118,16 @@ func StoreWithLogger(l *logger.Logger) StoreOptions {
return func(store Store) {
if bdb, ok := store.(*badgerDB); ok {
bdb.dbOpts = bdb.dbOpts.WithLogger(&badgerLog{
- delegated: l,
+ delegated: l.Named("normal-kv"),
})
}
}
}
-// StoreWithAutoGeneratedKey actives auto generation key to the Store
-func StoreWithAutoGeneratedKey() StoreOptions {
- return func(store Store) {
- if bdb, ok := store.(*badgerDB); ok {
- bdb.seqKey = "sequence"
- }
- }
-}
-
// OpenStore creates a new Store
-func OpenStore(path string, options ...StoreOptions) (Store, error) {
+func OpenStore(shardID int, path string, options ...StoreOptions) (Store, error) {
bdb := new(badgerDB)
+ bdb.shardID = shardID
bdb.dbOpts = badger.DefaultOptions(path)
for _, opt := range options {
opt(bdb)
diff --git a/banyand/series/idgen.go b/banyand/series/idgen.go
new file mode 100644
index 0000000..2acb2ba
--- /dev/null
+++ b/banyand/series/idgen.go
@@ -0,0 +1,60 @@
+// 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 series
+
+import (
+ "github.com/pkg/errors"
+
+ "github.com/apache/skywalking-banyandb/pkg/bytes"
+ "github.com/apache/skywalking-banyandb/pkg/convert"
+)
+
+var ErrInvalidID = errors.New("invalid id")
+
+type IDGen interface {
+ Next(shard uint, ts uint64) []byte
+ ParseShardID(ID []byte) (uint, error)
+ ParseTS(ID []byte) (uint64, error)
+}
+
+func NewIDGen() IDGen {
+ return &localID{}
+}
+
+var _ IDGen = (*localID)(nil)
+
+type localID struct {
+}
+
+func (l *localID) Next(shard uint, ts uint64) []byte {
+ return bytes.Join(convert.Uint32ToBytes(uint32(shard)), convert.Uint64ToBytes(ts))
+}
+
+func (l *localID) ParseShardID(ID []byte) (uint, error) {
+ if len(ID) < 12 {
+ return 0, ErrInvalidID
+ }
+ return uint(convert.BytesToUint32(ID[:4])), nil
+}
+
+func (l *localID) ParseTS(ID []byte) (uint64, error) {
+ if len(ID) < 12 {
+ return 0, ErrInvalidID
+ }
+ return convert.BytesToUint64(ID[4:]), nil
+}
diff --git a/banyand/series/schema/sw/index_rule.bin b/banyand/series/schema/sw/index_rule.bin
index 4d1204d..c45069e 100644
Binary files a/banyand/series/schema/sw/index_rule.bin and b/banyand/series/schema/sw/index_rule.bin differ
diff --git a/banyand/series/schema/sw/index_rule.json b/banyand/series/schema/sw/index_rule.json
index 7ecfbe2..978a952 100644
--- a/banyand/series/schema/sw/index_rule.json
+++ b/banyand/series/schema/sw/index_rule.json
@@ -30,6 +30,12 @@
},
{
"fields": [
+ "duration"
+ ],
+ "type": "Numerical"
+ },
+ {
+ "fields": [
"http.method"
],
"type": "Text"
diff --git a/banyand/series/schema/sw/trace_series.bin b/banyand/series/schema/sw/trace_series.bin
index a089d72..bd92821 100644
Binary files a/banyand/series/schema/sw/trace_series.bin and b/banyand/series/schema/sw/trace_series.bin differ
diff --git a/banyand/series/schema/sw/trace_series.json b/banyand/series/schema/sw/trace_series.json
index 7e6bd0c..22c1094 100644
--- a/banyand/series/schema/sw/trace_series.json
+++ b/banyand/series/schema/sw/trace_series.json
@@ -9,6 +9,10 @@
"type": "String"
},
{
+ "name": "state",
+ "type": "Int"
+ },
+ {
"name": "service_id",
"type": "String"
},
@@ -33,7 +37,11 @@
"type": "String"
},
{
- "name": "state",
+ "name": "duration",
+ "type": "Int"
+ },
+ {
+ "name": "start_time",
"type": "Int"
},
{
@@ -77,6 +85,14 @@
"val_error": "1"
}
},
+ "shard": {
+ "number": 2,
+ "routing_fields": [
+ "service_id",
+ "service_instance_id",
+ "endpoint_id"
+ ]
+ },
"duration": {
"val": 7,
"unit": "Day"
diff --git a/banyand/series/series.go b/banyand/series/series.go
index b09644f..e75db05 100644
--- a/banyand/series/series.go
+++ b/banyand/series/series.go
@@ -25,11 +25,10 @@ import (
"github.com/apache/skywalking-banyandb/api/data"
v1 "github.com/apache/skywalking-banyandb/api/fbs/v1"
"github.com/apache/skywalking-banyandb/banyand/series/schema"
- "github.com/apache/skywalking-banyandb/banyand/storage"
"github.com/apache/skywalking-banyandb/pkg/run"
)
-// TraceState represents the state of a trace link
+// TraceState represents the State of a traceSeries link
type TraceState int
const (
@@ -41,15 +40,15 @@ const (
//ScanOptions contain options
//nolint
type ScanOptions struct {
- projection []string
- state TraceState
- limit uint32
+ Projection []string
+ State TraceState
+ Limit uint32
}
-//TraceRepo contains trace and entity data
+//TraceRepo contains traceSeries and entity data
type TraceRepo interface {
//FetchTrace returns data.Trace by traceID
- FetchTrace(traceSeries common.Metadata, traceID string) (data.Trace, error)
+ FetchTrace(traceSeries common.Metadata, traceID string, opt ScanOptions) (data.Trace, error)
//FetchEntity returns data.Entity by ChunkID
FetchEntity(traceSeries common.Metadata, chunkIDs []common.ChunkID, opt ScanOptions) ([]data.Entity, error)
//ScanEntity returns data.Entity between a duration by ScanOptions
@@ -81,11 +80,5 @@ type Service interface {
UniModel
SchemaRepo
IndexFilter
- run.Config
run.PreRunner
}
-
-//NewService returns a new service
-func NewService(ctx context.Context, db storage.Database) (Service, error) {
- return &service{db: db}, nil
-}
diff --git a/banyand/series/trace/common_test.go b/banyand/series/trace/common_test.go
new file mode 100644
index 0000000..205b15e
--- /dev/null
+++ b/banyand/series/trace/common_test.go
@@ -0,0 +1,270 @@
+// 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 trace
+
+import (
+ "bytes"
+ "context"
+ "sort"
+ "testing"
+ "time"
+
+ flatbuffers "github.com/google/flatbuffers/go"
+ "github.com/stretchr/testify/assert"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/api/data"
+ v1 "github.com/apache/skywalking-banyandb/api/fbs/v1"
+ "github.com/apache/skywalking-banyandb/banyand/storage"
+ "github.com/apache/skywalking-banyandb/pkg/convert"
+ "github.com/apache/skywalking-banyandb/pkg/fb"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/partition"
+)
+
+var _ sort.Interface = (ByEntityID)(nil)
+
+type ByEntityID []data.Entity
+
+func (b ByEntityID) Len() int {
+ return len(b)
+}
+
+func (b ByEntityID) Less(i, j int) bool {
+ return bytes.Compare(b[i].EntityId(), b[j].EntityId()) < 0
+}
+
+func (b ByEntityID) Swap(i, j int) {
+ b[i], b[j] = b[j], b[i]
+}
+
+func setup(t *testing.T) (*traceSeries, func()) {
+ _ = logger.Bootstrap()
+ db, err := storage.NewDB(context.TODO(), nil)
+ assert.NoError(t, err)
+ assert.NoError(t, db.FlagSet().Parse(nil))
+ svc, err := NewService(context.TODO(), db)
+ assert.NoError(t, err)
+ assert.NoError(t, svc.PreRun())
+ assert.NoError(t, db.PreRun())
+ traceSVC := svc.(*service)
+ b := flatbuffers.NewBuilder(0)
+ name := b.CreateString("sw")
+ group := b.CreateString("default")
+ v1.MetadataStart(b)
+ v1.MetadataAddName(b, name)
+ v1.MetadataAddGroup(b, group)
+ b.Finish(v1.MetadataEnd(b))
+ meta := new(v1.Metadata)
+ meta.Init(b.Bytes, b.Offset())
+ ts, err := traceSVC.getSeries(common.Metadata{
+ Spec: v1.GetRootAsMetadata(b.FinishedBytes(), 0),
+ })
+ assert.NoError(t, err)
+ return ts, db.GracefulStop
+}
+
+type seriesEntity struct {
+ seriesID string
+ entity entity
+}
+
+type wantEntity struct {
+ entityID string
+ dataBinary []byte
+ fieldsSize int
+}
+
+var gap = uint64(500 * time.Millisecond)
+
+func testData(ts uint64) []seriesEntity {
+
+ return []seriesEntity{
+ {
+ seriesID: "webapp_10.0.0.1",
+ entity: getEntityWithTS("1", []byte{11}, ts,
+ "trace_id-xxfff.111323",
+ 0,
+ "webapp_id",
+ "10.0.0.1_id",
+ "/home_id",
+ "webapp",
+ "10.0.0.1",
+ "/home",
+ 300,
+ 1622933202000000000,
+ ),
+ },
+ {
+ seriesID: "gateway_10.0.0.2",
+ entity: getEntityWithTS("2", []byte{12}, ts+gap,
+ "trace_id-xxfff.111323a",
+ 1,
+ ),
+ },
+ {
+ seriesID: "httpserver_10.0.0.3",
+ entity: getEntityWithTS("3", []byte{13}, ts+2*gap,
+ "trace_id-xxfff.111323",
+ 1,
+ "httpserver_id",
+ "10.0.0.3_id",
+ "/home_id",
+ "httpserver",
+ "10.0.0.3",
+ "/home",
+ 300,
+ 1622933202000000000,
+ "GET",
+ "200",
+ ),
+ },
+ {
+ seriesID: "database_10.0.0.4",
+ entity: getEntityWithTS("4", []byte{14}, ts+3*gap,
+ "trace_id-xxfff.111323",
+ 0,
+ "database_id",
+ "10.0.0.4_id",
+ "/home_id",
+ "database",
+ "10.0.0.4",
+ "/home",
+ 300,
+ 1622933202000000000,
+ nil,
+ nil,
+ "MySQL",
+ "10.1.1.4",
+ ),
+ },
+ {
+ seriesID: "mq_10.0.0.5",
+ entity: getEntityWithTS("5", []byte{15}, ts+4*gap,
+ "trace_id-zzpp.111323",
+ 0,
+ "mq_id",
+ "10.0.0.5_id",
+ "/home_id",
+ "mq",
+ "10.0.0.5",
+ "/home",
+ 300,
+ 1622933202000000000,
+ nil,
+ nil,
+ nil,
+ nil,
+ "test_topic",
+ "10.0.0.5",
+ ),
+ },
+ {
+ seriesID: "database_10.0.0.6",
+ entity: getEntityWithTS("6", []byte{16}, ts+5*gap,
+ "trace_id-zzpp.111323",
+ 1,
+ "database_id",
+ "10.0.0.6_id",
+ "/home_id",
+ "database",
+ "10.0.0.6",
+ "/home",
+ 300,
+ 1622933202000000000,
+ nil,
+ nil,
+ "MySQL",
+ "10.1.1.6",
+ ),
+ },
+ {
+ seriesID: "mq_10.0.0.7",
+ entity: getEntityWithTS("7", []byte{17}, ts+6*gap,
+ "trace_id-zzpp.111323",
+ 0,
+ "nq_id",
+ "10.0.0.7_id",
+ "/home_id",
+ "mq",
+ "10.0.0.7",
+ "/home",
+ 300,
+ 1622933202000000000,
+ nil,
+ nil,
+ nil,
+ nil,
+ "test_topic",
+ "10.0.0.7",
+ ),
+ },
+ }
+}
+
+type entity struct {
+ id string
+ binary []byte
+ ts uint64
+ items []interface{}
+}
+
+func getEntity(id string, binary []byte, items ...interface{}) entity {
+ return entity{
+ id: id,
+ binary: binary,
+ items: items,
+ }
+}
+
+func getEntityWithTS(id string, binary []byte, ts uint64, items ...interface{}) entity {
+ return entity{
+ id: id,
+ binary: binary,
+ ts: ts,
+ items: items,
+ }
+}
+
+func setUpTestData(t *testing.T, ts *traceSeries, seriesEntities []seriesEntity) (chunkIDs []common.ChunkID) {
+ chunkIDs = make([]common.ChunkID, 0, len(seriesEntities))
+ for _, se := range seriesEntities {
+ seriesID := []byte(se.seriesID)
+ b := fb.NewWriteEntityBuilder()
+ items := make([]fb.ComponentBuilderFunc, 0, 2)
+ items = append(items, b.BuildMetaData("default", "sw"))
+ timestamp := se.entity.ts
+ items = append(items, b.BuildEntityWithTS(se.entity.id, se.entity.binary, timestamp, se.entity.items...))
+ builder, err := b.BuildWriteEntity(
+ items...,
+ )
+ assert.NoError(t, err)
+ we := v1.GetRootAsWriteEntity(builder.FinishedBytes(), 0)
+ got, err := ts.Write(common.SeriesID(convert.Hash(seriesID)), partition.ShardID(seriesID, 2), data.EntityValue{
+ EntityValue: we.Entity(nil),
+ })
+ if err != nil {
+ t.Error("Write() got error")
+ }
+ if len([]byte(got)) < 1 {
+ t.Error("Write() got empty chunkID")
+ }
+ chunkIDs = append(chunkIDs, got)
+ }
+ return chunkIDs
+}
diff --git a/banyand/series/trace/query.go b/banyand/series/trace/query.go
new file mode 100644
index 0000000..3967fc5
--- /dev/null
+++ b/banyand/series/trace/query.go
@@ -0,0 +1,253 @@
+// 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 trace
+
+import (
+ "encoding/hex"
+ "time"
+
+ flatbuffers "github.com/google/flatbuffers/go"
+ "github.com/pkg/errors"
+ "go.uber.org/multierr"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/api/data"
+ v1 "github.com/apache/skywalking-banyandb/api/fbs/v1"
+ "github.com/apache/skywalking-banyandb/banyand/kv"
+ "github.com/apache/skywalking-banyandb/banyand/series"
+ "github.com/apache/skywalking-banyandb/pkg/convert"
+ "github.com/apache/skywalking-banyandb/pkg/fb"
+ "github.com/apache/skywalking-banyandb/pkg/partition"
+)
+
+func (t *traceSeries) FetchTrace(traceID string, opt series.ScanOptions) (trace data.Trace, err error) {
+ if traceID == "" {
+ return trace, ErrInvalidTraceID
+ }
+ traceIDBytes := []byte(traceID)
+ traceIDShardID := partition.ShardID(traceIDBytes, t.shardNum)
+ bb, errTraceID := t.reader.TimeSeriesReader(traceIDShardID, traceIndex, 0, 0).GetAll(traceIDBytes)
+ if errTraceID != nil {
+ return trace, errTraceID
+ }
+ t.l.Debug().Uint("shard_id", traceIDShardID).
+ Str("trace_id", traceID).
+ Hex("trace_id_bytes", traceIDBytes).
+ Int("chunk_num", len(bb)).Msg("fetch Trace by trace_id")
+ if len(bb) < 1 {
+ return trace, nil
+ }
+ chunkIDs := make([]common.ChunkID, len(bb))
+ for i, b := range bb {
+ chunkIDs[i] = b
+ }
+ entities, errEntity := t.FetchEntity(chunkIDs, opt)
+ if errEntity != nil {
+ return trace, errEntity
+ }
+ return data.Trace{
+ KindVersion: data.TraceKindVersion,
+ Entities: entities,
+ }, err
+}
+
+func (t *traceSeries) ScanEntity(startTime, endTime uint64, opt series.ScanOptions) ([]data.Entity, error) {
+ total := opt.Limit
+ if total < 1 {
+ total = 10
+ }
+ states := make([]byte, 0, 2)
+ switch opt.State {
+ case series.TraceStateSuccess:
+ states = append(states, StateSuccess)
+ case series.TraceStateError:
+ states = append(states, StateError)
+ default:
+ states = append(states, StateSuccess, StateError)
+ }
+ seekKeys := make([][]byte, 0, len(states))
+ startTimeBytes := convert.Uint64ToBytes(startTime)
+ for _, state := range states {
+ key := make([]byte, 8+1)
+ key[0] = state
+ copy(key[1:], startTimeBytes)
+ seekKeys = append(seekKeys, key)
+ }
+ chunkIDs := make([]common.ChunkID, 0, total)
+ var num uint32
+ opts := kv.DefaultScanOpts
+ opts.PrefetchValues = false
+ opts.PrefetchSize = int(total)
+ var errAll error
+ for i := uint(0); i < t.shardNum; i++ {
+ for _, seekKey := range seekKeys {
+ state := seekKey[0]
+ err := t.reader.Reader(i, startTimeIndex, startTime, endTime).Scan(
+ seekKey,
+ opts,
+ func(shardID int, key []byte, _ func() ([]byte, error)) error {
+ if len(key) <= 9 {
+ return errors.Wrapf(ErrInvalidKey, "key:%s", hex.EncodeToString(key))
+ }
+ if key[0] != state {
+ return kv.ErrStopScan
+ }
+ ts := convert.BytesToUint64(key[1 : 8+1])
+ if ts > endTime {
+ return nil
+ }
+ chunk := make([]byte, len(key)-8-1)
+ copy(chunk, key[8+1:])
+ chunkID := common.ChunkID(chunk)
+ chunkIDs = append(chunkIDs, chunkID)
+ num++
+ if num > total {
+ return kv.ErrStopScan
+ }
+ return nil
+ })
+ if err != nil {
+ errAll = multierr.Append(errAll, err)
+ }
+ }
+ }
+ if len(chunkIDs) < 1 {
+ return nil, errAll
+ }
+ entities, err := t.FetchEntity(chunkIDs, opt)
+ if err != nil {
+ errAll = multierr.Append(errAll, err)
+ }
+ return entities, errAll
+}
+
+func (t *traceSeries) FetchEntity(chunkIDs []common.ChunkID, opt series.ScanOptions) (entities []data.Entity, err error) {
+ chunkIDsLen := len(chunkIDs)
+ if chunkIDsLen < 1 {
+ return nil, ErrChunkIDsEmpty
+ }
+ entities = make([]data.Entity, 0, len(chunkIDs))
+ fetchDataBinary, fetchFieldsIndices, errInfo := t.parseFetchInfo(opt)
+ if errInfo != nil {
+ return nil, errInfo
+ }
+ if !fetchDataBinary && len(fetchFieldsIndices) < 1 {
+ return nil, ErrProjectionEmpty
+ }
+ for _, chunkID := range chunkIDs {
+ shardID, errParseID := t.idGen.ParseShardID(chunkID)
+ if errParseID != nil {
+ err = multierr.Append(err, errParseID)
+ }
+ ts, errParseTS := t.idGen.ParseTS(chunkID)
+ if errParseTS != nil {
+ err = multierr.Append(err, errParseTS)
+ }
+ ref, chunkErr := t.reader.Reader(shardID, chunkIDMapping, ts, ts).Get(chunkID)
+ if chunkErr != nil {
+ err = multierr.Append(err, chunkErr)
+ continue
+ }
+ sRef := ref[:len(ref)-8]
+ seriesID := sRef[1:]
+ state := sRef[0]
+
+ t.l.Debug().
+ Hex("chunk_id", chunkID).
+ Hex("id", ref).
+ Uint64("series_id", convert.BytesToUint64(seriesID)).
+ Uint("shard_id", shardID).
+ Time("ts", time.Unix(0, int64(ts))).
+ Uint64("ts_int", ts).
+ Msg("fetch internal id by chunk_id")
+ entity, errGet := t.getEntityByInternalRef(seriesID, State(state), fetchDataBinary, fetchFieldsIndices, shardID, ts)
+ if errGet != nil {
+ err = multierr.Append(err, errGet)
+ continue
+ }
+ t.l.Debug().
+ Hex("entity_id", entity.EntityId()).
+ Int("fields_num", entity.FieldsLength()).
+ Int("data_binary_size_bytes", entity.DataBinaryLength()).
+ Msg("fetch entity")
+ entities = append(entities, entity)
+ }
+ return entities, err
+}
+
+func (t *traceSeries) parseFetchInfo(opt series.ScanOptions) (fetchDataBinary bool, fetchFieldsIndices []fb.FieldEntry, err error) {
+ fetchFieldsIndices = make([]fb.FieldEntry, 0)
+ for _, p := range opt.Projection {
+ if p == common.DataBinaryFieldName {
+ fetchDataBinary = true
+ t.l.Debug().Msg("to fetch data binary")
+ continue
+ }
+ index, ok := t.fieldIndex[p]
+ if !ok {
+ return false, nil, errors.Wrapf(ErrFieldNotFound, "field name:%s", p)
+ }
+ fetchFieldsIndices = append(fetchFieldsIndices, fb.FieldEntry{
+ Key: p,
+ Index: int(index),
+ })
+ t.l.Debug().Str("name", p).Uint("index", index).Msg("to fetch the field")
+ }
+ return fetchDataBinary, fetchFieldsIndices, nil
+}
+
+func (t *traceSeries) getEntityByInternalRef(seriesID []byte, state State, fetchDataBinary bool,
+ fetchFieldsIndices []fb.FieldEntry, shardID uint, ts uint64) (data.Entity, error) {
+ fieldsStore, dataStore, err := getStoreName(state)
+ if err != nil {
+ return data.Entity{}, err
+ }
+ b := flatbuffers.NewBuilder(0)
+ var fieldsOffset flatbuffers.UOffsetT
+ val, getErr := t.reader.TimeSeriesReader(shardID, fieldsStore, ts, ts).Get(seriesID, ts)
+ if getErr != nil {
+ return data.Entity{}, getErr
+ }
+ entityVal := v1.GetRootAsEntityValue(val, 0)
+ entityIDOffset := b.CreateByteString(entityVal.EntityId())
+ timestamp := entityVal.TimestampNanoseconds()
+ if len(fetchFieldsIndices) > 0 {
+ fieldsOffset = fb.Transform(entityVal, fetchFieldsIndices, b)
+ }
+ var dataBinary flatbuffers.UOffsetT
+ if fetchDataBinary {
+ val, getErr = t.reader.TimeSeriesReader(shardID, dataStore, ts, ts).Get(seriesID, ts)
+ if getErr != nil {
+ return data.Entity{}, getErr
+ }
+ dataBinary = b.CreateByteVector(val)
+ }
+ v1.EntityValueStart(b)
+ v1.EntityAddEntityId(b, entityIDOffset)
+ v1.EntityAddTimestampNanoseconds(b, timestamp)
+ if fieldsOffset > 0 {
+ v1.EntityAddFields(b, fieldsOffset)
+ }
+ if dataBinary > 0 {
+ v1.EntityAddDataBinary(b, dataBinary)
+ }
+ b.Finish(v1.EntityValueEnd(b))
+ return data.Entity{
+ Entity: v1.GetRootAsEntity(b.FinishedBytes(), 0),
+ }, nil
+}
diff --git a/banyand/series/trace/query_test.go b/banyand/series/trace/query_test.go
new file mode 100644
index 0000000..1892cbd
--- /dev/null
+++ b/banyand/series/trace/query_test.go
@@ -0,0 +1,301 @@
+// 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 trace
+
+import (
+ "math"
+ "sort"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/banyand/series"
+)
+
+func Test_traceSeries_FetchEntity(t *testing.T) {
+ type args struct {
+ chunkIDIndices []int
+ chunkIDs []string
+ opt series.ScanOptions
+ }
+ tests := []struct {
+ name string
+ args args
+ wantEntities []wantEntity
+ wantErr bool
+ }{
+ {
+ name: "golden path",
+ args: args{
+ chunkIDIndices: []int{0, 1, 2, 3, 4, 5, 6},
+ opt: series.ScanOptions{Projection: []string{"trace_id", "data_binary"}},
+ },
+ wantEntities: []wantEntity{
+ {entityID: "1", dataBinary: []byte{11}, fieldsSize: 1},
+ {entityID: "2", dataBinary: []byte{12}, fieldsSize: 1},
+ {entityID: "3", dataBinary: []byte{13}, fieldsSize: 1},
+ {entityID: "4", dataBinary: []byte{14}, fieldsSize: 1},
+ {entityID: "5", dataBinary: []byte{15}, fieldsSize: 1},
+ {entityID: "6", dataBinary: []byte{16}, fieldsSize: 1},
+ {entityID: "7", dataBinary: []byte{17}, fieldsSize: 1},
+ },
+ },
+ {
+ name: "multiple fields",
+ args: args{
+ chunkIDIndices: []int{0, 1, 2, 3, 4, 5, 6},
+ opt: series.ScanOptions{Projection: []string{"trace_id", "service_name", "state", "duration", "mq.queue"}},
+ },
+ wantEntities: []wantEntity{
+ {entityID: "1", fieldsSize: 4},
+ {entityID: "2", fieldsSize: 2},
+ {entityID: "3", fieldsSize: 4},
+ {entityID: "4", fieldsSize: 4},
+ {entityID: "5", fieldsSize: 5},
+ {entityID: "6", fieldsSize: 4},
+ {entityID: "7", fieldsSize: 5},
+ },
+ },
+ {
+ name: "data binary",
+ args: args{
+ chunkIDIndices: []int{0, 1, 2, 3, 4, 5, 6},
+ opt: series.ScanOptions{Projection: []string{"data_binary"}},
+ },
+ wantEntities: []wantEntity{
+ {entityID: "1", dataBinary: []byte{11}},
+ {entityID: "2", dataBinary: []byte{12}},
+ {entityID: "3", dataBinary: []byte{13}},
+ {entityID: "4", dataBinary: []byte{14}},
+ {entityID: "5", dataBinary: []byte{15}},
+ {entityID: "6", dataBinary: []byte{16}},
+ {entityID: "7", dataBinary: []byte{17}},
+ },
+ },
+ {
+ name: "invalid chunk ids",
+ args: args{
+ chunkIDs: []string{"x"},
+ opt: series.ScanOptions{Projection: []string{"trace_id", "data_binary"}},
+ },
+ wantErr: true,
+ },
+ {
+ name: "mix up invalid/valid ids",
+ args: args{
+ chunkIDs: []string{"undefined"},
+ chunkIDIndices: []int{0, 1},
+ opt: series.ScanOptions{Projection: []string{"trace_id", "data_binary"}},
+ },
+ wantEntities: []wantEntity{
+ {entityID: "1", dataBinary: []byte{11}, fieldsSize: 1},
+ {entityID: "2", dataBinary: []byte{12}, fieldsSize: 1},
+ },
+ wantErr: true,
+ },
+ {
+ name: "absent scan opt",
+ args: args{
+ chunkIDIndices: []int{0, 1},
+ },
+ wantErr: true,
+ },
+ {
+ name: "invalid opt absent",
+ args: args{
+ chunkIDIndices: []int{0, 1},
+ opt: series.ScanOptions{Projection: []string{"trace_id", "undefined"}},
+ },
+ wantErr: true,
+ },
+ }
+ ts, stopFunc := setup(t)
+ defer stopFunc()
+ chunkIDs := setUpTestData(t, ts, testData(uint64(time.Now().UnixNano())))
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ chunkIDCriteria := make([]common.ChunkID, 0, len(tt.args.chunkIDIndices))
+ for i := range tt.args.chunkIDIndices {
+ chunkIDCriteria = append(chunkIDCriteria, chunkIDs[i])
+ }
+ for _, id := range tt.args.chunkIDs {
+ chunkIDCriteria = append(chunkIDCriteria, common.ChunkID(id))
+ }
+ entities, err := ts.FetchEntity(chunkIDCriteria, tt.args.opt)
+ if (err != nil) != tt.wantErr {
+ t.Errorf("Write() error = %v, wantErr %v", err, tt.wantErr)
+ }
+ assert.Equal(t, len(tt.wantEntities), len(entities))
+ for i, e := range entities {
+ assert.EqualValues(t, tt.wantEntities[i].entityID, e.EntityId())
+ assert.Equal(t, tt.wantEntities[i].dataBinary, e.DataBinaryBytes())
+ assert.Equal(t, tt.wantEntities[i].fieldsSize, e.FieldsLength())
+ }
+ })
+ }
+}
+
+func Test_traceSeries_FetchTrace(t *testing.T) {
+ type args struct {
+ traceID string
+ }
+ tests := []struct {
+ name string
+ args args
+ wantEntities []wantEntity
+ wantErr bool
+ }{
+ {
+ name: "golden path",
+ args: args{
+ traceID: "trace_id-xxfff.111323",
+ },
+ wantEntities: []wantEntity{
+ {entityID: "1", dataBinary: []byte{11}, fieldsSize: 1},
+ {entityID: "2", dataBinary: []byte{12}, fieldsSize: 1},
+ {entityID: "3", dataBinary: []byte{13}, fieldsSize: 1},
+ {entityID: "4", dataBinary: []byte{14}, fieldsSize: 1},
+ },
+ },
+ {
+ name: "found nothing",
+ args: args{
+ traceID: "not_existed",
+ },
+ },
+ {
+ name: "absent Trace id",
+ args: args{},
+ wantErr: true,
+ },
+ }
+ ts, stopFunc := setup(t)
+ defer stopFunc()
+ setUpTestData(t, ts, testData(uint64(time.Now().UnixNano())))
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ traceData, err := ts.FetchTrace(tt.args.traceID, series.ScanOptions{Projection: []string{"data_binary", "trace_id"}})
+ if (err != nil) != tt.wantErr {
+ t.Errorf("Write() error = %v, wantErr %v", err, tt.wantErr)
+ }
+ var entities ByEntityID = traceData.Entities
+ assert.Equal(t, len(tt.wantEntities), len(entities))
+ sort.Sort(entities)
+ for i, e := range entities {
+ assert.EqualValues(t, tt.wantEntities[i].entityID, e.EntityId())
+ assert.Equal(t, tt.wantEntities[i].dataBinary, e.DataBinaryBytes())
+ assert.Equal(t, tt.wantEntities[i].fieldsSize, e.FieldsLength())
+ }
+ })
+ }
+}
+
+func Test_traceSeries_ScanEntity(t *testing.T) {
+ type args struct {
+ start uint64
+ end uint64
+ }
+ baseTS := uint64(time.Now().UnixNano())
+ tests := []struct {
+ name string
+ args args
+ wantEntities []wantEntity
+ wantErr bool
+ }{
+ {
+ name: "scan all",
+ args: args{
+ start: 0,
+ end: math.MaxUint64,
+ },
+ wantEntities: []wantEntity{
+ {entityID: "1", dataBinary: []byte{11}, fieldsSize: 1},
+ {entityID: "2", dataBinary: []byte{12}, fieldsSize: 1},
+ {entityID: "3", dataBinary: []byte{13}, fieldsSize: 1},
+ {entityID: "4", dataBinary: []byte{14}, fieldsSize: 1},
+ {entityID: "5", dataBinary: []byte{15}, fieldsSize: 1},
+ {entityID: "6", dataBinary: []byte{16}, fieldsSize: 1},
+ {entityID: "7", dataBinary: []byte{17}, fieldsSize: 1},
+ },
+ },
+ {
+ name: "scan range",
+ args: args{
+ start: baseTS,
+ end: baseTS + 6*gap,
+ },
+ wantEntities: []wantEntity{
+ {entityID: "1", dataBinary: []byte{11}, fieldsSize: 1},
+ {entityID: "2", dataBinary: []byte{12}, fieldsSize: 1},
+ {entityID: "3", dataBinary: []byte{13}, fieldsSize: 1},
+ {entityID: "4", dataBinary: []byte{14}, fieldsSize: 1},
+ {entityID: "5", dataBinary: []byte{15}, fieldsSize: 1},
+ {entityID: "6", dataBinary: []byte{16}, fieldsSize: 1},
+ {entityID: "7", dataBinary: []byte{17}, fieldsSize: 1},
+ },
+ },
+ {
+ name: "scan slice",
+ args: args{
+ start: baseTS + gap + 1,
+ end: baseTS + 5*gap - 2,
+ },
+ wantEntities: []wantEntity{
+ {entityID: "3", dataBinary: []byte{13}, fieldsSize: 1},
+ {entityID: "4", dataBinary: []byte{14}, fieldsSize: 1},
+ {entityID: "5", dataBinary: []byte{15}, fieldsSize: 1},
+ },
+ },
+ {
+ name: "single result",
+ args: args{
+ start: baseTS,
+ end: baseTS,
+ },
+ wantEntities: []wantEntity{
+ {entityID: "1", dataBinary: []byte{11}, fieldsSize: 1},
+ },
+ },
+ {
+ name: "found nothing",
+ args: args{},
+ },
+ }
+ ts, stopFunc := setup(t)
+ defer stopFunc()
+ setUpTestData(t, ts, testData(baseTS))
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ var entities ByEntityID
+ var err error
+ entities, err = ts.ScanEntity(tt.args.start, tt.args.end, series.ScanOptions{Projection: []string{"data_binary", "trace_id"}})
+ if (err != nil) != tt.wantErr {
+ t.Errorf("Write() error = %v, wantErr %v", err, tt.wantErr)
+ }
+ assert.Equal(t, len(tt.wantEntities), len(entities))
+ sort.Sort(entities)
+ for i, e := range entities {
+ assert.Equal(t, tt.wantEntities[i].entityID, string(e.EntityId()))
+ assert.Equal(t, tt.wantEntities[i].dataBinary, e.DataBinaryBytes())
+ assert.Equal(t, tt.wantEntities[i].fieldsSize, e.FieldsLength())
+ }
+ })
+ }
+}
diff --git a/banyand/series/service.go b/banyand/series/trace/schema.go
similarity index 73%
rename from banyand/series/service.go
rename to banyand/series/trace/schema.go
index 6c6a6d7..0918125 100644
--- a/banyand/series/service.go
+++ b/banyand/series/trace/schema.go
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-package series
+package trace
import (
"bytes"
@@ -25,20 +25,12 @@ import (
"go.uber.org/multierr"
"github.com/apache/skywalking-banyandb/api/common"
- "github.com/apache/skywalking-banyandb/api/data"
v1 "github.com/apache/skywalking-banyandb/api/fbs/v1"
+ "github.com/apache/skywalking-banyandb/banyand/series"
"github.com/apache/skywalking-banyandb/banyand/series/schema"
"github.com/apache/skywalking-banyandb/banyand/series/schema/sw"
- "github.com/apache/skywalking-banyandb/banyand/storage"
- "github.com/apache/skywalking-banyandb/pkg/run"
)
-var _ Service = (*service)(nil)
-
-type service struct {
- db storage.Database
-}
-
//Methods for query objects in the schema
func (s *service) TraceSeries() schema.TraceSeries {
@@ -53,7 +45,7 @@ func (s *service) IndexRuleBinding() schema.IndexRuleBinding {
return sw.NewIndexRuleBinding()
}
-func (s *service) IndexRules(ctx context.Context, subject v1.Series, filter IndexObjectFilter) ([]v1.IndexRule, error) {
+func (s *service) IndexRules(ctx context.Context, subject v1.Series, filter series.IndexObjectFilter) ([]v1.IndexRule, error) {
group := subject.Series(nil).Group()
var groupStr string
if group != nil {
@@ -68,7 +60,7 @@ func (s *service) IndexRules(ctx context.Context, subject v1.Series, filter Inde
return nil, nil
}
now := uint64(time.Now().UnixNano())
- foundRules := make([]v1.Metadata, 0)
+ foundRules := make([]*v1.Metadata, 0)
for _, binding := range bindings {
spec := binding.Spec
if spec.BeginAtNanoseconds() > now ||
@@ -85,7 +77,7 @@ func (s *service) IndexRules(ctx context.Context, subject v1.Series, filter Inde
bytes.Equal(s1.Group(), subjectSeries.Group()) {
ruleRef := spec.RuleRef(nil)
if ruleRef != nil {
- foundRules = append(foundRules, *ruleRef)
+ foundRules = append(foundRules, ruleRef)
}
}
break
@@ -119,31 +111,3 @@ func (s *service) IndexRules(ctx context.Context, subject v1.Series, filter Inde
}
return result, indexRuleErr
}
-
-func (s *service) FetchTrace(traceSeries common.Metadata, traceID string) (data.Trace, error) {
- panic("implement me")
-}
-
-func (s *service) FetchEntity(traceSeries common.Metadata, chunkIDs []common.ChunkID, opt ScanOptions) ([]data.Entity, error) {
- panic("implement me")
-}
-
-func (s *service) ScanEntity(traceSeries common.Metadata, startTime, endTime uint64, opt ScanOptions) ([]data.Entity, error) {
- panic("implement me")
-}
-
-func (s *service) Name() string {
- panic("implement me")
-}
-
-func (s *service) FlagSet() *run.FlagSet {
- panic("implement me")
-}
-
-func (s *service) Validate() error {
- panic("implement me")
-}
-
-func (s *service) PreRun() error {
- panic("implement me")
-}
diff --git a/banyand/series/service_test.go b/banyand/series/trace/schema_test.go
similarity index 95%
rename from banyand/series/service_test.go
rename to banyand/series/trace/schema_test.go
index 0acbb98..57c0d1b 100644
--- a/banyand/series/service_test.go
+++ b/banyand/series/trace/schema_test.go
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-package series
+package trace
import (
"context"
@@ -26,13 +26,14 @@ import (
"github.com/apache/skywalking-banyandb/api/common"
v1 "github.com/apache/skywalking-banyandb/api/fbs/v1"
+ "github.com/apache/skywalking-banyandb/banyand/series"
"github.com/apache/skywalking-banyandb/banyand/series/schema/sw"
)
func Test_service_RulesBySubject(t *testing.T) {
type args struct {
series v1.Series
- filter IndexObjectFilter
+ filter series.IndexObjectFilter
}
tests := []struct {
name string
@@ -90,7 +91,7 @@ func getIndexRule(name, group string) []v1.IndexRule {
b := flatbuffers.NewBuilder(0)
b.Finish(createMetadata(b, name, group))
md := v1.GetRootAsMetadata(b.FinishedBytes(), 0)
- indexRule, _ := sw.NewIndexRule().Get(context.Background(), common.Metadata{KindVersion: common.MetadataKindVersion, Spec: *md})
+ indexRule, _ := sw.NewIndexRule().Get(context.Background(), common.Metadata{KindVersion: common.MetadataKindVersion, Spec: md})
return []v1.IndexRule{indexRule.Spec}
}
diff --git a/banyand/series/trace/trace.go b/banyand/series/trace/trace.go
new file mode 100644
index 0000000..3bd6c89
--- /dev/null
+++ b/banyand/series/trace/trace.go
@@ -0,0 +1,368 @@
+// 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 trace
+
+import (
+ "bytes"
+ "context"
+ "fmt"
+ "strconv"
+
+ flatbuffers "github.com/google/flatbuffers/go"
+ "github.com/pkg/errors"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/api/data"
+ v1 "github.com/apache/skywalking-banyandb/api/fbs/v1"
+ apischema "github.com/apache/skywalking-banyandb/api/schema"
+ "github.com/apache/skywalking-banyandb/banyand/series"
+ "github.com/apache/skywalking-banyandb/banyand/series/schema"
+ "github.com/apache/skywalking-banyandb/banyand/storage"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+)
+
+const (
+ traceSeriesIDTemp = "%s:%s"
+ // KV stores
+ chunkIDMapping = "chunkIDMapping"
+ startTimeIndex = "startTimeIndex"
+ // Time series stores
+ successDataStore = "successDataStore"
+ successFieldsStore = "successFieldsStore"
+ errorDataStore = "errorDataStore"
+ errorFieldsStore = "errorFieldsStore"
+
+ traceIndex = "traceIndex"
+)
+
+var (
+ _ storage.Plugin = (*traceSeries)(nil)
+ _ series.TraceRepo = (*service)(nil)
+ ErrTraceSeriesNotFound = errors.New("failed to find Trace series")
+ ErrFieldSchemaNotFound = errors.New("failed to find trace_id specification")
+ ErrFieldNotFound = errors.New("failed to find specific field")
+ ErrProjectionEmpty = errors.New("projection is empty")
+ ErrChunkIDsEmpty = errors.New("chunkID is empty")
+ ErrInvalidTraceID = errors.New("invalid Trace id")
+ ErrUnsupportedFieldType = errors.New("unsupported field type")
+ ErrUnknownFieldValue = errors.New("unknown field value")
+ ErrInvalidKey = errors.New("invalid key")
+ ErrUnknownState = errors.New("unknown state value")
+)
+
+type State byte
+
+const (
+ StateSuccess = 0
+ StateError = 1
+)
+
+var _ series.Service = (*service)(nil)
+
+type service struct {
+ db storage.Database
+ schemaMap map[string]*traceSeries
+ l *logger.Logger
+}
+
+//NewService returns a new service
+func NewService(_ context.Context, db storage.Database) (series.Service, error) {
+ return &service{db: db}, nil
+}
+
+func (s *service) Name() string {
+ return "trace-series"
+}
+
+func (s *service) PreRun() error {
+ schemas, err := s.TraceSeries().List(context.Background(), schema.ListOpt{})
+ if err != nil {
+ return err
+ }
+ s.schemaMap = make(map[string]*traceSeries, len(schemas))
+ s.l = logger.GetLogger(s.Name())
+ for _, sa := range schemas {
+ ts, errTS := newTraceSeries(sa, s.l)
+ if errTS != nil {
+ return errTS
+ }
+ s.db.Register(ts)
+ id := fmt.Sprintf(traceSeriesIDTemp, ts.name, ts.group)
+ s.schemaMap[id] = ts
+ s.l.Info().Str("id", id).Msg("initialize Trace series")
+ }
+ return err
+}
+
+func (s *service) FetchTrace(traceSeries common.Metadata, traceID string, opt series.ScanOptions) (data.Trace, error) {
+ ts, err := s.getSeries(traceSeries)
+ if err != nil {
+ return data.Trace{}, err
+ }
+ return ts.FetchTrace(traceID, opt)
+}
+
+func (s *service) FetchEntity(traceSeries common.Metadata, chunkIDs []common.ChunkID, opt series.ScanOptions) ([]data.Entity, error) {
+ ts, err := s.getSeries(traceSeries)
+ if err != nil {
+ return nil, err
+ }
+ return ts.FetchEntity(chunkIDs, opt)
+}
+
+func (s *service) ScanEntity(traceSeries common.Metadata, startTime, endTime uint64, opt series.ScanOptions) ([]data.Entity, error) {
+ ts, err := s.getSeries(traceSeries)
+ if err != nil {
+ return nil, err
+ }
+ return ts.ScanEntity(startTime, endTime, opt)
+}
+
+func (s *service) getSeries(traceSeries common.Metadata) (*traceSeries, error) {
+ id := getTraceSeriesID(traceSeries)
+ s.l.Debug().Str("id", id).Msg("got Trace series")
+ ts, ok := s.schemaMap[id]
+ if !ok {
+ return nil, errors.Wrapf(ErrTraceSeriesNotFound, "series id:%s, map:%v", id, s.schemaMap)
+ }
+ return ts, nil
+}
+
+func getTraceSeriesID(traceSeries common.Metadata) string {
+ return fmt.Sprintf(traceSeriesIDTemp, string(traceSeries.Spec.Name()), string(traceSeries.Spec.Group()))
+}
+
+type traceSeries struct {
+ name string
+ group string
+ idGen series.IDGen
+ l *logger.Logger
+ schema apischema.TraceSeries
+ reader storage.StoreRepo
+ writePoint storage.GetWritePoint
+ shardNum uint
+ fieldIndex map[string]uint
+ traceIDIndex uint
+ traceIDFieldName string
+ stateFieldName string
+ stateFieldType v1.FieldType
+ strStateSuccessVal string
+ strStateErrorVal string
+ intStateSuccessVal int64
+ intStateErrorVal int64
+ stateIndex uint
+}
+
+func newTraceSeries(schema apischema.TraceSeries, l *logger.Logger) (*traceSeries, error) {
+ t := &traceSeries{
+ schema: schema,
+ idGen: series.NewIDGen(),
+ l: l,
+ }
+ meta := t.schema.Spec.Metadata(nil)
+ shardInfo := t.schema.Spec.Shard(nil)
+ t.shardNum = uint(shardInfo.Number())
+ t.name = string(meta.Name())
+ t.group = string(meta.Group())
+ if err := t.buildFieldIndex(); err != nil {
+ return nil, err
+ }
+ traceID, ok := t.fieldIndex[t.traceIDFieldName]
+ if !ok {
+ return nil, errors.Wrapf(ErrFieldSchemaNotFound, "trace_id field name:%s\n field index:%v",
+ t.traceIDFieldName, t.fieldIndex)
+ }
+ t.traceIDIndex = traceID
+ state, ok := t.fieldIndex[t.stateFieldName]
+ if !ok {
+ return nil, errors.Wrapf(ErrFieldSchemaNotFound, "state field name:%s\n field index:%v",
+ t.traceIDFieldName, t.fieldIndex)
+ }
+ t.stateIndex = state
+ return t, nil
+}
+
+func (t *traceSeries) Meta() storage.PluginMeta {
+ return storage.PluginMeta{
+ ID: t.name,
+ Group: t.group,
+ ShardNumber: t.shardNum,
+ KVSpecs: []storage.KVSpec{
+ {
+ Name: chunkIDMapping,
+ Type: storage.KVTypeNormal,
+ },
+ {
+ Name: startTimeIndex,
+ Type: storage.KVTypeNormal,
+ },
+
+ {
+ Name: successDataStore,
+ Type: storage.KVTypeTimeSeries,
+ CompressLevel: 3,
+ },
+ {
+ Name: successFieldsStore,
+ Type: storage.KVTypeTimeSeries,
+ CompressLevel: 3,
+ },
+ {
+ Name: errorDataStore,
+ Type: storage.KVTypeTimeSeries,
+ CompressLevel: 3,
+ },
+ {
+ Name: errorFieldsStore,
+ Type: storage.KVTypeTimeSeries,
+ CompressLevel: 3,
+ },
+ {
+ Name: traceIndex,
+ Type: storage.KVTypeTimeSeries,
+ CompressLevel: -1,
+ },
+ },
+ }
+}
+
+func (t *traceSeries) Init(repo storage.StoreRepo, point storage.GetWritePoint) {
+ t.reader = repo
+ t.writePoint = point
+}
+
+func (t *traceSeries) buildFieldIndex() error {
+ spec := t.schema.Spec
+ reservedMap := spec.ReservedFieldsMap(nil)
+ t.traceIDFieldName = string(reservedMap.TraceId())
+ state := reservedMap.State(nil)
+ stateFieldName := state.Field()
+
+ fieldsLen := spec.FieldsLength()
+ index := make(map[string]uint, fieldsLen)
+ for i := 0; i < fieldsLen; i++ {
+ fieldSpec := new(v1.FieldSpec)
+ if !spec.Fields(fieldSpec, i) {
+ continue
+ }
+ if bytes.Equal(fieldSpec.Name(), stateFieldName) {
+ t.stateFieldType = fieldSpec.Type()
+ }
+ index[string(fieldSpec.Name())] = uint(i)
+ }
+ switch t.stateFieldType {
+ case v1.FieldTypeString:
+ t.strStateSuccessVal = string(state.ValSuccess())
+ t.strStateErrorVal = string(state.ValError())
+ case v1.FieldTypeInt:
+ intSVal, err := strconv.ParseInt(string(state.ValSuccess()), 10, 64)
+ if err != nil {
+ return err
+ }
+ t.intStateSuccessVal = intSVal
+ intEVal, err := strconv.ParseInt(string(state.ValError()), 10, 64)
+ if err != nil {
+ return err
+ }
+ t.intStateErrorVal = intEVal
+ default:
+ return errors.Wrapf(ErrUnsupportedFieldType, "type:%s, supported type: Int and String", t.stateFieldType.String())
+ }
+ t.stateFieldName = string(stateFieldName)
+
+ t.fieldIndex = index
+ return nil
+}
+
+func (t *traceSeries) getTraceID(entityValue *v1.EntityValue) ([]byte, error) {
+ f := new(v1.Field)
+ if !entityValue.Fields(f, int(t.traceIDIndex)) {
+ return nil, errors.Wrapf(ErrFieldNotFound, "trace_id index :%d", t.traceIDIndex)
+ }
+ if f.ValueType() != v1.ValueTypeString {
+ return nil, errors.Wrapf(ErrUnsupportedFieldType, "type:%s, supported type: String", f.ValueType().String())
+ }
+ unionTable := new(flatbuffers.Table)
+ f.Value(unionTable)
+ stringValue := new(v1.String)
+ stringValue.Init(unionTable.Bytes, unionTable.Pos)
+ return stringValue.Value(), nil
+}
+
+func (t *traceSeries) getState(entityValue *v1.EntityValue) (state State, fieldStoreName, dataStoreName string, err error) {
+ f := new(v1.Field)
+ if !entityValue.Fields(f, int(t.stateIndex)) {
+ err = errors.Wrapf(ErrFieldNotFound, "state index :%d", t.traceIDIndex)
+ return
+ }
+ switch f.ValueType() {
+ case v1.ValueTypeInt:
+ if t.stateFieldType != v1.FieldTypeInt {
+ err = errors.Wrapf(ErrUnsupportedFieldType, "type:%s, supported type: Int", f.ValueType().String())
+ return
+ }
+ unionTable := new(flatbuffers.Table)
+ f.Value(unionTable)
+ intValue := new(v1.Int)
+ intValue.Init(unionTable.Bytes, unionTable.Pos)
+ switch intValue.Value() {
+ case t.intStateSuccessVal:
+ state = StateSuccess
+ case t.intStateErrorVal:
+ state = StateError
+ default:
+ err = errors.Wrapf(ErrUnknownFieldValue, "value:%d, supported value: %d, %d",
+ intValue.Value(), t.intStateSuccessVal, t.intStateErrorVal)
+ return
+ }
+ case v1.ValueTypeString:
+ if t.stateFieldType != v1.FieldTypeString {
+ err = errors.Wrapf(ErrUnsupportedFieldType, "type:%s, supported type: String", f.ValueType().String())
+ return
+ }
+ unionTable := new(flatbuffers.Table)
+ f.Value(unionTable)
+ stringValue := new(v1.String)
+ stringValue.Init(unionTable.Bytes, unionTable.Pos)
+ switch string(stringValue.Value()) {
+ case t.strStateSuccessVal:
+ state = StateSuccess
+ case t.strStateErrorVal:
+ state = StateError
+ default:
+ err = errors.Wrapf(ErrUnknownFieldValue, "value:%s, supported value: %s, %s",
+ string(stringValue.Value()), t.strStateSuccessVal, t.strStateErrorVal)
+ return
+ }
+ default:
+ err = errors.Wrapf(ErrUnsupportedFieldType, "type:%s, supported type: String and Int", f.ValueType().String())
+ return
+ }
+ fieldStoreName, dataStoreName, err = getStoreName(state)
+ return
+}
+
+func getStoreName(state State) (string, string, error) {
+ switch state {
+ case StateSuccess:
+ return successFieldsStore, successDataStore, nil
+ case StateError:
+ return errorFieldsStore, errorDataStore, nil
+ }
+ return "", "", ErrUnknownState
+}
diff --git a/banyand/series/trace/write.go b/banyand/series/trace/write.go
new file mode 100644
index 0000000..e7e7b22
--- /dev/null
+++ b/banyand/series/trace/write.go
@@ -0,0 +1,90 @@
+// 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 trace
+
+import (
+ "time"
+
+ flatbuffers "github.com/google/flatbuffers/go"
+ "github.com/pkg/errors"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/api/data"
+ bydb_bytes "github.com/apache/skywalking-banyandb/pkg/bytes"
+ "github.com/apache/skywalking-banyandb/pkg/convert"
+ "github.com/apache/skywalking-banyandb/pkg/fb"
+ "github.com/apache/skywalking-banyandb/pkg/partition"
+)
+
+//TODO: To add WAL to handle crashing
+//TODO: To replace timestamp with monotonic ID in case of server time reset
+func (t *traceSeries) Write(seriesID common.SeriesID, shardID uint, entity data.EntityValue) (common.ChunkID, error) {
+ entityVal := entity.EntityValue
+ traceID, err := t.getTraceID(entityVal)
+ if err != nil {
+ return nil, err
+ }
+ state, fieldsStoreName, dataStoreName, errGetState := t.getState(entityVal)
+ if errGetState != nil {
+ return nil, errGetState
+ }
+ stateBytes := []byte{byte(state)}
+ chunkID := t.idGen.Next(shardID, entity.TimestampNanoseconds())
+ ts, errParseTS := t.idGen.ParseTS(chunkID)
+ if errParseTS != nil {
+ return nil, errors.Wrap(errParseTS, "failed to parse timestamp from chunk id")
+ }
+ tsBytes := convert.Uint64ToBytes(ts)
+ intSeriesID := uint64(seriesID)
+ seriesIDBytes := convert.Uint64ToBytes(intSeriesID)
+ wp := t.writePoint(ts)
+
+ err = wp.TimeSeriesWriter(shardID, dataStoreName).Put(seriesIDBytes, entityVal.DataBinaryBytes(), ts)
+ if err != nil {
+ return nil, errors.Wrap(err, "failed to write traceSeries data")
+ }
+ builder := flatbuffers.NewBuilder(0)
+ builder.Finish(fb.CopyFields(entityVal, builder))
+ err = wp.TimeSeriesWriter(shardID, fieldsStoreName).Put(seriesIDBytes, builder.FinishedBytes(), ts)
+ if err != nil {
+ return nil, errors.Wrap(err, "failed to write traceSeries fields")
+ }
+
+ if err = wp.Writer(shardID, chunkIDMapping).Put(chunkID, bydb_bytes.Join(stateBytes, seriesIDBytes, tsBytes)); err != nil {
+ return nil, errors.Wrap(err, "failed to write chunkID index")
+ }
+ traceIDShardID := partition.ShardID(traceID, t.shardNum)
+ if err = wp.TimeSeriesWriter(traceIDShardID, traceIndex).Put(traceID, chunkID, entity.TimestampNanoseconds()); err != nil {
+ return nil, errors.Wrap(err, "failed to Trace index")
+ }
+ err = wp.Writer(shardID, startTimeIndex).Put(bydb_bytes.Join(stateBytes, tsBytes, chunkID), nil)
+ if err != nil {
+ return nil, errors.Wrap(err, "failed to write start time index")
+ }
+ t.l.Debug().Hex("chunk_id", chunkID).
+ Uint64("series_id", intSeriesID).
+ Time("ts", time.Unix(0, int64(ts))).
+ Uint64("ts_int", ts).
+ Int("data_size", entityVal.DataBinaryLength()).
+ Int("fields_num", entityVal.FieldsLength()).
+ Hex("trace_id", traceID).
+ Uint("trace_shard_id", traceIDShardID).
+ Uint("shard_id", shardID).
+ Msg("written to Trace series")
+ return chunkID, err
+}
diff --git a/banyand/series/trace/write_test.go b/banyand/series/trace/write_test.go
new file mode 100644
index 0000000..feb3e2f
--- /dev/null
+++ b/banyand/series/trace/write_test.go
@@ -0,0 +1,207 @@
+// 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 trace
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/api/data"
+ v1 "github.com/apache/skywalking-banyandb/api/fbs/v1"
+ "github.com/apache/skywalking-banyandb/pkg/convert"
+ "github.com/apache/skywalking-banyandb/pkg/fb"
+ "github.com/apache/skywalking-banyandb/pkg/partition"
+)
+
+func Test_traceSeries_Write(t *testing.T) {
+ ts, stopFunc := setup(t)
+ defer stopFunc()
+
+ tests := []struct {
+ name string
+ args seriesEntity
+ wantErr bool
+ }{
+ {
+ name: "golden path",
+ args: seriesEntity{
+ seriesID: "webapp_10.0.0.1",
+ entity: getEntity("1231.dfd.123123ssf", []byte{12},
+ "trace_id-xxfff.111323",
+ 0,
+ "webapp_id",
+ "10.0.0.1_id",
+ "/home_id",
+ "webapp",
+ "10.0.0.1",
+ "/home",
+ 300,
+ 1622933202000000000,
+ ),
+ },
+ },
+ {
+ name: "minimal",
+ args: seriesEntity{
+ seriesID: "webapp_10.0.0.1",
+ entity: getEntity("1231.dfd.123123ssf", []byte{12},
+ "trace_id-xxfff.111323",
+ 1,
+ ),
+ },
+ },
+ {
+ name: "http",
+ args: seriesEntity{
+ seriesID: "webapp_10.0.0.1",
+ entity: getEntity("1231.dfd.123123ssf", []byte{12},
+ "trace_id-xxfff.111323",
+ 0,
+ "webapp_id",
+ "10.0.0.1_id",
+ "/home_id",
+ "webapp",
+ "10.0.0.1",
+ "/home",
+ 300,
+ 1622933202000000000,
+ "GET",
+ "200",
+ ),
+ },
+ },
+ {
+ name: "database",
+ args: seriesEntity{
+ seriesID: "webapp_10.0.0.1",
+ entity: getEntity("1231.dfd.123123ssf", []byte{12},
+ "trace_id-xxfff.111323",
+ 0,
+ "webapp_id",
+ "10.0.0.1_id",
+ "/home_id",
+ "webapp",
+ "10.0.0.1",
+ "/home",
+ 300,
+ 1622933202000000000,
+ nil,
+ nil,
+ "MySQL",
+ "10.1.1.2",
+ ),
+ },
+ },
+ {
+ name: "mq",
+ args: seriesEntity{
+ seriesID: "webapp_10.0.0.1",
+ entity: getEntity("1231.dfd.123123ssf", []byte{12},
+ "trace_id-xxfff.111323",
+ 1,
+ "webapp_id",
+ "10.0.0.1_id",
+ "/home_id",
+ "webapp",
+ "10.0.0.1",
+ "/home",
+ 300,
+ 1622933202000000000,
+ nil,
+ nil,
+ nil,
+ nil,
+ "test_topic",
+ "10.0.0.1",
+ ),
+ },
+ },
+ {
+ name: "absent Trace id",
+ args: seriesEntity{
+ seriesID: "webapp_10.0.0.1",
+ entity: getEntity("1231.dfd.123123ssf", []byte{12},
+ nil,
+ 0,
+ ),
+ },
+ wantErr: true,
+ },
+ {
+ name: "invalid Trace id",
+ args: seriesEntity{
+ seriesID: "webapp_10.0.0.1",
+ entity: getEntity("1231.dfd.123123ssf", []byte{12},
+ 1212323,
+ 1,
+ ),
+ },
+ wantErr: true,
+ },
+ {
+ name: "absent State",
+ args: seriesEntity{
+ seriesID: "webapp_10.0.0.1",
+ entity: getEntity("1231.dfd.123123ssf", []byte{12},
+ "trace_id-xxfff.111323",
+ nil,
+ ),
+ },
+ wantErr: true,
+ },
+ {
+ name: "invalid State",
+ args: seriesEntity{
+ seriesID: "webapp_10.0.0.1",
+ entity: getEntity("1231.dfd.123123ssf", []byte{12},
+ "trace_id-xxfff.111323",
+ 6,
+ ),
+ },
+ wantErr: true,
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+
+ seriesID := []byte(tt.args.seriesID)
+ b := fb.NewWriteEntityBuilder()
+ items := make([]fb.ComponentBuilderFunc, 0, 2)
+ items = append(items, b.BuildMetaData("default", "sw"))
+ items = append(items, b.BuildEntity(tt.args.entity.id, tt.args.entity.binary, tt.args.entity.items...))
+ builder, err := b.BuildWriteEntity(
+ items...,
+ )
+ assert.NoError(t, err)
+ we := v1.GetRootAsWriteEntity(builder.FinishedBytes(), 0)
+ got, err := ts.Write(common.SeriesID(convert.Hash(seriesID)), partition.ShardID(seriesID, 2), data.EntityValue{
+ EntityValue: we.Entity(nil),
+ })
+ if (err != nil) != tt.wantErr {
+ t.Errorf("Write() error = %v, wantErr %v", err, tt.wantErr)
+ return
+ }
+ if err == nil && len([]byte(got)) < 1 {
+ t.Error("Write() got empty chunkID")
+ }
+ })
+ }
+}
diff --git a/banyand/storage/block.go b/banyand/storage/block.go
index 028e6d6..8bbf50f 100644
--- a/banyand/storage/block.go
+++ b/banyand/storage/block.go
@@ -20,29 +20,28 @@ package storage
import (
"fmt"
- "go.uber.org/atomic"
-
"github.com/apache/skywalking-banyandb/banyand/kv"
"github.com/apache/skywalking-banyandb/pkg/logger"
)
type block struct {
- path string
- plugins []Plugin
+ path string
+ plugin Plugin
l *logger.Logger
stores map[string]kv.Store
tsStores map[string]kv.TimeSeriesStore
- refCount atomic.Uint32
+ shardID int
}
-func newBlock(path string, plugins []Plugin) (*block, error) {
+func newBlock(shardID int, path string, plugin Plugin) (*block, error) {
l := logger.GetLogger("block")
return &block{
+ shardID: shardID,
path: path,
- plugins: plugins,
+ plugin: plugin,
l: l,
stores: make(map[string]kv.Store),
tsStores: make(map[string]kv.TimeSeriesStore),
@@ -50,28 +49,20 @@ func newBlock(path string, plugins []Plugin) (*block, error) {
}
func (b *block) init() (err error) {
- for _, plugin := range b.plugins {
- kvDefines := plugin.Init()
- if kvDefines == nil || len(kvDefines) < 1 {
- continue
- }
- if err = b.createKV(plugin.ID(), kvDefines); err != nil {
- return err
- }
- plugin.Start(func() AccessPoint {
- b.refCount.Inc()
- return &staticAccessPoint{
- b: b,
- pluginID: plugin.ID(),
- }
- })
+ meta := b.plugin.Meta()
+ kvDefines := meta.KVSpecs
+ if kvDefines == nil || len(kvDefines) < 1 {
+ return nil
+ }
+ if err = b.createKV(kvDefines); err != nil {
+ return err
}
return nil
}
-func (b *block) createKV(pluginID string, defines []KVSpec) (err error) {
+func (b *block) createKV(defines []KVSpec) (err error) {
for _, define := range defines {
- storeID := getStoreID(pluginID, define.Name)
+ storeID := define.Name
path := fmt.Sprintf("%s/%s", b.path, storeID)
b.l.Info().Str("path", path).Uint8("type", uint8(define.Type)).Msg("open kv store")
switch define.Type {
@@ -79,16 +70,13 @@ func (b *block) createKV(pluginID string, defines []KVSpec) (err error) {
var s kv.Store
opts := make([]kv.StoreOptions, 0)
opts = append(opts, kv.StoreWithLogger(b.l))
- if define.AutoGenKey {
- opts = append(opts, kv.StoreWithAutoGeneratedKey())
- }
- if s, err = kv.OpenStore(path, opts...); err != nil {
+ if s, err = kv.OpenStore(b.shardID, path, opts...); err != nil {
return fmt.Errorf("failed to open normal store: %w", err)
}
b.stores[storeID] = s
case KVTypeTimeSeries:
var s kv.TimeSeriesStore
- if s, err = kv.OpenTimeSeriesStore(path, define.TimeSeriesHook, kv.TSSWithLogger(b.l)); err != nil {
+ if s, err = kv.OpenTimeSeriesStore(b.shardID, path, define.CompressLevel, define.ValueSize, kv.TSSWithLogger(b.l)); err != nil {
return fmt.Errorf("failed to open time series store: %w", err)
}
b.tsStores[storeID] = s
@@ -105,33 +93,3 @@ func (b *block) close() {
_ = store.Close()
}
}
-
-func getStoreID(pluginID string, defineName string) string {
- return fmt.Sprintf("%s-%s", pluginID, defineName)
-}
-
-var _ AccessPoint = (*staticAccessPoint)(nil)
-
-type staticAccessPoint struct {
- pluginID string
- b *block
-}
-
-func (s *staticAccessPoint) Close() error {
- s.b.refCount.Dec()
- return nil
-}
-
-func (s *staticAccessPoint) Store(name string) kv.Store {
- if s, ok := s.b.stores[getStoreID(s.pluginID, name)]; ok {
- return s
- }
- return nil
-}
-
-func (s *staticAccessPoint) TimeSeriesStore(name string) kv.TimeSeriesStore {
- if s, ok := s.b.tsStores[getStoreID(s.pluginID, name)]; ok {
- return s
- }
- return nil
-}
diff --git a/banyand/storage/database.go b/banyand/storage/database.go
index 8bf76bf..f69af0f 100644
--- a/banyand/storage/database.go
+++ b/banyand/storage/database.go
@@ -26,9 +26,12 @@ import (
"sync"
"time"
+ "go.uber.org/atomic"
"go.uber.org/multierr"
"github.com/apache/skywalking-banyandb/banyand/discovery"
+ "github.com/apache/skywalking-banyandb/banyand/kv"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/run"
)
@@ -51,12 +54,12 @@ var _ Database = (*DB)(nil)
// Notice: The current status of DB is under WIP. It only contains feats support to verify the series module.
type DB struct {
root string
- shards int
- sLst []*shard
+ seriesLst []*series
repo discovery.ServiceRepo
pluginLst []Plugin
stopCh chan struct{}
+ log *logger.Logger
}
func (d *DB) Serve() error {
@@ -66,10 +69,12 @@ func (d *DB) Serve() error {
}
func (d *DB) GracefulStop() {
- for _, s := range d.sLst {
+ for _, s := range d.seriesLst {
s.stop()
}
- close(d.stopCh)
+ if d.stopCh != nil {
+ close(d.stopCh)
+ }
}
func (d *DB) Register(plugin Plugin) {
@@ -83,7 +88,6 @@ func (d *DB) Name() string {
func (d *DB) FlagSet() *run.FlagSet {
flagS := run.NewFlagSet("storage")
flagS.StringVar(&d.root, "root-path", "/tmp", "the root path of database")
- flagS.IntVar(&d.shards, "shards", 1, "total shards size")
return flagS
}
@@ -92,52 +96,140 @@ func (d *DB) Validate() error {
}
func (d *DB) PreRun() error {
+ d.log = logger.GetLogger("database")
if err := d.init(); err != nil {
return fmt.Errorf("failed to initialize db: %v", err)
}
return nil
}
-func (d *DB) init() (err error) {
- if _, err = mkdir(d.root); err != nil {
- return fmt.Errorf("failed to create %s: %v", d.root, err)
- }
- var entris []fs.FileInfo
- if entris, err = ioutil.ReadDir(d.root); err != nil {
- return fmt.Errorf("failed to read directory contents failed: %v", err)
+type series struct {
+ sLst []*shard
+ init bool
+ location string
+}
+
+func (s *series) Reader(shard uint, name string, start, end uint64) kv.Reader {
+ //TODO: find targets in all blocks
+ b, ok := s.sLst[shard].activeBlock.Load().(*block)
+ if !ok {
+ return nil
}
- if len(entris) < 1 {
- return d.createShards()
+ return b.stores[name]
+}
+
+func (s *series) TimeSeriesReader(shard uint, name string, start, end uint64) kv.TimeSeriesReader {
+ //TODO: find targets in all blocks
+ b, ok := s.sLst[shard].activeBlock.Load().(*block)
+ if !ok {
+ return nil
}
- return d.loadShards(entris)
+ return b.tsStores[name]
}
-func (d *DB) loadShards(_ []fs.FileInfo) (err error) {
- //TODO load existing shards
- return nil
+func (s *series) load(meta PluginMeta) error {
+ //TODO: to implement load instead of removing old contents
+ return os.RemoveAll(s.location)
}
-func (d *DB) createShards() (err error) {
- for i := 0; i < d.shards; i++ {
+func (s *series) create(meta PluginMeta, plugin Plugin) (err error) {
+ if _, err = mkdir(s.location); err != nil {
+ return err
+ }
+
+ for i := 0; i < int(meta.ShardNumber); i++ {
var shardLocation string
- if shardLocation, err = mkdir(shardTemplate, d.root, i); err != nil {
+ if shardLocation, err = mkdir(shardTemplate, s.location, i); err != nil {
return err
}
- s := newShard(i, shardLocation)
- if sErr := s.init(d.pluginLst); sErr != nil {
+ so := newShard(i, shardLocation)
+ if sErr := so.init(plugin); sErr != nil {
err = multierr.Append(err, sErr)
continue
}
- d.sLst = append(d.sLst, s)
+ s.sLst = append(s.sLst, so)
}
return err
+
+}
+
+func (s *series) stop() {
+ for _, sa := range s.sLst {
+ sa.stop()
+ }
+}
+
+func (d *DB) init() (err error) {
+ if _, err = mkdir(d.root); err != nil {
+ return fmt.Errorf("failed to create %s: %v", d.root, err)
+ }
+ d.log.Info().Str("path", d.root).Msg("initialize database")
+ var entries []fs.FileInfo
+ if entries, err = ioutil.ReadDir(d.root); err != nil {
+ return fmt.Errorf("failed to read directory contents failed: %v", err)
+ }
+ for _, plugin := range d.pluginLst {
+ meta := plugin.Meta()
+
+ s := &series{location: d.root + "/" + meta.ID}
+ d.seriesLst = append(d.seriesLst, s)
+ for _, entry := range entries {
+ if entry.Name() == meta.ID && entry.IsDir() {
+ err = s.load(meta)
+ if err != nil {
+ return err
+ }
+ d.log.Info().Str("ID", meta.ID).Msg("loaded series")
+ break
+ }
+ }
+ if !s.init {
+ err = s.create(meta, plugin)
+ if err != nil {
+ return err
+ }
+ d.log.Info().Str("ID", meta.ID).Msg("created series")
+ }
+ plugin.Init(s, func(_ uint64) WritePoint {
+ bLst := make([]*block, len(s.sLst))
+ for i, sa := range s.sLst {
+ b, ok := sa.activeBlock.Load().(*block)
+ if !ok {
+ continue
+ }
+ bLst[i] = b
+ }
+ return &writePoint{bLst: bLst}
+ })
+ d.log.Info().Str("ID", meta.ID).Msg("initialized plugin")
+ }
+ return err
+}
+
+var _ WritePoint = (*writePoint)(nil)
+
+type writePoint struct {
+ bLst []*block
+}
+
+func (w *writePoint) Close() error {
+ return nil
+}
+
+func (w *writePoint) Writer(shard uint, name string) kv.Writer {
+ return w.bLst[shard].stores[name]
+}
+
+func (w *writePoint) TimeSeriesWriter(shard uint, name string) kv.TimeSeriesWriter {
+ return w.bLst[shard].tsStores[name]
}
type shard struct {
id int
lst []*segment
sync.Mutex
- location string
+ location string
+ activeBlock atomic.Value
}
func newShard(id int, location string) *shard {
@@ -147,23 +239,29 @@ func newShard(id int, location string) *shard {
}
}
-func (s *shard) newSeg(path string) *segment {
+func (s *shard) newSeg(shardID int, path string) *segment {
s.Lock()
defer s.Unlock()
seg := &segment{
- path: path,
+ path: path,
+ shardID: shardID,
}
s.lst = append(s.lst, seg)
return seg
}
-func (s *shard) init(plugins []Plugin) error {
+func (s *shard) init(plugin Plugin) error {
segPath, err := mkdir(segTemplate, s.location, time.Now().Format(segFormat))
if err != nil {
return fmt.Errorf("failed to make segment directory: %v", err)
}
- seg := s.newSeg(segPath)
- return seg.init(plugins)
+
+ seg := s.newSeg(s.id, segPath)
+ return seg.init(plugin, s.updateActiveBlock)
+}
+
+func (s *shard) updateActiveBlock(newBlock *block) {
+ s.activeBlock.Store(newBlock)
}
func (s *shard) stop() {
@@ -183,7 +281,8 @@ func mkdir(format string, a ...interface{}) (path string, err error) {
type segment struct {
lst []*block
sync.Mutex
- path string
+ path string
+ shardID int
}
func (s *segment) addBlock(b *block) {
@@ -192,21 +291,22 @@ func (s *segment) addBlock(b *block) {
s.lst = append(s.lst, b)
}
-func (s *segment) init(plugins []Plugin) error {
+func (s *segment) init(plugin Plugin, activeBlock func(newBlock *block)) error {
blockPath, err := mkdir(blockTemplate, s.path, time.Now().Format(blockFormat))
if err != nil {
return fmt.Errorf("failed to make block directory: %v", err)
}
var b *block
- if b, err = newBlock(blockPath, plugins); err != nil {
+ if b, err = newBlock(s.shardID, blockPath, plugin); err != nil {
return fmt.Errorf("failed to create segment: %v", err)
}
+ activeBlock(b)
s.addBlock(b)
return b.init()
}
func (s *segment) close() {
- for _, block := range s.lst {
- block.close()
+ for _, b := range s.lst {
+ b.close()
}
}
diff --git a/banyand/storage/database_test.go b/banyand/storage/database_test.go
index d772173..754c25b 100644
--- a/banyand/storage/database_test.go
+++ b/banyand/storage/database_test.go
@@ -18,7 +18,6 @@
package storage
import (
- "bytes"
"context"
"fmt"
"io/ioutil"
@@ -26,20 +25,32 @@ import (
"testing"
"time"
- "github.com/dgraph-io/badger/v3/y"
"github.com/golang/mock/gomock"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
- "github.com/apache/skywalking-banyandb/banyand/kv"
- "github.com/apache/skywalking-banyandb/pkg/convert"
"github.com/apache/skywalking-banyandb/pkg/logger"
)
func TestDB_Create_Directory(t *testing.T) {
- tempDir, _ := setUp(t, nil)
+ ctrl := gomock.NewController(t)
+ p := NewMockPlugin(ctrl)
+ p.EXPECT().Meta().Return(PluginMeta{
+ ID: "sw",
+ Group: "default",
+ ShardNumber: 1,
+ KVSpecs: []KVSpec{
+ {
+ Name: "test",
+ Type: KVTypeTimeSeries,
+ CompressLevel: 3,
+ },
+ },
+ }).AnyTimes()
+ p.EXPECT().Init(gomock.Any(), gomock.Any()).AnyTimes()
+ tempDir, _ := setUp(t, p)
defer removeDir(tempDir)
- shardPath := fmt.Sprintf(shardTemplate, tempDir, 0)
+ shardPath := fmt.Sprintf(shardTemplate, tempDir+"/sw", 0)
validateDirectory(t, shardPath)
now := time.Now()
segPath := fmt.Sprintf(segTemplate, shardPath, now.Format(segFormat))
@@ -50,9 +61,12 @@ func TestDB_Create_Directory(t *testing.T) {
func TestDB_Store(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
- var ap AccessPoint
- p := mockPlugin(ctrl, func(get GetAccessPoint) {
- ap = get()
+ now := uint64(time.Now().UnixNano())
+ var ap WritePoint
+ var repo StoreRepo
+ p := mockPlugin(ctrl, func(r StoreRepo, get GetWritePoint) {
+ ap = get(now)
+ repo = r
})
tempDir, db := setUp(t, p)
@@ -61,90 +75,44 @@ func TestDB_Store(t *testing.T) {
removeDir(tempDir)
}()
- s := ap.Store("normal")
- assert.NoError(t, s.Put([]byte("key1"), []byte{12}))
- val, err := s.Get([]byte("key1"))
+ assert.NoError(t, ap.Writer(0, "normal").Put([]byte("key1"), []byte{12}))
+ val, err := repo.Reader(0, "normal", now, now).Get([]byte("key1"))
assert.NoError(t, err)
assert.Equal(t, []byte{12}, val)
- s = ap.Store("auto-gen")
- key, addErr := s.Add([]byte{11})
- assert.NoError(t, addErr)
- val, err = s.Get(convert.Uint64ToBytes(key))
- assert.NoError(t, err)
- assert.Equal(t, []byte{11}, val)
-
- tss := ap.TimeSeriesStore("time-series")
- assert.NoError(t, tss.Put([]byte("key11"), []byte{33}, 1))
- val, err = tss.Get([]byte("key11"), 1)
+ assert.NoError(t, ap.TimeSeriesWriter(1, "time-series").Put([]byte("key11"), []byte{33}, 1))
+ val, err = repo.TimeSeriesReader(1, "time-series", now, now).Get([]byte("key11"), 1)
assert.NoError(t, err)
assert.Equal(t, []byte{33}, val)
- vals, allErr := tss.GetAll([]byte("key11"))
+ vals, allErr := repo.TimeSeriesReader(1, "time-series", now, now).GetAll([]byte("key11"))
assert.NoError(t, allErr)
assert.Equal(t, [][]byte{{33}}, vals)
}
-func mockPlugin(ctrl *gomock.Controller, f func(get GetAccessPoint)) Plugin {
+func mockPlugin(ctrl *gomock.Controller, f func(repo StoreRepo, get GetWritePoint)) Plugin {
p := NewMockPlugin(ctrl)
- p.
- EXPECT().
- ID().
- Return("foo").
- AnyTimes()
- p.
- EXPECT().
- Init().
- Return([]KVSpec{
+ p.EXPECT().Meta().Return(PluginMeta{
+ ID: "sw",
+ Group: "default",
+ ShardNumber: 2,
+ KVSpecs: []KVSpec{
{
Name: "normal",
Type: KVTypeNormal,
},
{
- Name: "auto-gen",
- Type: KVTypeNormal,
- AutoGenKey: true,
- },
- {
- Name: "time-series",
- Type: KVTypeTimeSeries,
- TimeSeriesHook: mockHook(ctrl),
+ Name: "time-series",
+ Type: KVTypeTimeSeries,
+ CompressLevel: 3,
},
- }).
- AnyTimes()
- p.
- EXPECT().
- Start(gomock.Any()).
- Do(f).
- AnyTimes()
+ },
+ }).AnyTimes()
+ p.EXPECT().Init(gomock.Any(), gomock.Any()).Do(func(r StoreRepo, wp GetWritePoint) {
+ f(r, wp)
+ }).AnyTimes()
return p
}
-func mockHook(ctrl *gomock.Controller) kv.Hook {
- h := kv.NewMockHook(ctrl)
- h.
- EXPECT().
- Reduce(gomock.Any(), gomock.Any()).
- DoAndReturn(func(left bytes.Buffer, right y.ValueStruct) bytes.Buffer {
- return left
- }).
- AnyTimes()
- h.
- EXPECT().
- Extract(gomock.Any(), gomock.Any()).
- DoAndReturn(func(raw []byte, ts uint64) ([]byte, error) {
- return raw, nil
- }).
- AnyTimes()
- h.
- EXPECT().
- Split(gomock.Any()).
- DoAndReturn(func(raw []byte) ([][]byte, error) {
- return [][]byte{raw}, nil
- }).
- AnyTimes()
- return h
-}
-
func setUp(t *testing.T, p Plugin) (tempDir string, db Database) {
require.NoError(t, logger.Init(logger.Logging{
Env: "dev",
@@ -158,7 +126,7 @@ func setUp(t *testing.T, p Plugin) (tempDir string, db Database) {
tempDir, tempDirErr = ioutil.TempDir("", "banyandb-test-*")
require.Nil(t, tempDirErr)
- require.NoError(t, db.FlagSet().Parse([]string{"--root-path", tempDir, "--shards", "1"}))
+ require.NoError(t, db.FlagSet().Parse([]string{"--root-path", tempDir}))
if p != nil {
db.Register(p)
}
diff --git a/banyand/storage/storage.go b/banyand/storage/storage.go
index 15bffc0..a40c1af 100644
--- a/banyand/storage/storage.go
+++ b/banyand/storage/storage.go
@@ -42,6 +42,10 @@ type Database interface {
run.Config
run.PreRunner
run.Service
+ PluginRepo
+}
+
+type PluginRepo interface {
// Register a Plugin into a Database
Register(plugin Plugin)
}
@@ -55,32 +59,47 @@ func NewDB(_ context.Context, repo discovery.ServiceRepo) (Database, error) {
// Trace, metric and log series could be registered as Plugin to save data into kv regardless of where the kv
// are placed on the physical devices.
type Plugin interface {
- // ID is unique identifier
- ID() string
+ Meta() PluginMeta
// Init the Plugin, and generate KVSpec to indicate how kv storages are built
- Init() []KVSpec
- // Start the Plugin and inject the a func to get the AccessPoint
- Start(GetAccessPoint)
+ Init(StoreRepo, GetWritePoint)
+}
+
+type PluginMeta struct {
+ ID string
+ Group string
+ ShardNumber uint
+ KVSpecs []KVSpec
+}
+
+type CompressSpec struct {
+ //Level closest matches the compression ratio of a specific zstd compression level.
+ Level int
+ Size int
}
// KVSpec defines the behaviours of a KV store.
// Database get these specs which are generated by Plugin to create underlying kv storage.
type KVSpec struct {
- Name string
- Type KVType
- AutoGenKey bool
- TimeSeriesHook kv.Hook
+ Name string
+ Type KVType
+ CompressLevel int
+ ValueSize int
+}
+
+type StoreRepo interface {
+ Reader(shard uint, name string, start, end uint64) kv.Reader
+ TimeSeriesReader(shard uint, name string, start, end uint64) kv.TimeSeriesReader
}
-// AccessPoint is a reference to a underlying area.
-// A single write will be split into several kv writes. These writes can be placed into the same area though an AccessPoint
+// WritePoint is a reference to a underlying area.
+// A single write will be split into several kv writes which can be placed into the same area though an WritePoint
// even though the area rotation(a new area replaces the old area) takes place.
-// When a new area is created, the old one can be closed until all AccessPoint refers to it are closed.
-type AccessPoint interface {
+// When a new area is created, the old one can be closed until all WritePoint refers to it are closed.
+type WritePoint interface {
io.Closer
- Store(name string) kv.Store
- TimeSeriesStore(name string) kv.TimeSeriesStore
+ Writer(shard uint, name string) kv.Writer
+ TimeSeriesWriter(shard uint, name string) kv.TimeSeriesWriter
}
-// GetAccessPoint returns the AccessPoint which refers to the current active area.
-type GetAccessPoint func() AccessPoint
+// GetWritePoint returns the WritePoint which refers to the current active area.
+type GetWritePoint func(ts uint64) WritePoint
diff --git a/go.mod b/go.mod
index 8dc8ac8..2f971df 100644
--- a/go.mod
+++ b/go.mod
@@ -3,6 +3,7 @@ module github.com/apache/skywalking-banyandb
go 1.16
require (
+ github.com/cespare/xxhash v1.1.0
github.com/dgraph-io/badger/v3 v3.2011.1
github.com/golang/mock v1.3.1
github.com/google/flatbuffers v2.0.0+incompatible
@@ -19,4 +20,4 @@ require (
google.golang.org/grpc v1.37.0
)
-replace github.com/dgraph-io/badger/v3 v3.2011.1 => github.com/SkyAPM/badger/v3 v3.0.0-20210527215642-f1c960d2de88
+replace github.com/dgraph-io/badger/v3 v3.2011.1 => github.com/SkyAPM/badger/v3 v3.0.0-20210624023741-bd2dcfcaaa74
diff --git a/go.sum b/go.sum
index 8940496..24a16ba 100644
--- a/go.sum
+++ b/go.sum
@@ -14,12 +14,10 @@ dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7
github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ=
github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo=
-github.com/DataDog/zstd v1.4.6-0.20210216161059-8cb8bacba7ba h1:3qB2yylqW3kVPr9QoPZtTJOXsJOUdNWT2CrZcifhs5g=
-github.com/DataDog/zstd v1.4.6-0.20210216161059-8cb8bacba7ba/go.mod h1:g4AWEaM3yOg3HYfnJ3YIawPnVdXJh9QME85blwSAmyw=
github.com/OneOfOne/xxhash v1.2.2 h1:KMrpdQIwFcEqXDklaen+P1axHaj9BSKzvpUUfnHldSE=
github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU=
-github.com/SkyAPM/badger/v3 v3.0.0-20210527215642-f1c960d2de88 h1:S/pluxBSFbM+1s7RMazOkHimDLNBHDhDQBUwMb18568=
-github.com/SkyAPM/badger/v3 v3.0.0-20210527215642-f1c960d2de88/go.mod h1:NiMHhAd/Elhx154lGQ4oJennK/f4ADMqhbggxs/mAew=
+github.com/SkyAPM/badger/v3 v3.0.0-20210624023741-bd2dcfcaaa74 h1:BFHSkDBSYCtPxMgxGz07DfNRYS76KFVDlocQ2U9rY7E=
+github.com/SkyAPM/badger/v3 v3.0.0-20210624023741-bd2dcfcaaa74/go.mod h1:XieWaNygSGj5ZzSsZO4tQe/2wwLjCvESus4twFqxOKc=
github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
github.com/armon/circbuf v0.0.0-20150827004946-bbbad097214e/go.mod h1:3U/XgcO3hCbHZ8TKRvWD2dDTCfh9M9ya+I9JpbB7O8o=
@@ -97,8 +95,8 @@ github.com/golang/protobuf v1.4.0/go.mod h1:jodUvKwWbYaEsadDk5Fwe5c77LiNKVO9IDvq
github.com/golang/protobuf v1.4.1/go.mod h1:U8fpvMrcmy5pZrNK1lt4xCsGvpyWQ/VVv6QDs8UjoX8=
github.com/golang/protobuf v1.4.2 h1:+Z5KGCizgyZCbGh1KZqA0fcLLkwbsjIzS4aV2v7wJX0=
github.com/golang/protobuf v1.4.2/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
-github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4=
-github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
+github.com/golang/snappy v0.0.3 h1:fHPg5GQYlCeLIPB9BZqMVR5nR9A+IM5zcgeTdjMYmLA=
+github.com/golang/snappy v0.0.3/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
github.com/google/btree v1.0.0/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
github.com/google/flatbuffers v1.12.0/go.mod h1:1AeVuKshWv4vARoZatz6mlQ0JxURH0Kv5+zNeJKJCa8=
@@ -157,6 +155,8 @@ github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7V
github.com/kisielk/errcheck v1.1.0/go.mod h1:EZBBE59ingxPouuu3KfxchcWSUPOHkagtvWXihfKN4Q=
github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
+github.com/klauspost/compress v1.12.3 h1:G5AfA94pHPysR56qqrkO2pxEexdDzrpFJ6yt/VqWxVU=
+github.com/klauspost/compress v1.12.3/go.mod h1:8dP1Hq4DHOhN9w426knH3Rhby4rFm6D8eO+e+Dq5Gzg=
github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc=
github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
diff --git a/api/common/id.go b/pkg/convert/hash.go
similarity index 88%
copy from api/common/id.go
copy to pkg/convert/hash.go
index 64d4fbf..e7014fe 100644
--- a/api/common/id.go
+++ b/pkg/convert/hash.go
@@ -15,6 +15,10 @@
// specific language governing permissions and limitations
// under the License.
-package common
+package convert
-type ChunkID uint64
+import "github.com/cespare/xxhash"
+
+func Hash(key []byte) uint64 {
+ return xxhash.Sum64(key)
+}
diff --git a/pkg/convert/number.go b/pkg/convert/number.go
index 88c80fc..1f1636c 100644
--- a/pkg/convert/number.go
+++ b/pkg/convert/number.go
@@ -26,7 +26,7 @@ func Uint64ToBytes(u uint64) []byte {
}
func Uint32ToBytes(u uint32) []byte {
- bs := make([]byte, 8)
+ bs := make([]byte, 4)
binary.BigEndian.PutUint32(bs, u)
return bs
}
@@ -35,6 +35,10 @@ func BytesToUint64(b []byte) uint64 {
return binary.BigEndian.Uint64(b)
}
+func BytesToUint32(b []byte) uint32 {
+ return binary.BigEndian.Uint32(b)
+}
+
func IntToInt64(numbers ...int) []int64 {
var arr []int64
for i := 0; i < len(numbers); i++ {
diff --git a/pkg/fb/api.go b/pkg/fb/api.go
index defc42d..c41c7b8 100644
--- a/pkg/fb/api.go
+++ b/pkg/fb/api.go
@@ -50,8 +50,12 @@ func (b *WriteEntityBuilder) BuildMetaData(group, name string) ComponentBuilderF
}
func (b *WriteEntityBuilder) BuildEntity(id string, binary []byte, items ...interface{}) ComponentBuilderFunc {
+ return b.BuildEntityWithTS(id, binary, uint64(time.Now().UnixNano()), items...)
+}
+
+func (b *WriteEntityBuilder) BuildEntityWithTS(id string, binary []byte, ts uint64, items ...interface{}) ComponentBuilderFunc {
entityID := b.Builder.CreateString(id)
- binaryOffset := b.buildDataBinary(binary)
+ binaryOffset := b.CreateByteVector(binary)
l := len(items)
var fieldOffsets []flatbuffers.UOffsetT
for i := 0; i < l; i++ {
@@ -59,14 +63,13 @@ func (b *WriteEntityBuilder) BuildEntity(id string, binary []byte, items ...inte
fieldOffsets = append(fieldOffsets, o)
}
v1.EntityStartFieldsVector(b.Builder, len(fieldOffsets))
- for i := 0; i < len(fieldOffsets); i++ {
+ for i := len(fieldOffsets) - 1; i >= 0; i-- {
b.PrependUOffsetT(fieldOffsets[i])
}
fields := b.EndVector(len(fieldOffsets))
v1.EntityStart(b.Builder)
v1.EntityAddEntityId(b.Builder, entityID)
- t := uint64(time.Now().UnixNano())
- v1.EntityAddTimestampNanoseconds(b.Builder, t)
+ v1.EntityAddTimestampNanoseconds(b.Builder, ts)
v1.EntityAddDataBinary(b.Builder, binaryOffset)
v1.EntityAddFields(b.Builder, fields)
entity := v1.EntityEnd(b.Builder)
@@ -76,53 +79,68 @@ func (b *WriteEntityBuilder) BuildEntity(id string, binary []byte, items ...inte
}
func (b *WriteEntityBuilder) buildField(val interface{}) flatbuffers.UOffsetT {
- var ValueTypeOffset flatbuffers.UOffsetT
+ if val == nil {
+ v1.FieldStart(b.Builder)
+ v1.FieldAddValueType(b.Builder, v1.ValueTypeNONE)
+ return v1.FieldEnd(b.Builder)
+ }
+ var valueTypeOffset flatbuffers.UOffsetT
var valType v1.ValueType
switch v := val.(type) {
case int:
- ValueTypeOffset = b.buildInt(int64(v))
+ valueTypeOffset = b.buildInt(int64(v))
valType = v1.ValueTypeInt
case []int:
- ValueTypeOffset = b.buildInt(convert.IntToInt64(v...)...)
+ valueTypeOffset = b.buildInt(convert.IntToInt64(v...)...)
valType = v1.ValueTypeIntArray
case int64:
- ValueTypeOffset = b.buildInt(v)
+ valueTypeOffset = b.buildInt(v)
valType = v1.ValueTypeInt
case []int64:
- ValueTypeOffset = b.buildInt(v...)
+ valueTypeOffset = b.buildInt(v...)
valType = v1.ValueTypeIntArray
case string:
- ValueTypeOffset = b.buildStrValueType(v)
+ valueTypeOffset = b.buildStrValueType(v)
valType = v1.ValueTypeString
case []string:
- ValueTypeOffset = b.buildStrValueType(v...)
+ valueTypeOffset = b.buildStrValueType(v...)
valType = v1.ValueTypeStringArray
default:
panic("not supported value")
}
v1.FieldStart(b.Builder)
- v1.FieldAddValue(b.Builder, ValueTypeOffset)
+ v1.FieldAddValue(b.Builder, valueTypeOffset)
v1.FieldAddValueType(b.Builder, valType)
return v1.FieldEnd(b.Builder)
}
func (b *WriteEntityBuilder) buildStrValueType(values ...string) flatbuffers.UOffsetT {
- var strOffsets []flatbuffers.UOffsetT
+ strOffsets := make([]flatbuffers.UOffsetT, 0)
for i := 0; i < len(values); i++ {
strOffsets = append(strOffsets, b.CreateString(values[i]))
}
+ if len(values) == 1 {
+ v1.StringStart(b.Builder)
+ v1.StringAddValue(b.Builder, strOffsets[0])
+ return v1.StringEnd(b.Builder)
+ }
v1.StringArrayStartValueVector(b.Builder, len(values))
- for i := 0; i < len(strOffsets); i++ {
+ for i := len(strOffsets) - 1; i >= 0; i-- {
b.Builder.PrependUOffsetT(strOffsets[i])
}
int64Arr := b.Builder.EndVector(len(values))
- v1.IntArrayStart(b.Builder)
- v1.IntArrayAddValue(b.Builder, int64Arr)
- return v1.IntArrayEnd(b.Builder)
+ v1.StringArrayStart(b.Builder)
+ v1.StringArrayAddValue(b.Builder, int64Arr)
+ return v1.StringArrayEnd(b.Builder)
}
func (b *WriteEntityBuilder) buildInt(values ...int64) flatbuffers.UOffsetT {
+ if len(values) == 1 {
+ v1.IntStart(b.Builder)
+ v1.IntAddValue(b.Builder, values[0])
+ return v1.IntEnd(b.Builder)
+ }
v1.IntArrayStartValueVector(b.Builder, len(values))
for i := 0; i < len(values); i++ {
b.Builder.PrependInt64(values[i])
@@ -134,17 +152,6 @@ func (b *WriteEntityBuilder) buildInt(values ...int64) flatbuffers.UOffsetT {
return v1.IntArrayEnd(b.Builder)
}
-func (b *WriteEntityBuilder) buildDataBinary(binary []byte) flatbuffers.UOffsetT {
- dataBinaryLength := len(binary)
- v1.EntityStartDataBinaryVector(b.Builder, dataBinaryLength)
- for i := dataBinaryLength; i >= 0; i-- {
- b.Builder.PrependByte(byte(i))
- }
- dataBinaryOffset := b.Builder.EndVector(dataBinaryLength)
-
- return dataBinaryOffset
-}
-
func (b *WriteEntityBuilder) BuildWriteEntity(funcs ...ComponentBuilderFunc) (*flatbuffers.Builder, error) {
v1.WriteEntityStart(b.Builder)
for _, fun := range funcs {
@@ -227,7 +234,7 @@ func (b *criteriaBuilder) BuildProjection(keyNames ...string) ComponentBuilderFu
keyNamesOffsets = append(keyNamesOffsets, b.Builder.CreateString(keyNames[i]))
}
v1.ProjectionStartKeyNamesVector(b.Builder, len(keyNames))
- for i := 0; i < len(keyNamesOffsets); i++ {
+ for i := len(keyNamesOffsets) - 1; i >= 0; i-- {
b.Builder.PrependUOffsetT(keyNamesOffsets[i])
}
strArr := b.Builder.EndVector(len(keyNames))
@@ -239,86 +246,86 @@ func (b *criteriaBuilder) BuildProjection(keyNames ...string) ComponentBuilderFu
}
}
-func (b *criteriaBuilder) buildIntPair(key string, values ...int64) flatbuffers.UOffsetT {
- v1.IntPairStartValuesVector(b.Builder, len(values))
- for i := 0; i < len(values); i++ {
- b.Builder.PrependInt64(values[i])
+func buildIntPair(b *flatbuffers.Builder, key string, values ...int64) flatbuffers.UOffsetT {
+ v1.IntPairStartValuesVector(b, len(values))
+ for i := len(values) - 1; i >= 0; i-- {
+ b.PrependInt64(values[i])
}
- int64Arr := b.Builder.EndVector(len(values))
+ int64Arr := b.EndVector(len(values))
keyOffset := b.CreateString(key)
- v1.IntPairStart(b.Builder)
- v1.IntPairAddKey(b.Builder, keyOffset)
- v1.IntPairAddValues(b.Builder, int64Arr)
- return v1.IntPairEnd(b.Builder)
+ v1.IntPairStart(b)
+ v1.IntPairAddKey(b, keyOffset)
+ v1.IntPairAddValues(b, int64Arr)
+ return v1.IntPairEnd(b)
}
-func (b *criteriaBuilder) buildStrPair(key string, values ...string) flatbuffers.UOffsetT {
+func buildStrPair(b *flatbuffers.Builder, key string, values ...string) flatbuffers.UOffsetT {
var strOffsets []flatbuffers.UOffsetT
for i := 0; i < len(values); i++ {
strOffsets = append(strOffsets, b.CreateString(values[i]))
}
- v1.StrPairStartValuesVector(b.Builder, len(values))
- for i := 0; i < len(strOffsets); i++ {
- b.Builder.PrependUOffsetT(strOffsets[i])
+ v1.StrPairStartValuesVector(b, len(values))
+ for i := len(strOffsets) - 1; i >= 0; i-- {
+ b.PrependUOffsetT(strOffsets[i])
}
- int64Arr := b.Builder.EndVector(len(values))
+ int64Arr := b.EndVector(len(values))
keyOffset := b.CreateString(key)
- v1.IntPairStart(b.Builder)
- v1.IntPairAddKey(b.Builder, keyOffset)
- v1.IntPairAddValues(b.Builder, int64Arr)
- return v1.IntPairEnd(b.Builder)
+ v1.IntPairStart(b)
+ v1.IntPairAddKey(b, keyOffset)
+ v1.IntPairAddValues(b, int64Arr)
+ return v1.IntPairEnd(b)
}
-func (b *criteriaBuilder) buildCondition(key string, value interface{}) flatbuffers.UOffsetT {
+func buildPair(b *flatbuffers.Builder, key string, value interface{}) flatbuffers.UOffsetT {
var pairOffset flatbuffers.UOffsetT
var pairType v1.TypedPair
switch v := value.(type) {
case int:
- pairOffset = b.buildIntPair(key, int64(v))
+ pairOffset = buildIntPair(b, key, int64(v))
pairType = v1.TypedPairIntPair
case []int:
- pairOffset = b.buildIntPair(key, convert.IntToInt64(v...)...)
+ pairOffset = buildIntPair(b, key, convert.IntToInt64(v...)...)
pairType = v1.TypedPairIntPair
case int8:
- pairOffset = b.buildIntPair(key, int64(v))
+ pairOffset = buildIntPair(b, key, int64(v))
pairType = v1.TypedPairIntPair
case []int8:
- pairOffset = b.buildIntPair(key, convert.Int8ToInt64(v...)...)
+ pairOffset = buildIntPair(b, key, convert.Int8ToInt64(v...)...)
pairType = v1.TypedPairIntPair
case int16:
- pairOffset = b.buildIntPair(key, int64(v))
+ pairOffset = buildIntPair(b, key, int64(v))
pairType = v1.TypedPairIntPair
case []int16:
- pairOffset = b.buildIntPair(key, convert.Int16ToInt64(v...)...)
+ pairOffset = buildIntPair(b, key, convert.Int16ToInt64(v...)...)
pairType = v1.TypedPairIntPair
case int32:
- pairOffset = b.buildIntPair(key, int64(v))
+ pairOffset = buildIntPair(b, key, int64(v))
pairType = v1.TypedPairIntPair
case []int32:
- pairOffset = b.buildIntPair(key, convert.Int32ToInt64(v...)...)
+ pairOffset = buildIntPair(b, key, convert.Int32ToInt64(v...)...)
pairType = v1.TypedPairIntPair
case int64:
- pairOffset = b.buildIntPair(key, v)
+ pairOffset = buildIntPair(b, key, v)
pairType = v1.TypedPairIntPair
case []int64:
- pairOffset = b.buildIntPair(key, v...)
+ pairOffset = buildIntPair(b, key, v...)
pairType = v1.TypedPairIntPair
case string:
- pairOffset = b.buildStrPair(key, v)
+ pairOffset = buildStrPair(b, key, v)
pairType = v1.TypedPairStrPair
case []string:
- pairOffset = b.buildStrPair(key, v...)
+ pairOffset = buildStrPair(b, key, v...)
pairType = v1.TypedPairStrPair
default:
panic("not supported values")
}
- v1.PairStart(b.Builder)
- v1.PairAddPair(b.Builder, pairOffset)
- v1.PairAddPairType(b.Builder, pairType)
- return v1.PairEnd(b.Builder)
+ v1.PairStart(b)
+ v1.PairAddPair(b, pairOffset)
+ v1.PairAddPairType(b, pairType)
+ return v1.PairEnd(b)
}
func (b *criteriaBuilder) BuildFields(items ...interface{}) ComponentBuilderFunc {
@@ -329,7 +336,7 @@ func (b *criteriaBuilder) BuildFields(items ...interface{}) ComponentBuilderFunc
var pairQueryOffsets []flatbuffers.UOffsetT
for i := 0; i < l; i++ {
key, op, values := items[i*3+0], items[i*3+1], items[i*3+2]
- condition := b.buildCondition(key.(string), values)
+ condition := buildPair(b.Builder, key.(string), values)
v1.PairQueryStart(b.Builder)
// add op
v1.PairQueryAddOp(b.Builder, binaryOpsMap[op.(string)])
@@ -338,7 +345,7 @@ func (b *criteriaBuilder) BuildFields(items ...interface{}) ComponentBuilderFunc
pairQueryOffsets = append(pairQueryOffsets, v1.PairQueryEnd(b.Builder))
}
v1.EntityCriteriaStartFieldsVector(b.Builder, l)
- for i := 0; i < len(pairQueryOffsets); i++ {
+ for i := len(pairQueryOffsets) - 1; i >= 0; i-- {
b.PrependUOffsetT(pairQueryOffsets[i])
}
fields := b.EndVector(l)
@@ -369,3 +376,63 @@ func (b *criteriaBuilder) BuildQueryEntity(funcs ...ComponentBuilderFunc) (*flat
return b.Builder, nil
}
+
+type queryEntityBuilder struct {
+ *flatbuffers.Builder
+}
+
+func NewQueryEntityBuilder() *queryEntityBuilder {
+ return &queryEntityBuilder{
+ flatbuffers.NewBuilder(1024),
+ }
+}
+
+func (b *queryEntityBuilder) BuildEntityID(entityID string) ComponentBuilderFunc {
+ eID := b.Builder.CreateString(entityID)
+ return func(b *flatbuffers.Builder) {
+ v1.EntityAddDataBinary(b, eID)
+ }
+}
+
+func (b *queryEntityBuilder) BuildTimeStamp(t time.Time) ComponentBuilderFunc {
+ return func(b *flatbuffers.Builder) {
+ v1.EntityAddTimestampNanoseconds(b, uint64(t.UnixNano()))
+ }
+}
+
+func (b *queryEntityBuilder) BuildFields(items ...interface{}) ComponentBuilderFunc {
+ if len(items)%2 != 0 {
+ panic("invalid fields list")
+ }
+
+ l := len(items) / 2
+
+ var pairOffsets []flatbuffers.UOffsetT
+ for i := 0; i < l; i++ {
+ key, values := items[i*2+0], items[i*2+1]
+ pair := buildPair(b.Builder, key.(string), values)
+ pairOffsets = append(pairOffsets, pair)
+ }
+
+ v1.EntityStartFieldsVector(b.Builder, l)
+ for i := len(pairOffsets) - 1; i >= 0; i-- {
+ b.PrependUOffsetT(pairOffsets[i])
+ }
+ fields := b.EndVector(l)
+
+ return func(b *flatbuffers.Builder) {
+ v1.EntityAddFields(b, fields)
+ }
+}
+
+func (b *queryEntityBuilder) BuildEntity(funcs ...ComponentBuilderFunc) *v1.Entity {
+ v1.EntityStart(b.Builder)
+ for _, fun := range funcs {
+ fun(b.Builder)
+ }
+ criteriaOffset := v1.EntityEnd(b.Builder)
+ b.Builder.Finish(criteriaOffset)
+
+ buf := b.Bytes[b.Head():]
+ return v1.GetRootAsEntity(buf, 0)
+}
diff --git a/pkg/fb/util.go b/pkg/fb/util.go
new file mode 100644
index 0000000..2664794
--- /dev/null
+++ b/pkg/fb/util.go
@@ -0,0 +1,228 @@
+// 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 fb
+
+import (
+ flatbuffers "github.com/google/flatbuffers/go"
+
+ v1 "github.com/apache/skywalking-banyandb/api/fbs/v1"
+)
+
+func Transform(entityValue *v1.EntityValue, fieldsIndices []FieldEntry, builder *flatbuffers.Builder) flatbuffers.UOffsetT {
+ fieldsLen := len(fieldsIndices)
+ result := make([]flatbuffers.UOffsetT, 0, fieldsLen)
+
+ VisitFields(entityValue, fieldsIndices, func(key string, valueType v1.ValueType, value interface{}) {
+ keyOffset := builder.CreateString(key)
+ var pairOffset flatbuffers.UOffsetT
+ var typedPair v1.TypedPair
+ switch valueType {
+ case v1.ValueTypeNONE:
+ typedPair = v1.TypedPairNONE
+ pairOffset = 0
+ case v1.ValueTypeString, v1.ValueTypeStringArray:
+ typedPair = v1.TypedPairStrPair
+ stringValue, isStr := value.(*v1.String)
+ var l int
+ var stringArrayValue *v1.StringArray
+ if isStr {
+ l = 1
+ } else {
+ stringArrayValue = value.(*v1.StringArray)
+ l = stringArrayValue.ValueLength()
+ }
+ strPos := make([]flatbuffers.UOffsetT, 0, l)
+ for j := 0; j < l; j++ {
+ var o flatbuffers.UOffsetT
+ if isStr {
+ o = builder.CreateByteString(stringValue.Value())
+ } else {
+ o = builder.CreateByteString(stringArrayValue.Value(j))
+ }
+ strPos = append(strPos, o)
+ }
+ v1.StringArrayStartValueVector(builder, l)
+ for j := l - 1; j >= 0; j-- {
+ builder.PrependUOffsetT(strPos[j])
+ }
+ strArray := builder.EndVector(l)
+ v1.StrPairStart(builder)
+ v1.StrPairAddKey(builder, keyOffset)
+ v1.StrPairAddValues(builder, strArray)
+ pairOffset = v1.StrPairEnd(builder)
+ case v1.ValueTypeInt, v1.ValueTypeIntArray:
+ typedPair = v1.TypedPairIntPair
+ intValue, isInt := value.(*v1.Int)
+ var l int
+ var intArrayValue *v1.IntArray
+ if isInt {
+ l = 1
+ } else {
+ intArrayValue = value.(*v1.IntArray)
+ l = intArrayValue.ValueLength()
+ }
+ v1.IntArrayStartValueVector(builder, l)
+ for j := l - 1; j >= 0; j-- {
+ var v int64
+ if isInt {
+ v = intValue.Value()
+ } else {
+ v = intArrayValue.Value(j)
+ }
+ builder.PrependInt64(v)
+ }
+ intArray := builder.EndVector(l)
+ v1.IntPairStart(builder)
+ v1.IntPairAddKey(builder, keyOffset)
+ v1.IntPairAddValues(builder, intArray)
+ pairOffset = v1.IntPairEnd(builder)
+ }
+ if pairOffset <= 0 {
+ return
+ }
+ v1.PairStart(builder)
+ v1.PairAddPairType(builder, typedPair)
+ v1.PairAddPair(builder, pairOffset)
+ result = append(result, v1.PairEnd(builder))
+ })
+ l := len(result)
+ if l < 1 {
+ return 0
+ }
+ v1.EntityStartFieldsVector(builder, l)
+ for i := l - 1; i >= 0; i-- {
+ builder.PrependUOffsetT(result[i])
+ }
+ return builder.EndVector(l)
+}
+
+func CopyFields(entityValue *v1.EntityValue, builder *flatbuffers.Builder) flatbuffers.UOffsetT {
+ fieldsLen := entityValue.FieldsLength()
+ result := make([]flatbuffers.UOffsetT, 0, fieldsLen)
+ VisitFields(entityValue, nil, func(_ string, valueType v1.ValueType, value interface{}) {
+ var valueOffset flatbuffers.UOffsetT
+ switch valueType {
+ case v1.ValueTypeNONE:
+ valueOffset = 0
+ case v1.ValueTypeString:
+ stringValue := value.(*v1.String)
+ pos := builder.CreateByteString(stringValue.Value())
+ v1.StringStart(builder)
+ v1.StringAddValue(builder, pos)
+ valueOffset = v1.StringEnd(builder)
+ case v1.ValueTypeInt:
+ intValue := value.(*v1.Int)
+ v1.IntStart(builder)
+ v1.IntAddValue(builder, intValue.Value())
+ valueOffset = v1.IntEnd(builder)
+ case v1.ValueTypeStringArray:
+ stringArrayValue := value.(*v1.StringArray)
+ l := stringArrayValue.ValueLength()
+ v1.StringArrayStartValueVector(builder, l)
+ strPos := make([]flatbuffers.UOffsetT, 0, l)
+ for j := 0; j < l; j++ {
+ strPos = append(strPos, builder.CreateByteString(stringArrayValue.Value(j)))
+ }
+ for j := l; j >= 0; j-- {
+ builder.PrependUOffsetT(strPos[j])
+ }
+ valueOffset = builder.EndVector(l)
+ case v1.ValueTypeIntArray:
+ intArrayValue := value.(*v1.IntArray)
+ l := intArrayValue.ValueLength()
+ v1.IntArrayStartValueVector(builder, l)
+ for j := l; j >= 0; j-- {
+ builder.PrependInt64(intArrayValue.Value(j))
+ }
+ valueOffset = builder.EndVector(l)
+ }
+ v1.FieldStart(builder)
+ v1.FieldAddValueType(builder, valueType)
+ if valueOffset > 0 {
+ v1.FieldAddValue(builder, valueOffset)
+ }
+ result = append(result, v1.FieldEnd(builder))
+ })
+
+ v1.EntityValueStartFieldsVector(builder, fieldsLen)
+ for i := fieldsLen - 1; i >= 0; i-- {
+ builder.PrependUOffsetT(result[i])
+ }
+ o := builder.EndVector(fieldsLen)
+ entityID := builder.CreateByteString(entityValue.EntityId())
+ v1.EntityValueStart(builder)
+ v1.EntityValueAddFields(builder, o)
+ v1.EntityValueAddEntityId(builder, entityID)
+ v1.EntityValueAddTimestampNanoseconds(builder, entityValue.TimestampNanoseconds())
+ return v1.EntityValueEnd(builder)
+}
+
+func VisitFields(entityValue *v1.EntityValue, fieldsIndices []FieldEntry, visit func(key string, valueType v1.ValueType, value interface{})) {
+ var fieldsLen int
+ if fieldsIndices == nil {
+ fieldsLen = entityValue.FieldsLength()
+ } else {
+ fieldsLen = len(fieldsIndices)
+ }
+ for i := 0; i < fieldsLen; i++ {
+ var index int
+ var key string
+ if fieldsIndices == nil {
+ index = i
+ } else {
+ index = fieldsIndices[i].Index
+ key = fieldsIndices[i].Key
+ }
+ if index >= entityValue.FieldsLength() {
+ visit(key, v1.ValueTypeNONE, nil)
+ continue
+ }
+ f := &v1.Field{}
+ if !entityValue.Fields(f, index) {
+ continue
+ }
+
+ unionTable := new(flatbuffers.Table)
+ f.Value(unionTable)
+ switch f.ValueType() {
+ case v1.ValueTypeNONE:
+ visit(key, f.ValueType(), nil)
+ case v1.ValueTypeString:
+ stringValue := new(v1.String)
+ stringValue.Init(unionTable.Bytes, unionTable.Pos)
+ visit(key, f.ValueType(), stringValue)
+ case v1.ValueTypeInt:
+ intValue := new(v1.Int)
+ intValue.Init(unionTable.Bytes, unionTable.Pos)
+ visit(key, f.ValueType(), intValue)
+ case v1.ValueTypeStringArray:
+ stringArrayValue := new(v1.StringArray)
+ stringArrayValue.Init(unionTable.Bytes, unionTable.Pos)
+ visit(key, f.ValueType(), stringArrayValue)
+ case v1.ValueTypeIntArray:
+ intArrayValue := new(v1.IntArray)
+ intArrayValue.Init(unionTable.Bytes, unionTable.Pos)
+ visit(key, f.ValueType(), intArrayValue)
+ }
+ }
+}
+
+type FieldEntry struct {
+ Key string
+ Index int
+}
diff --git a/pkg/logical/analyzer_test.go b/pkg/logical/analyzer_test.go
index 50af8e7..a0bf129 100644
--- a/pkg/logical/analyzer_test.go
+++ b/pkg/logical/analyzer_test.go
@@ -50,7 +50,7 @@ func TestAnalyzer_SimpleTimeScan(t *testing.T) {
metadata := &common.Metadata{
KindVersion: apischema.SeriesKindVersion,
- Spec: *criteria.Metadata(nil),
+ Spec: criteria.Metadata(nil),
}
schema, err := ana.BuildTraceSchema(context.TODO(), *metadata)
@@ -90,7 +90,7 @@ func TestAnalyzer_ComplexQuery(t *testing.T) {
metadata := &common.Metadata{
KindVersion: apischema.SeriesKindVersion,
- Spec: *criteria.Metadata(nil),
+ Spec: criteria.Metadata(nil),
}
schema, err := ana.BuildTraceSchema(context.TODO(), *metadata)
@@ -132,7 +132,7 @@ func TestAnalyzer_TraceIDQuery(t *testing.T) {
metadata := &common.Metadata{
KindVersion: apischema.SeriesKindVersion,
- Spec: *criteria.Metadata(nil),
+ Spec: criteria.Metadata(nil),
}
schema, err := ana.BuildTraceSchema(context.TODO(), *metadata)
@@ -156,7 +156,7 @@ func TestAnalyzer_Fields_FieldNotDefined(t *testing.T) {
criteria := builder.BuildEntityCriteria(
fb.AddLimit(5),
fb.AddOffset(10),
- builder.BuildMetaData("default", "trace"),
+ builder.BuildMetaData("default", "sw"),
builder.BuildTimeStampNanoSeconds(time.Now().Add(-3*time.Hour), time.Now()),
builder.BuildFields("duration", ">", 500),
builder.BuildOrderBy("service_instance_id", apiv1.SortDESC),
@@ -165,7 +165,7 @@ func TestAnalyzer_Fields_FieldNotDefined(t *testing.T) {
metadata := &common.Metadata{
KindVersion: apischema.SeriesKindVersion,
- Spec: *criteria.Metadata(nil),
+ Spec: criteria.Metadata(nil),
}
schema, err := ana.BuildTraceSchema(context.TODO(), *metadata)
@@ -192,7 +192,7 @@ func TestAnalyzer_OrderBy_FieldNotDefined(t *testing.T) {
metadata := &common.Metadata{
KindVersion: apischema.SeriesKindVersion,
- Spec: *criteria.Metadata(nil),
+ Spec: criteria.Metadata(nil),
}
schema, err := ana.BuildTraceSchema(context.TODO(), *metadata)
@@ -211,14 +211,14 @@ func TestAnalyzer_Projection_FieldNotDefined(t *testing.T) {
criteria := builder.BuildEntityCriteria(
fb.AddLimit(5),
fb.AddOffset(10),
- builder.BuildMetaData("default", "trace"),
+ builder.BuildMetaData("default", "sw"),
builder.BuildTimeStampNanoSeconds(time.Now().Add(-3*time.Hour), time.Now()),
- builder.BuildProjection("duration", "service_id"),
+ builder.BuildProjection("duration", "service_id", "unknown"),
)
metadata := &common.Metadata{
KindVersion: apischema.SeriesKindVersion,
- Spec: *criteria.Metadata(nil),
+ Spec: criteria.Metadata(nil),
}
schema, err := ana.BuildTraceSchema(context.TODO(), *metadata)
@@ -245,7 +245,7 @@ func TestAnalyzer_Fields_IndexNotDefined(t *testing.T) {
metadata := &common.Metadata{
KindVersion: apischema.SeriesKindVersion,
- Spec: *criteria.Metadata(nil),
+ Spec: criteria.Metadata(nil),
}
schema, err := ana.BuildTraceSchema(context.TODO(), *metadata)
diff --git a/api/common/id.go b/pkg/partition/route.go
similarity index 80%
copy from api/common/id.go
copy to pkg/partition/route.go
index 64d4fbf..e6eae3f 100644
--- a/api/common/id.go
+++ b/pkg/partition/route.go
@@ -15,6 +15,11 @@
// specific language governing permissions and limitations
// under the License.
-package common
+package partition
-type ChunkID uint64
+import "github.com/apache/skywalking-banyandb/pkg/convert"
+
+func ShardID(key []byte, shardNum uint) uint {
+ encodeKey := convert.Hash(key)
+ return uint(encodeKey % uint64(shardNum))
+}
[skywalking-banyandb] 02/02: Add local discovery and shard event
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch series-trace
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit ccb0d430c5222a7dfbc0a90ca694082c631401a8
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Mon Jul 12 09:13:34 2021 +0800
Add local discovery and shard event
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
api/event/discovery.go | 5 +++
api/fbs/v1/database.fbs | 4 +-
api/fbs/v1/database_generated.go | 61 +++++++++++++++++++++++--------
banyand/discovery/discovery.go | 36 ++++++++++++++----
banyand/internal/cmd/standalone.go | 4 +-
banyand/liaison/grpc/grpc.go | 32 ++++++++++++++--
banyand/liaison/liaison.go | 6 ++-
banyand/queue/local.go | 18 ++++-----
banyand/queue/queue.go | 6 +--
banyand/series/series.go | 1 +
banyand/series/trace/common_test.go | 2 +-
banyand/series/trace/trace.go | 41 ++++++++++++++++++++-
{banyand/internal => pkg}/bus/bus.go | 0
{banyand/internal => pkg}/bus/bus_test.go | 0
pkg/convert/hash.go | 4 ++
pkg/fb/database.go | 59 ++++++++++++++++++++++++++++++
16 files changed, 234 insertions(+), 45 deletions(-)
diff --git a/api/event/discovery.go b/api/event/discovery.go
index 6916c65..40360c4 100644
--- a/api/event/discovery.go
+++ b/api/event/discovery.go
@@ -20,6 +20,7 @@ package event
import (
"github.com/apache/skywalking-banyandb/api/common"
v1 "github.com/apache/skywalking-banyandb/api/fbs/v1"
+ "github.com/apache/skywalking-banyandb/pkg/bus"
)
var (
@@ -27,6 +28,10 @@ var (
Version: "v1",
Kind: "event-shard",
}
+ TopicShardEvent = bus.Topic{
+ ID: ShardEventKindVersion.String(),
+ Type: bus.ChTypeUnidirectional,
+ }
)
type Shard struct {
diff --git a/api/fbs/v1/database.fbs b/api/fbs/v1/database.fbs
index 836553c..d71c1f1 100644
--- a/api/fbs/v1/database.fbs
+++ b/api/fbs/v1/database.fbs
@@ -25,8 +25,10 @@ table Node {
}
table Shard {
- id:int64;
+ id:uint64;
+ series:Metadata;
node:Node;
+ total:uint8;
update_time:int64;
create_time:int64;
}
diff --git a/api/fbs/v1/database_generated.go b/api/fbs/v1/database_generated.go
index 0021dc6..4236dff 100644
--- a/api/fbs/v1/database_generated.go
+++ b/api/fbs/v1/database_generated.go
@@ -145,23 +145,36 @@ func (rcv *Shard) Table() flatbuffers.Table {
return rcv._tab
}
-func (rcv *Shard) Id() int64 {
+func (rcv *Shard) Id() uint64 {
o := flatbuffers.UOffsetT(rcv._tab.Offset(4))
if o != 0 {
- return rcv._tab.GetInt64(o + rcv._tab.Pos)
+ return rcv._tab.GetUint64(o + rcv._tab.Pos)
}
return 0
}
-func (rcv *Shard) MutateId(n int64) bool {
- return rcv._tab.MutateInt64Slot(4, n)
+func (rcv *Shard) MutateId(n uint64) bool {
+ return rcv._tab.MutateUint64Slot(4, n)
}
-func (rcv *Shard) Node(obj *Node) *Node {
+func (rcv *Shard) Series(obj *Metadata) *Metadata {
o := flatbuffers.UOffsetT(rcv._tab.Offset(6))
if o != 0 {
x := rcv._tab.Indirect(o + rcv._tab.Pos)
if obj == nil {
+ obj = new(Metadata)
+ }
+ obj.Init(rcv._tab.Bytes, x)
+ return obj
+ }
+ return nil
+}
+
+func (rcv *Shard) Node(obj *Node) *Node {
+ o := flatbuffers.UOffsetT(rcv._tab.Offset(8))
+ if o != 0 {
+ x := rcv._tab.Indirect(o + rcv._tab.Pos)
+ if obj == nil {
obj = new(Node)
}
obj.Init(rcv._tab.Bytes, x)
@@ -170,8 +183,20 @@ func (rcv *Shard) Node(obj *Node) *Node {
return nil
}
+func (rcv *Shard) Total() byte {
+ o := flatbuffers.UOffsetT(rcv._tab.Offset(10))
+ if o != 0 {
+ return rcv._tab.GetByte(o + rcv._tab.Pos)
+ }
+ return 0
+}
+
+func (rcv *Shard) MutateTotal(n byte) bool {
+ return rcv._tab.MutateByteSlot(10, n)
+}
+
func (rcv *Shard) UpdateTime() int64 {
- o := flatbuffers.UOffsetT(rcv._tab.Offset(8))
+ o := flatbuffers.UOffsetT(rcv._tab.Offset(12))
if o != 0 {
return rcv._tab.GetInt64(o + rcv._tab.Pos)
}
@@ -179,11 +204,11 @@ func (rcv *Shard) UpdateTime() int64 {
}
func (rcv *Shard) MutateUpdateTime(n int64) bool {
- return rcv._tab.MutateInt64Slot(8, n)
+ return rcv._tab.MutateInt64Slot(12, n)
}
func (rcv *Shard) CreateTime() int64 {
- o := flatbuffers.UOffsetT(rcv._tab.Offset(10))
+ o := flatbuffers.UOffsetT(rcv._tab.Offset(14))
if o != 0 {
return rcv._tab.GetInt64(o + rcv._tab.Pos)
}
@@ -191,23 +216,29 @@ func (rcv *Shard) CreateTime() int64 {
}
func (rcv *Shard) MutateCreateTime(n int64) bool {
- return rcv._tab.MutateInt64Slot(10, n)
+ return rcv._tab.MutateInt64Slot(14, n)
}
func ShardStart(builder *flatbuffers.Builder) {
- builder.StartObject(4)
+ builder.StartObject(6)
+}
+func ShardAddId(builder *flatbuffers.Builder, id uint64) {
+ builder.PrependUint64Slot(0, id, 0)
}
-func ShardAddId(builder *flatbuffers.Builder, id int64) {
- builder.PrependInt64Slot(0, id, 0)
+func ShardAddSeries(builder *flatbuffers.Builder, series flatbuffers.UOffsetT) {
+ builder.PrependUOffsetTSlot(1, flatbuffers.UOffsetT(series), 0)
}
func ShardAddNode(builder *flatbuffers.Builder, node flatbuffers.UOffsetT) {
- builder.PrependUOffsetTSlot(1, flatbuffers.UOffsetT(node), 0)
+ builder.PrependUOffsetTSlot(2, flatbuffers.UOffsetT(node), 0)
+}
+func ShardAddTotal(builder *flatbuffers.Builder, total byte) {
+ builder.PrependByteSlot(3, total, 0)
}
func ShardAddUpdateTime(builder *flatbuffers.Builder, updateTime int64) {
- builder.PrependInt64Slot(2, updateTime, 0)
+ builder.PrependInt64Slot(4, updateTime, 0)
}
func ShardAddCreateTime(builder *flatbuffers.Builder, createTime int64) {
- builder.PrependInt64Slot(3, createTime, 0)
+ builder.PrependInt64Slot(5, createTime, 0)
}
func ShardEnd(builder *flatbuffers.Builder) flatbuffers.UOffsetT {
return builder.EndObject()
diff --git a/banyand/discovery/discovery.go b/banyand/discovery/discovery.go
index 9e7197c..8856166 100644
--- a/banyand/discovery/discovery.go
+++ b/banyand/discovery/discovery.go
@@ -20,17 +20,39 @@ package discovery
import (
"context"
- "github.com/apache/skywalking-banyandb/banyand/internal/bus"
+ bus2 "github.com/apache/skywalking-banyandb/pkg/bus"
"github.com/apache/skywalking-banyandb/pkg/run"
)
type ServiceRepo interface {
- run.Config
- run.Service
- bus.Subscriber
- bus.Publisher
+ NodeID() string
+ run.Unit
+ bus2.Subscriber
+ bus2.Publisher
}
-func NewServiceRepo(ctx context.Context) (ServiceRepo, error) {
- return nil, nil
+type repo struct {
+ local *bus2.Bus
+}
+
+func (r *repo) NodeID() string {
+ return "local"
+}
+
+func (r *repo) Name() string {
+ return "service-discovery"
+}
+
+func (r *repo) Subscribe(topic bus2.Topic, listener bus2.MessageListener) error {
+ return r.local.Subscribe(topic, listener)
+}
+
+func (r *repo) Publish(topic bus2.Topic, message ...bus2.Message) (bus2.Future, error) {
+ return r.local.Publish(topic, message...)
+}
+
+func NewServiceRepo(_ context.Context) (ServiceRepo, error) {
+ return &repo{
+ local: bus2.NewBus(),
+ }, nil
}
diff --git a/banyand/internal/cmd/standalone.go b/banyand/internal/cmd/standalone.go
index 69e069e..2d50601 100644
--- a/banyand/internal/cmd/standalone.go
+++ b/banyand/internal/cmd/standalone.go
@@ -61,7 +61,7 @@ func newStandaloneCmd() *cobra.Command {
if err != nil {
l.Fatal().Err(err).Msg("failed to initiate index builder")
}
- traceSeries, err := trace.NewService(ctx, db)
+ traceSeries, err := trace.NewService(ctx, db, repo)
if err != nil {
l.Fatal().Err(err).Msg("failed to initiate trace series")
}
@@ -69,7 +69,7 @@ func newStandaloneCmd() *cobra.Command {
if err != nil {
l.Fatal().Err(err).Msg("failed to initiate query executor")
}
- tcp, err := liaison.NewEndpoint(ctx, pipeline)
+ tcp, err := liaison.NewEndpoint(ctx, pipeline, repo)
if err != nil {
l.Fatal().Err(err).Msg("failed to initiate Endpoint transport layer")
}
diff --git a/banyand/liaison/grpc/grpc.go b/banyand/liaison/grpc/grpc.go
index 36f658c..deade08 100644
--- a/banyand/liaison/grpc/grpc.go
+++ b/banyand/liaison/grpc/grpc.go
@@ -25,7 +25,11 @@ import (
grpclib "google.golang.org/grpc"
"google.golang.org/grpc/encoding"
+ "github.com/apache/skywalking-banyandb/api/event"
+ v1 "github.com/apache/skywalking-banyandb/api/fbs/v1"
+ "github.com/apache/skywalking-banyandb/banyand/discovery"
"github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/pkg/bus"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/run"
)
@@ -35,10 +39,33 @@ type Server struct {
log *logger.Logger
ser *grpclib.Server
pipeline queue.Queue
+ repo discovery.ServiceRepo
}
-func NewServer(ctx context.Context, pipeline queue.Queue) *Server {
- return &Server{pipeline: pipeline}
+func (s *Server) Rev(message bus.Message) (resp bus.Message) {
+ data, ok := message.Data().([]byte)
+ if !ok {
+ s.log.Warn().Msg("invalid event data type")
+ return
+ }
+ shardEvent := v1.GetRootAsShardEvent(data, 0)
+ s.log.Info().
+ Str("action", shardEvent.Action().String()).
+ Uint64("shardID", shardEvent.Shard(nil).Id()).
+ Msg("received a shard event")
+ return
+}
+
+func (s *Server) PreRun() error {
+ s.log = logger.GetLogger("liaison-grpc")
+ return s.repo.Subscribe(event.TopicShardEvent, s)
+}
+
+func NewServer(ctx context.Context, pipeline queue.Queue, repo discovery.ServiceRepo) *Server {
+ return &Server{
+ pipeline: pipeline,
+ repo: repo,
+ }
}
func (s *Server) Name() string {
@@ -56,7 +83,6 @@ func (s *Server) Validate() error {
}
func (s *Server) Serve() error {
- s.log = logger.GetLogger("grpc")
lis, err := net.Listen("tcp", s.addr)
if err != nil {
s.log.Fatal().Err(err).Msg("Failed to listen")
diff --git a/banyand/liaison/liaison.go b/banyand/liaison/liaison.go
index 117b361..8eb9274 100644
--- a/banyand/liaison/liaison.go
+++ b/banyand/liaison/liaison.go
@@ -20,6 +20,7 @@ package liaison
import (
"context"
+ "github.com/apache/skywalking-banyandb/banyand/discovery"
"github.com/apache/skywalking-banyandb/banyand/liaison/grpc"
"github.com/apache/skywalking-banyandb/banyand/queue"
"github.com/apache/skywalking-banyandb/pkg/run"
@@ -27,9 +28,10 @@ import (
type Endpoint interface {
run.Config
+ run.PreRunner
run.Service
}
-func NewEndpoint(ctx context.Context, pipeline queue.Queue) (Endpoint, error) {
- return grpc.NewServer(ctx, pipeline), nil
+func NewEndpoint(ctx context.Context, pipeline queue.Queue, repo discovery.ServiceRepo) (Endpoint, error) {
+ return grpc.NewServer(ctx, pipeline, repo), nil
}
diff --git a/banyand/queue/local.go b/banyand/queue/local.go
index 2a4fe7a..45bad1d 100644
--- a/banyand/queue/local.go
+++ b/banyand/queue/local.go
@@ -21,7 +21,7 @@ import (
"go.uber.org/multierr"
"github.com/apache/skywalking-banyandb/banyand/discovery"
- "github.com/apache/skywalking-banyandb/banyand/internal/bus"
+ bus2 "github.com/apache/skywalking-banyandb/pkg/bus"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/run"
)
@@ -34,33 +34,33 @@ type Component interface {
type DataSubscriber interface {
Component
- Sub(subscriber bus.Subscriber) error
+ Sub(subscriber bus2.Subscriber) error
}
type DataPublisher interface {
Component
- Pub(publisher bus.Publisher) error
+ Pub(publisher bus2.Publisher) error
}
var _ run.PreRunner = (*Local)(nil)
var _ run.Config = (*Local)(nil)
-var _ bus.Publisher = (*Local)(nil)
-var _ bus.Subscriber = (*Local)(nil)
+var _ bus2.Publisher = (*Local)(nil)
+var _ bus2.Subscriber = (*Local)(nil)
type Local struct {
logger *logger.Logger
test string
- dataBus *bus.Bus
+ dataBus *bus2.Bus
dps []DataPublisher
dss []DataSubscriber
repo discovery.ServiceRepo
}
-func (e *Local) Subscribe(topic bus.Topic, listener bus.MessageListener) error {
+func (e *Local) Subscribe(topic bus2.Topic, listener bus2.MessageListener) error {
return nil
}
-func (e *Local) Publish(topic bus.Topic, message ...bus.Message) (bus.Future, error) {
+func (e *Local) Publish(topic bus2.Topic, message ...bus2.Message) (bus2.Future, error) {
return nil, nil
}
@@ -82,7 +82,7 @@ func (e Local) Name() string {
func (e *Local) PreRun() error {
var err error
- e.dataBus = bus.NewBus()
+ e.dataBus = bus2.NewBus()
for _, dp := range e.dps {
err = multierr.Append(err, dp.Pub(e.dataBus))
}
diff --git a/banyand/queue/queue.go b/banyand/queue/queue.go
index ff59cee..e2cd6af 100644
--- a/banyand/queue/queue.go
+++ b/banyand/queue/queue.go
@@ -21,15 +21,15 @@ import (
"context"
"github.com/apache/skywalking-banyandb/banyand/discovery"
- "github.com/apache/skywalking-banyandb/banyand/internal/bus"
+ bus2 "github.com/apache/skywalking-banyandb/pkg/bus"
"github.com/apache/skywalking-banyandb/pkg/run"
)
type Queue interface {
run.Config
run.PreRunner
- bus.Subscriber
- bus.Publisher
+ bus2.Subscriber
+ bus2.Publisher
}
func NewQueue(ctx context.Context, repo discovery.ServiceRepo) (Queue, error) {
diff --git a/banyand/series/series.go b/banyand/series/series.go
index e75db05..db280f3 100644
--- a/banyand/series/series.go
+++ b/banyand/series/series.go
@@ -81,4 +81,5 @@ type Service interface {
SchemaRepo
IndexFilter
run.PreRunner
+ run.Service
}
diff --git a/banyand/series/trace/common_test.go b/banyand/series/trace/common_test.go
index 205b15e..7616ebd 100644
--- a/banyand/series/trace/common_test.go
+++ b/banyand/series/trace/common_test.go
@@ -58,7 +58,7 @@ func setup(t *testing.T) (*traceSeries, func()) {
db, err := storage.NewDB(context.TODO(), nil)
assert.NoError(t, err)
assert.NoError(t, db.FlagSet().Parse(nil))
- svc, err := NewService(context.TODO(), db)
+ svc, err := NewService(context.TODO(), db, nil)
assert.NoError(t, err)
assert.NoError(t, svc.PreRun())
assert.NoError(t, db.PreRun())
diff --git a/banyand/series/trace/trace.go b/banyand/series/trace/trace.go
index 3bd6c89..a785c7b 100644
--- a/banyand/series/trace/trace.go
+++ b/banyand/series/trace/trace.go
@@ -22,17 +22,22 @@ import (
"context"
"fmt"
"strconv"
+ "time"
flatbuffers "github.com/google/flatbuffers/go"
"github.com/pkg/errors"
"github.com/apache/skywalking-banyandb/api/common"
"github.com/apache/skywalking-banyandb/api/data"
+ "github.com/apache/skywalking-banyandb/api/event"
v1 "github.com/apache/skywalking-banyandb/api/fbs/v1"
apischema "github.com/apache/skywalking-banyandb/api/schema"
+ "github.com/apache/skywalking-banyandb/banyand/discovery"
"github.com/apache/skywalking-banyandb/banyand/series"
"github.com/apache/skywalking-banyandb/banyand/series/schema"
"github.com/apache/skywalking-banyandb/banyand/storage"
+ "github.com/apache/skywalking-banyandb/pkg/bus"
+ "github.com/apache/skywalking-banyandb/pkg/fb"
"github.com/apache/skywalking-banyandb/pkg/logger"
)
@@ -78,11 +83,16 @@ type service struct {
db storage.Database
schemaMap map[string]*traceSeries
l *logger.Logger
+ repo discovery.ServiceRepo
+ stopCh chan struct{}
}
//NewService returns a new service
-func NewService(_ context.Context, db storage.Database) (series.Service, error) {
- return &service{db: db}, nil
+func NewService(_ context.Context, db storage.Database, repo discovery.ServiceRepo) (series.Service, error) {
+ return &service{
+ db: db,
+ repo: repo,
+ }, nil
}
func (s *service) Name() string {
@@ -109,6 +119,33 @@ func (s *service) PreRun() error {
return err
}
+func (s *service) Serve() error {
+ for _, sMeta := range s.schemaMap {
+ for i := 0; i < int(sMeta.shardNum); i++ {
+ now := time.Now().UnixNano()
+ _, err := s.repo.Publish(event.TopicShardEvent, bus.NewMessage(bus.MessageID(now), fb.BuildShardEvent(
+ s.repo.NodeID(),
+ sMeta.name,
+ sMeta.group,
+ uint(i),
+ sMeta.shardNum,
+ )))
+ if err != nil {
+ return err
+ }
+ }
+ }
+ s.stopCh = make(chan struct{})
+ <-s.stopCh
+ return nil
+}
+
+func (s *service) GracefulStop() {
+ if s.stopCh != nil {
+ close(s.stopCh)
+ }
+}
+
func (s *service) FetchTrace(traceSeries common.Metadata, traceID string, opt series.ScanOptions) (data.Trace, error) {
ts, err := s.getSeries(traceSeries)
if err != nil {
diff --git a/banyand/internal/bus/bus.go b/pkg/bus/bus.go
similarity index 100%
rename from banyand/internal/bus/bus.go
rename to pkg/bus/bus.go
diff --git a/banyand/internal/bus/bus_test.go b/pkg/bus/bus_test.go
similarity index 100%
rename from banyand/internal/bus/bus_test.go
rename to pkg/bus/bus_test.go
diff --git a/pkg/convert/hash.go b/pkg/convert/hash.go
index e7014fe..b08c2a5 100644
--- a/pkg/convert/hash.go
+++ b/pkg/convert/hash.go
@@ -22,3 +22,7 @@ import "github.com/cespare/xxhash"
func Hash(key []byte) uint64 {
return xxhash.Sum64(key)
}
+
+func HashStr(key string) uint64 {
+ return xxhash.Sum64String(key)
+}
diff --git a/pkg/fb/database.go b/pkg/fb/database.go
new file mode 100644
index 0000000..0d18fc0
--- /dev/null
+++ b/pkg/fb/database.go
@@ -0,0 +1,59 @@
+// 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 fb
+
+import (
+ "time"
+
+ flatbuffers "github.com/google/flatbuffers/go"
+
+ v1 "github.com/apache/skywalking-banyandb/api/fbs/v1"
+)
+
+func BuildShardEvent(nodeID, seriesName, seriesGroup string, shardID, totalShards uint) []byte {
+ now := time.Now().UnixNano()
+ b := flatbuffers.NewBuilder(0)
+ nID := b.CreateString(nodeID)
+ addr := b.CreateString("localhost")
+ v1.NodeStart(b)
+ v1.NodeAddId(b, nID)
+ v1.NodeAddAddr(b, addr)
+ v1.NodeAddCreateTime(b, now)
+ v1.NodeAddUpdateTime(b, now)
+ node := v1.NodeEnd(b)
+ name := b.CreateString(seriesName)
+ group := b.CreateString(seriesGroup)
+ v1.MetadataStart(b)
+ v1.MetadataAddName(b, name)
+ v1.MetadataAddGroup(b, group)
+ md := v1.MetadataEnd(b)
+ v1.ShardStart(b)
+ v1.ShardAddId(b, uint64(shardID))
+ v1.ShardAddNode(b, node)
+ v1.ShardAddSeries(b, md)
+ v1.ShardAddTotal(b, uint8(totalShards))
+ v1.ShardAddCreateTime(b, now)
+ v1.ShardAddUpdateTime(b, now)
+ shard := v1.ShardEnd(b)
+ v1.ShardEventStart(b)
+ v1.ShardEventAddShard(b, shard)
+ v1.ShardEventAddTime(b, time.Now().UnixNano())
+ v1.ShardEventAddAction(b, v1.ActionPut)
+ b.Finish(v1.ShardEventEnd(b))
+ return b.FinishedBytes()
+}