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/14 05:58:54 UTC
[skywalking-banyandb] branch main updated: Add series event (#17)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
The following commit(s) were added to refs/heads/main by this push:
new 01c28bc Add series event (#17)
01c28bc is described below
commit 01c28bc0b8cccc927780a2e1cad7293cf71f9276
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Wed Jul 14 13:58:42 2021 +0800
Add series event (#17)
* Add series event
Signed-off-by: Gao Hongtao <ha...@gmail.com>
* Use build method to build topics
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
api/event/discovery.go | 13 +++--
api/fbs/v1/database.fbs | 7 +++
api/fbs/v1/database_generated.go | 103 +++++++++++++++++++++++++++++++++++++++
banyand/liaison/grpc/grpc.go | 51 +++++++++++++++----
banyand/series/trace/trace.go | 60 ++++++++++++++---------
pkg/fb/database.go | 27 ++++++++++
6 files changed, 224 insertions(+), 37 deletions(-)
diff --git a/api/event/discovery.go b/api/event/discovery.go
index 40360c4..0dad5c5 100644
--- a/api/event/discovery.go
+++ b/api/event/discovery.go
@@ -28,10 +28,12 @@ var (
Version: "v1",
Kind: "event-shard",
}
- TopicShardEvent = bus.Topic{
- ID: ShardEventKindVersion.String(),
- Type: bus.ChTypeUnidirectional,
+ TopicShardEvent = bus.UniTopic(ShardEventKindVersion.String())
+ SeriesEventKindVersion = common.KindVersion{
+ Version: "v1",
+ Kind: "event-series",
}
+ TopicSeriesEvent = bus.UniTopic(SeriesEventKindVersion.String())
)
type Shard struct {
@@ -39,6 +41,7 @@ type Shard struct {
Payload v1.ShardEvent
}
-func NewShard() *Shard {
- return &Shard{KindVersion: ShardEventKindVersion}
+type Series struct {
+ common.KindVersion
+ Payload v1.SeriesEvent
}
diff --git a/api/fbs/v1/database.fbs b/api/fbs/v1/database.fbs
index d71c1f1..62ff40c 100644
--- a/api/fbs/v1/database.fbs
+++ b/api/fbs/v1/database.fbs
@@ -41,6 +41,13 @@ table ShardEvent {
time:int64;
}
+table SeriesEvent {
+ series:Metadata;
+ field_names_composite_series_id:[string];
+ action:Action;
+ time:int64;
+}
+
// Metadata is for multi-tenant, multi-model use
table Metadata {
// group contains a set of options, like retention policy, max
diff --git a/api/fbs/v1/database_generated.go b/api/fbs/v1/database_generated.go
index 4236dff..2494815 100644
--- a/api/fbs/v1/database_generated.go
+++ b/api/fbs/v1/database_generated.go
@@ -324,6 +324,109 @@ func ShardEventEnd(builder *flatbuffers.Builder) flatbuffers.UOffsetT {
return builder.EndObject()
}
+type SeriesEvent struct {
+ _tab flatbuffers.Table
+}
+
+func GetRootAsSeriesEvent(buf []byte, offset flatbuffers.UOffsetT) *SeriesEvent {
+ n := flatbuffers.GetUOffsetT(buf[offset:])
+ x := &SeriesEvent{}
+ x.Init(buf, n+offset)
+ return x
+}
+
+func GetSizePrefixedRootAsSeriesEvent(buf []byte, offset flatbuffers.UOffsetT) *SeriesEvent {
+ n := flatbuffers.GetUOffsetT(buf[offset+flatbuffers.SizeUint32:])
+ x := &SeriesEvent{}
+ x.Init(buf, n+offset+flatbuffers.SizeUint32)
+ return x
+}
+
+func (rcv *SeriesEvent) Init(buf []byte, i flatbuffers.UOffsetT) {
+ rcv._tab.Bytes = buf
+ rcv._tab.Pos = i
+}
+
+func (rcv *SeriesEvent) Table() flatbuffers.Table {
+ return rcv._tab
+}
+
+func (rcv *SeriesEvent) Series(obj *Metadata) *Metadata {
+ o := flatbuffers.UOffsetT(rcv._tab.Offset(4))
+ 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 *SeriesEvent) FieldNamesCompositeSeriesId(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 *SeriesEvent) FieldNamesCompositeSeriesIdLength() int {
+ o := flatbuffers.UOffsetT(rcv._tab.Offset(6))
+ if o != 0 {
+ return rcv._tab.VectorLen(o)
+ }
+ return 0
+}
+
+func (rcv *SeriesEvent) Action() Action {
+ o := flatbuffers.UOffsetT(rcv._tab.Offset(8))
+ if o != 0 {
+ return Action(rcv._tab.GetInt8(o + rcv._tab.Pos))
+ }
+ return 0
+}
+
+func (rcv *SeriesEvent) MutateAction(n Action) bool {
+ return rcv._tab.MutateInt8Slot(8, int8(n))
+}
+
+func (rcv *SeriesEvent) Time() int64 {
+ o := flatbuffers.UOffsetT(rcv._tab.Offset(10))
+ if o != 0 {
+ return rcv._tab.GetInt64(o + rcv._tab.Pos)
+ }
+ return 0
+}
+
+func (rcv *SeriesEvent) MutateTime(n int64) bool {
+ return rcv._tab.MutateInt64Slot(10, n)
+}
+
+func SeriesEventStart(builder *flatbuffers.Builder) {
+ builder.StartObject(4)
+}
+func SeriesEventAddSeries(builder *flatbuffers.Builder, series flatbuffers.UOffsetT) {
+ builder.PrependUOffsetTSlot(0, flatbuffers.UOffsetT(series), 0)
+}
+func SeriesEventAddFieldNamesCompositeSeriesId(builder *flatbuffers.Builder, fieldNamesCompositeSeriesId flatbuffers.UOffsetT) {
+ builder.PrependUOffsetTSlot(1, flatbuffers.UOffsetT(fieldNamesCompositeSeriesId), 0)
+}
+func SeriesEventStartFieldNamesCompositeSeriesIdVector(builder *flatbuffers.Builder, numElems int) flatbuffers.UOffsetT {
+ return builder.StartVector(4, numElems, 4)
+}
+func SeriesEventAddAction(builder *flatbuffers.Builder, action Action) {
+ builder.PrependInt8Slot(2, int8(action), 0)
+}
+func SeriesEventAddTime(builder *flatbuffers.Builder, time int64) {
+ builder.PrependInt64Slot(3, time, 0)
+}
+func SeriesEventEnd(builder *flatbuffers.Builder) flatbuffers.UOffsetT {
+ return builder.EndObject()
+}
+
type Metadata struct {
_tab flatbuffers.Table
}
diff --git a/banyand/liaison/grpc/grpc.go b/banyand/liaison/grpc/grpc.go
index deade08..0c8a2a0 100644
--- a/banyand/liaison/grpc/grpc.go
+++ b/banyand/liaison/grpc/grpc.go
@@ -35,14 +35,20 @@ import (
)
type Server struct {
- addr string
- log *logger.Logger
- ser *grpclib.Server
- pipeline queue.Queue
- repo discovery.ServiceRepo
+ addr string
+ log *logger.Logger
+ ser *grpclib.Server
+ pipeline queue.Queue
+ repo discovery.ServiceRepo
+ shardInfo *shardInfo
+ seriesInfo *seriesInfo
}
-func (s *Server) Rev(message bus.Message) (resp bus.Message) {
+type shardInfo struct {
+ log *logger.Logger
+}
+
+func (s *shardInfo) Rev(message bus.Message) (resp bus.Message) {
data, ok := message.Data().([]byte)
if !ok {
s.log.Warn().Msg("invalid event data type")
@@ -56,15 +62,42 @@ func (s *Server) Rev(message bus.Message) (resp bus.Message) {
return
}
+type seriesInfo struct {
+ log *logger.Logger
+}
+
+func (s *seriesInfo) Rev(message bus.Message) (resp bus.Message) {
+ data, ok := message.Data().([]byte)
+ if !ok {
+ s.log.Warn().Msg("invalid event data type")
+ return
+ }
+ seriesEvent := v1.GetRootAsSeriesEvent(data, 0)
+ s.log.Info().
+ Str("action", seriesEvent.Action().String()).
+ Str("name", string(seriesEvent.Series(nil).Name())).
+ Str("group", string(seriesEvent.Series(nil).Group())).
+ Msg("received a shard event")
+ return
+}
+
func (s *Server) PreRun() error {
s.log = logger.GetLogger("liaison-grpc")
- return s.repo.Subscribe(event.TopicShardEvent, s)
+ s.shardInfo.log = s.log
+ s.seriesInfo.log = s.log
+ err := s.repo.Subscribe(event.TopicShardEvent, s.shardInfo)
+ if err != nil {
+ return err
+ }
+ return s.repo.Subscribe(event.TopicSeriesEvent, s.seriesInfo)
}
func NewServer(ctx context.Context, pipeline queue.Queue, repo discovery.ServiceRepo) *Server {
return &Server{
- pipeline: pipeline,
- repo: repo,
+ pipeline: pipeline,
+ repo: repo,
+ shardInfo: &shardInfo{},
+ seriesInfo: &seriesInfo{},
}
}
diff --git a/banyand/series/trace/trace.go b/banyand/series/trace/trace.go
index a785c7b..e6aa39d 100644
--- a/banyand/series/trace/trace.go
+++ b/banyand/series/trace/trace.go
@@ -120,18 +120,26 @@ func (s *service) PreRun() error {
}
func (s *service) Serve() error {
+ now := time.Now().UnixNano()
for _, sMeta := range s.schemaMap {
+ _, err := s.repo.Publish(event.TopicSeriesEvent, bus.NewMessage(bus.MessageID(now), fb.BuildSeriesEvent(
+ sMeta.name,
+ sMeta.group,
+ sMeta.fieldsNamesCompositeSeriesID,
+ )))
+ if err != nil {
+ return err
+ }
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(
+ _, errShard := 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
+ if errShard != nil {
+ return errShard
}
}
}
@@ -185,24 +193,25 @@ func getTraceSeriesID(traceSeries common.Metadata) string {
}
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
+ 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
+ fieldsNamesCompositeSeriesID [][]byte
}
func newTraceSeries(schema apischema.TraceSeries, l *logger.Logger) (*traceSeries, error) {
@@ -292,6 +301,7 @@ func (t *traceSeries) buildFieldIndex() error {
fieldsLen := spec.FieldsLength()
index := make(map[string]uint, fieldsLen)
+ t.fieldIndex = index
for i := 0; i < fieldsLen; i++ {
fieldSpec := new(v1.FieldSpec)
if !spec.Fields(fieldSpec, i) {
@@ -322,7 +332,11 @@ func (t *traceSeries) buildFieldIndex() error {
}
t.stateFieldName = string(stateFieldName)
- t.fieldIndex = index
+ t.fieldsNamesCompositeSeriesID = make([][]byte, 0, reservedMap.SeriesIdLength())
+ for i := 0; i < reservedMap.SeriesIdLength(); i++ {
+ t.fieldsNamesCompositeSeriesID = append(t.fieldsNamesCompositeSeriesID, reservedMap.SeriesId(i))
+ }
+
return nil
}
diff --git a/pkg/fb/database.go b/pkg/fb/database.go
index 0d18fc0..cce084b 100644
--- a/pkg/fb/database.go
+++ b/pkg/fb/database.go
@@ -57,3 +57,30 @@ func BuildShardEvent(nodeID, seriesName, seriesGroup string, shardID, totalShard
b.Finish(v1.ShardEventEnd(b))
return b.FinishedBytes()
}
+
+func BuildSeriesEvent(seriesName, seriesGroup string, fieldNames [][]byte) []byte {
+ now := time.Now().UnixNano()
+ b := flatbuffers.NewBuilder(0)
+ name := b.CreateString(seriesName)
+ group := b.CreateString(seriesGroup)
+ v1.MetadataStart(b)
+ v1.MetadataAddName(b, name)
+ v1.MetadataAddGroup(b, group)
+ md := v1.MetadataEnd(b)
+ var nameOffsets []flatbuffers.UOffsetT
+ for _, fieldName := range fieldNames {
+ nameOffsets = append(nameOffsets, b.CreateByteString(fieldName))
+ }
+ v1.SeriesEventStartFieldNamesCompositeSeriesIdVector(b, len(fieldNames))
+ for i := len(nameOffsets) - 1; i >= 0; i-- {
+ b.PrependUOffsetT(nameOffsets[i])
+ }
+ names := b.EndVector(len(fieldNames))
+ v1.SeriesEventStart(b)
+ v1.SeriesEventAddSeries(b, md)
+ v1.SeriesEventAddFieldNamesCompositeSeriesId(b, names)
+ v1.SeriesEventAddAction(b, v1.ActionPut)
+ v1.SeriesEventAddTime(b, now)
+ b.Finish(v1.SeriesEventEnd(b))
+ return b.FinishedBytes()
+}