Skip to content

Commit 24da846

Browse files
hwjiangkaihwjiangkai
committed
feat: support event tracing
Signed-off-by: jyjiangkai <[email protected]>
1 parent 895dd83 commit 24da846

File tree

30 files changed

+2779
-351
lines changed

30 files changed

+2779
-351
lines changed

‎client/client.go

+2
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@ import (
2525
"github.com/vanus-labs/vanus/observability/log"
2626
"github.com/vanus-labs/vanus/observability/tracing"
2727
"github.com/vanus-labs/vanus/pkg/cluster"
28+
"go.opentelemetry.io/otel/trace"
2829
"google.golang.org/grpc/credentials/insecure"
2930

3031
// this project.
@@ -109,6 +110,7 @@ func Connect(endpoints []string) Client {
109110
}
110111
return &client{
111112
Endpoints: endpoints,
113+
tracer: tracing.NewTracer("client.client", trace.SpanKindClient),
112114
}
113115
}
114116

‎client/internal/vanus/eventbus/name_service.go

+1-6
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ package eventbus
1717
import (
1818
// standard libraries.
1919
"context"
20+
2021
// third-party libraries.
2122
"go.opentelemetry.io/otel/trace"
2223
"google.golang.org/grpc/credentials/insecure"
@@ -46,9 +47,6 @@ type NameService struct {
4647
}
4748

4849
func (ns *NameService) LookupWritableLogs(ctx context.Context, eventbusID uint64) ([]*record.Eventlog, error) {
49-
ctx, span := ns.tracer.Start(ctx, "LookupWritableLogs")
50-
defer span.End()
51-
5250
req := &wrapperspb.UInt64Value{
5351
Value: eventbusID,
5452
}
@@ -62,9 +60,6 @@ func (ns *NameService) LookupWritableLogs(ctx context.Context, eventbusID uint64
6260
}
6361

6462
func (ns *NameService) LookupReadableLogs(ctx context.Context, eventbusID uint64) ([]*record.Eventlog, error) {
65-
ctx, span := ns.tracer.Start(ctx, "LookupReadableLogs")
66-
defer span.End()
67-
6863
req := &wrapperspb.UInt64Value{
6964
Value: eventbusID,
7065
}

‎client/internal/vanus/eventlog/name_service.go

-6
Original file line numberDiff line numberDiff line change
@@ -49,9 +49,6 @@ type NameService struct {
4949
}
5050

5151
func (ns *NameService) LookupWritableSegment(ctx context.Context, logID uint64) (*record.Segment, error) {
52-
ctx, span := ns.tracer.Start(ctx, "LookupWritableSegment")
53-
defer span.End()
54-
5552
req := &ctrlpb.GetAppendableSegmentRequest{
5653
EventlogId: logID,
5754
Limited: 1,
@@ -74,9 +71,6 @@ func (ns *NameService) LookupWritableSegment(ctx context.Context, logID uint64)
7471
}
7572

7673
func (ns *NameService) LookupReadableSegments(ctx context.Context, logID uint64) ([]*record.Segment, error) {
77-
ctx, span := ns.tracer.Start(ctx, "LookupReadableSegments")
78-
defer span.End()
79-
8074
req := &ctrlpb.ListSegmentRequest{
8175
EventlogId: logID,
8276
StartOffset: 0,

‎client/internal/vanus/net/rpc/bare/client.go

+1-3
Original file line numberDiff line numberDiff line change
@@ -72,13 +72,11 @@ func (c *client) Get(ctx context.Context) (interface{}, error) {
7272
if c.closed.Load() {
7373
return nil, errors.ErrClosed
7474
}
75-
_ctx, span := c.tracer.Start(ctx, "Get")
76-
defer span.End()
7775

7876
if client := c.cachedClient(); client != nil {
7977
return client, nil
8078
}
81-
return c.refreshClient(_ctx, false)
79+
return c.refreshClient(ctx, false)
8280
}
8381

8482
func (c *client) cachedClient() interface{} {

‎client/internal/vanus/store/block_store.go

+2-11
Original file line numberDiff line numberDiff line change
@@ -67,9 +67,6 @@ func (s *BlockStore) Close() {
6767
func (s *BlockStore) Read(
6868
ctx context.Context, block uint64, offset int64, size int16, pollingTimeout uint32,
6969
) (*cloudevents.CloudEventBatch, error) {
70-
ctx, span := s.tracer.Start(ctx, "Read")
71-
defer span.End()
72-
7370
req := &segpb.ReadFromBlockRequest{
7471
BlockId: block,
7572
Offset: offset,
@@ -90,9 +87,6 @@ func (s *BlockStore) Read(
9087
}
9188

9289
func (s *BlockStore) LookupOffset(ctx context.Context, blockID uint64, t time.Time) (int64, error) {
93-
ctx, span := s.tracer.Start(ctx, "LookupOffset")
94-
defer span.End()
95-
9690
req := &segpb.LookupOffsetInBlockRequest{
9791
BlockId: blockID,
9892
Stime: t.UnixMilli(),
@@ -111,20 +105,17 @@ func (s *BlockStore) LookupOffset(ctx context.Context, blockID uint64, t time.Ti
111105
}
112106

113107
func (s *BlockStore) Append(ctx context.Context, block uint64, events *cloudevents.CloudEventBatch) ([]int64, error) {
114-
_ctx, span := s.tracer.Start(ctx, "Append")
115-
defer span.End()
116-
117108
req := &segpb.AppendToBlockRequest{
118109
BlockId: block,
119110
Events: events,
120111
}
121112

122-
client, err := s.client.Get(_ctx)
113+
client, err := s.client.Get(ctx)
123114
if err != nil {
124115
return nil, err
125116
}
126117

127-
res, err := client.(segpb.SegmentServerClient).AppendToBlock(_ctx, req)
118+
res, err := client.(segpb.SegmentServerClient).AppendToBlock(ctx, req)
128119
if err != nil {
129120
return nil, err
130121
}

‎client/pkg/eventbus/eventbus.go

+6-30
Original file line numberDiff line numberDiff line change
@@ -68,13 +68,10 @@ func NewEventbus(cfg *eb.Config) *eventbus {
6868
break
6969
}
7070

71-
ctx, span := bus.tracer.Start(context.Background(), "updateWritableLogsTask")
7271
if bus.writableWatcher != nil {
73-
bus.updateWritableLogs(ctx, re)
72+
bus.updateWritableLogs(context.Background(), re)
7473
}
75-
7674
bus.writableWatcher.Wakeup()
77-
span.End()
7875
}
7976
}()
8077
bus.writableWatcher.Start()
@@ -88,13 +85,10 @@ func NewEventbus(cfg *eb.Config) *eventbus {
8885
break
8986
}
9087

91-
ctx, span := bus.tracer.Start(context.Background(), "updateReadableLogsTask")
9288
if bus.readableWatcher != nil {
93-
bus.updateReadableLogs(ctx, re)
89+
bus.updateReadableLogs(context.Background(), re)
9490
}
95-
9691
bus.readableWatcher.Wakeup()
97-
span.End()
9892
}
9993
}()
10094
bus.readableWatcher.Start()
@@ -272,9 +266,6 @@ func (b *eventbus) isNeedUpdateWritableLogs(err error) bool {
272266
}
273267

274268
func (b *eventbus) updateWritableLogs(ctx context.Context, re *WritableLogsResult) {
275-
_, span := b.tracer.Start(ctx, "updateWritableLogs")
276-
defer span.End()
277-
278269
if !b.isNeedUpdateWritableLogs(re.Err) {
279270
return
280271
}
@@ -333,10 +324,7 @@ func (b *eventbus) getWritableLog(ctx context.Context, logID uint64) eventlog.Ev
333324
}
334325

335326
func (b *eventbus) refreshWritableLogs(ctx context.Context) {
336-
_ctx, span := b.tracer.Start(ctx, "refreshWritableLogs")
337-
defer span.End()
338-
339-
_ = b.writableWatcher.Refresh(_ctx)
327+
_ = b.writableWatcher.Refresh(ctx)
340328
}
341329

342330
func (b *eventbus) getReadableState() error {
@@ -364,9 +352,6 @@ func (b *eventbus) isNeedUpdateReadableLogs(err error) bool {
364352
}
365353

366354
func (b *eventbus) updateReadableLogs(ctx context.Context, re *ReadableLogsResult) {
367-
_, span := b.tracer.Start(ctx, "updateReadableLogs")
368-
defer span.End()
369-
370355
if !b.isNeedUpdateReadableLogs(re.Err) {
371356
return
372357
}
@@ -425,10 +410,7 @@ func (b *eventbus) getReadableLog(ctx context.Context, logID uint64) eventlog.Ev
425410
}
426411

427412
func (b *eventbus) refreshReadableLogs(ctx context.Context) {
428-
_ctx, span := b.tracer.Start(ctx, "refreshReadableLogs")
429-
defer span.End()
430-
431-
_ = b.readableWatcher.Refresh(_ctx)
413+
_ = b.readableWatcher.Refresh(ctx)
432414
}
433415

434416
type busWriter struct {
@@ -479,15 +461,12 @@ func (w *busWriter) Bus() api.Eventbus {
479461
}
480462

481463
func (w *busWriter) pickWritableLog(ctx context.Context, opts *api.WriteOptions) (eventlog.LogWriter, error) {
482-
_ctx, span := w.tracer.Start(ctx, "pickWritableLog")
483-
defer span.End()
484-
485464
l, err := opts.Policy.NextLog(ctx)
486465
if err != nil {
487466
return nil, err
488467
}
489468

490-
lw := w.ebus.getWritableLog(_ctx, l.ID())
469+
lw := w.ebus.getWritableLog(ctx, l.ID())
491470
if lw == nil {
492471
return nil, stderrors.New("can not pick writable log")
493472
}
@@ -554,14 +533,11 @@ func (r *busReader) Bus() api.Eventbus {
554533
}
555534

556535
func (r *busReader) pickReadableLog(ctx context.Context, opts *api.ReadOptions) (eventlog.LogReader, error) {
557-
_ctx, span := r.tracer.Start(ctx, "pickReadableLog")
558-
defer span.End()
559-
560536
l, err := opts.Policy.NextLog(ctx)
561537
if err != nil {
562538
return nil, err
563539
}
564-
lr := r.ebus.getReadableLog(_ctx, l.ID())
540+
lr := r.ebus.getReadableLog(ctx, l.ID())
565541
if lr == nil {
566542
return nil, stderrors.New("can not pick readable log")
567543
}

‎client/pkg/eventlog/eventlog_impl.go

+3-8
Original file line numberDiff line numberDiff line change
@@ -63,13 +63,10 @@ func NewEventlog(cfg *el.Config) Eventlog {
6363
break
6464
}
6565

66-
ctx, span := l.tracer.Start(context.Background(), "updateReadableSegmentsTask")
6766
if r != nil {
68-
l.updateWritableSegment(ctx, r)
67+
l.updateWritableSegment(context.Background(), r)
6968
}
70-
7169
l.writableWatcher.Wakeup()
72-
span.End()
7370
}
7471
}()
7572
l.writableWatcher.Start()
@@ -84,13 +81,11 @@ func NewEventlog(cfg *el.Config) Eventlog {
8481
Msg("eventlog quits readable watcher")
8582
break
8683
}
87-
ctx, span := l.tracer.Start(context.Background(), "updateReadableSegmentsTask")
84+
8885
if rs != nil {
89-
l.updateReadableSegments(ctx, rs)
86+
l.updateReadableSegments(context.Background(), rs)
9087
}
91-
9288
l.readableWatcher.Wakeup()
93-
span.End()
9489
}
9590
}()
9691
l.readableWatcher.Start()

‎client/pkg/eventlog/log_segment.go

+1-10
Original file line numberDiff line numberDiff line change
@@ -128,9 +128,6 @@ func (s *segment) Update(ctx context.Context, r *record.Segment, towrite bool) e
128128
return nil
129129
}
130130

131-
_, span := s.tracer.Start(ctx, "Update")
132-
defer span.End()
133-
134131
switchBlock := func() bool {
135132
if towrite {
136133
if s.prefer.id != r.LeaderBlockID {
@@ -155,14 +152,11 @@ func (s *segment) Update(ctx context.Context, r *record.Segment, towrite bool) e
155152
}
156153

157154
func (s *segment) Append(ctx context.Context, event *cloudevents.CloudEventBatch) ([]int64, error) {
158-
_ctx, span := s.tracer.Start(ctx, "Append")
159-
defer span.End()
160-
161155
b := s.preferSegmentBlock()
162156
if b == nil {
163157
return nil, errors.ErrNotLeader
164158
}
165-
offs, err := b.Append(_ctx, event)
159+
offs, err := b.Append(ctx, event)
166160
if err != nil {
167161
return nil, err
168162
}
@@ -176,9 +170,6 @@ func (s *segment) Read(ctx context.Context, from int64, size int16, pollingTimeo
176170
if from < s.startOffset {
177171
return nil, errors.ErrOffsetUnderflow
178172
}
179-
ctx, span := s.tracer.Start(ctx, "Read")
180-
defer span.End()
181-
182173
if eo := s.endOffset.Load(); eo >= 0 {
183174
if from > eo {
184175
return nil, errors.ErrOffsetOverflow

‎cmd/gateway/main.go

+4-4
Original file line numberDiff line numberDiff line change
@@ -38,16 +38,16 @@ func main() {
3838
}
3939

4040
ctx := signal.SetupSignalContext()
41-
ga := gateway.NewGateway(*cfg)
41+
cfg.Observability.T.ServerName = "Vanus Gateway"
42+
_ = observability.Initialize(ctx, cfg.Observability, metrics.GetGatewayMetrics)
4243

44+
ga := gateway.NewGateway(*cfg)
4345
if err = ga.Start(ctx); err != nil {
4446
log.Error().Err(err).Msg("start gateway failed")
4547
os.Exit(-1)
4648
}
47-
48-
cfg.Observability.T.ServerName = "Vanus Gateway"
49-
_ = observability.Initialize(ctx, cfg.Observability, metrics.GetGatewayMetrics)
5049
log.Info(ctx).Msg("Gateway has started")
50+
5151
select {
5252
case <-ctx.Done():
5353
log.Info(ctx).Msg("received system signal, preparing exit")

‎cmd/trigger/main.go

+1
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,7 @@ func main() {
4949
os.Exit(-1)
5050
}
5151
ctx := signal.SetupSignalContext()
52+
cfg.Observability.T.ServerName = "Vanus Trigger"
5253
_ = observability.Initialize(ctx, cfg.Observability, metrics.GetTriggerMetrics)
5354
var opts []grpc.ServerOption
5455
grpcServer := grpc.NewServer(opts...)

‎go.mod

+3-1
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,8 @@ require (
4848
k8s.io/utils v0.0.0-20230406110748-d93618cff8a2
4949
)
5050

51+
require github.com/vanus-labs/sdk/golang v0.4.7 // indirect
52+
5153
require (
5254
cloud.google.com/go/compute v1.19.0 // indirect
5355
cloud.google.com/go/compute/metadata v0.2.3 // indirect
@@ -127,4 +129,4 @@ replace (
127129
github.com/vanus-labs/vanus/raft => ./raft
128130
)
129131

130-
replace github.com/vanus-labs/sdk/golang => ./FORBIDDEN_DEPENDENCY
132+
// replace github.com/vanus-labs/sdk/golang => ./FORBIDDEN_DEPENDENCY

‎go.sum

+2
Original file line numberDiff line numberDiff line change
@@ -325,6 +325,8 @@ github.com/tidwall/match v1.1.1/go.mod h1:eRSPERbgtNPcGhD8UCthc6PmLEQXEWd3PRB5JT
325325
github.com/tidwall/pretty v1.2.0 h1:RWIZEg2iJ8/g6fDDYzMpobmaoGh5OLl4AXtGUGPcqCs=
326326
github.com/tidwall/pretty v1.2.0/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU=
327327
github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw=
328+
github.com/vanus-labs/sdk/golang v0.4.7 h1:SIWuyguOX4t0Jve9U7g6A2A4knFjoIW+9UhMsji8jIY=
329+
github.com/vanus-labs/sdk/golang v0.4.7/go.mod h1:QpmncLBj1i1rtZmqSsoZyWN4l6odMHpSZmbad4GBuQQ=
328330
github.com/vigneshuvi/GoDateFormat v0.0.0-20210204121036-67364dc23c79 h1:37VzBuFO88QQnCEu+G41v9IqgJNBXR+4vR9vGwVqJ00=
329331
github.com/vigneshuvi/GoDateFormat v0.0.0-20210204121036-67364dc23c79/go.mod h1:190gFTWxRNREiiPal7zWZlNrwFSpv3BxDmOfgYqoYCY=
330332
github.com/yuin/goldmark v1.1.25/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=

‎internal/convert/convert.go

+10
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,8 @@ package convert
1616

1717
import (
1818
// standard libraries.
19+
"encoding/base64"
20+
"encoding/binary"
1921
"errors"
2022

2123
// third-party libraries.
@@ -539,3 +541,11 @@ func ToPbTransformer(transformer *primitive.Transformer) *pb.Transformer {
539541
Template: transformer.Template.Template,
540542
}
541543
}
544+
545+
func DecodeEventID(eid string) (uint64, uint64, error) {
546+
bytes, err := base64.StdEncoding.DecodeString(eid)
547+
if err != nil {
548+
return 0, 0, err
549+
}
550+
return binary.BigEndian.Uint64(bytes[0:8]), binary.BigEndian.Uint64(bytes[8:16]), nil
551+
}

0 commit comments

Comments
 (0)